fix(manifest): protect DataFile metadata from external mutation

data_file_refs.go

@@ -0,0 +1,36 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package iceberg
+
+import "github.com/apache/iceberg-go/internal"
+
+// DataFileStatsRef returns statistics maps owned by the data file. The token
+// restricts this zero-copy accessor to trusted in-module callers; the public
+// DataFile getters continue returning defensive copies. Callers must treat the
+// returned maps and the byte slices stored in the bounds maps as read-only.
+func (d *dataFile) DataFileStatsRef(_ internal.DataFileRef) (
+	valueCounts map[int]int64,
+	nullCounts map[int]int64,
+	nanCounts map[int]int64,
+	lowerBounds map[int][]byte,
+	upperBounds map[int][]byte,
+) {
+	d.initColumnStatsData()
+
+	return d.valCntMap, d.nullCntMap, d.nanCntMap, d.lowerBoundMap, d.upperBoundMap
+}

internal/data_file_ref.go

@@ -0,0 +1,23 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package internal
+
+// DataFileRef authorizes zero-copy access to immutable DataFile state from
+// trusted packages within this module. Go's internal-package rule prevents
+// external callers from constructing this token.
+type DataFileRef struct{}

manifest.go

@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"io"
 	"iter"
+	"maps"
 	"math"
 	"math/big"
 	"reflect"
@@ -2201,7 +2202,45 @@ func (d *dataFile) FileFormat() FileFormat            { return d.Format }
 func (d *dataFile) Partition() map[int]any {
 	d.initPartitionData()
 
-	return d.fieldIDToPartitionData
+	return clonePartitionMap(d.fieldIDToPartitionData)
+}
+
+func clonePartitionMap(src map[int]any) map[int]any {
+	if src == nil {
+		return nil
+	}
+
+	out := maps.Clone(src)
+	for id, value := range out {
+		if bytes, ok := value.([]byte); ok {
+			out[id] = slices.Clone(bytes)
+		}
+	}
+
+	return out
+}
+
+func cloneByteMap(src map[int][]byte) map[int][]byte {
+	if src == nil {
+		return nil
+	}
+
+	out := make(map[int][]byte, len(src))
+	for id, value := range src {
+		out[id] = slices.Clone(value)
+	}
+
+	return out
+}
+
+func clonePointer[T any](value *T) *T {
+	if value == nil {
+		return nil
+	}
+
+	cloned := *value
+
+	return &cloned
 }
 
 func (d *dataFile) Count() int64         { return d.RecordCount }
@@ -2211,75 +2250,75 @@ func (d *dataFile) SpecID() int32        { return d.specID }
 func (d *dataFile) ColumnSizes() map[int]int64 {
 	d.initColumnStatsData()
 
-	return d.colSizeMap
+	return maps.Clone(d.colSizeMap)
 }
 
 func (d *dataFile) ValueCounts() map[int]int64 {
 	d.initColumnStatsData()
 
-	return d.valCntMap
+	return maps.Clone(d.valCntMap)
 }
 
 func (d *dataFile) NullValueCounts() map[int]int64 {
 	d.initColumnStatsData()
 
-	return d.nullCntMap
+	return maps.Clone(d.nullCntMap)
 }
 
 func (d *dataFile) NaNValueCounts() map[int]int64 {
 	d.initColumnStatsData()
 
-	return d.nanCntMap
+	return maps.Clone(d.nanCntMap)
 }
 
 func (d *dataFile) DistinctValueCounts() map[int]int64 {
 	d.initColumnStatsData()
 
-	return d.distinctCntMap
+	return maps.Clone(d.distinctCntMap)
 }
 
 func (d *dataFile) LowerBoundValues() map[int][]byte {
 	d.initColumnStatsData()
 
-	return d.lowerBoundMap
+	return cloneByteMap(d.lowerBoundMap)
 }
 
 func (d *dataFile) UpperBoundValues() map[int][]byte {
 	d.initColumnStatsData()
 
-	return d.upperBoundMap
+	return cloneByteMap(d.upperBoundMap)
 }
 
 func (d *dataFile) KeyMetadata() []byte {
 	if d.Key == nil {
 		return nil
 	}
 
-	return *d.Key
+	return slices.Clone(*d.Key)
 }
 
 func (d *dataFile) SplitOffsets() []int64 {
 	if d.Splits == nil {
 		return nil
 	}
 
-	return *d.Splits
+	return slices.Clone(*d.Splits)
 }
 
 func (d *dataFile) EqualityFieldIDs() []int {
 	if d.EqualityIDs == nil {
 		return nil
 	}
 
-	return *d.EqualityIDs
+	return slices.Clone(*d.EqualityIDs)
 }
 
-func (d *dataFile) SortOrderID() *int { return d.SortOrder }
+func (d *dataFile) SortOrderID() *int { return clonePointer(d.SortOrder) }
 
-func (d *dataFile) FirstRowID() *int64          { return d.FirstRowIDField }
-func (d *dataFile) ReferencedDataFile() *string { return d.ReferencedDataFileField }
-func (d *dataFile) ContentSizeInBytes() *int64  { return d.ContentSizeInBytesField }
-func (d *dataFile) ContentOffset() *int64       { return d.ContentOffsetField }
+func (d *dataFile) FirstRowID() *int64          { return clonePointer(d.FirstRowIDField) }
+func (d *dataFile) ReferencedDataFile() *string { return clonePointer(d.ReferencedDataFileField) }
+func (d *dataFile) ContentSizeInBytes() *int64  { return clonePointer(d.ContentSizeInBytesField) }
+func (d *dataFile) ContentOffset() *int64       { return clonePointer(d.ContentOffsetField) }
 
 type ManifestEntryBuilder struct {
 	m *manifestEntry
@@ -2414,6 +2453,9 @@ type DataFileBuilder struct {
 // before calling [DataFileBuilder.Build] to construct the object.
 // The fieldIDToFixedSize argument is retained for source compatibility and is
 // ignored; manifest schemas determine decimal fixed widths during encoding.
+// Byte-slice partition values are cloned. Other partition values must be the
+// immutable scalar or literal values produced by Iceberg's manifest decoder;
+// mutable caller-defined values are retained by reference.
 func NewDataFileBuilder(
 	spec PartitionSpec,
 	content ManifestEntryContent,
@@ -2476,7 +2518,7 @@ func NewDataFileBuilder(
 			RecordCount:            recordCount,
 			FileSize:               fileSize,
 			specID:                 int32(spec.id),
-			fieldIDToPartitionData: fieldIDToPartitionData,
+			fieldIDToPartitionData: clonePartitionMap(fieldIDToPartitionData),
 			fieldNameToID:          fieldNameToID,
 			fieldIDToLogicalType:   fieldIDToLogicalType,
 		},
@@ -2534,35 +2576,38 @@ func (b *DataFileBuilder) DistinctValueCounts(counts map[int]int64) *DataFileBui
 
 // LowerBoundValues sets the lower bound values for the data file.
 func (b *DataFileBuilder) LowerBoundValues(bounds map[int][]byte) *DataFileBuilder {
-	b.d.LowerBounds = mapToAvroColMap(bounds)
+	b.d.LowerBounds = mapToAvroColMap(cloneByteMap(bounds))
 
 	return b
 }
 
 // UpperBoundValues sets the upper bound values for the data file.
 func (b *DataFileBuilder) UpperBoundValues(bounds map[int][]byte) *DataFileBuilder {
-	b.d.UpperBounds = mapToAvroColMap(bounds)
+	b.d.UpperBounds = mapToAvroColMap(cloneByteMap(bounds))
 
 	return b
 }
 
 // KeyMetadata sets the key metadata for the data file.
 func (b *DataFileBuilder) KeyMetadata(key []byte) *DataFileBuilder {
-	b.d.Key = &key
+	cloned := slices.Clone(key)
+	b.d.Key = &cloned
 
 	return b
 }
 
 // SplitOffsets sets the split offsets for the data file.
 func (b *DataFileBuilder) SplitOffsets(offsets []int64) *DataFileBuilder {
-	b.d.Splits = &offsets
+	cloned := slices.Clone(offsets)
+	b.d.Splits = &cloned
 
 	return b
 }
 
 // EqualityFieldIDs sets the equality field ids for the data file.
 func (b *DataFileBuilder) EqualityFieldIDs(ids []int) *DataFileBuilder {
-	b.d.EqualityIDs = &ids
+	cloned := slices.Clone(ids)
+	b.d.EqualityIDs = &cloned
 
 	return b
 }

manifest_test.go

@@ -2969,6 +2969,82 @@ func (m *ManifestTestSuite) TestManifestEntryPresentEmptyListSurvivesRewrite() {
 		"present-empty column_sizes must survive a decode -> re-encode rewrite as a present array")
 }
 
+func (m *ManifestTestSuite) TestDataFileMetadataIsIsolatedFromExternalMutation() {
+	partition := []byte{0x01, 0x02}
+	partitionData := map[int]any{1000: partition}
+	spec := NewPartitionSpec(PartitionField{SourceIDs: []int{1}, FieldID: 1000, Name: "part", Transform: IdentityTransform{}})
+	builder, err := NewDataFileBuilder(spec, EntryContentData, "s3://bucket/file.parquet", ParquetFile,
+		partitionData, nil, nil, 1, 10)
+	m.Require().NoError(err)
+
+	columnSizes := map[int]int64{1: 10}
+	valueCounts := map[int]int64{1: 2}
+	nullCounts := map[int]int64{1: 1}
+	nanCounts := map[int]int64{1: 0}
+	distinctCounts := map[int]int64{1: 2}
+	lower := map[int][]byte{1: {0x03, 0x04}}
+	upper := map[int][]byte{1: {0x05, 0x06}}
+	key := []byte{0x07, 0x08}
+	splits := []int64{10, 20}
+	equalityIDs := []int{1, 2}
+
+	dataFile := builder.
+		ColumnSizes(columnSizes).
+		ValueCounts(valueCounts).
+		NullValueCounts(nullCounts).
+		NaNValueCounts(nanCounts).
+		DistinctValueCounts(distinctCounts).
+		LowerBoundValues(lower).
+		UpperBoundValues(upper).
+		KeyMetadata(key).
+		SplitOffsets(splits).
+		EqualityFieldIDs(equalityIDs).
+		SortOrderID(3).
+		FirstRowID(4).
+		ReferencedDataFile("data.parquet").
+		ContentOffset(5).
+		ContentSizeInBytes(6).
+		Build()
+
+	partition[0], lower[1][0], upper[1][0], key[0], splits[0], equalityIDs[0] = 0xff, 0xff, 0xff, 0xff, 99, 99
+	partitionData[1000] = []byte{0xff}
+	columnSizes[1], valueCounts[1], nullCounts[1], nanCounts[1], distinctCounts[1] = 99, 99, 99, 99, 99
+
+	dataFile.Partition()[1000].([]byte)[0] = 0xff
+	dataFile.ColumnSizes()[1] = 99
+	dataFile.ValueCounts()[1] = 99
+	dataFile.NullValueCounts()[1] = 99
+	dataFile.NaNValueCounts()[1] = 99
+	dataFile.DistinctValueCounts()[1] = 99
+	dataFile.LowerBoundValues()[1][0] = 0xff
+	dataFile.UpperBoundValues()[1][0] = 0xff
+	dataFile.KeyMetadata()[0] = 0xff
+	dataFile.SplitOffsets()[0] = 99
+	dataFile.EqualityFieldIDs()[0] = 99
+	*dataFile.SortOrderID() = 99
+	*dataFile.FirstRowID() = 99
+	*dataFile.ReferencedDataFile() = "changed"
+	*dataFile.ContentOffset() = 99
+	*dataFile.ContentSizeInBytes() = 99
+
+	m.Equal([]byte{0x01, 0x02}, dataFile.Partition()[1000])
+	m.Equal(map[int]int64{1: 10}, dataFile.ColumnSizes())
+	m.Equal(map[int]int64{1: 2}, dataFile.ValueCounts())
+	m.Equal(map[int]int64{1: 1}, dataFile.NullValueCounts())
+	m.Equal(map[int]int64{1: 0}, dataFile.NaNValueCounts())
+	m.Equal(map[int]int64{1: 2}, dataFile.DistinctValueCounts())
+	m.Equal([]byte{0x03, 0x04}, dataFile.LowerBoundValues()[1])
+	m.Equal([]byte{0x05, 0x06}, dataFile.UpperBoundValues()[1])
+	m.Equal([]byte{0x07, 0x08}, dataFile.KeyMetadata())
+	m.Equal([]int64{10, 20}, dataFile.SplitOffsets())
+	m.Equal([]int{1, 2}, dataFile.EqualityFieldIDs())
+	m.Equal(3, *dataFile.SortOrderID())
+	m.Equal(int64(4), *dataFile.FirstRowID())
+	m.Equal("data.parquet", *dataFile.ReferencedDataFile())
+	m.Equal(int64(5), *dataFile.ContentOffset())
+	m.Equal(int64(6), *dataFile.ContentSizeInBytes())
+}
+
 func (m *ManifestTestSuite) TestWriteManifestListClosesWriterOnError() {
 	// A v2 manifest list cannot reference v3 manifests because the v2 entry
 	// schema has no first_row_id column; this gives us a deterministic