Repository: apache/iceberg-go
Pull request #317 (2025-03-01 19:21:52)

feat(table): Adds updateSnapshotSummary internal function
Next step to getting writes up and working

zeroshade: I've addressed @Fokko's comments in general and the bigger point around the V1/V2 handling is shifted to a new PR so I'm going to merge this in the interests of continuing to move forward.

Patch:
diff --git a/internal/utils.go b/internal/utils.go
index c7eb231..aa3c841 100644
--- a/internal/utils.go
+++ b/internal/utils.go
@@ -19,6 +19,7 @@ package internal
 
 import (
 	"cmp"
+	"io"
 	"slices"
 )
 
@@ -130,3 +131,15 @@ func (s *SlicePacker[T]) PackEnd(items []T, weightFunc func(T) int64) [][]T {
 
 	return result
 }
+
+type CountingWriter struct {
+	Count int64
+	W     io.Writer
+}
+
+func (w *CountingWriter) Write(p []byte) (int, error) {
+	n, err := w.W.Write(p)
+	w.Count += int64(n)
+
+	return n, err
+}
diff --git a/manifest.go b/manifest.go
index 25dada1..1589257 100644
--- a/manifest.go
+++ b/manifest.go
@@ -85,12 +85,6 @@ func NewManifestV1Builder(path string, length int64, partitionSpecID int32, adde
 	}
 }
 
-func (b *ManifestV1Builder) PartitionType(st *StructType) *ManifestV1Builder {
-	b.m.partitionType = st
-
-	return b
-}
-
 func (b *ManifestV1Builder) AddedFiles(cnt int32) *ManifestV1Builder {
 	b.m.AddedFilesCount = &cnt
 
@@ -144,10 +138,6 @@ func (b *ManifestV1Builder) KeyMetadata(km []byte) *ManifestV1Builder {
 // a pointer to the constructed manifest file. Further calls to the modifier
 // methods after calling build would modify the constructed ManifestFile.
 func (b *ManifestV1Builder) Build() ManifestFile {
-	if b.m.partitionType == nil {
-		b.m.partitionType = UnpartitionedSpec.PartitionType(nil)
-	}
-
 	return b.m
 }
 
@@ -163,7 +153,6 @@ func (f *fallbackManifestFileV1) toManifest() *manifestFileV1 {
 }
 
 type manifestFileV1 struct {
-	partitionType      *StructType
 	Path               string          `avro:"manifest_path"`
 	Len                int64           `avro:"manifest_length"`
 	SpecID             int32           `avro:"partition_spec_id"`
@@ -261,11 +250,6 @@ func (m *manifestFileV1) FetchEntries(fs iceio.IO, discardDeleted bool) ([]Manif
 	return fetchManifestEntries(m, fs, discardDeleted)
 }
 
-// // WriteEntries writes a list of manifest entries to an avro file.
-// func (m *manifestFileV1) WriteEntries(out io.Writer, entries []ManifestEntry) error {
-// 	return writeManifestEntries(out, m.partitionType, entries, m.Version())
-// }
-
 // ManifestV2Builder is a helper for building a V2 manifest file
 // struct which will conform to the ManifestFile interface.
 type ManifestV2Builder struct {
@@ -288,12 +272,6 @@ func NewManifestV2Builder(path string, length int64, partitionSpecID int32, cont
 	}
 }
 
-func (b *ManifestV2Builder) PartitionSchema(st *StructType) *ManifestV2Builder {
-	b.m.partitionType = st
-
-	return b
-}
-
 func (b *ManifestV2Builder) SequenceNum(num, minSeqNum int64) *ManifestV2Builder {
 	b.m.SeqNumber, b.m.MinSeqNumber = num, minSeqNum
 
@@ -353,15 +331,10 @@ func (b *ManifestV2Builder) KeyMetadata(km []byte) *ManifestV2Builder {
 // a pointer to the constructed manifest file. Further calls to the modifier
 // methods after calling build would modify the constructed ManifestFile.
 func (b *ManifestV2Builder) Build() ManifestFile {
-	if b.m.partitionType == nil {
-		b.m.partitionType = UnpartitionedSpec.PartitionType(nil)
-	}
-
 	return b.m
 }
 
 type manifestFileV2 struct {
-	partitionType      *StructType
 	Path               string          `avro:"manifest_path"`
 	Len                int64           `avro:"manifest_length"`
 	SpecID             int32           `avro:"partition_spec_id"`
@@ -437,11 +410,6 @@ func (m *manifestFileV2) FetchEntries(fs iceio.IO, discardDeleted bool) ([]Manif
 	return fetchManifestEntries(m, fs, discardDeleted)
 }
 
-// // WriteEntries writes a list of manifest entries to an avro file.
-// func (m *manifestFileV2) WriteEntries(out io.Writer, entries []ManifestEntry) error {
-// 	return writeManifestEntries(out, m.partitionType, entries, m.Version())
-// }
-
 func getFieldIDMap(sc avro.Schema) (map[string]int, map[int]avro.LogicalType) {
 	getField := func(rs *avro.RecordSchema, name string) *avro.Field {
 		for _, f := range rs.Fields() {
@@ -538,7 +506,7 @@ func fetchManifestEntries(m ManifestFile, fs iceio.IO, discardDeleted bool) ([]M
 		}
 
 		if !discardDeleted || tmp.Status() != EntryStatusDELETED {
-			tmp.inheritSeqNum(m)
+			tmp.inherit(m)
 			if fieldToIDMap, ok := tmp.DataFile().(hasFieldToIDMap); ok {
 				fieldToIDMap.setFieldNameToIDMap(fieldNameToID)
 				fieldToIDMap.setFieldIDToLogicalTypeMap(fieldIDToLogicalType)
@@ -1218,14 +1186,17 @@ func WriteManifestList(version int, out io.Writer, snapshotID int64, parentSnaps
 }
 
 func WriteManifest(
+	filename string,
 	out io.Writer,
 	version int,
 	spec PartitionSpec,
 	schema *Schema,
 	snapshotID int64,
 	entries []ManifestEntry,
-) (func(string, int64) (ManifestFile, error), error) {
-	w, err := NewManifestWriter(version, out, spec, schema, snapshotID)
+) (ManifestFile, error) {
+	cnt := &internal.CountingWriter{W: out}
+
+	w, err := NewManifestWriter(version, cnt, spec, schema, snapshotID)
 	if err != nil {
 		return nil, err
 	}
@@ -1236,7 +1207,7 @@ func WriteManifest(
 		}
 	}
 
-	return w.ToManifestFile, w.Close()
+	return w.ToManifestFile(filename, cnt.Count)
 }
 
 // ManifestEntryStatus defines constants for the entry status of
@@ -1376,6 +1347,7 @@ type dataFile struct {
 	fieldNameToID        map[string]int
 	fieldIDToLogicalType map[int]avro.LogicalType
 
+	specID   int32
 	initMaps sync.Once
 }
 
@@ -1408,6 +1380,7 @@ func (d *dataFile) Partition() map[string]any {
 
 func (d *dataFile) Count() int64         { return d.RecordCount }
 func (d *dataFile) FileSizeBytes() int64 { return d.FileSize }
+func (d *dataFile) SpecID() int32        { return d.specID }
 
 func (d *dataFile) ColumnSizes() map[int]int64 {
 	d.initializeMapData()
@@ -1519,7 +1492,13 @@ func (f *fallbackManifestEntryV1) toEntry() *manifestEntryV1 {
 	return &f.manifestEntryV1
 }
 
-func (m *manifestEntryV1) inheritSeqNum(manifest ManifestFile) {}
+func (m *manifestEntryV1) inherit(manifest ManifestFile) {
+	if m.Snapshot <= 0 {
+		m.Snapshot = manifest.SnapshotID()
+	}
+
+	m.Data.(*dataFile).specID = manifest.PartitionSpecID()
+}
 
 func (m *manifestEntryV1) Status() ManifestEntryStatus { return m.EntryStatus }
 func (m *manifestEntryV1) SnapshotID() int64           { return m.Snapshot }
@@ -1601,7 +1580,7 @@ type manifestEntryV2 struct {
 	Data        DataFile            `avro:"data_file"`
 }
 
-func (m *manifestEntryV2) inheritSeqNum(manifest ManifestFile) {
+func (m *manifestEntryV2) inherit(manifest ManifestFile) {
 	if m.Snapshot == nil {
 		snap := manifest.SnapshotID()
 		m.Snapshot = &snap
@@ -1615,6 +1594,8 @@ func (m *manifestEntryV2) inheritSeqNum(manifest ManifestFile) {
 	if m.FileSeqNum == nil && (manifestSequenceNum == 0 || m.EntryStatus == EntryStatusADDED) {
 		m.FileSeqNum = &manifestSequenceNum
 	}
+
+	m.Data.(*dataFile).specID = manifest.PartitionSpecID()
 }
 
 func (m *manifestEntryV2) Status() ManifestEntryStatus { return m.EntryStatus }
@@ -1655,6 +1636,16 @@ func (m *manifestEntryV2) wrap(status ManifestEntryStatus, snapshotID int64, seq
 	return m
 }
 
+func NewManifestEntry(status ManifestEntryStatus, snapshotID int64, seqNum, fileSeqNum *int64, df DataFile) ManifestEntry {
+	return &manifestEntryV2{
+		EntryStatus: status,
+		Snapshot:    &snapshotID,
+		SeqNum:      seqNum,
+		FileSeqNum:  fileSeqNum,
+		Data:        df,
+	}
+}
+
 // DataFileBuilder is a helper for building a data file struct which will
 // conform to the DataFile interface.
 type DataFileBuilder struct {
@@ -1665,6 +1656,7 @@ type DataFileBuilder struct {
 // all of the optional fields to be set by calling the corresponding methods
 // before calling [DataFileBuilder.Build] to construct the object.
 func NewDataFileBuilder(
+	spec PartitionSpec,
 	content ManifestEntryContent,
 	path string,
 	format FileFormat,
@@ -1706,6 +1698,7 @@ func NewDataFileBuilder(
 			PartitionData: partitionData,
 			RecordCount:   recordCount,
 			FileSize:      fileSize,
+			specID:        int32(spec.id),
 		},
 	}, nil
 }
@@ -1857,6 +1850,9 @@ type DataFile interface {
 	// SortOrderID returns the id representing the sort order for this
 	// file, or nil if there is no sort order.
 	SortOrderID() *int
+	// SpecID returns the partition spec id for this data file, inherited
+	// from the manifest that the data file was read from
+	SpecID() int32
 }
 
 // ManifestEntry is an interface for both v1 and v2 manifest entries.
@@ -1879,7 +1875,7 @@ type ManifestEntry interface {
 	// by this manifest entry.
 	DataFile() DataFile
 
-	inheritSeqNum(manifest ManifestFile)
+	inherit(manifest ManifestFile)
 	wrap(status ManifestEntryStatus, snapshotID int64, seqNum int64, fileSeqNum *int64, datafile DataFile) ManifestEntry
 }
 
diff --git a/table/snapshots.go b/table/snapshots.go
index d06bc24..b12c74a 100644
--- a/table/snapshots.go
+++ b/table/snapshots.go
@@ -22,10 +22,12 @@ import (
 	"errors"
 	"fmt"
 	"maps"
+	"slices"
 	"strconv"
+	"strings"
 
 	"github.com/apache/iceberg-go"
-	"github.com/apache/iceberg-go/io"
+	iceio "github.com/apache/iceberg-go/io"
 )
 
 type Operation string
@@ -53,14 +55,130 @@ func ValidOperation(s string) (Operation, error) {
 	return "", fmt.Errorf("%w: found '%s'", ErrInvalidOperation, s)
 }
 
-const operationKey = "operation"
+const (
+	operationKey = "operation"
+
+	addedDataFilesKey         = "added-data-files"
+	addedDeleteFilesKey       = "added-delete-files"
+	addedEqDeletesKey         = "added-equality-deletes"
+	addedFileSizeKey          = "added-files-size"
+	addedPosDeletesKey        = "added-position-deletes"
+	addedPosDeleteFilesKey    = "added-position-delete-files"
+	addedRecordsKey           = "added-records"
+	addedEqDeleteFilesKey     = "added-equality-delete-files"
+	deletedDataFilesKey       = "deleted-data-files"
+	deletedRecordsKey         = "deleted-records"
+	removedDeleteFilesKey     = "removed-delete-files"
+	removedEqDeletesKey       = "removed-equality-deletes"
+	removedEqDeleteFilesKey   = "removed-equality-delete-files"
+	removedFileSizeKey        = "removed-files-size"
+	removedPosDeletesKey      = "removed-position-deletes"
+	removedPosDeleteFilesKey  = "removed-position-delete-files"
+	totalEqDeletesKey         = "total-equality-deletes"
+	totalPosDeletesKey        = "total-position-deletes"
+	totalDataFilesKey         = "total-data-files"
+	totalDeleteFilesKey       = "total-delete-files"
+	totalRecordsKey           = "total-records"
+	totalFileSizeKey          = "total-files-size"
+	changedPartitionCountProp = "changed-partition-count"
+	changedPartitionPrefix    = "partitions."
+)
+
+type updateMetrics struct {
+	addedFileSize         int64
+	removedFileSize       int64
+	addedDataFiles        int64
+	removedDataFiles      int64
+	addedEqDeleteFiles    int64
+	removedEqDeleteFiles  int64
+	addedPosDeleteFiles   int64
+	removedPosDeleteFiles int64
+	addedDeleteFiles      int64
+	removedDeleteFiles    int64
+	addedRecords          int64
+	deletedRecords        int64
+	addedPosDeletes       int64
+	removedPosDeletes     int64
+	addedEqDeletes        int64
+	removedEqDeletes      int64
+}
+
+func (m *updateMetrics) addDataFile(df iceberg.DataFile) error {
+	m.addedFileSize += df.FileSizeBytes()
+	switch df.ContentType() {
+	case iceberg.EntryContentData:
+		m.addedDataFiles++
+		m.addedRecords += df.Count()
+	case iceberg.EntryContentPosDeletes:
+		m.addedDeleteFiles++
+		m.addedPosDeleteFiles++
+		m.addedPosDeletes += df.Count()
+	case iceberg.EntryContentEqDeletes:
+		m.addedDeleteFiles++
+		m.addedEqDeleteFiles++
+		m.addedEqDeletes += df.Count()
+	default:
+		return fmt.Errorf("unknown data file content: %s", df.ContentType())
+	}
+
+	return nil
+}
+
+func (m *updateMetrics) removeFile(df iceberg.DataFile) error {
+	m.removedFileSize += df.FileSizeBytes()
+	switch df.ContentType() {
+	case iceberg.EntryContentData:
+		m.removedDataFiles++
+		m.deletedRecords += df.Count()
+	case iceberg.EntryContentPosDeletes:
+		m.removedDeleteFiles++
+		m.removedPosDeleteFiles++
+		m.removedPosDeletes += df.Count()
+	case iceberg.EntryContentEqDeletes:
+		m.removedDeleteFiles++
+		m.removedEqDeleteFiles++
+		m.removedEqDeletes += df.Count()
+	default:
+		return fmt.Errorf("unknown data file content: %s", df.ContentType())
+	}
+
+	return nil
+}
+
+func setWhenPositive(props iceberg.Properties, key string, value int64) {
+	if value > 0 {
+		props[key] = strconv.FormatInt(value, 10)
+	}
+}
+
+func (m *updateMetrics) toProps() iceberg.Properties {
+	props := iceberg.Properties{}
+	setWhenPositive(props, addedFileSizeKey, m.addedFileSize)
+	setWhenPositive(props, removedFileSizeKey, m.removedFileSize)
+	setWhenPositive(props, addedDataFilesKey, m.addedDataFiles)
+	setWhenPositive(props, deletedDataFilesKey, m.removedDataFiles)
+	setWhenPositive(props, addedEqDeleteFilesKey, m.addedEqDeleteFiles)
+	setWhenPositive(props, removedEqDeleteFilesKey, m.removedEqDeleteFiles)
+	setWhenPositive(props, addedPosDeleteFilesKey, m.addedPosDeleteFiles)
+	setWhenPositive(props, removedPosDeleteFilesKey, m.removedPosDeleteFiles)
+	setWhenPositive(props, addedDeleteFilesKey, m.addedDeleteFiles)
+	setWhenPositive(props, removedDeleteFilesKey, m.removedDeleteFiles)
+	setWhenPositive(props, addedRecordsKey, m.addedRecords)
+	setWhenPositive(props, deletedRecordsKey, m.deletedRecords)
+	setWhenPositive(props, addedPosDeletesKey, m.addedPosDeletes)
+	setWhenPositive(props, removedPosDeletesKey, m.removedPosDeletes)
+	setWhenPositive(props, addedEqDeletesKey, m.addedEqDeletes)
+	setWhenPositive(props, removedEqDeletesKey, m.removedEqDeletes)
+
+	return props
+}
 
 // Summary stores the summary information for a snapshot indicating
 // the operation that created the snapshot, and various properties
 // which might exist in the summary.
 type Summary struct {
 	Operation  Operation
-	Properties map[string]string
+	Properties iceberg.Properties
 }
 
 func (s *Summary) String() string {
@@ -174,7 +292,7 @@ func (s Snapshot) Equals(other Snapshot) bool {
 		s.Summary.Equals(other.Summary)
 }
 
-func (s Snapshot) Manifests(fio io.IO) ([]iceberg.ManifestFile, error) {
+func (s Snapshot) Manifests(fio iceio.IO) ([]iceberg.ManifestFile, error) {
 	if s.ManifestList != "" {
 		f, err := fio.Open(s.ManifestList)
 		if err != nil {
@@ -197,3 +315,161 @@ type SnapshotLogEntry struct {
 	SnapshotID  int64 `json:"snapshot-id"`
 	TimestampMs int64 `json:"timestamp-ms"`
 }
+
+type SnapshotSummaryCollector struct {
+	metrics                          updateMetrics
+	partitionMetrics                 map[string]updateMetrics
+	maxChangedPartitionsForSummaries int
+}
+
+func (s *SnapshotSummaryCollector) setPartitionSummaryLimit(limit int) {
+	s.maxChangedPartitionsForSummaries = limit
+}
+
+func (s *SnapshotSummaryCollector) updatePartitionMetrics(partitionPath string, df iceberg.DataFile, isAddFile bool) error {
+	if s.partitionMetrics == nil {
+		s.partitionMetrics = make(map[string]updateMetrics)
+	}
+
+	metrics := s.partitionMetrics[partitionPath]
+	if isAddFile {
+		if err := metrics.addDataFile(df); err != nil {
+			return err
+		}
+	} else {
+		if err := metrics.removeFile(df); err != nil {
+			return err
+		}
+	}
+	s.partitionMetrics[partitionPath] = metrics
+
+	return nil
+}
+
+func (s *SnapshotSummaryCollector) addFile(df iceberg.DataFile, sc *iceberg.Schema, spec iceberg.PartitionSpec) error {
+	if err := s.metrics.addDataFile(df); err != nil {
+		return err
+	}
+
+	if len(df.Partition()) > 0 {
+		partitionPath := spec.PartitionToPath(
+			getPartitionRecord(df, spec.PartitionType(sc)), sc)
+
+		return s.updatePartitionMetrics(partitionPath, df, true)
+	}
+
+	return nil
+}
+
+func (s *SnapshotSummaryCollector) removeFile(df iceberg.DataFile, sc *iceberg.Schema, spec iceberg.PartitionSpec) error {
+	if err := s.metrics.removeFile(df); err != nil {
+		return err
+	}
+
+	if len(df.Partition()) > 0 {
+		partitionPath := spec.PartitionToPath(
+			getPartitionRecord(df, spec.PartitionType(sc)), sc)
+
+		return s.updatePartitionMetrics(partitionPath, df, false)
+	}
+
+	return nil
+}
+
+func (s *SnapshotSummaryCollector) partitionSummary(metrics *updateMetrics) string {
+	props := metrics.toProps()
+
+	return strings.Join(slices.Sorted(func(yield func(s string) bool) {
+		for k, v := range props {
+			if !yield(fmt.Sprintf("%s=%s", k, v)) {
+				return
+			}
+		}
+	}), ",")
+}
+
+func (s *SnapshotSummaryCollector) build() iceberg.Properties {
+	props := s.metrics.toProps()
+	changedPartitionsSize := len(s.partitionMetrics)
+	setWhenPositive(props, changedPartitionCountProp, int64(changedPartitionsSize))
+	if changedPartitionsSize <= s.maxChangedPartitionsForSummaries {
+		for partPath, updateMetricsPart := range s.partitionMetrics {
+			if summary := s.partitionSummary(&updateMetricsPart); len(summary) > 0 {
+				props[changedPartitionPrefix+partPath] = summary
+			}
+		}
+	}
+
+	return props
+}
+
+func truncateTableSummary(sum Summary, previous iceberg.Properties) Summary {
+	keys := []string{
+		totalDataFilesKey, totalDeleteFilesKey, totalRecordsKey,
+		totalFileSizeKey, totalPosDeletesKey, totalEqDeletesKey,
+	}
+	for _, prop := range keys {
+		sum.Properties[prop] = "0"
+	}
+
+	updateProp := func(prop, prevProp string) {
+		if val := previous.GetInt(prevProp, 0); val > 0 {
+			sum.Properties[prop] = strconv.Itoa(val)
+		}
+	}
+
+	updateProp(deletedDataFilesKey, totalDataFilesKey)
+	updateProp(removedDeleteFilesKey, totalDeleteFilesKey)
+	updateProp(deletedRecordsKey, totalRecordsKey)
+	updateProp(removedFileSizeKey, totalFileSizeKey)
+	updateProp(removedPosDeletesKey, totalPosDeletesKey)
+	updateProp(removedEqDeletesKey, totalEqDeletesKey)
+
+	return sum
+}
+
+func updateSnapshotSummaries(sum Summary, previous iceberg.Properties, truncateFullTable bool) (Summary, error) {
+	switch sum.Operation {
+	case OpAppend, OpOverwrite, OpDelete:
+	default:
+		return sum, fmt.Errorf("%w: operation: %s", iceberg.ErrNotImplemented, sum.Operation)
+	}
+
+	if sum.Properties == nil {
+		sum.Properties = make(iceberg.Properties)
+	}
+
+	if truncateFullTable && sum.Operation == OpOverwrite && previous != nil {
+		sum = truncateTableSummary(sum, previous)
+	}
+
+	if previous == nil {
+		previous = iceberg.Properties{
+			totalDataFilesKey:   "0",
+			totalDeleteFilesKey: "0",
+			totalRecordsKey:     "0",
+			totalFileSizeKey:    "0",
+			totalPosDeletesKey:  "0",
+			totalEqDeletesKey:   "0",
+		}
+	}
+
+	updateTotals := func(totalProp, addedProp, removedProp string) {
+		newTotal := previous.GetInt(totalProp, 0)
+		newTotal += sum.Properties.GetInt(addedProp, 0)
+		newTotal -= sum.Properties.GetInt(removedProp, 0)
+
+		if newTotal >= 0 {
+			sum.Properties[totalProp] = strconv.Itoa(newTotal)
+		}
+	}
+
+	updateTotals(totalDataFilesKey, addedDataFilesKey, deletedDataFilesKey)
+	updateTotals(totalDeleteFilesKey, addedDeleteFilesKey, removedDeleteFilesKey)
+	updateTotals(totalRecordsKey, addedRecordsKey, deletedRecordsKey)
+	updateTotals(totalFileSizeKey, addedFileSizeKey, removedFileSizeKey)
+	updateTotals(totalPosDeletesKey, addedPosDeletesKey, removedPosDeletesKey)
+	updateTotals(totalEqDeletesKey, addedEqDeletesKey, removedEqDeletesKey)
+
+	return sum, nil
+}


Test patch:
diff --git a/manifest_test.go b/manifest_test.go
index 6eb68c9..53f6fe6 100644
--- a/manifest_test.go
+++ b/manifest_test.go
@@ -422,16 +422,12 @@ func (m *ManifestTestSuite) writeManifestEntries() {
 		PartitionField{FieldID: 1000, SourceID: 1, Name: "VendorID", Transform: IdentityTransform{}},
 		PartitionField{FieldID: 1001, SourceID: 2, Name: "tpep_pickup_datetime", Transform: IdentityTransform{}})
 
-	fn, err := WriteManifest(&m.v1ManifestEntries, 1, partitionSpec, testSchema, entrySnapshotID, manifestEntryV1Recs)
+	_, err := WriteManifest("/home/iceberg/warehouse/nyc/taxis_partitioned/metadata/0125c686-8aa6-4502-bdcc-b6d17ca41a3b-m0.avro",
+		&m.v1ManifestEntries, 1, partitionSpec, testSchema, entrySnapshotID, manifestEntryV1Recs)
 	m.Require().NoError(err)
 
-	_, err = fn("/home/iceberg/warehouse/nyc/taxis_partitioned/metadata/0125c686-8aa6-4502-bdcc-b6d17ca41a3b-m0.avro", int64(m.v1ManifestList.Len()))
-	m.Require().NoError(err)
-
-	fn, err = WriteManifest(&m.v2ManifestEntries, 2, partitionSpec, testSchema, entrySnapshotID, manifestEntryV2Recs)
-	m.Require().NoError(err)
-
-	_, err = fn("/home/iceberg/warehouse/nyc/taxis_partitioned/metadata/0125c686-8aa6-4502-bdcc-b6d17ca41a3b-m0.avro", int64(m.v1ManifestList.Len()))
+	_, err = WriteManifest("/home/iceberg/warehouse/nyc/taxis_partitioned/metadata/0125c686-8aa6-4502-bdcc-b6d17ca41a3b-m0.avro",
+		&m.v2ManifestEntries, 2, partitionSpec, testSchema, entrySnapshotID, manifestEntryV2Recs)
 	m.Require().NoError(err)
 }
 
@@ -784,6 +780,7 @@ func (m *ManifestTestSuite) TestManifestEntriesV2() {
 
 func (m *ManifestTestSuite) TestManifestEntryBuilder() {
 	dataFileBuilder, err := NewDataFileBuilder(
+		NewPartitionSpec(),
 		EntryContentData,
 		"sample.parquet",
 		ParquetFile,
diff --git a/table/evaluators_test.go b/table/evaluators_test.go
index 83e5e72..df14a49 100644
--- a/table/evaluators_test.go
+++ b/table/evaluators_test.go
@@ -1081,6 +1081,8 @@ type mockDataFile struct {
 	nanCounts   map[int]int64
 	lowerBounds map[int][]byte
 	upperBounds map[int][]byte
+
+	specid int32
 }
 
 func (*mockDataFile) ContentType() iceberg.ManifestEntryContent { return iceberg.EntryContentData }
@@ -1100,6 +1102,7 @@ func (*mockDataFile) KeyMetadata() []byte                       { return nil }
 func (*mockDataFile) SplitOffsets() []int64                     { return nil }
 func (*mockDataFile) EqualityFieldIDs() []int                   { return nil }
 func (*mockDataFile) SortOrderID() *int                         { return nil }
+func (m *mockDataFile) SpecID() int32                           { return m.specid }
 
 type InclusiveMetricsTestSuite struct {
 	suite.Suite
diff --git a/table/snapshots_internal_test.go b/table/snapshots_internal_test.go
new file mode 100644
index 0000000..4063ad8
--- /dev/null
+++ b/table/snapshots_internal_test.go
@@ -0,0 +1,204 @@
+// 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 table
+
+import (
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var tableSchemaSimple = iceberg.NewSchemaWithIdentifiers(1,
+	[]int{2},
+	iceberg.NestedField{ID: 1, Name: "foo", Type: iceberg.PrimitiveTypes.String},
+	iceberg.NestedField{ID: 2, Name: "bar", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+	iceberg.NestedField{ID: 3, Name: "baz", Type: iceberg.PrimitiveTypes.Bool},
+)
+
+func must[T any](v T, err error) T {
+	if err != nil {
+		panic(err)
+	}
+
+	return v
+}
+
+func TestSnapshotSummaryCollector(t *testing.T) {
+	var ssc SnapshotSummaryCollector
+
+	assert.Equal(t, iceberg.Properties{}, ssc.build())
+
+	dataFile, err := iceberg.NewDataFileBuilder(*iceberg.UnpartitionedSpec,
+		iceberg.EntryContentData, "/path/to/file.parquet", iceberg.ParquetFile, nil, 100, 1234)
+	require.NoError(t, err)
+	require.NoError(t, ssc.addFile(dataFile.Build(), tableSchemaSimple, *iceberg.UnpartitionedSpec))
+
+	assert.Equal(t, iceberg.Properties{
+		"added-data-files": "1",
+		"added-files-size": "1234",
+		"added-records":    "100",
+	}, ssc.build())
+}
+
+func TestSnapshotSummaryCollectorWithPartition(t *testing.T) {
+	var ssc SnapshotSummaryCollector
+
+	assert.Equal(t, iceberg.Properties{}, ssc.build())
+	sc := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "bool_field", Type: iceberg.PrimitiveTypes.Bool},
+		iceberg.NestedField{ID: 2, Name: "string_field", Type: iceberg.PrimitiveTypes.String},
+		iceberg.NestedField{ID: 3, Name: "int_field", Type: iceberg.PrimitiveTypes.Int32},
+	)
+	spec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+		Name: "int_field", SourceID: 3, FieldID: 1001, Transform: iceberg.IdentityTransform{},
+	})
+
+	dataFile1 := must(iceberg.NewDataFileBuilder(
+		spec, iceberg.EntryContentData, "/path/to/file1.parquet",
+		iceberg.ParquetFile, map[string]any{"int_field": int32(1)}, 100, 1234)).Build()
+
+	dataFile2 := must(iceberg.NewDataFileBuilder(
+		spec, iceberg.EntryContentData, "/path/to/file2.parquet",
+		iceberg.ParquetFile, map[string]any{"int_field": int32(2)}, 200, 4321)).Build()
+
+	ssc.addFile(dataFile1, sc, spec)
+	ssc.removeFile(dataFile1, sc, spec)
+	ssc.removeFile(dataFile2, sc, spec)
+
+	assert.Equal(t, iceberg.Properties{
+		"added-files-size":        "1234",
+		"removed-files-size":      "5555",
+		"added-data-files":        "1",
+		"deleted-data-files":      "2",
+		"added-records":           "100",
+		"deleted-records":         "300",
+		"changed-partition-count": "2",
+	}, ssc.build())
+
+	ssc.setPartitionSummaryLimit(10)
+	assert.Equal(t, iceberg.Properties{
+		"added-files-size":        "1234",
+		"removed-files-size":      "5555",
+		"added-data-files":        "1",
+		"deleted-data-files":      "2",
+		"added-records":           "100",
+		"deleted-records":         "300",
+		"changed-partition-count": "2",
+		"partitions.int_field=1":  "added-data-files=1,added-files-size=1234,added-records=100,deleted-data-files=1,deleted-records=100,removed-files-size=1234",
+		"partitions.int_field=2":  "deleted-data-files=1,deleted-records=200,removed-files-size=4321",
+	}, ssc.build())
+}
+
+func TestMergeSnapshotSummaries(t *testing.T) {
+	tests := []struct {
+		sum           Summary
+		previous      iceberg.Properties
+		truncateTable bool
+		expected      Summary
+	}{
+		{
+			Summary{Operation: OpAppend},
+			nil, false,
+			Summary{Operation: OpAppend, Properties: iceberg.Properties{
+				"total-data-files":       "0",
+				"total-delete-files":     "0",
+				"total-records":          "0",
+				"total-files-size":       "0",
+				"total-position-deletes": "0",
+				"total-equality-deletes": "0",
+			}},
+		},
+		{
+			Summary{Operation: OpAppend, Properties: iceberg.Properties{
+				"added-data-files":       "1",
+				"added-delete-files":     "2",
+				"added-equality-deletes": "3",
+				"added-files-size":       "4",
+				"added-position-deletes": "5",
+				"added-records":          "6",
+			}},
+			nil, false,
+			Summary{Operation: OpAppend, Properties: iceberg.Properties{
+				"added-data-files":       "1",
+				"added-delete-files":     "2",
+				"added-equality-deletes": "3",
+				"added-files-size":       "4",
+				"added-position-deletes": "5",
+				"added-records":          "6",
+				"total-data-files":       "1",
+				"total-delete-files":     "2",
+				"total-records":          "6",
+				"total-files-size":       "4",
+				"total-position-deletes": "5",
+				"total-equality-deletes": "3",
+			}},
+		},
+		{
+			Summary{Operation: OpOverwrite, Properties: iceberg.Properties{
+				"added-data-files":       "1",
+				"added-delete-files":     "2",
+				"added-equality-deletes": "3",
+				"added-files-size":       "4",
+				"added-position-deletes": "5",
+				"added-records":          "6",
+			}},
+			iceberg.Properties{
+				"total-data-files":       "1",
+				"total-delete-files":     "1",
+				"total-equality-deletes": "1",
+				"total-files-size":       "1",
+				"total-position-deletes": "1",
+				"total-records":          "1",
+			},
+			true,
+			Summary{Operation: OpOverwrite, Properties: iceberg.Properties{
+				"added-data-files":         "1",
+				"added-delete-files":       "2",
+				"added-equality-deletes":   "3",
+				"added-files-size":         "4",
+				"added-position-deletes":   "5",
+				"added-records":            "6",
+				"total-data-files":         "1",
+				"total-records":            "6",
+				"total-delete-files":       "2",
+				"total-equality-deletes":   "3",
+				"total-files-size":         "4",
+				"total-position-deletes":   "5",
+				"deleted-data-files":       "1",
+				"removed-delete-files":     "1",
+				"deleted-records":          "1",
+				"removed-files-size":       "1",
+				"removed-position-deletes": "1",
+				"removed-equality-deletes": "1",
+			}},
+		},
+	}
+
+	for _, tt := range tests {
+		result, err := updateSnapshotSummaries(tt.sum, tt.previous, tt.truncateTable)
+		require.NoError(t, err)
+		assert.Equal(t, tt.expected, result)
+	}
+}
+
+func TestInvalidOperation(t *testing.T) {
+	_, err := updateSnapshotSummaries(Summary{Operation: OpReplace}, nil, false)
+	assert.ErrorIs(t, err, iceberg.ErrNotImplemented)
+}