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 7aee8e973 feat(table): build transformed dynamic partition predicates
(#1675)
7aee8e973 is described below
commit 7aee8e9731cdf97e245de9c53e150434033ff004
Author: Minh Vu <[email protected]>
AuthorDate: Wed Sep 30 18:59:07 2026 +0200
feat(table): build transformed dynamic partition predicates (#1675)
* feat(expr): bind predicates over transform terms
Signed-off-by: Minh Vu <[email protected]>
* fix(expr): reject unknown transform predicates
Signed-off-by: Minh Vu <[email protected]>
* feat(table): match transformed dynamic partitions
Signed-off-by: Minh Vu <[email protected]>
* fix(expr): satisfy dynamic transform lint
Signed-off-by: Minh Vu <[email protected]>
* test(table): cover transformed rewrite filters
Signed-off-by: Minh Vu <[email protected]>
* fix(table): preserve dynamic partition source paths
Signed-off-by: Minh Vu <[email protected]>
* fix(table): validate dynamic partition transforms before matching
Signed-off-by: Minh Vu <[email protected]>
* test(table): cover transformed overwrite predicate evaluation
Signed-off-by: Minh Vu <[email protected]>
* test(table): satisfy partition predicate lint
Signed-off-by: Minh Vu <[email protected]>
* fix(table): validate dynamic overwrite partition values
Signed-off-by: Minh Vu <[email protected]>
* fix(table): support source-less void partitions
Signed-off-by: Minh Vu <[email protected]>
* fix(table): trim dynamic partition predicate validation
Signed-off-by: Minh Vu <[email protected]>
* style(table): satisfy nlreturn
Signed-off-by: Minh Vu <[email protected]>
* fix(codec): support ns timestamps in schema fingerprint
Signed-off-by: Minh Vu <[email protected]>
* test(codec): align ns timestamp cache validation
Signed-off-by: Minh Vu <[email protected]>
* ci: refresh integration checks against latest main
Signed-off-by: Minh Vu <[email protected]>
---------
Signed-off-by: Minh Vu <[email protected]>
---
partitions.go | 8 +-
partitions_test.go | 6 +
table/internal/partition_predicate.go | 194 +++++++++--
table/internal/partition_predicate_test.go | 524 ++++++++++++++++++++++++++---
table/transaction_internal_test.go | 15 +
5 files changed, 671 insertions(+), 76 deletions(-)
diff --git a/partitions.go b/partitions.go
index 4c87b7c39..3cf043e18 100644
--- a/partitions.go
+++ b/partitions.go
@@ -777,7 +777,13 @@ func (ps PartitionSpec) IsUnpartitioned() bool {
}
for _, f := range ps.fields {
- if _, ok := f.Transform.(VoidTransform); !ok {
+ switch transform := f.Transform.(type) {
+ case VoidTransform:
+ case *VoidTransform:
+ if transform == nil {
+ return false
+ }
+ default:
return false
}
}
diff --git a/partitions_test.go b/partitions_test.go
index 3ca53f754..a1b5a4b74 100644
--- a/partitions_test.go
+++ b/partitions_test.go
@@ -318,6 +318,12 @@ func TestUnpartitionedWithVoidField(t *testing.T) {
assert.True(t, spec.IsUnpartitioned())
+ pointerSpec := iceberg.NewPartitionSpec(iceberg.PartitionField{
+ SourceIDs: []int{3}, FieldID: 1001, Name: "void", Transform:
&iceberg.VoidTransform{},
+ })
+
+ assert.True(t, pointerSpec.IsUnpartitioned())
+
spec2 := iceberg.NewPartitionSpec(iceberg.PartitionField{
SourceIDs: []int{3}, FieldID: 1001, Name: "void", Transform:
iceberg.VoidTransform{},
}, iceberg.PartitionField{
diff --git a/table/internal/partition_predicate.go
b/table/internal/partition_predicate.go
index 663ed97fb..609d5e627 100644
--- a/table/internal/partition_predicate.go
+++ b/table/internal/partition_predicate.go
@@ -39,46 +39,62 @@ import (
// result is an OR across distinct partitions, each clause an AND across the
// spec's fields:
//
-// source == value when the partition value is present
-// IsNaN(source) when the value is a floating-point NaN (x == NaN is
never true)
-// IsNull(source) when the partition value is absent or nil
+// transform(source) == value when the partition value is present
+// IsNaN(transform(source)) when the value is a floating-point NaN (x ==
NaN is never true)
+// IsNull(transform(source)) when the partition value is explicitly nil
+// Missing partition field IDs are rejected with ErrInvalidArgument.
//
// Duplicate tuples collapse to a single clause, and an empty input yields
// AlwaysFalse (matching nothing). Callers are expected to pass a partitioned
// spec; dynamic partition overwrite rejects unpartitioned tables upstream.
+// Void fields are also accepted when their source column has been dropped (the
+// spec represents those tombstones with source ID 0), because void always
+// produces a null partition value and does not need a source column.
//
-// Because only identity transforms are accepted (see below), the partition
-// value equals the source-column value, so "source == value" selects exactly
-// the rows in that partition. Non-identity transforms (bucket, truncate, the
-// time transforms) cannot be matched by a source-column predicate and need
-// partition-level matching instead; they are rejected here and tracked as a
-// follow-up under issue #1215.
+// The transform is retained in the row predicate so the post-transform
+// partition value is compared against transform(source), not against the raw
+// source column. This is phase 1 of issue #1216. The current overwrite path
+// cannot execute non-identity predicates for partial-file rewrites because
+// source-column metrics are conservative and the Substrait row-filter
converter
+// rejects transformed terms; partition-level strict matching is still needed
+// before this helper can drive those rewrites (tracked in issue #1215).
func BuildPartitionMatchPredicate(spec iceberg.PartitionSpec, schema
*iceberg.Schema, partitions []map[int]any) (iceberg.BooleanExpression, error) {
type fieldRef struct {
- id int
- name string
+ id int
+ name string
+ transform iceberg.Transform
+ resultType iceberg.Type
+ isVoid bool
}
var fields []fieldRef
for _, f := range spec.Fields() {
- if _, ok := f.Transform.(iceberg.IdentityTransform); !ok {
- return nil, fmt.Errorf("%w: dynamic partition overwrite
supports identity-transform partition fields only, got %s on %q (tracked in
https://github.com/apache/iceberg-go/issues/1215)",
- iceberg.ErrNotImplemented, f.Transform, f.Name)
- }
-
- // Identity transforms always have exactly one source column.
+ // Partition transforms currently have exactly one source
column.
if len(f.SourceIDs) != 1 {
- return nil, fmt.Errorf("%w: identity partition field %q
must have exactly one source id, got %d",
+ return nil, fmt.Errorf("%w: partition field %q must
have exactly one source id, got %d",
iceberg.ErrInvalidArgument, f.Name,
len(f.SourceIDs))
}
+ if isVoidTransform(f.Transform) {
+ // A void field can survive source-column removal as a
source-less
+ // tombstone. Its output is always null, so resolving
or binding the
+ // source would add no information and would reject
source ID 0.
+ fields = append(fields, fieldRef{id: f.FieldID, name:
f.Name, transform: f.Transform, isVoid: true})
+
+ continue
+ }
- src, ok := schema.FindFieldByID(f.SourceIDs[0])
+ sourceName, ok := schema.FindColumnName(f.SourceIDs[0])
if !ok {
return nil, fmt.Errorf("%w: partition field %q
references unknown source id %d",
iceberg.ErrInvalidArgument, f.Name,
f.SourceIDs[0])
}
-
- fields = append(fields, fieldRef{id: f.FieldID, name: src.Name})
+ bound, err := iceberg.NewUnboundTransform(f.Transform,
iceberg.Reference(sourceName)).Bind(schema, true)
+ if err != nil {
+ return nil, fmt.Errorf("partition field %q: %w",
f.Name, err)
+ }
+ fields = append(fields, fieldRef{
+ id: f.FieldID, name: sourceName, transform:
f.Transform, resultType: bound.Type(),
+ })
}
// NewAnd/NewOr fold away the AlwaysTrue/AlwaysFalse seeds, so a
single-field,
@@ -96,20 +112,47 @@ func BuildPartitionMatchPredicate(spec
iceberg.PartitionSpec, schema *iceberg.Sc
sigParts := make([]string, 0, len(fields))
for _, fr := range fields {
- ref := iceberg.Reference(fr.name)
-
- // A missing key (!ok) is treated like an explicit nil.
In practice
// DataFile.Partition() stores a null partition value
as a nil-valued
- // entry rather than an absent key; !ok is a defensive
guard.
+ // entry. A missing field ID is malformed and must fail
closed rather
+ // than silently under-deleting.
val, ok := part[fr.id]
- if !ok || val == nil {
- clause = iceberg.NewAnd(clause,
iceberg.IsNull(ref))
+ if !ok {
+ return nil, fmt.Errorf("%w: partition field %q
(field id %d) is missing from partition tuple",
+ iceberg.ErrInvalidArgument, fr.name,
fr.id)
+ }
+ if val == nil {
+ if fr.isVoid {
+ // Void transforms produce null for
every source value, so their
+ // null predicate is AlwaysTrue.
Keeping the clause unchanged
+ // avoids inventing a reference for a
source-less tombstone.
+ sigParts = append(sigParts,
strconv.Itoa(fr.id)+":null")
+
+ continue
+ }
+
+ term := partitionTerm(fr.transform, fr.name)
+ clause = iceberg.NewAnd(clause,
iceberg.IsNull(term))
sigParts = append(sigParts,
strconv.Itoa(fr.id)+":null")
continue
}
- if isNaN(val) {
+ if fr.isVoid {
+ return nil, fmt.Errorf("%w: partition value for
void field %q must be nil",
+ iceberg.ErrInvalidArgument, fr.name)
+ }
+ term := partitionTerm(fr.transform, fr.name)
+
+ lit, err := LiteralForPartitionValue(val)
+ if err != nil {
+ return nil, fmt.Errorf("partition field %q:
%w", fr.name, err)
+ }
+ lit, err = validatePartitionValue(fr.transform,
fr.resultType, lit)
+ if err != nil {
+ return nil, fmt.Errorf("partition field %q:
%w", fr.name, err)
+ }
+
+ if isNaN(lit.Any()) {
// x == NaN is never true (IEEE 754), so a NaN
partition value must
// match via IsNaN. This intentionally diverges
from PyIceberg's
// _build_partition_predicate, which emits
EqualTo for every value:
@@ -119,18 +162,13 @@ func BuildPartitionMatchPredicate(spec
iceberg.PartitionSpec, schema *iceberg.Sc
// NaN has many valid bit patterns, so the
dedup signature uses a
// fixed sentinel (not MarshalBinary) to keep
duplicate NaN
// partitions collapsing to a single clause.
- clause = iceberg.NewAnd(clause,
iceberg.IsNaN(ref))
+ clause = iceberg.NewAnd(clause,
iceberg.IsNaN(term))
sigParts = append(sigParts,
strconv.Itoa(fr.id)+":nan")
continue
}
- lit, err := LiteralForPartitionValue(val)
- if err != nil {
- return nil, fmt.Errorf("partition field %q:
%w", fr.name, err)
- }
-
- clause = iceberg.NewAnd(clause,
iceberg.LiteralPredicate(iceberg.OpEQ, ref, lit))
+ clause = iceberg.NewAnd(clause,
iceberg.LiteralPredicate(iceberg.OpEQ, term, lit))
enc, err := lit.MarshalBinary()
if err != nil {
@@ -151,6 +189,92 @@ func BuildPartitionMatchPredicate(spec
iceberg.PartitionSpec, schema *iceberg.Sc
return result, nil
}
+func partitionTerm(transform iceberg.Transform, name string)
iceberg.UnboundTerm {
+ ref := iceberg.Reference(name)
+ if isIdentityTransform(transform) {
+ return ref
+ }
+
+ return iceberg.NewUnboundTransform(transform, ref)
+}
+
+func isIdentityTransform(transform iceberg.Transform) bool {
+ switch t := transform.(type) {
+ case iceberg.IdentityTransform:
+ return true
+ case *iceberg.IdentityTransform:
+ return t != nil
+ default:
+ return false
+ }
+}
+
+func isVoidTransform(transform iceberg.Transform) bool {
+ switch t := transform.(type) {
+ case iceberg.VoidTransform:
+ return true
+ case *iceberg.VoidTransform:
+ return t != nil
+ default:
+ return false
+ }
+}
+
+func isTruncateTransform(transform iceberg.Transform) bool {
+ switch t := transform.(type) {
+ case iceberg.TruncateTransform:
+ return true
+ case *iceberg.TruncateTransform:
+ return t != nil
+ default:
+ return false
+ }
+}
+
+func validatePartitionValue(transform iceberg.Transform, resultType
iceberg.Type, lit iceberg.Literal) (iceberg.Literal, error) {
+ normalized, err := lit.To(resultType)
+ if err != nil {
+ return nil, fmt.Errorf("%w: partition value type %s cannot be
converted to transform result type %s: %w",
+ iceberg.ErrInvalidArgument, lit.Type(), resultType, err)
+ }
+
+ switch normalized.(type) {
+ case iceberg.AboveMaxLiteral, iceberg.BelowMinLiteral:
+ return nil, fmt.Errorf("%w: partition value %s is outside
transform result type %s",
+ iceberg.ErrInvalidArgument, normalized, resultType)
+ }
+
+ switch t := transform.(type) {
+ case iceberg.BucketTransform:
+ if err := validateBucketPartitionValue(t.NumBuckets,
normalized); err != nil {
+ return nil, err
+ }
+ }
+
+ if (isIdentityTransform(transform) || isTruncateTransform(transform))
&& !isNaN(normalized.Any()) {
+ applied :=
transform.Apply(iceberg.Optional[iceberg.Literal]{Valid: true, Val: normalized})
+ if !applied.Valid || !applied.Val.Equals(normalized) {
+ return nil, fmt.Errorf("%w: partition value %s is not
in the range of transform %s",
+ iceberg.ErrInvalidArgument, normalized,
transform)
+ }
+ }
+
+ return normalized, nil
+}
+
+func validateBucketPartitionValue(numBuckets int, lit iceberg.Literal) error {
+ value, ok := lit.(iceberg.Int32Literal)
+ if !ok {
+ return fmt.Errorf("%w: bucket partition value must be int32,
got %s", iceberg.ErrInvalidArgument, lit.Type())
+ }
+ if value.Value() < 0 || int64(value.Value()) >= int64(numBuckets) {
+ return fmt.Errorf("%w: bucket partition value %d is outside [0,
%d)",
+ iceberg.ErrInvalidArgument, value.Value(), numBuckets)
+ }
+
+ return nil
+}
+
// isNaN reports whether v is a floating-point NaN.
func isNaN(v any) bool {
switch f := v.(type) {
diff --git a/table/internal/partition_predicate_test.go
b/table/internal/partition_predicate_test.go
index ff6f2f8ea..530d4ae7b 100644
--- a/table/internal/partition_predicate_test.go
+++ b/table/internal/partition_predicate_test.go
@@ -20,6 +20,7 @@ package internal
import (
"math"
"testing"
+ "time"
"github.com/apache/arrow-go/v18/arrow/decimal128"
"github.com/apache/iceberg-go"
@@ -43,12 +44,30 @@ func identityField(sourceID, fieldID int, name string)
iceberg.PartitionField {
return iceberg.PartitionField{SourceIDs: []int{sourceID}, FieldID:
fieldID, Name: name, Transform: iceberg.IdentityTransform{}}
}
-func identitySpec(fields ...iceberg.PartitionField) iceberg.PartitionSpec {
+func specWithFields(fields ...iceberg.PartitionField) iceberg.PartitionSpec {
return iceberg.NewPartitionSpec(fields...)
}
+type partitionPredicateRow []any
+
+func (r partitionPredicateRow) Size() int { return len(r) }
+func (r partitionPredicateRow) Get(pos int) any { return r[pos] }
+func (r partitionPredicateRow) Set(pos int, val any) {
+ r[pos] = val
+}
+
+func timestampAtUTC(year, month, day, hour, minute, second, microsecond int)
iceberg.Timestamp {
+ return iceberg.Timestamp(time.Date(year, time.Month(month), day, hour,
minute, second,
+ microsecond*int(time.Microsecond), time.UTC).UnixMicro())
+}
+
+func dateAtUTC(year, month, day int) iceberg.Date {
+ return iceberg.Date(time.Date(year, time.Month(month), day, 0, 0, 0, 0,
time.UTC).Unix() /
+ int64((24*time.Hour)/time.Second))
+}
+
func TestBuildPartitionMatchPredicate_EmptyInput(t *testing.T) {
- spec := identitySpec(identityField(1, 1000, "id"))
+ spec := specWithFields(identityField(1, 1000, "id"))
expr, err := BuildPartitionMatchPredicate(spec,
dynamicOverwriteSchema(), nil)
require.NoError(t, err)
@@ -56,7 +75,7 @@ func TestBuildPartitionMatchPredicate_EmptyInput(t
*testing.T) {
}
func TestBuildPartitionMatchPredicate_SingleField(t *testing.T) {
- spec := identitySpec(identityField(1, 1000, "id"))
+ spec := specWithFields(identityField(1, 1000, "id"))
expr, err := BuildPartitionMatchPredicate(spec,
dynamicOverwriteSchema(), []map[int]any{
{1000: int32(5)},
@@ -68,7 +87,7 @@ func TestBuildPartitionMatchPredicate_SingleField(t
*testing.T) {
}
func TestBuildPartitionMatchPredicate_MultipleFields(t *testing.T) {
- spec := identitySpec(
+ spec := specWithFields(
identityField(1, 1000, "id"),
identityField(2, 1001, "category"),
)
@@ -86,7 +105,7 @@ func TestBuildPartitionMatchPredicate_MultipleFields(t
*testing.T) {
}
func TestBuildPartitionMatchPredicate_MultiplePartitions(t *testing.T) {
- spec := identitySpec(identityField(1, 1000, "id"))
+ spec := specWithFields(identityField(1, 1000, "id"))
expr, err := BuildPartitionMatchPredicate(spec,
dynamicOverwriteSchema(), []map[int]any{
{1000: int32(5)},
@@ -102,7 +121,7 @@ func TestBuildPartitionMatchPredicate_MultiplePartitions(t
*testing.T) {
}
func TestBuildPartitionMatchPredicate_MultipleFieldsAndPartitions(t
*testing.T) {
- spec := identitySpec(
+ spec := specWithFields(
identityField(1, 1000, "id"),
identityField(2, 1001, "category"),
)
@@ -127,30 +146,25 @@ func
TestBuildPartitionMatchPredicate_MultipleFieldsAndPartitions(t *testing.T)
}
func TestBuildPartitionMatchPredicate_NullValue(t *testing.T) {
- spec := identitySpec(identityField(2, 1001, "category"))
+ spec := specWithFields(identityField(2, 1001, "category"))
- // Both an explicit nil and a missing key represent a null partition
value.
- cases := []struct {
- name string
- partitions []map[int]any
- }{
- {"explicit nil", []map[int]any{{1001: nil}}},
- {"missing key", []map[int]any{{}}},
- }
+ expr, err := BuildPartitionMatchPredicate(spec,
dynamicOverwriteSchema(), []map[int]any{{1001: nil}})
+ require.NoError(t, err)
- for _, tc := range cases {
- t.Run(tc.name, func(t *testing.T) {
- expr, err := BuildPartitionMatchPredicate(spec,
dynamicOverwriteSchema(), tc.partitions)
- require.NoError(t, err)
+ want := iceberg.IsNull(iceberg.Reference("category"))
+ assert.True(t, expr.Equals(want), "want %s, got %s", want, expr)
+}
- want := iceberg.IsNull(iceberg.Reference("category"))
- assert.True(t, expr.Equals(want), "want %s, got %s",
want, expr)
- })
- }
+func TestBuildPartitionMatchPredicate_MissingFieldID(t *testing.T) {
+ spec := specWithFields(identityField(2, 1001, "category"))
+
+ _, err := BuildPartitionMatchPredicate(spec, dynamicOverwriteSchema(),
[]map[int]any{{}})
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+ assert.Contains(t, err.Error(), "missing from partition tuple")
}
func TestBuildPartitionMatchPredicate_NullAndNonNullInSameTuple(t *testing.T) {
- spec := identitySpec(
+ spec := specWithFields(
identityField(1, 1000, "id"),
identityField(2, 1001, "category"),
)
@@ -182,7 +196,7 @@ func TestBuildPartitionMatchPredicate_NaNValue(t
*testing.T) {
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
- spec := identitySpec(tc.field)
+ spec := specWithFields(tc.field)
expr, err := BuildPartitionMatchPredicate(spec,
dynamicOverwriteSchema(), []map[int]any{
{tc.field.FieldID: tc.nan},
@@ -196,7 +210,7 @@ func TestBuildPartitionMatchPredicate_NaNValue(t
*testing.T) {
}
func TestBuildPartitionMatchPredicate_NaNDeduplicates(t *testing.T) {
- spec := identitySpec(identityField(4, 1003, "score"))
+ spec := specWithFields(identityField(4, 1003, "score"))
// NaN has many valid bit patterns; distinct payloads must still
collapse to a
// single IsNaN clause (the dedup key uses a sentinel, not the raw
float bits).
@@ -215,7 +229,7 @@ func TestBuildPartitionMatchPredicate_NaNDeduplicates(t
*testing.T) {
}
func TestBuildPartitionMatchPredicate_DeduplicatesPartitions(t *testing.T) {
- spec := identitySpec(identityField(1, 1000, "id"))
+ spec := specWithFields(identityField(1, 1000, "id"))
// Many data files land in the same partition; the predicate must not
repeat it.
expr, err := BuildPartitionMatchPredicate(spec,
dynamicOverwriteSchema(), []map[int]any{
@@ -230,7 +244,7 @@ func
TestBuildPartitionMatchPredicate_DeduplicatesPartitions(t *testing.T) {
}
func TestBuildPartitionMatchPredicate_DeduplicatesMultiFieldPartitions(t
*testing.T) {
- spec := identitySpec(
+ spec := specWithFields(
identityField(1, 1000, "id"),
identityField(2, 1001, "category"),
)
@@ -252,7 +266,7 @@ func
TestBuildPartitionMatchPredicate_DeduplicatesMultiFieldPartitions(t *testin
// not collide into a single clause. A naive "%d=%v"-joined signature would map
// both of these to "1001=5/1002=three/1002=x" and silently drop one partition.
func TestBuildPartitionMatchPredicate_NoSeparatorCollision(t *testing.T) {
- spec := identitySpec(
+ spec := specWithFields(
identityField(2, 1001, "category"),
identityField(3, 1002, "sub"),
)
@@ -276,30 +290,398 @@ func
TestBuildPartitionMatchPredicate_NoSeparatorCollision(t *testing.T) {
assert.True(t, expr.Equals(want), "distinct tuples must not be deduped,
got %s", expr)
}
-func TestBuildPartitionMatchPredicate_NonIdentityTransformRejected(t
*testing.T) {
+func TestBuildPartitionMatchPredicate_UsesNonIdentityTransform(t *testing.T) {
cases := []struct {
name string
- transform iceberg.Transform
+ field iceberg.PartitionField
+ source string
+ value any
+ wantValue iceberg.Literal
}{
- {"bucket", iceberg.BucketTransform{NumBuckets: 4}},
- {"truncate", iceberg.TruncateTransform{Width: 10}},
- {"day", iceberg.DayTransform{}},
+ {
+ name: "bucket",
+ field: iceberg.PartitionField{SourceIDs: []int{1},
FieldID: 1000, Name: "id_part", Transform: iceberg.BucketTransform{NumBuckets:
4}},
+ source: "id",
+ value: int32(1),
+ wantValue: iceberg.Int32Literal(1),
+ },
+ {
+ name: "truncate",
+ field: iceberg.PartitionField{SourceIDs: []int{2},
FieldID: 1000, Name: "category_part", Transform:
iceberg.TruncateTransform{Width: 3}},
+ source: "category",
+ value: "boo",
+ wantValue: iceberg.StringLiteral("boo"),
+ },
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
- spec := identitySpec(iceberg.PartitionField{SourceIDs:
[]int{1}, FieldID: 1000, Name: "id_part", Transform: tc.transform})
+ spec := specWithFields(tc.field)
- _, err := BuildPartitionMatchPredicate(spec,
dynamicOverwriteSchema(), []map[int]any{{1000: int32(1)}})
- require.Error(t, err)
- assert.ErrorIs(t, err, iceberg.ErrNotImplemented)
+ expr, err := BuildPartitionMatchPredicate(spec,
dynamicOverwriteSchema(), []map[int]any{{1000: tc.value}})
+ require.NoError(t, err)
+
+ want := iceberg.LiteralPredicate(iceberg.OpEQ,
+ iceberg.NewUnboundTransform(tc.field.Transform,
iceberg.Reference(tc.source)),
+ tc.wantValue,
+ )
+ assert.True(t, expr.Equals(want), "want %s, got %s",
want, expr)
})
}
}
+func bucketValueForInt32(t *testing.T, transform iceberg.BucketTransform,
value int32) int32 {
+ t.Helper()
+
+ result := transform.Apply(iceberg.Optional[iceberg.Literal]{
+ Valid: true,
+ Val: iceberg.Int32Literal(value),
+ })
+ require.True(t, result.Valid)
+
+ return result.Val.(iceberg.Int32Literal).Value()
+}
+
+func TestBuildPartitionMatchPredicate_EvaluatesTransforms(t *testing.T) {
+ bucketTransform := iceberg.BucketTransform{NumBuckets: 4}
+ bucketPartition := bucketValueForInt32(t, bucketTransform, 1)
+ bucketNonMatch := int32(2)
+ for bucketValueForInt32(t, bucketTransform, bucketNonMatch) ==
bucketPartition {
+ bucketNonMatch++
+ }
+
+ cases := []struct {
+ name string
+ sourceType iceberg.Type
+ transform iceberg.Transform
+ partitionValue any
+ matchingValues []any
+ nonMatchingVals []any
+ }{
+ {
+ name: "bucket",
+ sourceType: iceberg.PrimitiveTypes.Int32,
+ transform: bucketTransform,
+ partitionValue: bucketPartition,
+ matchingValues: []any{int32(1)},
+ nonMatchingVals: []any{bucketNonMatch},
+ },
+ {
+ name: "truncate",
+ sourceType: iceberg.PrimitiveTypes.String,
+ transform: iceberg.TruncateTransform{Width: 3},
+ partitionValue: "boo",
+ matchingValues: []any{"boo", "books", "booster"},
+ nonMatchingVals: []any{"bar", "science"},
+ },
+ {
+ name: "year",
+ sourceType: iceberg.PrimitiveTypes.Timestamp,
+ transform: iceberg.YearTransform{},
+ partitionValue: int32(50), // 2020 - 1970
+ matchingValues: []any{
+ timestampAtUTC(2020, 1, 1, 0, 0, 0, 0),
+ timestampAtUTC(2020, 12, 31, 23, 59, 59,
999999),
+ },
+ nonMatchingVals: []any{
+ timestampAtUTC(2019, 12, 31, 23, 59, 59,
999999),
+ timestampAtUTC(2021, 1, 1, 0, 0, 0, 0),
+ },
+ },
+ {
+ name: "month",
+ sourceType: iceberg.PrimitiveTypes.Timestamp,
+ transform: iceberg.MonthTransform{},
+ partitionValue: int32(601), // 2020-02, relative to
1970-01
+ matchingValues: []any{
+ timestampAtUTC(2020, 2, 1, 0, 0, 0, 0),
+ timestampAtUTC(2020, 2, 29, 23, 59, 59, 999999),
+ },
+ nonMatchingVals: []any{
+ timestampAtUTC(2020, 1, 31, 23, 59, 59, 999999),
+ timestampAtUTC(2020, 3, 1, 0, 0, 0, 0),
+ },
+ },
+ {
+ name: "day",
+ sourceType: iceberg.PrimitiveTypes.Timestamp,
+ transform: iceberg.DayTransform{},
+ partitionValue: dateAtUTC(2020, 2, 29),
+ matchingValues: []any{
+ timestampAtUTC(2020, 2, 29, 0, 0, 0, 0),
+ timestampAtUTC(2020, 2, 29, 23, 59, 59, 999999),
+ },
+ nonMatchingVals: []any{
+ timestampAtUTC(2020, 2, 28, 23, 59, 59, 999999),
+ timestampAtUTC(2020, 3, 1, 0, 0, 0, 0),
+ },
+ },
+ {
+ name: "hour",
+ sourceType: iceberg.PrimitiveTypes.Timestamp,
+ transform: iceberg.HourTransform{},
+ partitionValue: int32(439714), // 2020-02-29 10:00 UTC,
relative to 1970
+ matchingValues: []any{
+ timestampAtUTC(2020, 2, 29, 10, 0, 0, 0),
+ timestampAtUTC(2020, 2, 29, 10, 59, 59, 999999),
+ },
+ nonMatchingVals: []any{
+ timestampAtUTC(2020, 2, 29, 9, 59, 59, 999999),
+ timestampAtUTC(2020, 2, 29, 11, 0, 0, 0),
+ },
+ },
+ {
+ name: "hour before epoch",
+ sourceType: iceberg.PrimitiveTypes.Timestamp,
+ transform: iceberg.HourTransform{},
+ partitionValue: int32(-1), // 1969-12-31 23:00 UTC
+ matchingValues: []any{
+ timestampAtUTC(1969, 12, 31, 23, 0, 0, 0),
+ timestampAtUTC(1969, 12, 31, 23, 59, 59,
999999),
+ },
+ nonMatchingVals: []any{
+ timestampAtUTC(1969, 12, 31, 22, 59, 59,
999999),
+ timestampAtUTC(1970, 1, 1, 0, 0, 0, 0),
+ },
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ schema := iceberg.NewSchema(0, iceberg.NestedField{
+ ID: 1, Name: "value", Type: tc.sourceType,
+ })
+ spec := specWithFields(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name:
"value_part", Transform: tc.transform,
+ })
+
+ expr, err := BuildPartitionMatchPredicate(spec, schema,
[]map[int]any{{1000: tc.partitionValue}})
+ require.NoError(t, err)
+
+ eval, err := iceberg.ExpressionEvaluator(schema, expr,
true)
+ require.NoError(t, err)
+
+ for _, value := range tc.matchingValues {
+ matched, err :=
eval(partitionPredicateRow{value})
+ require.NoError(t, err)
+ assert.True(t, matched, "source value %v should
match partition value %v", value, tc.partitionValue)
+ }
+ for _, value := range tc.nonMatchingVals {
+ matched, err :=
eval(partitionPredicateRow{value})
+ require.NoError(t, err)
+ assert.False(t, matched, "source value %v
should not match partition value %v", value, tc.partitionValue)
+ }
+ })
+ }
+}
+
+func TestBuildPartitionMatchPredicate_EvaluatesBucketCollision(t *testing.T) {
+ transform := iceberg.BucketTransform{NumBuckets: 4}
+ valuesByBucket := make(map[int32][]int32)
+ var collisionBucket int32
+ var collisionValues []int32
+
+ for value := range int32(10000) {
+ bucket := bucketValueForInt32(t, transform, value)
+ valuesByBucket[bucket] = append(valuesByBucket[bucket], value)
+ if len(valuesByBucket[bucket]) == 2 {
+ collisionBucket = bucket
+ collisionValues = valuesByBucket[bucket]
+
+ break
+ }
+ }
+ require.Len(t, collisionValues, 2)
+ assert.NotEqual(t, collisionValues[0], collisionValues[1])
+
+ nonMatchingValue := int32(-1)
+ for value := range int32(10000) {
+ if bucketValueForInt32(t, transform, value) != collisionBucket {
+ nonMatchingValue = value
+
+ break
+ }
+ }
+ require.NotEqual(t, int32(-1), nonMatchingValue)
+
+ schema := iceberg.NewSchema(0, iceberg.NestedField{
+ ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32,
+ })
+ spec := specWithFields(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "id_bucket",
Transform: transform,
+ })
+ expr, err := BuildPartitionMatchPredicate(spec, schema,
[]map[int]any{{1000: collisionBucket}})
+ require.NoError(t, err)
+
+ eval, err := iceberg.ExpressionEvaluator(schema, expr, true)
+ require.NoError(t, err)
+ for _, value := range collisionValues {
+ matched, err := eval(partitionPredicateRow{value})
+ require.NoError(t, err)
+ assert.True(t, matched, "source value %d should match bucket
%d", value, collisionBucket)
+ }
+ matched, err := eval(partitionPredicateRow{nonMatchingValue})
+ require.NoError(t, err)
+ assert.False(t, matched, "source value %d should not match bucket %d",
nonMatchingValue, collisionBucket)
+}
+
+func TestBuildPartitionMatchPredicate_EvaluatesTransformedNull(t *testing.T) {
+ cases := []struct {
+ name string
+ sourceType iceberg.Type
+ transform iceberg.Transform
+ nonNil any
+ nonNilMatches bool
+ }{
+ {name: "bucket", sourceType: iceberg.PrimitiveTypes.Int32,
transform: iceberg.BucketTransform{NumBuckets: 4}, nonNil: int32(1),
nonNilMatches: false},
+ {name: "truncate", sourceType: iceberg.PrimitiveTypes.String,
transform: iceberg.TruncateTransform{Width: 3}, nonNil: "books", nonNilMatches:
false},
+ {name: "year", sourceType: iceberg.PrimitiveTypes.Timestamp,
transform: iceberg.YearTransform{}, nonNil: timestampAtUTC(2020, 1, 1, 0, 0, 0,
0), nonNilMatches: false},
+ {name: "month", sourceType: iceberg.PrimitiveTypes.Timestamp,
transform: iceberg.MonthTransform{}, nonNil: timestampAtUTC(2020, 1, 1, 0, 0,
0, 0), nonNilMatches: false},
+ {name: "day", sourceType: iceberg.PrimitiveTypes.Timestamp,
transform: iceberg.DayTransform{}, nonNil: timestampAtUTC(2020, 1, 1, 0, 0, 0,
0), nonNilMatches: false},
+ {name: "hour", sourceType: iceberg.PrimitiveTypes.Timestamp,
transform: iceberg.HourTransform{}, nonNil: timestampAtUTC(2020, 1, 1, 0, 0, 0,
0), nonNilMatches: false},
+ {name: "void", sourceType: iceberg.PrimitiveTypes.String,
transform: iceberg.VoidTransform{}, nonNil: "books", nonNilMatches: true},
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ schema := iceberg.NewSchema(0, iceberg.NestedField{
+ ID: 1, Name: "value", Type: tc.sourceType,
+ })
+ spec := specWithFields(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name:
"value_part", Transform: tc.transform,
+ })
+
+ expr, err := BuildPartitionMatchPredicate(spec, schema,
[]map[int]any{{1000: nil}})
+ require.NoError(t, err)
+
+ eval, err := iceberg.ExpressionEvaluator(schema, expr,
true)
+ require.NoError(t, err)
+
+ matched, err := eval(partitionPredicateRow{nil})
+ require.NoError(t, err)
+ assert.True(t, matched, "null source value should match
null partition value")
+
+ matched, err = eval(partitionPredicateRow{tc.nonNil})
+ require.NoError(t, err)
+ assert.Equal(t, tc.nonNilMatches, matched, "unexpected
match for non-null source value")
+ })
+ }
+}
+
+func TestBuildPartitionMatchPredicate_VoidBindsAlwaysTrue(t *testing.T) {
+ spec := specWithFields(iceberg.PartitionField{
+ SourceIDs: []int{0}, FieldID: 1001, Name: "category_void",
Transform: iceberg.VoidTransform{},
+ })
+
+ expr, err := BuildPartitionMatchPredicate(spec,
dynamicOverwriteSchema(), []map[int]any{{1001: nil}})
+ require.NoError(t, err)
+
+ bound, err := iceberg.BindExpr(dynamicOverwriteSchema(), expr, true)
+ require.NoError(t, err)
+ assert.True(t, bound.Equals(iceberg.AlwaysTrue{}), "void-only predicate
should bind to AlwaysTrue, got %s", bound)
+}
+
+func TestBuildPartitionMatchPredicate_SourceLessVoidDoesNotDropOtherFields(t
*testing.T) {
+ schema := dynamicOverwriteSchema()
+ spec := specWithFields(
+ iceberg.PartitionField{
+ SourceIDs: []int{0}, FieldID: 1000, Name:
"old_category", Transform: iceberg.VoidTransform{},
+ },
+ identityField(1, 1001, "id"),
+ )
+
+ expr, err := BuildPartitionMatchPredicate(spec, schema, []map[int]any{
+ {1000: nil, 1001: int32(5)},
+ })
+ require.NoError(t, err)
+
+ want := iceberg.EqualTo(iceberg.Reference("id"), int32(5))
+ assert.True(t, expr.Equals(want), "want %s, got %s", want, expr)
+
+ bound, err := iceberg.BindExpr(schema, expr, true)
+ require.NoError(t, err)
+
+ eval, err := iceberg.ExpressionEvaluator(schema, expr, true)
+ require.NoError(t, err)
+
+ matched, err := eval(partitionPredicateRow{int32(5), nil, nil, nil,
nil})
+ require.NoError(t, err)
+ assert.True(t, matched)
+
+ wantBound, err := iceberg.BindExpr(schema, want, true)
+ require.NoError(t, err)
+ assert.True(t, bound.Equals(wantBound),
+ "bound predicate should retain non-void field, got %s", bound)
+}
+
+func TestBuildPartitionMatchPredicate_CoercesRawDayValue(t *testing.T) {
+ schema := iceberg.NewSchema(0, iceberg.NestedField{
+ ID: 1, Name: "value", Type: iceberg.PrimitiveTypes.Timestamp,
+ })
+ spec := specWithFields(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "value_day",
Transform: iceberg.DayTransform{},
+ })
+ rawDay := int32(dateAtUTC(2020, 2, 29))
+
+ expr, err := BuildPartitionMatchPredicate(spec, schema,
[]map[int]any{{1000: rawDay}})
+ require.NoError(t, err)
+
+ want := iceberg.LiteralPredicate(iceberg.OpEQ,
+ iceberg.NewUnboundTransform(iceberg.DayTransform{},
iceberg.Reference("value")),
+ iceberg.DateLiteral(rawDay),
+ )
+ assert.True(t, expr.Equals(want), "want %s, got %s", want, expr)
+
+ eval, err := iceberg.ExpressionEvaluator(schema, expr, true)
+ require.NoError(t, err)
+
+ matched, err := eval(partitionPredicateRow{timestampAtUTC(2020, 2, 29,
12, 0, 0, 0)})
+ require.NoError(t, err)
+ assert.True(t, matched)
+}
+
+func TestBuildPartitionMatchPredicate_UsesNestedSourcePath(t *testing.T) {
+ schema := iceberg.NewSchema(0,
+ iceberg.NestedField{ID: 10, Name: "location", Type:
&iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 11, Name: "category", Type:
iceberg.PrimitiveTypes.String},
+ },
+ }},
+ iceberg.NestedField{ID: 12, Name: "category", Type:
iceberg.PrimitiveTypes.String},
+ )
+ spec := specWithFields(iceberg.PartitionField{
+ SourceIDs: []int{11}, FieldID: 1000, Name: "category_part",
+ Transform: iceberg.TruncateTransform{Width: 3},
+ })
+
+ expr, err := BuildPartitionMatchPredicate(spec, schema,
[]map[int]any{{1000: "boo"}})
+ require.NoError(t, err)
+ _, err = iceberg.BindExpr(schema, expr, true)
+ require.NoError(t, err)
+
+ want := iceberg.EqualTo(
+ iceberg.NewUnboundTransform(iceberg.TruncateTransform{Width:
3}, iceberg.Reference("location.category")),
+ "boo",
+ )
+ assert.True(t, expr.Equals(want), "want %s, got %s", want, expr)
+}
+
+func TestBuildPartitionMatchPredicate_RecognizesPointerIdentityTransform(t
*testing.T) {
+ spec := specWithFields(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name: "id_part",
+ Transform: &iceberg.IdentityTransform{},
+ })
+
+ expr, err := BuildPartitionMatchPredicate(spec,
dynamicOverwriteSchema(), []map[int]any{{1000: int32(1)}})
+ require.NoError(t, err)
+
+ want := iceberg.EqualTo(iceberg.Reference("id"), int32(1))
+ assert.True(t, expr.Equals(want), "want %s, got %s", want, expr)
+}
+
func TestBuildPartitionMatchPredicate_UnknownSourceID(t *testing.T) {
// Partition field points at a source id that is not present in the
schema.
- spec := identitySpec(identityField(999, 1000, "ghost"))
+ spec := specWithFields(identityField(999, 1000, "ghost"))
_, err := BuildPartitionMatchPredicate(spec, dynamicOverwriteSchema(),
[]map[int]any{{1000: int32(1)}})
require.Error(t, err)
@@ -307,7 +689,7 @@ func TestBuildPartitionMatchPredicate_UnknownSourceID(t
*testing.T) {
}
func TestBuildPartitionMatchPredicate_UnsupportedValueType(t *testing.T) {
- spec := identitySpec(identityField(1, 1000, "id"))
+ spec := specWithFields(identityField(1, 1000, "id"))
_, err := BuildPartitionMatchPredicate(spec, dynamicOverwriteSchema(),
[]map[int]any{
{1000: struct{}{}},
@@ -358,3 +740,65 @@ func TestLiteralForPartitionValue(t *testing.T) {
assert.ErrorIs(t, err, iceberg.ErrInvalidArgument)
})
}
+
+func TestBuildPartitionMatchPredicate_RejectsInvalidTransforms(t *testing.T) {
+ unknown, err := iceberg.ParseTransform("future_transform")
+ require.NoError(t, err)
+ for _, tc := range []struct {
+ name string
+ transform iceberg.Transform
+ want error
+ }{
+ {name: "nil", want: iceberg.ErrInvalidArgument},
+ {name: "typed nil identity", transform:
(*iceberg.IdentityTransform)(nil), want: iceberg.ErrInvalidArgument},
+ {name: "typed nil bucket", transform:
(*iceberg.BucketTransform)(nil), want: iceberg.ErrInvalidArgument},
+ {name: "unknown", transform: unknown, want:
iceberg.ErrNotImplemented},
+ {name: "invalid source type", transform:
iceberg.DayTransform{}, want: iceberg.ErrInvalidArgument},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ spec := specWithFields(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name:
"id_part", Transform: tc.transform,
+ })
+ _, err := BuildPartitionMatchPredicate(spec,
dynamicOverwriteSchema(), []map[int]any{{1000: int32(1)}})
+ require.ErrorIs(t, err, tc.want)
+ })
+ }
+}
+
+func TestBuildPartitionMatchPredicate_RejectsImpossiblePartitionValues(t
*testing.T) {
+ cases := []struct {
+ name string
+ schema *iceberg.Schema
+ transform iceberg.Transform
+ value any
+ }{
+ {
+ name: "bucket value outside range",
+ schema: iceberg.NewSchema(0, iceberg.NestedField{ID:
1, Name: "value", Type: iceberg.PrimitiveTypes.Int32}),
+ transform: iceberg.BucketTransform{NumBuckets: 4},
+ value: int32(4),
+ },
+ {
+ name: "truncate value is not fixed point",
+ schema: iceberg.NewSchema(0, iceberg.NestedField{ID:
1, Name: "value", Type: iceberg.PrimitiveTypes.String}),
+ transform: iceberg.TruncateTransform{Width: 3},
+ value: "books",
+ },
+ {
+ name: "void value is not nil",
+ schema: iceberg.NewSchema(0, iceberg.NestedField{ID:
1, Name: "value", Type: iceberg.PrimitiveTypes.String}),
+ transform: iceberg.VoidTransform{},
+ value: "books",
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ spec := specWithFields(iceberg.PartitionField{
+ SourceIDs: []int{1}, FieldID: 1000, Name:
"value_part", Transform: tc.transform,
+ })
+ _, err := BuildPartitionMatchPredicate(spec, tc.schema,
[]map[int]any{{1000: tc.value}})
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+ })
+ }
+}
diff --git a/table/transaction_internal_test.go
b/table/transaction_internal_test.go
index 7754928c3..ce2d4c2b0 100644
--- a/table/transaction_internal_test.go
+++ b/table/transaction_internal_test.go
@@ -31,6 +31,21 @@ import (
"github.com/stretchr/testify/require"
)
+func TestPrepareBatchFilterRejectsTransformedTerms(t *testing.T) {
+ schema := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "category", Type:
iceberg.PrimitiveTypes.String},
+ )
+ term := iceberg.NewUnboundTransform(
+ iceberg.TruncateTransform{Width: 3},
+ iceberg.Reference("category"),
+ )
+
+ _, err := prepareBatchFilter(iceberg.EqualTo(term, "boo"), schema, true)
+ require.Error(t, err)
+ assert.ErrorIs(t, err, iceberg.ErrNotImplemented)
+ assert.Contains(t, err.Error(), "transformed terms")
+}
+
func TestTransactionApplyKeepsDistinctRequirementsOfSameType(t *testing.T) {
txn := newTransactionWithSnapshotRefs(t)
