(iceberg-go) branch main updated: feat(table): set referenced_data_file on single-target v2 position deletes (#2189)

[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 28ff8d85c feat(table): set referenced_data_file on single-target v2 position deletes (#2189)
28ff8d85c is described below

commit 28ff8d85cc187292c23a2e6db904d12050995bcd
Author: Revanth <re...@gmail.com>
AuthorDate: Mon Oct 19 10:36:29 2026 -0500

    feat(table): set referenced_data_file on single-target v2 position deletes (#2189)
---
 table/internal/parquet_files.go  | 39 ++++++++++++++++++--
 table/pos_delete_metrics_test.go | 77 ++++++++++++++++++++++++++++++++++++++++
 2 files changed, 114 insertions(+), 2 deletions(-)

diff --git a/table/internal/parquet_files.go b/table/internal/parquet_files.go
index f8ee6b4f7..9805878dd 100644
--- a/table/internal/parquet_files.go
+++ b/table/internal/parquet_files.go
@@ -1263,7 +1263,7 @@ func (w *ParquetFileWriter) Close() (_ iceberg.DataFile, err error) {
 		return nil, err
 	}
 
-	return stats.ToDataFile(DataFileOpts{
+	opts := DataFileOpts{
 		Schema:          w.info.FileSchema,
 		Spec:            w.info.Spec,
 		Path:            w.info.FileName,
@@ -1272,7 +1272,42 @@ func (w *ParquetFileWriter) Close() (_ iceberg.DataFile, err error) {
 		FileSize:        w.counter.Count,
 		PartitionValues: w.partition,
 		SortOrderID:     w.info.SortOrderID,
-	}), nil
+	}
+	if w.info.Content == iceberg.EntryContentPosDeletes {
+		opts.ReferencedDataFile = singleReferencedDataFile(stats)
+	}
+
+	return stats.ToDataFile(opts), nil
+}
+
+// posDeleteFilePathFieldID is the reserved field ID of the file_path column
+// in a position delete file.
+const posDeleteFilePathFieldID = 2147483546
+
+// singleReferencedDataFile returns the data file path when every delete in a
+// position delete file targets that one file, or nil otherwise. It compares
+// the untruncated file_path min and max, which position delete writers always
+// collect (see table.positionDeleteMetricsProps). Java's v2
+// PositionDeleteWriter leaves the field unset; the spec allows it on position
+// delete files with a single target.
+func singleReferencedDataFile(stats *DataFileStatistics) *string {
+	agg, ok := stats.ColAggs[posDeleteFilePathFieldID]
+	if !ok {
+		return nil
+	}
+
+	lower, ok := agg.Min().(iceberg.StringLiteral)
+	if !ok || lower == "" {
+		return nil
+	}
+	upper, ok := agg.Max().(iceberg.StringLiteral)
+	if !ok || lower != upper {
+		return nil
+	}
+
+	path := string(lower)
+
+	return &path
 }
 
 func (w *ParquetFileWriter) Abort() error {
diff --git a/table/pos_delete_metrics_test.go b/table/pos_delete_metrics_test.go
index 258a1c42e..ed03a6f9c 100644
--- a/table/pos_delete_metrics_test.go
+++ b/table/pos_delete_metrics_test.go
@@ -130,3 +130,80 @@ func TestPositionDeleteFilesKeepFullFilePathBounds(t *testing.T) {
 		}
 	}
 }
+
+// TestPositionDeleteFilesSetReferencedDataFile pins #2167: a v2 position
+// delete file whose deletes all target one data file records that file in
+// referenced_data_file, and one spanning several data files leaves it nil.
+// PlanFiles reads the delete files back from v2 manifests, so the field must
+// also survive the manifest round trip.
+func TestPositionDeleteFilesSetReferencedDataFile(t *testing.T) {
+	tests := []struct {
+		name      string
+		filter    iceberg.BooleanExpression
+		wantSaved bool
+	}{
+		{"single target", iceberg.EqualTo(iceberg.Reference("id"), int64(2)), true},
+		{"multiple targets", iceberg.NewOr(
+			iceberg.EqualTo(iceberg.Reference("id"), int64(2)),
+			iceberg.EqualTo(iceberg.Reference("id"), int64(4))), false},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+
+			location := filepath.ToSlash(t.TempDir())
+			schema := iceberg.NewSchema(0,
+				iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+			)
+			meta, err := table.NewMetadata(schema, iceberg.UnpartitionedSpec, table.UnsortedSortOrder, location,
+				iceberg.Properties{
+					table.PropertyFormatVersion: "2",
+					table.WriteDeleteModeKey:    table.WriteModeMergeOnRead,
+				})
+			require.NoError(t, err)
+
+			metaLoc := location + "/metadata/v1.metadata.json"
+			fsF := func(context.Context) (iceio.IO, error) { return iceio.LocalFS{}, nil }
+			cat := &concurrentTestCatalog{metadata: meta, location: metaLoc, fsF: fsF}
+			tbl := table.New(table.Identifier{"db", "pos_delete_ref"}, meta, metaLoc, fsF, cat)
+
+			arrowSchema := arrow.NewSchema([]arrow.Field{
+				{Name: "id", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
+			}, nil)
+			for _, rows := range []string{`[{"id":1},{"id":2}]`, `[{"id":3},{"id":4}]`} {
+				data, err := array.TableFromJSON(memory.DefaultAllocator, arrowSchema, []string{rows})
+				require.NoError(t, err)
+				rdr := array.NewTableReader(data, -1)
+				tbl, err = tbl.Append(ctx, rdr, nil)
+				rdr.Release()
+				data.Release()
+				require.NoError(t, err)
+			}
+
+			tbl, err = tbl.Delete(ctx, tc.filter, nil)
+			require.NoError(t, err)
+
+			tasks, err := tbl.Scan().PlanFiles(ctx)
+			require.NoError(t, err)
+
+			deleteFiles := make(map[string]iceberg.DataFile)
+			for _, task := range tasks {
+				for _, df := range task.DeleteFiles {
+					require.Equal(t, iceberg.EntryContentPosDeletes, df.ContentType())
+					deleteFiles[df.FilePath()] = df
+					if tc.wantSaved {
+						require.NotNil(t, df.ReferencedDataFile())
+						require.Equal(t, task.File.FilePath(), *df.ReferencedDataFile())
+					}
+				}
+			}
+			require.Len(t, deleteFiles, 1)
+			if !tc.wantSaved {
+				for _, df := range deleteFiles {
+					require.Nil(t, df.ReferencedDataFile())
+				}
+			}
+		})
+	}
+}