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 <[email protected]>
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())
+ }
+ }
+ })
+ }
+}
