This is an automated email from the ASF dual-hosted git repository.
laskoviymishka 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 26e566b90 fix(table): keep full file_path bounds on v2 position delete
files (#2164)
26e566b90 is described below
commit 26e566b902ebcad7cd9d0cc1c808204af80bbbce
Author: Revanth <[email protected]>
AuthorDate: Sun Oct 11 13:17:57 2026 -0500
fix(table): keep full file_path bounds on v2 position delete files (#2164)
---
table/cow_delete_conflict_test.go | 42 ++++++------
table/pos_delete_metrics_test.go | 132 ++++++++++++++++++++++++++++++++++++++
table/rolling_data_writer.go | 2 +-
table/writer.go | 30 ++++++++-
table/writer_test.go | 22 +++++++
5 files changed, 206 insertions(+), 22 deletions(-)
diff --git a/table/cow_delete_conflict_test.go
b/table/cow_delete_conflict_test.go
index 693681be3..4a11118f2 100644
--- a/table/cow_delete_conflict_test.go
+++ b/table/cow_delete_conflict_test.go
@@ -177,27 +177,29 @@ func TestCopyOnWriteConflict_ConcurrentAppendCommits(t
*testing.T) {
// TestCopyOnWriteConflict_ConcurrentDeleteOnUntouchedFileCommits pins the
// validator's scope: a concurrent delete against a data file this commit does
// not remove must not reject it. v3 deletion vectors carry
-// referenced_data_file, so the delete resolves to its exact target; v2
position
-// deletes written here are partition-scoped and conservatively conflict.
+// referenced_data_file, and v2 position deletes keep full file_path bounds
+// (#2141), so either resolves to its exact target.
func TestCopyOnWriteConflict_ConcurrentDeleteOnUntouchedFileCommits(t
*testing.T) {
- for _, isolation := range cowIsolations {
- for _, op := range cowOps {
- t.Run(fmt.Sprintf("v3/%s/%s", isolation, op), func(t
*testing.T) {
- ctx := context.Background()
- tbl := appendTenRows(t,
newCoWConflictTestTable(t, "3", isolation))
- // A second data file that the id==2 removal
does not touch.
- tbl, err := tbl.Append(ctx, cowTestRecords(t,
`[{"id":11,"data":"k"},{"id":12,"data":"l"}]`), nil)
- require.NoError(t, err)
-
- txn := stageCopyOnWrite(t, tbl, op,
iceberg.EqualTo(iceberg.Reference("id"), int64(2)))
-
- _, err = tbl.Delete(ctx,
iceberg.EqualTo(iceberg.Reference("id"), int64(12)), nil)
- require.NoError(t, err)
-
- committed, err := txn.Commit(ctx)
- require.NoError(t, err)
- require.Equal(t, []int64{1, 3, 4, 5, 6, 7, 8,
9, 10, 11}, idsInTable(t, committed))
- })
+ for _, version := range cowFormatVersions {
+ for _, isolation := range cowIsolations {
+ for _, op := range cowOps {
+ t.Run(fmt.Sprintf("v%s/%s/%s", version,
isolation, op), func(t *testing.T) {
+ ctx := context.Background()
+ tbl := appendTenRows(t,
newCoWConflictTestTable(t, version, isolation))
+ // A second data file that the id==2
removal does not touch.
+ tbl, err := tbl.Append(ctx,
cowTestRecords(t, `[{"id":11,"data":"k"},{"id":12,"data":"l"}]`), nil)
+ require.NoError(t, err)
+
+ txn := stageCopyOnWrite(t, tbl, op,
iceberg.EqualTo(iceberg.Reference("id"), int64(2)))
+
+ _, err = tbl.Delete(ctx,
iceberg.EqualTo(iceberg.Reference("id"), int64(12)), nil)
+ require.NoError(t, err)
+
+ committed, err := txn.Commit(ctx)
+ require.NoError(t, err)
+ require.Equal(t, []int64{1, 3, 4, 5, 6,
7, 8, 9, 10, 11}, idsInTable(t, committed))
+ })
+ }
}
}
}
diff --git a/table/pos_delete_metrics_test.go b/table/pos_delete_metrics_test.go
new file mode 100644
index 000000000..258a1c42e
--- /dev/null
+++ b/table/pos_delete_metrics_test.go
@@ -0,0 +1,132 @@
+// 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_test
+
+import (
+ "context"
+ "maps"
+ "path/filepath"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
+ iceio "github.com/apache/iceberg-go/io"
+ "github.com/apache/iceberg-go/table"
+ "github.com/stretchr/testify/require"
+)
+
+// TestPositionDeleteFilesKeepFullFilePathBounds pins #2141: v2 position delete
+// files keep exact file_path bounds and pos bounds whatever the table metrics
+// settings are, as Java's MetricsConfig.forPositionDelete does. Data file
+// paths are far longer than 16 characters, so the table-defaults case would
+// truncate them under the default truncate(16) mode, and the override case
+// asks for no file_path or pos metrics at all.
+func TestPositionDeleteFilesKeepFullFilePathBounds(t *testing.T) {
+ identity := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{2}, FieldID: 1000, Transform:
iceberg.IdentityTransform{}, Name: "data",
+ })
+
+ specs := []struct {
+ name string
+ spec *iceberg.PartitionSpec
+ }{
+ {"unpartitioned", iceberg.UnpartitionedSpec},
+ {"partitioned", &identity},
+ }
+
+ metricsConfigs := []struct {
+ name string
+ props iceberg.Properties
+ }{
+ {"table defaults", nil},
+ {"file_path and pos none", iceberg.Properties{
+ table.MetricsModeColumnConfPrefix + ".file_path":
"none",
+ table.MetricsModeColumnConfPrefix + ".pos":
"none",
+ }},
+ }
+
+ filePathField, ok :=
iceberg.PositionalDeleteSchema.FindFieldByName("file_path")
+ require.True(t, ok)
+ posField, ok := iceberg.PositionalDeleteSchema.FindFieldByName("pos")
+ require.True(t, ok)
+
+ for _, tc := range specs {
+ for _, mc := range metricsConfigs {
+ t.Run(tc.name+"/"+mc.name, func(t *testing.T) {
+ ctx := context.Background()
+
+ location := filepath.ToSlash(t.TempDir())
+ props := iceberg.Properties{
+ table.PropertyFormatVersion: "2",
+ table.WriteDeleteModeKey:
table.WriteModeMergeOnRead,
+ }
+ maps.Copy(props, mc.props)
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "id",
Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ iceberg.NestedField{ID: 2, Name:
"data", Type: iceberg.PrimitiveTypes.String, Required: true},
+ )
+ meta, err := table.NewMetadata(schema, tc.spec,
table.UnsortedSortOrder, location, props)
+ 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_metrics"}, meta, metaLoc, fsF, cat)
+
+ arrowSchema := arrow.NewSchema([]arrow.Field{
+ {Name: "id", Type:
arrow.PrimitiveTypes.Int64, Nullable: false},
+ {Name: "data", Type:
arrow.BinaryTypes.String, Nullable: false},
+ }, nil)
+ data, err :=
array.TableFromJSON(memory.DefaultAllocator, arrowSchema, []string{
+
`[{"id":1,"data":"a"},{"id":2,"data":"a"},{"id":3,"data":"b"},{"id":4,"data":"b"}]`,
+ })
+ require.NoError(t, err)
+ defer data.Release()
+
+ rdr := array.NewTableReader(data, -1)
+ defer rdr.Release()
+
+ tbl, err = tbl.Append(ctx, rdr, nil)
+ require.NoError(t, err)
+ tbl, err = tbl.Delete(ctx,
iceberg.EqualTo(iceberg.Reference("id"), int64(2)), nil)
+ require.NoError(t, err)
+
+ tasks, err := tbl.Scan().PlanFiles(ctx)
+ require.NoError(t, err)
+
+ deleteFiles := 0
+ for _, task := range tasks {
+ dataPath := task.File.FilePath()
+ require.Greater(t, len(dataPath), 16)
+ for _, df := range task.DeleteFiles {
+ require.Equal(t,
iceberg.EntryContentPosDeletes, df.ContentType())
+ deleteFiles++
+
+ require.Equal(t,
[]byte(dataPath), df.LowerBoundValues()[filePathField.ID])
+ require.Equal(t,
[]byte(dataPath), df.UpperBoundValues()[filePathField.ID])
+ require.NotEmpty(t,
df.LowerBoundValues()[posField.ID])
+ require.NotEmpty(t,
df.UpperBoundValues()[posField.ID])
+ }
+ }
+ require.Equal(t, 1, deleteFiles)
+ })
+ }
+ }
+}
diff --git a/table/rolling_data_writer.go b/table/rolling_data_writer.go
index e431d0f2b..e87d6fc3d 100644
--- a/table/rolling_data_writer.go
+++ b/table/rolling_data_writer.go
@@ -240,7 +240,7 @@ func newWriterFactory(rootLocation string, args
recordWritingArgs, meta *Metadat
f.variantFieldIDs =
tblutils.VariantFieldIDsFromSchema(f.fileSchema)
}
- f.statsCols, err = computeStatsPlan(f.fileSchema, meta.props)
+ f.statsCols, err = computeStatsPlan(f.fileSchema,
metricsPropsFor(f.content, meta.props))
if err != nil {
stopCount()
diff --git a/table/writer.go b/table/writer.go
index 9246d7fdd..16b32cefa 100644
--- a/table/writer.go
+++ b/table/writer.go
@@ -138,7 +138,7 @@ func newDataFileWriter(rootLocation string, fs
io.WriteFileIO, meta *MetadataBui
apply(&w)
}
- statsCols, err := computeStatsPlan(w.fileSchema, meta.props)
+ statsCols, err := computeStatsPlan(w.fileSchema,
metricsPropsFor(w.content, meta.props))
if err != nil {
return nil, err
}
@@ -239,6 +239,34 @@ type dataFileWriter interface {
writeFile(ctx context.Context, partitionValues map[int]any, task
WriteTask) (iceberg.DataFile, error)
}
+// positionDeleteMetricsProps mirrors Java's MetricsConfig.forPositionDelete:
+// position delete files keep full file_path and pos bounds whatever the table
+// metrics settings are, so readers and conflict checks can resolve the data
+// file a delete file targets from equal file_path bounds.
+//
+// The default mode is none on purpose, as in Java: any column besides
+// file_path and pos, such as a future row column, gets no value counts, null
+// counts, column sizes or bounds. The table's write.metadata.metrics.*
+// settings do not apply to position delete files.
+//
+// The map is shared and handed to computeStatsPlan as is; treat it as
+// read-only.
+var positionDeleteMetricsProps = iceberg.Properties{
+ DefaultWriteMetricsModeKey:
string(tblutils.MetricModeNone),
+ MetricsModeColumnConfPrefix + ".file_path":
string(tblutils.MetricModeFull),
+ MetricsModeColumnConfPrefix + ".pos":
string(tblutils.MetricModeFull),
+}
+
+// metricsPropsFor returns the properties that drive column metrics for files
+// of the given content type.
+func metricsPropsFor(content iceberg.ManifestEntryContent, props
iceberg.Properties) iceberg.Properties {
+ if content == iceberg.EntryContentPosDeletes {
+ return positionDeleteMetricsProps
+ }
+
+ return props
+}
+
func newPositionDeleteWriter(rootLocation string, fs io.WriteFileIO, meta
*MetadataBuilder, props iceberg.Properties, opts ...dataFileWriterOption)
(*defaultDataFileWriter, error) {
// Always enforce the file schema to be the Positional Delete Schema by
appending the option at the very end
return newDataFileWriter(rootLocation, fs, meta, props, append(opts,
withFileSchema(iceberg.PositionalDeleteSchema),
withContent(iceberg.EntryContentPosDeletes))...)
diff --git a/table/writer_test.go b/table/writer_test.go
index 6a2891842..eac6c2d71 100644
--- a/table/writer_test.go
+++ b/table/writer_test.go
@@ -265,3 +265,25 @@ func TestWriteFileRejectsPosDeleteSortOrderClaim(t
*testing.T) {
assert.Nil(t, df)
assert.Contains(t, err.Error(), "position delete")
}
+
+func TestMetricsPropsFor(t *testing.T) {
+ tableProps := iceberg.Properties{
+ DefaultWriteMetricsModeKey: "counts",
+ MetricsModeColumnConfPrefix + ".file_path": "none",
+ }
+
+ tests := []struct {
+ content iceberg.ManifestEntryContent
+ want iceberg.Properties
+ }{
+ {iceberg.EntryContentData, tableProps},
+ {iceberg.EntryContentEqDeletes, tableProps},
+ {iceberg.EntryContentPosDeletes, positionDeleteMetricsProps},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.content.String(), func(t *testing.T) {
+ assert.Equal(t, tt.want, metricsPropsFor(tt.content,
tableProps))
+ })
+ }
+}
