feat(table): variant predicate pushdown and reject variant partition source

exprs.go

@@ -1278,7 +1278,7 @@ type BoundBBoxPredicate interface {
 // as a generic unary predicate (it would reach substrait and error, or be
 // dropped to AlwaysFalse when a column is absent). It has no record-filter or
 // REST-JSON form at all, so the two visitors that would otherwise rebuild a bound
-// predicate - columnNameTranslator.VisitBound and sanitizeVisitor.VisitBound -
+// predicate - scanTranslator.VisitBound and sanitizeVisitor.VisitBound -
 // special-case *boundBBoxPredicate and collapse it to AlwaysTrue. Data-file
 // pruning is done separately by inclusiveMetricsEval.
 type boundBBoxPredicate struct {

partitions.go

@@ -263,6 +263,9 @@ func (p *PartitionSpec) addSpecFieldInternal(targetName string, field NestedFiel
 	if err := validateTransform(transform); err != nil {
 		return err
 	}
+	if _, ok := field.Type.(VariantType); ok {
+		return fmt.Errorf("%w: cannot partition by %s source field: %s", ErrInvalidArgument, field.Type, targetName)
+	}
 	for _, existingField := range p.fields {
 		if existingField.Name == targetName {
 			return errors.New("duplicate partition name: " + targetName)

partitions_test.go

@@ -202,6 +202,18 @@ func TestPartitionSpecRejectsInvalidBucketTransform(t *testing.T) {
 	require.ErrorContains(t, err, "numBuckets > 0")
 }
 
+func TestPartitionSpecRejectsVariantSource(t *testing.T) {
+	schema := iceberg.NewSchema(0,
+		iceberg.NestedField{ID: 1, Name: "v", Type: iceberg.VariantType{}},
+	)
+
+	_, err := iceberg.NewPartitionSpecOpts(
+		iceberg.AddPartitionFieldBySourceID(1, "v_p", iceberg.IdentityTransform{}, schema, nil),
+	)
+	require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+	require.ErrorContains(t, err, "cannot partition by")
+}
+
 func TestPartitionSpecRejectsNegativeSpecID(t *testing.T) {
 	_, err := iceberg.NewPartitionSpecOpts(iceberg.WithSpecID(-1))
 

table/arrow_scanner.go

@@ -734,7 +734,7 @@ func (as *arrowScan) getRecordFilter(ctx context.Context, fileSchema *iceberg.Sc
 		return nil, false, nil
 	}
 
-	translatedFilter, err := iceberg.TranslateColumnNames(as.boundRowFilter, fileSchema)
+	translatedFilter, extracts, err := iceberg.TranslateColumnNamesForScan(as.boundRowFilter, fileSchema)
 	if err != nil {
 		return nil, false, err
 	}
@@ -743,23 +743,35 @@ func (as *arrowScan) getRecordFilter(ctx context.Context, fileSchema *iceberg.Sc
 		return nil, true, nil
 	}
 
-	translatedFilter, err = iceberg.BindExpr(fileSchema, translatedFilter, as.caseSensitive)
+	filterSchema := fileSchema
+	if len(extracts) > 0 {
+		filterSchema, err = augmentSchemaWithExtracts(fileSchema, extracts)
+		if err != nil {
+			return nil, false, err
+		}
+	}
+
+	translatedFilter, err = iceberg.BindExpr(filterSchema, translatedFilter, as.caseSensitive)
 	if err != nil {
 		return nil, false, err
 	}
 
-	if !translatedFilter.Equals(iceberg.AlwaysTrue{}) {
-		extSet, recordFilter, err := substrait.ConvertExpr(fileSchema, translatedFilter, as.caseSensitive)
-		if err != nil {
-			return nil, false, err
-		}
+	if translatedFilter.Equals(iceberg.AlwaysTrue{}) {
+		return nil, false, nil
+	}
 
-		ctx = exprs.WithExtensionIDSet(ctx, exprs.NewExtensionSetDefault(*extSet))
+	extSet, recordFilter, err := substrait.ConvertExpr(filterSchema, translatedFilter, as.caseSensitive)
+	if err != nil {
+		return nil, false, err
+	}
 
-		return filterRecords(ctx, recordFilter), false, nil
+	ctx = exprs.WithExtensionIDSet(ctx, exprs.NewExtensionSetDefault(*extSet))
+	base := filterRecords(ctx, recordFilter)
+	if len(extracts) == 0 {
+		return base, false, nil
 	}
 
-	return nil, false, nil
+	return as.extractResidualFilter(ctx, extracts, base), false, nil
 }
 
 // fieldIndexByID returns the index of the field carrying fieldID in its Arrow