[spark] Support nested fields in SparkFilterConverter

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

@@ -23,7 +23,10 @@
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
+import javax.annotation.Nullable;
+
 import java.io.Serializable;
+import java.util.Arrays;
 import java.util.Objects;
 
 /** A reference to a field in an input. */
@@ -34,19 +37,31 @@ public class FieldRef implements Serializable {
     private static final String FIELD_INDEX = "index";
     private static final String FIELD_NAME = "name";
     private static final String FIELD_TYPE = "type";
+    private static final String FIELD_NESTED_INDEXES = "nestedIndexes";
+    private static final String FIELD_NESTED_ARITIES = "nestedArities";
 
     private final int index;
     private final String name;
     private final DataType type;
+    @Nullable private final int[] nestedIndexes;
+    @Nullable private final int[] nestedArities;
+
+    public FieldRef(int index, String name, DataType type) {
+        this(index, name, type, null, null);
+    }
 
     @JsonCreator
     public FieldRef(
             @JsonProperty(FIELD_INDEX) int index,
             @JsonProperty(FIELD_NAME) String name,
-            @JsonProperty(FIELD_TYPE) DataType type) {
+            @JsonProperty(FIELD_TYPE) DataType type,
+            @JsonProperty(FIELD_NESTED_INDEXES) @Nullable int[] nestedIndexes,
+            @JsonProperty(FIELD_NESTED_ARITIES) @Nullable int[] nestedArities) {
         this.index = index;
         this.name = name;
         this.type = type;
+        this.nestedIndexes = nestedIndexes;
+        this.nestedArities = nestedArities;
     }
 
     @JsonProperty(FIELD_INDEX)
@@ -64,6 +79,38 @@ public DataType type() {
         return type;
     }
 
+    @JsonProperty(FIELD_NESTED_INDEXES)
+    @Nullable
+    public int[] nestedIndexes() {
+        return nestedIndexes;
+    }
+
+    @JsonProperty(FIELD_NESTED_ARITIES)
+    @Nullable
+    public int[] nestedArities() {
+        return nestedArities;
+    }
+
+    /**
+     * Returns a copy of this ref with the given top-level index, preserving name, type and nested
+     * path metadata.
+     */
+    public FieldRef withIndex(int newIndex) {
+        return new FieldRef(newIndex, name, type, nestedIndexes, nestedArities);
+    }
+
+    /**
+     * Returns the name of the top-level field this ref points at: {@link #name()} for a top-level
+     * ref, the first path segment for a nested ref (whose name is the full dotted path).
+     */
+    public String topLevelName() {
+        if (nestedIndexes == null || nestedIndexes.length == 0) {
+            return name;
+        }
+        int firstDot = name.indexOf('.');
+        return firstDot < 0 ? name : name.substring(0, firstDot);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -75,12 +122,17 @@ public boolean equals(Object o) {
         FieldRef fieldRef = (FieldRef) o;
         return index == fieldRef.index
                 && Objects.equals(name, fieldRef.name)
-                && Objects.equals(type, fieldRef.type);
+                && Objects.equals(type, fieldRef.type)
+                && Arrays.equals(nestedIndexes, fieldRef.nestedIndexes)
+                && Arrays.equals(nestedArities, fieldRef.nestedArities);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(index, name, type);
+        int result = Objects.hash(index, name, type);
+        result = 31 * result + Arrays.hashCode(nestedIndexes);
+        result = 31 * result + Arrays.hashCode(nestedArities);
+        return result;
     }
 
     @Override

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

@@ -71,7 +71,34 @@ public DataType outputType() {
 
     @Override
     public Object transform(InternalRow row) {
-        return get(row, fieldRef.index(), fieldRef.type());
+        if (fieldRef.nestedIndexes() == null || fieldRef.nestedIndexes().length == 0) {
+            return get(row, fieldRef.index(), fieldRef.type());
+        }
+
+        InternalRow currentRow = row;
+        if (currentRow == null) {
+            return null;
+        }
+        int[] indexes = fieldRef.nestedIndexes();
+        int[] arities = fieldRef.nestedArities();
+
+        if (currentRow.isNullAt(fieldRef.index())) {
+            return null;
+        }
+        currentRow = currentRow.getRow(fieldRef.index(), arities[0]);
+
+        for (int i = 0; i < indexes.length - 1; i++) {
+            if (currentRow == null || currentRow.isNullAt(indexes[i])) {
+                return null;
+            }
+            currentRow = currentRow.getRow(indexes[i], arities[i + 1]);
+        }
+
+        if (currentRow == null) {
+            return null;
+        }
+        int leafIndex = indexes[indexes.length - 1];
+        return get(currentRow, leafIndex, fieldRef.type());
     }
 
     @Override

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

@@ -146,6 +146,9 @@ public boolean test(
             return !(function instanceof AlwaysFalse);
         }
         FieldRef fieldRef = fieldRefOptional.get();
+        if (fieldRef.nestedIndexes() != null && fieldRef.nestedIndexes().length > 0) {
+            return true;
+        }
         int index = fieldRef.index();
         DataType type = fieldRef.type();
 

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

@@ -77,7 +77,7 @@ public Optional<Predicate> visit(LeafPredicate predicate) {
             if (input instanceof FieldRef) {
                 FieldRef ref = (FieldRef) input;
                 int partIdx = tableToPartitionMapping[ref.index()];
-                remappedInputs.add(new FieldRef(partIdx, ref.name(), ref.type()));
+                remappedInputs.add(ref.withIndex(partIdx));
             } else {
                 remappedInputs.add(input);
             }

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

@@ -221,7 +221,7 @@ public Predicate in(int idx, List<Object> literals) {
     public Predicate in(Transform transform, List<Object> literals) {
         // In the IN predicate, 20 literals are critical for performance.
         // If there are more than 20 literals, the performance will decrease.
-        if (literals.size() > 20) {
+        if (literals.size() > 20 || literals.isEmpty()) {
             return LeafPredicate.of(transform, In.INSTANCE, literals);
         }
 

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

@@ -92,7 +92,7 @@ public Optional<Predicate> visit(LeafPredicate predicate) {
                 FieldRef fieldRef = (FieldRef) input;
                 Integer mappedIndex = reversed.get(fieldRef.index());
                 if (mappedIndex != null) {
-                    newInputs.add(new FieldRef(mappedIndex, fieldRef.name(), fieldRef.type()));
+                    newInputs.add(fieldRef.withIndex(mappedIndex));
                 } else {
                     return Optional.empty();
                 }

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

@@ -0,0 +1,53 @@
+/*
+ * 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.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link FieldRef}. */
+class FieldRefTest {
+
+    @Test
+    void testTopLevelName() {
+        FieldRef topLevel = new FieldRef(0, "a.b", DataTypes.INT());
+        assertThat(topLevel.topLevelName()).isEqualTo("a.b");
+
+        FieldRef nested = new FieldRef(0, "a.b", DataTypes.INT(), new int[] {1}, new int[] {2});
+        assertThat(nested.topLevelName()).isEqualTo("a");
+
+        FieldRef deepNested =
+                new FieldRef(0, "a.b.c", DataTypes.INT(), new int[] {1, 0}, new int[] {2, 3});
+        assertThat(deepNested.topLevelName()).isEqualTo("a");
+    }
+
+    @Test
+    void testWithIndexPreservesNestedMetadata() {
+        FieldRef nested = new FieldRef(3, "a.b", DataTypes.INT(), new int[] {1}, new int[] {2});
+        FieldRef remapped = nested.withIndex(0);
+        assertThat(remapped.index()).isEqualTo(0);
+        assertThat(remapped.name()).isEqualTo("a.b");
+        assertThat(remapped.type()).isEqualTo(DataTypes.INT());
+        assertThat(remapped.nestedIndexes()).containsExactly(1);
+        assertThat(remapped.nestedArities()).containsExactly(2);
+    }
+}

paimon-core/src/main/java/org/apache/paimon/catalog/TableQueryAuthResult.java

@@ -187,16 +187,15 @@ private static Map<Integer, Transform> transformRemapping(
             for (Object input : transform.inputs()) {
                 if (input instanceof FieldRef) {
                     FieldRef ref = (FieldRef) input;
-                    int newIndex = outputRowType.getFieldIndex(ref.name());
+                    int newIndex = outputRowType.getFieldIndex(ref.topLevelName());
                     if (newIndex < 0) {
                         throw new IllegalArgumentException(
                                 "Column masking refers to field '"
                                         + ref.name()
                                         + "' which is not present in output row type "
                                         + outputRowType);
                     }
-                    DataType type = outputRowType.getTypeAt(newIndex);
-                    newInputs.add(new FieldRef(newIndex, ref.name(), type));
+                    newInputs.add(ref.withIndex(newIndex));
                 } else {
                     newInputs.add(input);
                 }
@@ -221,16 +220,14 @@ public Predicate visit(LeafPredicate predicate) {
             for (Object input : transform.inputs()) {
                 if (input instanceof FieldRef) {
                     FieldRef ref = (FieldRef) input;
-                    String fieldName = ref.name();
-                    int newIndex = outputRowType.getFieldIndex(fieldName);
+                    int newIndex = outputRowType.getFieldIndex(ref.topLevelName());
                     if (newIndex < 0) {
                         throw new RuntimeException(
                                 String.format(
                                         "Unable to read data without column %s when row filter enabled.",
-                                        fieldName));
+                                        ref.name()));
                     }
-                    DataType type = outputRowType.getTypeAt(newIndex);
-                    newInputs.add(new FieldRef(newIndex, fieldName, type));
+                    newInputs.add(ref.withIndex(newIndex));
                 } else {
                     newInputs.add(input);
                 }

paimon-core/src/test/java/org/apache/paimon/catalog/TableQueryAuthResultTest.java

@@ -0,0 +1,101 @@
+/*
+ * 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.catalog;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.predicate.Equal;
+import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.predicate.FieldTransform;
+import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.Transform;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.IteratorRecordReader;
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link TableQueryAuthResult}. */
+class TableQueryAuthResultTest {
+
+    private static final RowType NESTED_TYPE =
+            RowType.of(
+                    new DataType[] {DataTypes.INT(), DataTypes.STRING()}, new String[] {"b", "c"});
+
+    /** Output row type with top-level fields reordered relative to the table schema (id, s). */
+    private static final RowType OUTPUT_TYPE =
+            RowType.of(new DataType[] {NESTED_TYPE, DataTypes.INT()}, new String[] {"s", "id"});
+
+    private static Transform nestedFieldTransform() {
+        return new FieldTransform(
+                new FieldRef(1, "s.b", DataTypes.INT(), new int[] {0}, new int[] {2}));
+    }
+
+    @Test
+    void testNestedRowFilterRemappedByTopLevelField() throws Exception {
+        Predicate predicate =
+                LeafPredicate.of(
+                        nestedFieldTransform(), Equal.INSTANCE, Collections.singletonList(10));
+        TableQueryAuthResult authResult =
+                new TableQueryAuthResult(
+                        Collections.singletonList(JsonSerdeUtil.toJson(predicate)), null);
+
+        List<InternalRow> rows =
+                Arrays.asList(
+                        GenericRow.of(GenericRow.of(10, BinaryString.fromString("x")), 1),
+                        GenericRow.of(GenericRow.of(20, BinaryString.fromString("y")), 2));
+
+        List<Integer> ids = new ArrayList<>();
+        authResult
+                .doAuth(new IteratorRecordReader<>(rows.iterator()), OUTPUT_TYPE)
+                .forEachRemaining(row -> ids.add(row.getInt(1)));
+
+        assertThat(ids).containsExactly(1);
+    }
+
+    @Test
+    void testNestedColumnMaskingRemappedByTopLevelField() throws Exception {
+        Map<String, String> masking =
+                Collections.singletonMap("id", JsonSerdeUtil.toJson(nestedFieldTransform()));
+        TableQueryAuthResult authResult = new TableQueryAuthResult(null, masking);
+
+        List<InternalRow> rows =
+                Collections.singletonList(
+                        GenericRow.of(GenericRow.of(42, BinaryString.fromString("x")), 1));
+
+        List<Integer> ids = new ArrayList<>();
+        authResult
+                .doAuth(new IteratorRecordReader<>(rows.iterator()), OUTPUT_TYPE)
+                .forEachRemaining(row -> ids.add(row.getInt(1)));
+
+        assertThat(ids).containsExactly(42);
+    }
+}