[core][format][spark] Support nested field predicate pushdown

paimon-common/src/main/java/org/apache/paimon/predicate/NestedFieldTransform.java

@@ -0,0 +1,182 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.predicate;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.RowType;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+import static org.apache.paimon.utils.InternalRowUtils.get;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Transform that extracts a field nested inside a row-typed column, for example {@code addr.city}.
+ *
+ * <p>The transform keeps the enclosing top-level column as its only {@link #inputs() input}, so
+ * anything that rewrites field indices (schema projection, for instance) keeps working without
+ * knowing about nesting. The positions below that column are held separately in {@link #path()}.
+ *
+ * <p>Deliberately <b>not</b> a {@link FieldTransform}: {@link LeafPredicate#fieldRefOptional()}
+ * returns empty for it, which is what keeps every consumer that equates a leaf with a top-level
+ * column — min/max pruning, file index lookup, ORC pushdown, schema evolution — from silently
+ * reading the enclosing column's metadata as if it belonged to the nested field. Those consumers
+ * give up on this transform instead, which costs pruning but never rows.
+ */
+public class NestedFieldTransform implements Transform {
+
+    private static final long serialVersionUID = 1L;
+
+    public static final String NAME = "NESTED_FIELD_REF";
+
+    public static final String FIELD_FIELD_REF = "fieldRef";
+    public static final String FIELD_PATH = "path";
+
+    /** The top-level row-typed column the nested field lives in. */
+    private final FieldRef fieldRef;
+
+    /** Positions to descend, relative to {@code fieldRef}'s row type. Never empty. */
+    private final List<Integer> path;
+
+    private final String name;
+    private final DataType outputType;
+
+    @JsonCreator
+    public NestedFieldTransform(
+            @JsonProperty(FIELD_FIELD_REF) FieldRef fieldRef,
+            @JsonProperty(FIELD_PATH) List<Integer> path) {
+        checkArgument(path != null && !path.isEmpty(), "Nested field path must not be empty.");
+        this.fieldRef = fieldRef;
+        this.path = Collections.unmodifiableList(new ArrayList<>(path));
+
+        StringBuilder nameBuilder = new StringBuilder(fieldRef.name());
+        DataType current = fieldRef.type();
+        for (int position : this.path) {
+            checkArgument(
+                    current instanceof RowType,
+                    "Nested field path of '%s' descends into a non-row type %s.",
+                    fieldRef.name(),
+                    current);
+            RowType rowType = (RowType) current;
+            checkArgument(
+                    position >= 0 && position < rowType.getFieldCount(),
+                    "Nested field position %s is out of range for %s.",
+                    position,
+                    rowType);
+            nameBuilder.append('.').append(rowType.getFields().get(position).name());
+            current = rowType.getTypeAt(position);
+        }
+        this.name = nameBuilder.toString();
+        this.outputType = current;
+    }
+
+    @Override
+    public String name() {
+        return NAME;
+    }
+
+    @JsonProperty(FIELD_FIELD_REF)
+    public FieldRef fieldRef() {
+        return fieldRef;
+    }
+
+    @JsonProperty(FIELD_PATH)
+    public List<Integer> path() {
+        return path;
+    }
+
+    /** Dot-separated name from the top-level column down to the nested field, {@code addr.city}. */
+    @JsonIgnore
+    public String fieldName() {
+        return name;
+    }
+
+    @Override
+    @JsonIgnore
+    public List<Object> inputs() {
+        return Collections.singletonList(fieldRef);
+    }
+
+    @Override
+    @JsonIgnore
+    public DataType outputType() {
+        return outputType;
+    }
+
+    /**
+     * Reads the nested field out of {@code row}, which must match the row type {@link #fieldRef}
+     * was built against. A null anywhere along the path yields null, matching SQL semantics for
+     * field access on a null struct.
+     */
+    @Override
+    public Object transform(InternalRow row) {
+        int position = fieldRef.index();
+        if (row.isNullAt(position)) {
+            return null;
+        }
+        RowType currentType = (RowType) fieldRef.type();
+        InternalRow current = row.getRow(position, currentType.getFieldCount());
+
+        for (int i = 0; i < path.size() - 1; i++) {
+            position = path.get(i);
+            if (current.isNullAt(position)) {
+                return null;
+            }
+            RowType nextType = (RowType) currentType.getTypeAt(position);
+            current = current.getRow(position, nextType.getFieldCount());
+            currentType = nextType;
+        }
+
+        int leaf = path.get(path.size() - 1);
+        return get(current, leaf, currentType.getTypeAt(leaf));
+    }
+
+    @Override
+    public Transform copyWithNewInputs(List<Object> inputs) {
+        checkArgument(inputs.size() == 1);
+        return new NestedFieldTransform((FieldRef) inputs.get(0), path);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        NestedFieldTransform that = (NestedFieldTransform) o;
+        return Objects.equals(fieldRef, that.fieldRef) && Objects.equals(path, that.path);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(fieldRef, path);
+    }
+
+    @Override
+    public String toString() {
+        return name;
+    }
+}

paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java

@@ -278,6 +278,10 @@ public Predicate notIn(int idx, List<Object> literals) {
         return in(idx, literals).negate().get();
     }
 
+    public Predicate notIn(Transform transform, List<Object> literals) {
+        return in(transform, literals).negate().get();
+    }
+
     public Predicate between(int idx, Object includedLowerBound, Object includedUpperBound) {
         DataField field = rowType.getFields().get(idx);
         return new LeafPredicate(

paimon-common/src/main/java/org/apache/paimon/predicate/Transform.java

@@ -34,6 +34,7 @@
         property = Transform.FIELD_NAME)
 @JsonSubTypes({
     @JsonSubTypes.Type(value = FieldTransform.class, name = FieldTransform.NAME),
+    @JsonSubTypes.Type(value = NestedFieldTransform.class, name = NestedFieldTransform.NAME),
     @JsonSubTypes.Type(value = CastTransform.class, name = CastTransform.NAME),
     @JsonSubTypes.Type(value = ConcatTransform.class, name = ConcatTransform.NAME),
     @JsonSubTypes.Type(value = ConcatWsTransform.class, name = ConcatWsTransform.NAME),

paimon-common/src/test/java/org/apache/paimon/predicate/NestedFieldTransformTest.java

@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.predicate;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link NestedFieldTransform}. */
+class NestedFieldTransformTest {
+
+    // user STRUCT<id BIGINT, addr STRUCT<city STRING, zip STRING>>
+    private static final RowType ADDR_TYPE =
+            RowType.of(
+                    new org.apache.paimon.types.DataType[] {DataTypes.STRING(), DataTypes.STRING()},
+                    new String[] {"city", "zip"});
+    private static final RowType USER_TYPE =
+            RowType.of(
+                    new org.apache.paimon.types.DataType[] {DataTypes.BIGINT(), ADDR_TYPE},
+                    new String[] {"id", "addr"});
+    private static final RowType ROW_TYPE =
+            RowType.of(
+                    new org.apache.paimon.types.DataType[] {DataTypes.INT(), USER_TYPE},
+                    new String[] {"pk", "user"});
+
+    private static final FieldRef USER_REF = new FieldRef(1, "user", USER_TYPE);
+
+    private static GenericRow row(Object user) {
+        return GenericRow.of(1, user);
+    }
+
+    @Test
+    public void testReadOneLevel() {
+        NestedFieldTransform transform =
+                new NestedFieldTransform(USER_REF, Collections.singletonList(0));
+
+        assertThat(transform.fieldName()).isEqualTo("user.id");
+        assertThat(transform.outputType()).isEqualTo(DataTypes.BIGINT());
+        assertThat(transform.transform(row(GenericRow.of(42L, null)))).isEqualTo(42L);
+    }
+
+    @Test
+    public void testReadTwoLevels() {
+        NestedFieldTransform transform = new NestedFieldTransform(USER_REF, Arrays.asList(1, 0));
+
+        assertThat(transform.fieldName()).isEqualTo("user.addr.city");
+        assertThat(transform.outputType()).isEqualTo(DataTypes.STRING());
+
+        GenericRow addr =
+                GenericRow.of(
+                        BinaryString.fromString("Beijing"), BinaryString.fromString("100080"));
+        assertThat(transform.transform(row(GenericRow.of(42L, addr))))
+                .isEqualTo(BinaryString.fromString("Beijing"));
+    }
+
+    @Test
+    public void testNullAnywhereOnThePathYieldsNull() {
+        NestedFieldTransform transform = new NestedFieldTransform(USER_REF, Arrays.asList(1, 0));
+
+        // the top-level column is null
+        assertThat(transform.transform(row(null))).isNull();
+        // an intermediate struct is null
+        assertThat(transform.transform(row(GenericRow.of(42L, null)))).isNull();
+        // the leaf itself is null
+        assertThat(transform.transform(row(GenericRow.of(42L, GenericRow.of(null, null)))))
+                .isNull();
+    }
+
+    @Test
+    public void testPredicateOnNullEvaluatesFalse() {
+        PredicateBuilder builder = new PredicateBuilder(ROW_TYPE);
+        Predicate predicate =
+                builder.equal(
+                        new NestedFieldTransform(USER_REF, Arrays.asList(1, 0)),
+                        BinaryString.fromString("Beijing"));
+
+        assertThat(predicate.test(row(null))).isFalse();
+        assertThat(predicate.test(row(GenericRow.of(42L, null)))).isFalse();
+    }
+
+    /**
+     * The whole safety story rests on this: nothing that equates a leaf with a top-level column can
+     * mistake a nested field for one, because it never gets a {@link FieldRef} back.
+     */
+    @Test
+    public void testNoFieldRefIsExposed() {
+        LeafPredicate predicate =
+                (LeafPredicate)
+                        new PredicateBuilder(ROW_TYPE)
+                                .equal(
+                                        new NestedFieldTransform(
+                                                USER_REF, Collections.singletonList(0)),
+                                        42L);
+
+        assertThat(predicate.fieldRefOptional()).isEmpty();
+        // the enclosing column is what schema-level rewrites see
+        assertThat(predicate.fieldNames()).containsExactly("user");
+    }
+
+    /** Min/max of the enclosing column say nothing about the nested field, so nothing is pruned. */
+    @Test
+    public void testStatsNeverPrune() {
+        Predicate predicate =
+                new PredicateBuilder(ROW_TYPE)
+                        .equal(
+                                new NestedFieldTransform(USER_REF, Collections.singletonList(0)),
+                                42L);
+
+        assertThat(
+                        predicate.test(
+                                100L,
+                                GenericRow.of(1, null),
+                                GenericRow.of(10, null),
+                                new GenericArray(new Object[] {0L, 0L})))
+                .isTrue();
+    }
+
+    @Test
+    public void testProjectionKeepsThePath() {
+        Predicate predicate =
+                new PredicateBuilder(ROW_TYPE)
+                        .equal(new NestedFieldTransform(USER_REF, Arrays.asList(1, 0)), 42L);
+
+        // "user" moves from index 1 to index 0
+        Optional<Predicate> projected =
+                predicate.visit(PredicateProjectionConverter.fromProjection(new int[] {1}));
+
+        assertThat(projected).isPresent();
+        NestedFieldTransform transform =
+                (NestedFieldTransform) ((LeafPredicate) projected.get()).transform();
+        assertThat(transform.fieldRef().index()).isEqualTo(0);
+        assertThat(transform.path()).containsExactly(1, 0);
+        assertThat(transform.fieldName()).isEqualTo("user.addr.city");
+    }
+
+    @Test
+    public void testJsonRoundTrip() {
+        Predicate predicate =
+                new PredicateBuilder(ROW_TYPE)
+                        .equal(
+                                new NestedFieldTransform(USER_REF, Arrays.asList(1, 0)),
+                                BinaryString.fromString("Beijing"));
+
+        String json = JsonSerdeUtil.toJson(predicate);
+        assertThat(JsonSerdeUtil.fromJson(json, Predicate.class)).isEqualTo(predicate);
+    }
+
+    @Test
+    public void testRejectsPathThroughNonRowType() {
+        FieldRef arrayRef = new FieldRef(0, "tags", DataTypes.ARRAY(DataTypes.STRING()));
+        assertThatThrownBy(() -> new NestedFieldTransform(arrayRef, Collections.singletonList(0)))
+                .isInstanceOf(IllegalArgumentException.class);
+
+        assertThatThrownBy(() -> new NestedFieldTransform(USER_REF, Collections.emptyList()))
+                .isInstanceOf(IllegalArgumentException.class);
+        assertThatThrownBy(() -> new NestedFieldTransform(USER_REF, Collections.singletonList(9)))
+                .isInstanceOf(IllegalArgumentException.class);
+    }
+}