(iceberg-go) branch main updated: fix: manifest pruning when partition source columns are dropped (#1191)

[at...@apache.org]

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 ba4f3fa4 fix: manifest pruning when partition source columns are dropped (#1191)
ba4f3fa4 is described below

commit ba4f3fa42acc777fa7c901a219605c2843199c11
Author: Raul-Mircea Crivineanu <gf...@gmail.com>
AuthorDate: Tue Jun 16 19:43:11 2026 +0800

    fix: manifest pruning when partition source columns are dropped (#1191)
    
    ## Summary
    
    Closes https://github.com/apache/iceberg-go/issues/1190.
    
    Fix manifest-level pruning when a partition source column has been
    dropped from the current schema.
    
    Manifest partition summaries are positional and correspond to the full
    partition field list in the manifest's partition spec. Before this
    change, `newManifestEvaluator` used
    `PartitionSpec.PartitionType(schema)`, which drops partition fields
    whose source column is missing from the provided schema. That can shift
    later partition fields to the wrong position.
    
    For example, if the manifest was written with partition fields `[q_part,
    p_part]`, and the current schema has dropped `q`, then `p_part` could be
    evaluated against `q_part`'s summary slot. This can incorrectly prune a
    manifest and return incomplete results.
    
    This change keeps all partition fields when building the manifest
    evaluator schema. Dropped-source fields are represented as `UnknownType`
    placeholders so later fields remain positionally aligned.
    
    This matches Java Iceberg's partition type behavior, where dropped
    partition-source fields are represented with UnknownType in the
    partition type.
    
    ## Tests
    
    Added `TestManifestEvaluatorWithDroppedPartitionSource`, covering:
    
    - partition spec: `identity(q), identity(p)`
    - current schema: `q` dropped, `p` retained
    - row filter: `p = 2`, projected to `p_part`
    - manifest summaries: `q_part` at slot 0, `p_part` at slot 1
    
    Before the fix, the manifest can be incorrectly pruned. After the fix,
    it remains eligible.
    
    ## Validation
    
    - `go test ./table -run TestManifestEvaluatorWithDroppedPartitionSource
    -count=1`
    - `go test ./table`
    - `go test ./...`
    
    ---------
    
    Signed-off-by: Raul <gf...@gmail.com>
---
 partitions.go            |  9 ++++++
 table/evaluators.go      | 27 ++++++++++++++--
 table/evaluators_test.go | 80 ++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 114 insertions(+), 2 deletions(-)

diff --git a/partitions.go b/partitions.go
index 98873b8f..fcb525b0 100644
--- a/partitions.go
+++ b/partitions.go
@@ -491,6 +491,15 @@ func (ps *PartitionSpec) LastAssignedFieldID() int {
 // There is a case where we can guarantee that a partition field in the first
 // and only parittion spec that uses a required source column will never be
 // null, but it doesn't seem worth tracking this case.
+//
+// Note: the result is compacted. A partition field whose source column is not
+// present in schema (for example, the column was dropped) is omitted rather
+// than retained, so the returned struct does NOT positionally correspond to a
+// manifest's partition FieldSummary list, which is ordered by the full
+// partition spec. It must therefore not be used to index positional partition
+// summaries; build a schema that keeps every spec field (with an UnknownType
+// placeholder for dropped sources) for that purpose — see manifestPartitionFields
+// in the table package.
 func (ps *PartitionSpec) PartitionType(schema *Schema) *StructType {
 	nestedFields := []NestedField{}
 	for _, field := range ps.fields {
diff --git a/table/evaluators.go b/table/evaluators.go
index 7d5b0e0b..2cff27fe 100644
--- a/table/evaluators.go
+++ b/table/evaluators.go
@@ -39,8 +39,7 @@ const (
 // manifest file has rows that might or might not match a given partition filter by using
 // the stats provided in the partitions (UpperBound/LowerBound/ContainsNull/ContainsNaN).
 func newManifestEvaluator(spec iceberg.PartitionSpec, schema *iceberg.Schema, partitionFilter iceberg.BooleanExpression, caseSensitive bool) (func(iceberg.ManifestFile) (bool, error), error) {
-	partType := spec.PartitionType(schema)
-	partSchema := iceberg.NewSchema(0, partType.FieldList...)
+	partSchema := iceberg.NewSchema(0, manifestPartitionFields(spec, schema)...)
 	filter, err := iceberg.RewriteNotExpr(partitionFilter)
 	if err != nil {
 		return nil, err
@@ -54,6 +53,30 @@ func newManifestEvaluator(spec iceberg.PartitionSpec, schema *iceberg.Schema, pa
 	return (&manifestEvalVisitor{partitionFilter: boundFilter}).Eval, nil
 }
 
+// manifestPartitionFields builds the partition-struct fields for evaluating
+// manifest FieldSummary lists. Summaries are positional per the spec's full
+// field list, so unlike PartitionSpec.PartitionType (which compacts fields
+// whose source column is missing from the schema), every spec field must be
+// kept. A dropped-source field becomes an Unknown-typed placeholder that no
+// projected predicate can reference, preserving positions of later fields.
+func manifestPartitionFields(spec iceberg.PartitionSpec, schema *iceberg.Schema) []iceberg.NestedField {
+	fields := make([]iceberg.NestedField, 0, spec.NumFields())
+	for _, field := range spec.Fields() {
+		typ := iceberg.Type(iceberg.UnknownType{})
+		if sourceType, ok := schema.FindTypeByID(field.SourceID()); ok {
+			typ = field.Transform.ResultType(sourceType)
+		}
+		fields = append(fields, iceberg.NestedField{
+			ID:       field.FieldID,
+			Name:     field.Name,
+			Type:     typ,
+			Required: false,
+		})
+	}
+
+	return fields
+}
+
 type manifestEvalVisitor struct {
 	partitionFields []iceberg.FieldSummary
 	partitionFilter iceberg.BooleanExpression
diff --git a/table/evaluators_test.go b/table/evaluators_test.go
index b99da0c6..d63359d2 100644
--- a/table/evaluators_test.go
+++ b/table/evaluators_test.go
@@ -715,6 +715,86 @@ func TestManifestEvaluator(t *testing.T) {
 	})
 }
 
+func TestManifestEvaluatorWithDroppedPartitionSource(t *testing.T) {
+	// Manifest partition summaries are positional per the spec's full field
+	// list. When a partition field's source column is dropped from the current
+	// schema, the evaluator must keep that field's slot (as an Unknown-typed
+	// placeholder) rather than compacting it away; otherwise a later, present
+	// field is read against an earlier slot's bounds.
+	bound := func(lo, hi int64) iceberg.FieldSummary {
+		lower, err := iceberg.Int64Literal(lo).MarshalBinary()
+		require.NoError(t, err)
+		upper, err := iceberg.Int64Literal(hi).MarshalBinary()
+		require.NoError(t, err)
+
+		return iceberg.FieldSummary{ContainsNull: false, LowerBound: &lower, UpperBound: &upper}
+	}
+	manifestWith := func(summaries ...iceberg.FieldSummary) iceberg.ManifestFile {
+		return iceberg.NewManifestFile(1, "", 0, 0, 0).Partitions(summaries).Build()
+	}
+	// evalForCol projects (col == 2) onto the spec and returns the manifest
+	// evaluator. In every case below the dropped field's slot contains 2
+	// exactly when the present field's slot does not, so a compacted read of
+	// the wrong slot flips both the match and the prune assertion.
+	evalForCol := func(spec iceberg.PartitionSpec, schema *iceberg.Schema, col string) func(iceberg.ManifestFile) bool {
+		project := newInclusiveProjection(schema, spec, true)
+		partitionFilter, err := project(iceberg.EqualTo(iceberg.Reference(col), int64(2)))
+		require.NoError(t, err)
+		eval, err := newManifestEvaluator(spec, schema, partitionFilter, true)
+		require.NoError(t, err)
+
+		return func(mf iceberg.ManifestFile) bool {
+			res, err := eval(mf)
+			require.NoError(t, err)
+
+			return res
+		}
+	}
+
+	t.Run("dropped source at slot 0", func(t *testing.T) {
+		// q_part (slot 0) sources a dropped column; p_part (slot 1) sources the
+		// present column "p". Compaction would drop q_part and read p_part at
+		// slot 0.
+		spec := iceberg.NewPartitionSpec(
+			iceberg.PartitionField{SourceIDs: []int{3}, FieldID: 1000, Name: "q_part", Transform: iceberg.IdentityTransform{}},
+			iceberg.PartitionField{SourceIDs: []int{2}, FieldID: 1001, Name: "p_part", Transform: iceberg.IdentityTransform{}},
+		)
+		schema := iceberg.NewSchema(1,
+			iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+			iceberg.NestedField{ID: 2, Name: "p", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+		)
+		eval := evalForCol(spec, schema, "p")
+
+		assert.True(t, eval(manifestWith(bound(100, 200), bound(2, 2))),
+			"p=2 is in p_part's slot-1 bounds; must read slot 1, not q_part's slot 0")
+		assert.False(t, eval(manifestWith(bound(0, 5), bound(100, 200))),
+			"p=2 is outside p_part's slot-1 bounds; manifest must be pruned (reading slot 0 would wrongly keep it)")
+	})
+
+	t.Run("dropped sources around a middle field", func(t *testing.T) {
+		// b_part (slot 1) sources the only present column; a_part (slot 0) and
+		// c_part (slot 2) source dropped columns. Dropping the leading a_part
+		// shifts b_part to slot 0 under compaction.
+		spec := iceberg.NewPartitionSpec(
+			iceberg.PartitionField{SourceIDs: []int{3}, FieldID: 1000, Name: "a_part", Transform: iceberg.IdentityTransform{}},
+			iceberg.PartitionField{SourceIDs: []int{2}, FieldID: 1001, Name: "b_part", Transform: iceberg.IdentityTransform{}},
+			iceberg.PartitionField{SourceIDs: []int{4}, FieldID: 1002, Name: "c_part", Transform: iceberg.IdentityTransform{}},
+		)
+		schema := iceberg.NewSchema(1,
+			iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+			iceberg.NestedField{ID: 2, Name: "b", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+		)
+		eval := evalForCol(spec, schema, "b")
+
+		// Dropped slots 0 and 2 carry the opposite bounds to b_part, catching
+		// an off-by-one in either direction.
+		assert.True(t, eval(manifestWith(bound(100, 200), bound(2, 2), bound(100, 200))),
+			"b=2 is in b_part's slot-1 bounds; must read slot 1")
+		assert.False(t, eval(manifestWith(bound(0, 5), bound(100, 200), bound(0, 5))),
+			"b=2 is outside b_part's slot-1 bounds; manifest must be pruned")
+	})
+}
+
 type ProjectionTestSuite struct {
 	suite.Suite
 }