You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by at...@apache.org on 2026/09/01 10:56:15 UTC
(iceberg-go) branch main updated: fix(table): resolve dropped equality fields from schema history (#1952)
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 fc0e6fcaa fix(table): resolve dropped equality fields from schema history (#1952)
fc0e6fcaa is described below
commit fc0e6fcaa7d469d782ecd54e7d025481698dabd4
Author: Matt Faltyn <76...@users.noreply.github.com>
AuthorDate: Tue Sep 1 12:56:09 2026 +0200
fix(table): resolve dropped equality fields from schema history (#1952)
fix(table): resolve dropped equality fields from history
Signed-off-by: Matt Faltyn <fa...@gmail.com>
---
table/arrow_scanner.go | 16 ++++++++++-
table/equality_delete_reader.go | 39 ++++++++++++++++++++++----
table/equality_delete_reader_bench_test.go | 2 +-
table/equality_delete_reader_internal_test.go | 34 +++++++++++++++++++----
table/equality_delete_reader_test.go | 40 ++++++++++++++++++---------
5 files changed, 106 insertions(+), 25 deletions(-)
diff --git a/table/arrow_scanner.go b/table/arrow_scanner.go
index 567aa1052..cda8ad341 100644
--- a/table/arrow_scanner.go
+++ b/table/arrow_scanner.go
@@ -2189,8 +2189,22 @@ func (as *arrowScan) GetRecords(ctx context.Context, tasks []FileScanTask) (*arr
return nil, nil, err
}
+ var tableSchemas []*iceberg.Schema
+
+loadSchemaHistory:
+ for _, task := range tasks {
+ for _, deleteFile := range task.EqualityDeleteFiles {
+ for _, fieldID := range deleteFile.EqualityFieldIDs() {
+ if _, found := invariants.tableSchema.FindFieldByID(fieldID); !found {
+ tableSchemas = as.metadata.Schemas()
+
+ break loadSchemaHistory
+ }
+ }
+ }
+ }
equalityDeleteLoader, err := newLazyEqualityDeleteLoader(
- as.fs, invariants.tableSchema, invariants.nameMapping, tasks)
+ as.fs, invariants.tableSchema, tableSchemas, invariants.nameMapping, tasks)
if err != nil {
releasePerFilePosDeletes(deletesPerFile)
diff --git a/table/equality_delete_reader.go b/table/equality_delete_reader.go
index 7fd35b194..8bbcde74d 100644
--- a/table/equality_delete_reader.go
+++ b/table/equality_delete_reader.go
@@ -302,9 +302,35 @@ func newEqualityDeleteFileSet(id int, deleteSet *equalityDeleteSet) *equalityDel
}
}
+func schemaForEqualityFields(current *iceberg.Schema, schemas []*iceberg.Schema, fieldIDs []int) *iceberg.Schema {
+ hasAllFields := func(schema *iceberg.Schema) bool {
+ for _, fieldID := range fieldIDs {
+ if _, ok := schema.FindFieldByID(fieldID); !ok {
+ return false
+ }
+ }
+
+ return true
+ }
+
+ if hasAllFields(current) {
+ return current
+ }
+ // Scan tasks do not retain the equality delete's sequence number, so use
+ // the newest schema that can resolve the file's complete equality key.
+ for i := len(schemas) - 1; i >= 0; i-- {
+ if hasAllFields(schemas[i]) {
+ return schemas[i]
+ }
+ }
+
+ return current
+}
+
type lazyEqualityDeleteLoader struct {
fs iceio.IO
tableSchema *iceberg.Schema
+ tableSchemas []*iceberg.Schema
nameMapping iceberg.NameMapping
files map[string]*lazyEqualityDeleteFile
combinations sync.Map
@@ -328,14 +354,16 @@ type lazyEqualityDeleteCombination struct {
func newLazyEqualityDeleteLoader(
fs iceio.IO,
tableSchema *iceberg.Schema,
+ tableSchemas []*iceberg.Schema,
nameMapping iceberg.NameMapping,
tasks []FileScanTask,
) (*lazyEqualityDeleteLoader, error) {
loader := &lazyEqualityDeleteLoader{
- fs: fs,
- tableSchema: tableSchema,
- nameMapping: nameMapping,
- files: make(map[string]*lazyEqualityDeleteFile),
+ fs: fs,
+ tableSchema: tableSchema,
+ tableSchemas: tableSchemas,
+ nameMapping: nameMapping,
+ files: make(map[string]*lazyEqualityDeleteFile),
}
for _, task := range tasks {
@@ -383,8 +411,9 @@ func (l *lazyEqualityDeleteLoader) addFieldIDs(idset set[int]) {
func (l *lazyEqualityDeleteLoader) loadFile(ctx context.Context, file *lazyEqualityDeleteFile) (*equalityDeleteFileSet, error) {
file.once.Do(func() {
+ deleteSchema := schemaForEqualityFields(l.tableSchema, l.tableSchemas, file.fieldIDs)
keys, colNames, err := readEqualityDeleteFile(
- ctx, l.fs, l.tableSchema, l.nameMapping, file.dataFile, file.fieldIDs)
+ ctx, l.fs, deleteSchema, l.nameMapping, file.dataFile, file.fieldIDs)
if err != nil {
file.err = err
diff --git a/table/equality_delete_reader_bench_test.go b/table/equality_delete_reader_bench_test.go
index f3f3d2671..176f27c9f 100644
--- a/table/equality_delete_reader_bench_test.go
+++ b/table/equality_delete_reader_bench_test.go
@@ -168,7 +168,7 @@ func BenchmarkLazyEqualityDeleteLoading(b *testing.B) {
b.ResetTimer()
for b.Loop() {
loader, err := newLazyEqualityDeleteLoader(
- input.fs, input.tableSchema, nil, input.tasks)
+ input.fs, input.tableSchema, nil, nil, input.tasks)
if err != nil {
b.Fatal(err)
}
diff --git a/table/equality_delete_reader_internal_test.go b/table/equality_delete_reader_internal_test.go
index f4e3a4aeb..1ff4c3501 100644
--- a/table/equality_delete_reader_internal_test.go
+++ b/table/equality_delete_reader_internal_test.go
@@ -259,7 +259,7 @@ func TestReadAllEqualityDeleteFilesRejectsEmptyEqualityFieldIDs(t *testing.T) {
require.ErrorContains(t, err, "empty-equality-fields.parquet")
_, err = newLazyEqualityDeleteLoader(
- iceio.NewMemFS(), schema, nil,
+ iceio.NewMemFS(), schema, nil, nil,
[]FileScanTask{{EqualityDeleteFiles: []iceberg.DataFile{deleteFile}}})
require.ErrorIs(t, err, ErrEmptyEqualityFieldIDs)
require.ErrorContains(t, err, "empty-equality-fields.parquet")
@@ -285,7 +285,7 @@ func TestLazyEqualityDeleteLoaderLoadsFilesOnDemand(t *testing.T) {
{EqualityDeleteFiles: []iceberg.DataFile{deleteB, deleteA}},
}
- loader, err := newLazyEqualityDeleteLoader(fs, tableSchema, nil, tasks)
+ loader, err := newLazyEqualityDeleteLoader(fs, tableSchema, nil, nil, tasks)
require.NoError(t, err)
assert.Equal(t, int64(0), fs.opens.Load())
@@ -315,6 +315,30 @@ func TestLazyEqualityDeleteLoaderLoadsFilesOnDemand(t *testing.T) {
assert.Equal(t, int64(2), fs.attempts.Load())
}
+func TestLazyEqualityDeleteLoaderRejectsFieldAbsentFromSchemaHistory(t *testing.T) {
+ t.Parallel()
+
+ tableSchema := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+ historicalSchema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 2, Name: "data", Type: iceberg.PrimitiveTypes.String},
+ )
+ fs := iceio.NewMemFS()
+ deletePath := "mem://lazy-equality/unknown-field.parquet"
+ writeEqualityDeleteParquetToMemFS(t, fs, deletePath, `[{"id": 1}]`)
+ deleteFile := newEqualityDeleteSetAssemblyTestFile(t, deletePath, []int{99})
+ task := FileScanTask{EqualityDeleteFiles: []iceberg.DataFile{deleteFile}}
+
+ loader, err := newLazyEqualityDeleteLoader(
+ fs, tableSchema, []*iceberg.Schema{historicalSchema, tableSchema}, nil, []FileScanTask{task})
+ require.NoError(t, err)
+
+ _, err = loader.load(t.Context(), task)
+ require.ErrorContains(t, err, "equality delete field ID 99 not found in table schema")
+ require.ErrorContains(t, err, deletePath)
+}
+
func TestLazyEqualityDeleteLoaderReadsSharedFileOnceConcurrently(t *testing.T) {
t.Parallel()
@@ -327,7 +351,7 @@ func TestLazyEqualityDeleteLoaderReadsSharedFileOnceConcurrently(t *testing.T) {
deleteFile := newEqualityDeleteSetAssemblyTestFile(t, deletePath, []int{1})
tasks := []FileScanTask{{EqualityDeleteFiles: []iceberg.DataFile{deleteFile}}}
- loader, err := newLazyEqualityDeleteLoader(fs, tableSchema, nil, tasks)
+ loader, err := newLazyEqualityDeleteLoader(fs, tableSchema, nil, nil, tasks)
require.NoError(t, err)
const workers = 16
@@ -368,7 +392,7 @@ func TestLazyEqualityDeleteLoaderCachesReadErrors(t *testing.T) {
t, "mem://lazy-equality/missing.parquet", []int{1})
tasks := []FileScanTask{{EqualityDeleteFiles: []iceberg.DataFile{deleteFile}}}
- loader, err := newLazyEqualityDeleteLoader(fs, tableSchema, nil, tasks)
+ loader, err := newLazyEqualityDeleteLoader(fs, tableSchema, nil, nil, tasks)
require.NoError(t, err)
_, firstErr := loader.load(t.Context(), tasks[0])
_, secondErr := loader.load(t.Context(), tasks[0])
@@ -390,7 +414,7 @@ func TestLazyEqualityDeleteLoaderHonorsCanceledContext(t *testing.T) {
deleteFile := newEqualityDeleteSetAssemblyTestFile(t, deletePath, []int{1})
tasks := []FileScanTask{{EqualityDeleteFiles: []iceberg.DataFile{deleteFile}}}
- loader, err := newLazyEqualityDeleteLoader(fs, tableSchema, nil, tasks)
+ loader, err := newLazyEqualityDeleteLoader(fs, tableSchema, nil, nil, tasks)
require.NoError(t, err)
ctx, cancel := context.WithCancel(t.Context())
cancel()
diff --git a/table/equality_delete_reader_test.go b/table/equality_delete_reader_test.go
index d1d2db7ff..b022b95cc 100644
--- a/table/equality_delete_reader_test.go
+++ b/table/equality_delete_reader_test.go
@@ -143,24 +143,38 @@ func TestEqualityDeleteReadRoundTrip(t *testing.T) {
assert.Equal(t, []int64{1, 3, 5}, ids, "expected rows with id=2 and id=4 deleted")
- resultSchema, itr, err := tbl.Scan(table.WithSelectedFields("data")).ToArrowRecords(t.Context())
- require.NoError(t, err)
- require.Len(t, resultSchema.Fields(), 1)
- assert.Equal(t, "data", resultSchema.Fields()[0].Name)
-
- var data []string
- for rec, err := range itr {
+ scanData := func(tbl *table.Table) []string {
+ resultSchema, itr, err := tbl.Scan(table.WithSelectedFields("data")).ToArrowRecords(t.Context())
require.NoError(t, err)
- require.EqualValues(t, 1, rec.NumCols())
- col := rec.Column(0).(*array.String)
- for i := range col.Len() {
- data = append(data, col.Value(i))
+ require.Len(t, resultSchema.Fields(), 1)
+ assert.Equal(t, "data", resultSchema.Fields()[0].Name)
+
+ var data []string
+ for rec, err := range itr {
+ require.NoError(t, err)
+ require.EqualValues(t, 1, rec.NumCols())
+ col := rec.Column(0).(*array.String)
+ for i := range col.Len() {
+ data = append(data, col.Value(i))
+ }
+ rec.Release()
}
- rec.Release()
+
+ return data
}
- assert.Equal(t, []string{"alpha", "gamma", "epsilon"}, data,
+ assert.Equal(t, []string{"alpha", "gamma", "epsilon"}, scanData(tbl),
"expected equality deletes to apply when the equality field is not projected")
+
+ tx3 := tbl.NewTransaction()
+ require.NoError(t, tx3.UpdateSchema(true, false).DeleteColumn([]string{"id"}).Commit())
+ tbl, err = tx3.Commit(t.Context())
+ require.NoError(t, err)
+ _, currentHasEqualityField := tbl.Schema().FindFieldByID(1)
+ require.False(t, currentHasEqualityField)
+
+ assert.Equal(t, []string{"alpha", "gamma", "epsilon"}, scanData(tbl),
+ "expected equality deletes to apply when the equality field exists only in schema history")
}
func TestEqualityDeleteReadPrunesNonOverlappingDeleteFiles(t *testing.T) {