diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/FieldRef.java b/paimon-common/src/main/java/org/apache/paimon/predicate/FieldRef.java index 92a489d00eb8..2258dfbd47d3 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/FieldRef.java +++ b/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 diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/FieldTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/FieldTransform.java index 5136dedec530..27947188ef1e 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/FieldTransform.java +++ b/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 diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicate.java b/paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicate.java index f68771bb7ef5..9c2c2eef2709 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicate.java +++ b/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(); diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/PartitionValuePredicateVisitor.java b/paimon-common/src/main/java/org/apache/paimon/predicate/PartitionValuePredicateVisitor.java index 943b751870e9..a97e727a6ad1 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/PartitionValuePredicateVisitor.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/PartitionValuePredicateVisitor.java @@ -77,7 +77,7 @@ public Optional 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); } diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java index 05acce17290d..2cfae7d78337 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java @@ -221,7 +221,7 @@ public Predicate in(int idx, List literals) { public Predicate in(Transform transform, List 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); } diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateProjectionConverter.java b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateProjectionConverter.java index 78d7f2e117ec..62b972c70415 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateProjectionConverter.java +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateProjectionConverter.java @@ -92,7 +92,7 @@ public Optional 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(); } diff --git a/paimon-common/src/test/java/org/apache/paimon/predicate/FieldRefTest.java b/paimon-common/src/test/java/org/apache/paimon/predicate/FieldRefTest.java new file mode 100644 index 000000000000..ada3e466a923 --- /dev/null +++ b/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); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/TableQueryAuthResult.java b/paimon-core/src/main/java/org/apache/paimon/catalog/TableQueryAuthResult.java index b0bb9b1d520d..24f8878d13b2 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/TableQueryAuthResult.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/TableQueryAuthResult.java @@ -187,7 +187,7 @@ private static Map 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 '" @@ -195,8 +195,7 @@ private static Map transformRemapping( + "' 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); } diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/TableQueryAuthResultTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/TableQueryAuthResultTest.java new file mode 100644 index 000000000000..9bdd81129f79 --- /dev/null +++ b/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 rows = + Arrays.asList( + GenericRow.of(GenericRow.of(10, BinaryString.fromString("x")), 1), + GenericRow.of(GenericRow.of(20, BinaryString.fromString("y")), 2)); + + List 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 masking = + Collections.singletonMap("id", JsonSerdeUtil.toJson(nestedFieldTransform())); + TableQueryAuthResult authResult = new TableQueryAuthResult(null, masking); + + List rows = + Collections.singletonList( + GenericRow.of(GenericRow.of(42, BinaryString.fromString("x")), 1)); + + List ids = new ArrayList<>(); + authResult + .doAuth(new IteratorRecordReader<>(rows.iterator()), OUTPUT_TYPE) + .forEachRemaining(row -> ids.add(row.getInt(1))); + + assertThat(ids).containsExactly(42); + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkFilterConverter.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkFilterConverter.java index 31e4c8aec92a..e7263b865b88 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkFilterConverter.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkFilterConverter.java @@ -18,8 +18,11 @@ package org.apache.paimon.spark; +import org.apache.paimon.predicate.FieldRef; +import org.apache.paimon.predicate.FieldTransform; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.predicate.Transform; import org.apache.paimon.types.DataType; import org.apache.paimon.types.RowType; @@ -107,53 +110,55 @@ public Predicate convert(Filter filter) { return PredicateBuilder.alwaysFalse(); } else if (filter instanceof EqualTo) { EqualTo eq = (EqualTo) filter; - int index = fieldIndex(eq.attribute()); + FieldInfo fieldInfo = resolveField(eq.attribute()); if (isNaN(eq.value())) { - return builder.isNaN(index); + return builder.isNaN(fieldInfo.transform()); } - Object literal = convertLiteral(index, eq.value()); - return builder.equal(index, literal); + Object literal = convertLiteral(fieldInfo.type(), eq.value()); + return builder.equal(fieldInfo.transform(), literal); } else if (filter instanceof EqualNullSafe) { EqualNullSafe eq = (EqualNullSafe) filter; - int index = fieldIndex(eq.attribute()); + FieldInfo fieldInfo = resolveField(eq.attribute()); if (eq.value() == null) { - return builder.isNull(index); + return builder.isNull(fieldInfo.transform()); } else { - Object literal = convertLiteral(index, eq.value()); - return builder.equal(index, literal); + Object literal = convertLiteral(fieldInfo.type(), eq.value()); + return builder.equal(fieldInfo.transform(), literal); } } else if (filter instanceof GreaterThan) { GreaterThan gt = (GreaterThan) filter; - int index = fieldIndex(gt.attribute()); - Object literal = convertLiteral(index, gt.value()); - return builder.greaterThan(index, literal); + FieldInfo fieldInfo = resolveField(gt.attribute()); + Object literal = convertLiteral(fieldInfo.type(), gt.value()); + return builder.greaterThan(fieldInfo.transform(), literal); } else if (filter instanceof GreaterThanOrEqual) { GreaterThanOrEqual gt = (GreaterThanOrEqual) filter; - int index = fieldIndex(gt.attribute()); - Object literal = convertLiteral(index, gt.value()); - return builder.greaterOrEqual(index, literal); + FieldInfo fieldInfo = resolveField(gt.attribute()); + Object literal = convertLiteral(fieldInfo.type(), gt.value()); + return builder.greaterOrEqual(fieldInfo.transform(), literal); } else if (filter instanceof LessThan) { LessThan lt = (LessThan) filter; - int index = fieldIndex(lt.attribute()); - Object literal = convertLiteral(index, lt.value()); - return builder.lessThan(index, literal); + FieldInfo fieldInfo = resolveField(lt.attribute()); + Object literal = convertLiteral(fieldInfo.type(), lt.value()); + return builder.lessThan(fieldInfo.transform(), literal); } else if (filter instanceof LessThanOrEqual) { LessThanOrEqual lt = (LessThanOrEqual) filter; - int index = fieldIndex(lt.attribute()); - Object literal = convertLiteral(index, lt.value()); - return builder.lessOrEqual(index, literal); + FieldInfo fieldInfo = resolveField(lt.attribute()); + Object literal = convertLiteral(fieldInfo.type(), lt.value()); + return builder.lessOrEqual(fieldInfo.transform(), literal); } else if (filter instanceof In) { In in = (In) filter; - int index = fieldIndex(in.attribute()); + FieldInfo fieldInfo = resolveField(in.attribute()); return builder.in( - index, + fieldInfo.transform(), Arrays.stream(in.values()) - .map(v -> convertLiteral(index, v)) + .map(v -> convertLiteral(fieldInfo.type(), v)) .collect(Collectors.toList())); } else if (filter instanceof IsNull) { - return builder.isNull(fieldIndex(((IsNull) filter).attribute())); + FieldInfo fieldInfo = resolveField(((IsNull) filter).attribute()); + return builder.isNull(fieldInfo.transform()); } else if (filter instanceof IsNotNull) { - return builder.isNotNull(fieldIndex(((IsNotNull) filter).attribute())); + FieldInfo fieldInfo = resolveField(((IsNotNull) filter).attribute()); + return builder.isNotNull(fieldInfo.transform()); } else if (filter instanceof And) { And and = (And) filter; return PredicateBuilder.and(convert(and.left()), convert(and.right())); @@ -168,19 +173,19 @@ public Predicate convert(Filter filter) { } } else if (filter instanceof StringStartsWith) { StringStartsWith startsWith = (StringStartsWith) filter; - int index = fieldIndex(startsWith.attribute()); - Object literal = convertLiteral(index, startsWith.value()); - return builder.startsWith(index, literal); + FieldInfo fieldInfo = resolveField(startsWith.attribute()); + Object literal = convertLiteral(fieldInfo.type(), startsWith.value()); + return builder.startsWith(fieldInfo.transform(), literal); } else if (filter instanceof StringEndsWith) { StringEndsWith endsWith = (StringEndsWith) filter; - int index = fieldIndex(endsWith.attribute()); - Object literal = convertLiteral(index, endsWith.value()); - return builder.endsWith(index, literal); + FieldInfo fieldInfo = resolveField(endsWith.attribute()); + Object literal = convertLiteral(fieldInfo.type(), endsWith.value()); + return builder.endsWith(fieldInfo.transform(), literal); } else if (filter instanceof StringContains) { StringContains contains = (StringContains) filter; - int index = fieldIndex(contains.attribute()); - Object literal = convertLiteral(index, contains.value()); - return builder.contains(index, literal); + FieldInfo fieldInfo = resolveField(contains.attribute()); + Object literal = convertLiteral(fieldInfo.type(), contains.value()); + return builder.contains(fieldInfo.transform(), literal); } throw new UnsupportedOperationException( @@ -198,7 +203,8 @@ private static boolean isNaN(Object value) { } public Object convertLiteral(String field, Object value) { - return convertLiteral(fieldIndex(field), value); + FieldInfo fieldInfo = resolveField(field); + return convertLiteral(fieldInfo.type(), value); } public String convertString(String field, Object value) { @@ -206,18 +212,89 @@ public String convertString(String field, Object value) { return literal == null ? null : literal.toString(); } - private int fieldIndex(String field) { - int index = rowType.getFieldIndex(field); - // TODO: support nested field - if (index == -1) { + private static class FieldInfo { + private final Transform transform; + private final DataType type; + + public FieldInfo(Transform transform, DataType type) { + this.transform = transform; + this.type = type; + } + + public Transform transform() { + return transform; + } + + public DataType type() { + return type; + } + } + + private FieldInfo resolveField(String field) { + // Paimon schema does not forbid top-level column names containing '.', so always try an + // exact top-level lookup first and only fall back to nested path resolution when no such + // field exists. + int exactIndex = rowType.getFieldIndex(field); + if (exactIndex != -1) { + DataType fieldType = rowType.getTypeAt(exactIndex); + Transform transform = new FieldTransform(new FieldRef(exactIndex, field, fieldType)); + return new FieldInfo(transform, fieldType); + } + + String[] parts = field.split("\\."); + if (parts.length == 1) { + throw new UnsupportedOperationException( + String.format("Field '%s' is not found in table schema.", field)); + } + + int topLevelIndex = rowType.getFieldIndex(parts[0]); + if (topLevelIndex == -1) { + throw new UnsupportedOperationException( + String.format("Field '%s' is not found in table schema.", field)); + } + + DataType fieldType = getNestedFieldType(rowType, parts); + if (fieldType == null) { throw new UnsupportedOperationException( String.format("Nested field '%s' is unsupported.", field)); } - return index; + int[] nestedIndexes = new int[parts.length - 1]; + int[] nestedArities = new int[parts.length - 1]; + DataType currentType = rowType.getTypeAt(topLevelIndex); + for (int i = 0; i < parts.length - 1; i++) { + RowType currentSelection = (RowType) currentType; + nestedArities[i] = currentSelection.getFieldCount(); + String nextPart = parts[i + 1]; + int nextIndex = currentSelection.getFieldIndex(nextPart); + nestedIndexes[i] = nextIndex; + currentType = currentSelection.getTypeAt(nextIndex); + } + + Transform transform = + new FieldTransform( + new FieldRef( + topLevelIndex, field, fieldType, nestedIndexes, nestedArities)); + return new FieldInfo(transform, fieldType); + } + + private DataType getNestedFieldType(RowType rowType, String[] path) { + DataType currentType = rowType; + for (String part : path) { + if (currentType instanceof RowType) { + RowType currentSelection = (RowType) currentType; + int idx = currentSelection.getFieldIndex(part); + if (idx == -1) { + return null; + } + currentType = currentSelection.getTypeAt(idx); + } else { + return null; + } + } + return currentType; } - private Object convertLiteral(int index, Object value) { - DataType type = rowType.getTypeAt(index); + private Object convertLiteral(DataType type, Object value) { return convertJavaObject(type, value); } } diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkFilterConverterTest.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkFilterConverterTest.java index 8b5457c9dff6..b3ed4ed5cda7 100644 --- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkFilterConverterTest.java +++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkFilterConverterTest.java @@ -277,4 +277,44 @@ public void testIgnoreFailure() { assertThat(converter.convert(not, true)).isNull(); assertThat(converter.convertIgnoreFailure(not)).isNull(); } + + @Test + public void testNestedField() { + RowType nestedType = + new RowType( + Arrays.asList( + new DataField(0, "b", new IntType()), + new DataField(1, "c", new VarCharType(10)))); + RowType rowType = new RowType(Arrays.asList(new DataField(0, "a", nestedType))); + SparkFilterConverter converter = new SparkFilterConverter(rowType); + + EqualTo eq = EqualTo.apply("a.b", 1); + Predicate actual = converter.convert(eq); + assertThat(actual.toString()).isEqualTo("Equal(a.b, 1)"); + + IsNull isNull = IsNull.apply("a.c"); + Predicate actualIsNull = converter.convert(isNull); + assertThat(actualIsNull.toString()).isEqualTo("IsNull(a.c)"); + + GenericRow nestedRow = new GenericRow(2); + nestedRow.setField(0, 1); + nestedRow.setField(1, fromString("paimon")); + GenericRow row = new GenericRow(1); + row.setField(0, nestedRow); + + assertThat(actual.test(row)).isTrue(); + assertThat(actualIsNull.test(row)).isFalse(); + + nestedRow.setField(0, 2); + assertThat(actual.test(row)).isFalse(); + + nestedRow.setField(1, null); + assertThat(actualIsNull.test(row)).isTrue(); + + GenericRow min = new GenericRow(1); + min.setField(0, nestedRow); + GenericRow max = new GenericRow(1); + max.setField(0, nestedRow); + assertThat(actual.test(1L, min, max, new GenericArray(new long[] {0L}))).isTrue(); + } }