(iceberg-go) branch main updated: fix(manifest): commit writer state after encode (#1306)

[ze...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git


The following commit(s) were added to refs/heads/main by this push:
     new 2a831b00 fix(manifest): commit writer state after encode (#1306)
2a831b00 is described below

commit 2a831b00b65d5955243764eaa85aef06f7e3ae05
Author: Minh Vu <vu...@gmail.com>
AuthorDate: Wed Jun 24 22:54:36 2026 +0200

    fix(manifest): commit writer state after encode (#1306)
    
    ManifestWriter.addEntry updated file counts, row counts, partition summaries, and the caller's DataFile partition data before prepareEntry and writer.Encode could fail, so a rejected entry left the writer looking as if it had been written. Encode the entry (through an Avro-field clone so partition conversion does not mutate the caller's DataFile) before updating any writer counters or summaries, keeping those mutations on the success side of Encode.
---
 manifest.go      | 66 +++++++++++++++++++++++++++-----------------------------
 manifest_test.go | 54 ++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 86 insertions(+), 34 deletions(-)

diff --git a/manifest.go b/manifest.go
index d887f7bd..91e1a179 100644
--- a/manifest.go
+++ b/manifest.go
@@ -1341,54 +1341,52 @@ func (w *ManifestWriter) addEntry(entry *manifestEntry) error {
 		return errors.New("cannot add entry to closed manifest writer")
 	}
 
-	switch entry.Status() {
-	case EntryStatusADDED:
-		w.addedFiles++
-		w.addedRows += entry.DataFile().Count()
-	case EntryStatusEXISTING:
-		w.existingFiles++
-		w.existingRows += entry.DataFile().Count()
-	case EntryStatusDELETED:
-		w.deletedFiles++
-		w.deletedRows += entry.DataFile().Count()
+	status := entry.Status()
+	switch status {
+	case EntryStatusADDED, EntryStatusEXISTING, EntryStatusDELETED:
 	default:
-		return fmt.Errorf("unknown entry status: %v", entry.Status())
+		return fmt.Errorf("unknown entry status: %v", status)
 	}
+	count := entry.DataFile().Count()
+	partition := entry.Data.Partition()
 
-	if setter, ok := entry.DataFile().(hasFieldToIDMap); ok {
-		setter.setFieldNameToIDMap(w.partFieldNameToID)
-		setter.setFieldIDToLogicalTypeMap(w.partFieldIDToType)
+	entryToEncode := *entry
+	if dataFile, ok := entry.DataFile().(*dataFile); ok {
+		encodeDataFile := cloneDataFileAvroFields(dataFile)
+		encodeDataFile.PartitionData = avroEncodePartitionData(partition, w.partFieldNameToID, w.partFieldIDToType)
+		entryToEncode.Data = encodeDataFile
 	}
 
-	w.partitions = append(w.partitions, entry.Data.Partition())
-	partitionData := avroPartitionData(entry.Data.Partition(), w.partFieldIDToType)
+	toEncode, err := w.impl.prepareEntry(&entryToEncode, w.snapshotID)
+	if err != nil {
+		return err
+	}
 
-	if dataFile, ok := entry.DataFile().(*dataFile); ok {
-		convertedPartitionData := make(map[string]any)
-		for fieldID, convertedValue := range partitionData {
-			for fieldName, id := range w.partFieldNameToID {
-				if id == fieldID {
-					convertedPartitionData[fieldName] = convertedValue
+	if err := w.writer.Encode(toEncode); err != nil {
+		return err
+	}
 
-					break
-				}
-			}
-		}
-		dataFile.PartitionData = convertedPartitionData
+	switch status {
+	case EntryStatusADDED:
+		w.addedFiles++
+		w.addedRows += count
+	case EntryStatusEXISTING:
+		w.existingFiles++
+		w.existingRows += count
+	case EntryStatusDELETED:
+		w.deletedFiles++
+		w.deletedRows += count
 	}
 
-	if entry.Status() == EntryStatusADDED || entry.Status() == EntryStatusEXISTING {
+	w.partitions = append(w.partitions, partition)
+
+	if status == EntryStatusADDED || status == EntryStatusEXISTING {
 		if seq := entry.SequenceNum(); seq >= 0 && (w.minSeqNum < 0 || seq < w.minSeqNum) {
 			w.minSeqNum = seq
 		}
 	}
 
-	toEncode, err := w.impl.prepareEntry(entry, w.snapshotID)
-	if err != nil {
-		return err
-	}
-
-	return w.writer.Encode(toEncode)
+	return nil
 }
 
 func (w *ManifestWriter) Add(entry ManifestEntry) error {
diff --git a/manifest_test.go b/manifest_test.go
index 1ec5fc25..228d891c 100644
--- a/manifest_test.go
+++ b/manifest_test.go
@@ -2084,6 +2084,60 @@ func (w *limitedWriter) Write(p []byte) (int, error) {
 	return len(p), nil
 }
 
+func (m *ManifestTestSuite) TestManifestWriterDoesNotCommitStateOnEncodeError() {
+	schema := NewSchema(0,
+		NestedField{ID: 1, Name: "dt", Type: PrimitiveTypes.Date},
+	)
+	partitionSpec := NewPartitionSpecID(1,
+		PartitionField{FieldID: 1000, SourceIDs: []int{1}, Name: "dt", Transform: IdentityTransform{}},
+	)
+
+	var out bytes.Buffer
+	writer, err := NewManifestWriter(2, &out, partitionSpec, schema, snapshotID)
+	m.Require().NoError(err)
+
+	badPartitionValue := struct{ Value string }{Value: "not-a-date"}
+	entry := newDatePartitionManifestEntry(m.T(), partitionSpec, snapshotID, badPartitionValue)
+	dataFile := entry.DataFile().(*dataFile)
+	originalPartitionData := map[string]any{"dt": badPartitionValue}
+	m.Equal(originalPartitionData, dataFile.PartitionData)
+
+	err = writer.Add(entry)
+	m.Require().Error(err)
+
+	m.Zero(writer.addedFiles)
+	m.Zero(writer.addedRows)
+	m.Zero(writer.existingFiles)
+	m.Zero(writer.existingRows)
+	m.Zero(writer.deletedFiles)
+	m.Zero(writer.deletedRows)
+	m.Empty(writer.partitions)
+	m.Equal(int64(-1), writer.minSeqNum)
+	m.Equal(originalPartitionData, dataFile.PartitionData)
+}
+
+func newDatePartitionManifestEntry(t *testing.T, partitionSpec PartitionSpec, snapshotID int64, partitionValue any) ManifestEntry {
+	t.Helper()
+
+	seqNum := int64(1)
+	builder, err := NewDataFileBuilder(
+		partitionSpec,
+		EntryContentData,
+		"s3://bucket/ns/table/data/date-partition.parquet",
+		ParquetFile,
+		map[int]any{1000: partitionValue},
+		nil,
+		nil,
+		5,
+		1024,
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return NewManifestEntry(EntryStatusADDED, &snapshotID, &seqNum, &seqNum, builder.Build())
+}
+
 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