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 774b4be6d perf(table): simplify equality delete conflict filters
(#2380)
774b4be6d is described below
commit 774b4be6da83c730500995614fc7e1e47c9ba558
Author: Minh Vu <[email protected]>
AuthorDate: Wed Nov 4 18:55:39 2026 +0100
perf(table): simplify equality delete conflict filters (#2380)
* perf(table): simplify equality delete conflict filters
Signed-off-by: Minh Vu <[email protected]>
* refactor(table): address equality delete filter review
Restore conflict-filter rationale and reuse source field types.
Remove arbitrary grouping threshold and consolidate test coverage.
Signed-off-by: Minh Vu <[email protected]>
* perf(table): group high-precision decimal equality delete partitions
Compare decimal scale against the source field instead of literal precision.
Avoid unused group maps and add cast and decimal regression coverage.
Signed-off-by: Minh Vu <[email protected]>
* style(table): fix nlreturn in equality delete literal guard
---------
Signed-off-by: Minh Vu <[email protected]>
---
table/conflict_validation.go | 124 +++++++++++++++++++---
table/partition_conflict_bench_test.go | 27 +++++
table/partition_conflict_test.go | 188 ++++++++++++++++++++++++++++++++-
3 files changed, 320 insertions(+), 19 deletions(-)
diff --git a/table/conflict_validation.go b/table/conflict_validation.go
index 9d0a4ca33..2343a3f7b 100644
--- a/table/conflict_validation.go
+++ b/table/conflict_validation.go
@@ -53,6 +53,7 @@ import (
"errors"
"fmt"
"io/fs"
+ "math"
"slices"
"sort"
"strconv"
@@ -872,8 +873,66 @@ type equalityDeletePartitionSpecInfo struct {
}
type equalityDeleteSourceFieldInfo struct {
- name string
- found bool
+ name string
+ fieldType iceberg.Type
+ found bool
+}
+
+// NaNs are not equal to themselves as Go map keys, while the existing
+// equality evaluator compares two NaNs as equal. Binary sets use hash keys,
+// which can collide, so retain individual predicates for those values too.
+// Keep literals that require a cast on the equality path.
+func equalityDeleteLiteralCanUseSet(lit iceberg.Literal, fieldType
iceberg.Type) bool {
+ switch value := lit.(type) {
+ case iceberg.Float32Literal:
+ return !math.IsNaN(float64(value)) &&
lit.Type().Equals(fieldType)
+ case iceberg.Float64Literal:
+ return !math.IsNaN(float64(value)) &&
lit.Type().Equals(fieldType)
+ case iceberg.DecimalLiteral:
+ // DecimalLiteral.Type reports a placeholder precision of 9.
+ // Equal scales bind without conversion for any declared
precision.
+ decimalType, ok := fieldType.(iceberg.DecimalType)
+
+ return ok && value.Scale == decimalType.Scale()
+ case iceberg.BoolLiteral, iceberg.Int32Literal, iceberg.Int64Literal,
+ iceberg.DateLiteral, iceberg.TimeLiteral,
iceberg.TimestampLiteral,
+ iceberg.TimestampNsLiteral, iceberg.StringLiteral,
iceberg.UUIDLiteral:
+ return lit.Type().Equals(fieldType)
+ default:
+ return false
+ }
+}
+
+type equalityDeleteIdentityGroupKey struct {
+ specID int
+ fieldID int
+}
+
+type equalityDeleteIdentityGroup struct {
+ fieldName string
+ index int
+ literals []iceberg.Literal
+}
+
+// balancedConflictDisjunction combines partition terms pairwise so the OR
+// expression tree has logarithmic depth while preserving term order.
+func balancedConflictDisjunction(terms []iceberg.BooleanExpression)
iceberg.BooleanExpression {
+ if len(terms) == 0 {
+ return iceberg.AlwaysTrue{}
+ }
+ for len(terms) > 1 {
+ next := terms[:0]
+ for i := 0; i < len(terms); i += 2 {
+ if i+1 == len(terms) {
+ next = append(next, terms[i])
+ } else {
+ next = append(next, iceberg.NewOr(terms[i],
terms[i+1]))
+ }
+ }
+ terms = next
+ }
+
+ return terms[0]
}
// eqDeletePartitionsToFilter converts equality-delete data files into an
@@ -896,6 +955,8 @@ type equalityDeleteSourceFieldInfo struct {
func eqDeletePartitionsToFilter(files []iceberg.DataFile, meta Metadata)
(iceberg.BooleanExpression, error) {
terms := make([]iceberg.BooleanExpression, 0, len(files))
specInfoByID := make(map[int]*equalityDeletePartitionSpecInfo)
+ var identityGroups
map[equalityDeleteIdentityGroupKey]*equalityDeleteIdentityGroup
+ var groups []*equalityDeleteIdentityGroup
seen := make(map[string]struct{}, len(files))
var currentSchema *iceberg.Schema
@@ -962,6 +1023,7 @@ func eqDeletePartitionsToFilter(files []iceberg.DataFile,
meta Metadata) (iceber
}
conjuncts := make([]iceberg.BooleanExpression, 0, len(p))
+ grouped := false
for _, partFieldID := range fieldIDs {
pf, ok := specInfo.fields[partFieldID]
if !ok {
@@ -976,7 +1038,7 @@ func eqDeletePartitionsToFilter(files []iceberg.DataFile,
meta Metadata) (iceber
currentSchema = meta.CurrentSchema()
}
field, found :=
currentSchema.FindFieldByID(pf.FirstSourceID())
- sourceField =
equalityDeleteSourceFieldInfo{name: field.Name, found: found}
+ sourceField =
equalityDeleteSourceFieldInfo{name: field.Name, fieldType: field.Type, found:
found}
specInfo.sourceFields[partFieldID] = sourceField
}
if !sourceField.found {
@@ -995,25 +1057,57 @@ func eqDeletePartitionsToFilter(files
[]iceberg.DataFile, meta Metadata) (iceber
return nil, fmt.Errorf("partition field %q:
%w", sourceField.name, err)
}
- conjuncts = append(conjuncts,
iceberg.LiteralPredicate(iceberg.OpEQ, iceberg.Reference(sourceField.name),
lit))
+ // Keep values needing a cast on the equality path,
including
+ // strings that can cast to NaN floats during binding.
+ if len(fieldIDs) == 1 &&
equalityDeleteLiteralCanUseSet(lit, sourceField.fieldType) {
+ key := equalityDeleteIdentityGroupKey{specID:
specID, fieldID: partFieldID}
+ group := identityGroups[key]
+ if group == nil {
+ if identityGroups == nil {
+ identityGroups =
make(map[equalityDeleteIdentityGroupKey]*equalityDeleteIdentityGroup)
+ }
+ group =
&equalityDeleteIdentityGroup{fieldName: sourceField.name, index: len(terms)}
+ identityGroups[key] = group
+ groups = append(groups, group)
+ terms = append(terms, nil)
+ }
+ group.literals = append(group.literals, lit)
+ grouped = true
+
+ break
+ }
+
+ conjuncts = append(conjuncts,
+ iceberg.LiteralPredicate(iceberg.OpEQ,
iceberg.Reference(sourceField.name), lit))
+ }
+ if grouped {
+ continue
}
+ var term iceberg.BooleanExpression
if len(conjuncts) == 1 {
- terms = append(terms, conjuncts[0])
+ term = conjuncts[0]
} else {
- terms = append(terms, iceberg.NewAnd(conjuncts[0],
conjuncts[1], conjuncts[2:]...))
+ term = iceberg.NewAnd(conjuncts[0], conjuncts[1],
conjuncts[2:]...)
+ }
+ terms = append(terms, term)
+ }
+
+ for _, group := range groups {
+ // SetPredicate keeps singleton groups as ordinary equality
predicates.
+ // Keep larger IN sets within the metrics evaluator's per-value
limit.
+ for start := 0; start < len(group.literals); start +=
inPredicateLimit {
+ end := min(start+inPredicateLimit, len(group.literals))
+ term := iceberg.SetPredicate(iceberg.OpIn,
iceberg.Reference(group.fieldName), group.literals[start:end])
+ if start == 0 {
+ terms[group.index] = term
+ } else {
+ terms = append(terms, term)
+ }
}
}
- if len(terms) == 0 {
- return iceberg.AlwaysTrue{}, nil
- }
-
- if len(terms) == 1 {
- return terms[0], nil
- }
-
- return iceberg.NewOr(terms[0], terms[1], terms[2:]...), nil
+ return balancedConflictDisjunction(terms), nil
}
// validateNoNewDeletesForRewrittenFiles rejects the commit if any
diff --git a/table/partition_conflict_bench_test.go
b/table/partition_conflict_bench_test.go
index 9ec32ff17..ebbc86ba3 100644
--- a/table/partition_conflict_bench_test.go
+++ b/table/partition_conflict_bench_test.go
@@ -35,9 +35,12 @@ func BenchmarkEqDeletePartitionsToFilter(b *testing.B) {
fileCount int
partitionCount int
}{
+ {fileCount: 1, partitionCount: 1},
+ {fileCount: 16, partitionCount: 16},
{fileCount: 1_000, partitionCount: 100},
{fileCount: 10_000, partitionCount: 100},
{fileCount: 1_000, partitionCount: 1_000},
+ {fileCount: 4_096, partitionCount: 4_096},
} {
name := fmt.Sprintf(
"fields=%d/files=%d/partitions=%d",
@@ -55,6 +58,9 @@ func BenchmarkEqDeletePartitionsToFilter(b *testing.B) {
b.Run("after", func(b *testing.B) {
benchmarkEqDeletePartitionFilter(b,
files, meta, eqDeletePartitionsToFilter)
})
+ b.Run("after_bind", func(b *testing.B) {
+
benchmarkEqDeletePartitionFilterAndBind(b, files, meta)
+ })
})
}
}
@@ -83,6 +89,27 @@ func benchmarkEqDeletePartitionFilter(
}
}
+// benchmarkEqDeletePartitionFilterAndBind includes schema binding in the
+// measured work without performing manifest reads or commit IO.
+func benchmarkEqDeletePartitionFilterAndBind(b *testing.B, files
[]iceberg.DataFile, meta Metadata) {
+ b.Helper()
+ b.ReportAllocs()
+ schema := meta.CurrentSchema()
+ b.ResetTimer()
+
+ for range b.N {
+ filter, err := eqDeletePartitionsToFilter(files, meta)
+ if err != nil {
+ b.Fatal(err)
+ }
+ bound, err := iceberg.BindExpr(schema, filter, true)
+ if err != nil {
+ b.Fatal(err)
+ }
+ equalityDeleteConflictFilterBenchmarkSink = bound
+ }
+}
+
func equalityDeleteConflictBenchmarkInput(
b *testing.B,
fieldCount, fileCount, partitionCount int,
diff --git a/table/partition_conflict_test.go b/table/partition_conflict_test.go
index c09018793..3b344cb18 100644
--- a/table/partition_conflict_test.go
+++ b/table/partition_conflict_test.go
@@ -21,9 +21,11 @@ import (
"bytes"
"context"
"fmt"
+ "math"
"path/filepath"
"testing"
+ "github.com/apache/arrow-go/v18/arrow/decimal128"
"github.com/apache/iceberg-go"
iceio "github.com/apache/iceberg-go/io"
"github.com/google/uuid"
@@ -241,10 +243,10 @@ func
TestEqDeletePartitionsToFilterDeduplicatesPartitions(t *testing.T) {
filter, err := eqDeletePartitionsToFilter(files, countingMeta)
require.NoError(t, err)
- expected := iceberg.NewOr(
- iceberg.EqualTo(iceberg.Reference("region"), "us-east-1"),
- iceberg.EqualTo(iceberg.Reference("region"), "eu-west-1"),
- )
+ expected := iceberg.SetPredicate(iceberg.OpIn,
iceberg.Reference("region"), []iceberg.Literal{
+ iceberg.StringLiteral("us-east-1"),
+ iceberg.StringLiteral("eu-west-1"),
+ })
assert.True(t, filter.Equals(expected), "duplicate equality-delete
partitions should produce one filter term")
assert.Equal(t, 1, countingMeta.currentSchemaCalls,
"the current schema should be cloned once per filter")
@@ -803,3 +805,181 @@ func
TestRowDeltaValidate_DayTransformFallsBackToAlwaysTrue(t *testing.T) {
require.Error(t, err, "day-partitioned eq-delete must trigger
conservative AlwaysTrue fallback even for a different day")
assert.ErrorIs(t, err, ErrConflictingDataFiles)
}
+
+func TestEqDeleteConflictSingleFieldWithNullAndSeparatedValues(t *testing.T) {
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "region", Type:
iceberg.PrimitiveTypes.String},
+ )
+ meta := partitionedConflictMeta(t, schema, 1, nil)
+ spec := meta.PartitionSpec()
+ fieldID := spec.Field(0).FieldID
+
+ var files []iceberg.DataFile
+ values := []any{"west", "east", nil, "central"}
+ for i := range 32 {
+ value := values[i%len(values)]
+ files = append(files, equalityDeleteTestFile(t, spec,
+ fmt.Sprintf("delete-%d.parquet", i),
map[int]any{fieldID: value}))
+ }
+ filter, err := eqDeletePartitionsToFilter(files, meta)
+ require.NoError(t, err)
+
+ want := iceberg.NewOr(
+ iceberg.SetPredicate(iceberg.OpIn, iceberg.Reference("region"),
+ []iceberg.Literal{
+ iceberg.StringLiteral("west"),
+ iceberg.StringLiteral("east"),
+ iceberg.StringLiteral("central"),
+ }),
+ iceberg.IsNull(iceberg.Reference("region")),
+ )
+ require.True(t, filter.Equals(want), "single-field values must be
grouped without dropping nulls")
+ _, err = iceberg.BindExpr(meta.CurrentSchema(), filter, true)
+ require.NoError(t, err)
+}
+
+func TestEqDeleteConflictLargeMultiFieldPredicateBinds(t *testing.T) {
+ const fieldCount = 3
+ fields := make([]iceberg.NestedField, fieldCount)
+ partFields := make([]iceberg.PartitionField, fieldCount)
+ for i := range fields {
+ fields[i] = iceberg.NestedField{
+ ID: i + 1, Name: fmt.Sprintf("field_%d", i), Type:
iceberg.PrimitiveTypes.String,
+ }
+ partFields[i] = iceberg.PartitionField{
+ FieldID: 1000 + i, SourceIDs: []int{i + 1},
+ Name: fmt.Sprintf("field_%d", i), Transform:
iceberg.IdentityTransform{},
+ }
+ }
+ schema := iceberg.NewSchema(0, fields...)
+ spec := iceberg.NewPartitionSpec(partFields...)
+ meta, err := NewMetadata(schema, &spec, UnsortedSortOrder,
"file:///tmp/conflict-balanced",
+ iceberg.Properties{PropertyFormatVersion: "2"})
+ require.NoError(t, err)
+
+ files := make([]iceberg.DataFile, 256)
+ for i := range files {
+ p := make(map[int]any, fieldCount)
+ for field := range fieldCount {
+ p[1000+field] = fmt.Sprintf("v-%d-%d", i, field)
+ }
+ files[i] = equalityDeleteTestFile(t, spec,
fmt.Sprintf("del-%d.parquet", i), p)
+ }
+
+ filter, err := eqDeletePartitionsToFilter(files, meta)
+ require.NoError(t, err)
+ require.Equal(t, iceberg.OpOr, filter.Op())
+ _, err = iceberg.BindExpr(meta.CurrentSchema(), filter, true)
+ require.NoError(t, err)
+}
+
+func TestEqDeleteConflictLargeSparseSetPreservesMetricsPruning(t *testing.T) {
+ schema := iceberg.NewSchema(0, iceberg.NestedField{ID: 1, Name:
"value", Type: iceberg.PrimitiveTypes.Int32})
+ meta := partitionedConflictMeta(t, schema, 1, nil)
+ spec := meta.PartitionSpec()
+ fieldID := spec.Field(0).FieldID
+ files := make([]iceberg.DataFile, 401)
+ for i := range files {
+ files[i] = equalityDeleteTestFile(t, spec,
fmt.Sprintf("delete-%d.parquet", i), map[int]any{fieldID: int32(i * 2)})
+ }
+ filter, err := eqDeletePartitionsToFilter(files, meta)
+ require.NoError(t, err)
+ evaluate, err := newInclusiveMetricsEvaluator(schema, filter, true,
true)
+ require.NoError(t, err)
+ for _, tc := range []struct {
+ value int32
+ want bool
+ }{{0, true}, {1, false}, {199, false}, {200, true}, {399, false}, {400,
true}, {799, false}, {800, true}, {801, false}} {
+ raw, err := iceberg.NewLiteral(tc.value).MarshalBinary()
+ require.NoError(t, err)
+ got, err := evaluate(inclusiveMetricsInTestFile(t, raw, raw))
+ require.NoError(t, err)
+ require.Equalf(t, tc.want, got, "value=%d", tc.value)
+ }
+}
+
+// NaN equality must remain an equality predicate: Go map lookup cannot find
+// a NaN key, so placing it in an IN set can miss a conflicting partition.
+func TestEqDeleteConflictNaNPartition(t *testing.T) {
+ schema := iceberg.NewSchema(0, iceberg.NestedField{ID: 1, Name:
"value", Type: iceberg.PrimitiveTypes.Float64})
+ meta := partitionedConflictMeta(t, schema, 1, nil)
+ spec := meta.PartitionSpec()
+ fieldID := spec.Field(0).FieldID
+ values := []float64{1, 2, 3, 4, 5, 6, 7, math.NaN()}
+ files := make([]iceberg.DataFile, 32)
+ for i := range files {
+ value := values[i%len(values)]
+ files[i] = equalityDeleteTestFile(t, spec,
fmt.Sprintf("delete-%d.parquet", i), map[int]any{fieldID: value})
+ }
+ filter, err := eqDeletePartitionsToFilter(files, meta)
+ require.NoError(t, err)
+ filters := newKeyDefaultMapWrapErr(func(id int)
(iceberg.BooleanExpression, error) {
+ return buildPartitionProjection(id, meta, schema, filter, true)
+ })
+ evaluate, err := buildPartitionEvaluator(spec.ID(), meta, schema,
filters, true)
+ require.NoError(t, err)
+ for _, i := range []int{0, 6, 7} {
+ got, err := evaluate(files[i])
+ require.NoError(t, err)
+ require.True(t, got, "partition %d must match", i)
+ }
+}
+
+func TestEqDeleteConflictDecimalPrecisionPreservesGrouping(t *testing.T) {
+ // DecimalLiteral.Type reports decimal(9, scale), even when the source
+ // column has higher precision. Grouping must use the actual source
scale.
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "amount", Type:
iceberg.MustDecimalTypeOf(18, 2)},
+ )
+ meta := partitionedConflictMeta(t, schema, 1, nil)
+ spec := meta.PartitionSpec()
+ fieldID := spec.Field(0).FieldID
+
+ values := []iceberg.Decimal{
+ {Val: decimal128.FromI64(123), Scale: 2},
+ {Val: decimal128.FromI64(456), Scale: 2},
+ {Val: decimal128.FromI64(789), Scale: 2},
+ }
+ files := make([]iceberg.DataFile, len(values))
+ literals := make([]iceberg.Literal, len(values))
+ for i, value := range values {
+ files[i] = equalityDeleteTestFile(t, spec,
fmt.Sprintf("decimal-%d.parquet", i),
+ map[int]any{fieldID: value})
+ literals[i] = iceberg.NewLiteral(value)
+ }
+
+ filter, err := eqDeletePartitionsToFilter(files, meta)
+ require.NoError(t, err)
+ want := iceberg.SetPredicate(iceberg.OpIn, iceberg.Reference("amount"),
literals)
+ require.True(t, filter.Equals(want), "decimal precision must not
disable safe IN grouping")
+
+ _, err = iceberg.BindExpr(meta.CurrentSchema(), filter, true)
+ require.NoError(t, err)
+}
+
+func TestEqDeleteConflictCastValuePreservesEqualityPredicate(t *testing.T) {
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 1, Name: "value", Type:
iceberg.PrimitiveTypes.Float64},
+ )
+ meta := partitionedConflictMeta(t, schema, 1, nil)
+ spec := meta.PartitionSpec()
+ fieldID := spec.Field(0).FieldID
+
+ files := []iceberg.DataFile{
+ equalityDeleteTestFile(t, spec, "float-1.parquet",
map[int]any{fieldID: float64(1)}),
+ equalityDeleteTestFile(t, spec, "cast-nan.parquet",
map[int]any{fieldID: "NaN"}),
+ equalityDeleteTestFile(t, spec, "float-2.parquet",
map[int]any{fieldID: float64(2)}),
+ }
+
+ filter, err := eqDeletePartitionsToFilter(files, meta)
+ require.NoError(t, err)
+ want := iceberg.NewOr(
+ iceberg.SetPredicate(iceberg.OpIn, iceberg.Reference("value"),
[]iceberg.Literal{
+ iceberg.Float64Literal(1), iceberg.Float64Literal(2),
+ }),
+ iceberg.LiteralPredicate(iceberg.OpEQ,
iceberg.Reference("value"), iceberg.StringLiteral("NaN")),
+ )
+ require.True(t, filter.Equals(want), "cast values must retain
individual equality predicates")
+ _, err = iceberg.BindExpr(meta.CurrentSchema(), filter, true)
+ require.NoError(t, err)
+}
