diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/NestedFieldTransform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/NestedFieldTransform.java new file mode 100644 index 000000000000..2a201acf9df8 --- /dev/null +++ b/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}. + * + *

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()}. + * + *

Deliberately not 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 path; + + private final String name; + private final DataType outputType; + + @JsonCreator + public NestedFieldTransform( + @JsonProperty(FIELD_FIELD_REF) FieldRef fieldRef, + @JsonProperty(FIELD_PATH) List 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 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 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 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; + } +} 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 04e813cdcff9..81221a3599bc 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 @@ -278,6 +278,10 @@ public Predicate notIn(int idx, List literals) { return in(idx, literals).negate().get(); } + public Predicate notIn(Transform transform, List 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( diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/Transform.java b/paimon-common/src/main/java/org/apache/paimon/predicate/Transform.java index ad01afcfb7be..687e489388f0 100644 --- a/paimon-common/src/main/java/org/apache/paimon/predicate/Transform.java +++ b/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), diff --git a/paimon-common/src/test/java/org/apache/paimon/predicate/NestedFieldTransformTest.java b/paimon-common/src/test/java/org/apache/paimon/predicate/NestedFieldTransformTest.java new file mode 100644 index 000000000000..3a90f700494c --- /dev/null +++ b/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> + 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 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); + } +} diff --git a/paimon-format/src/main/java/org/apache/parquet/filter2/predicate/ParquetFilters.java b/paimon-format/src/main/java/org/apache/parquet/filter2/predicate/ParquetFilters.java index e28b8ee437e1..610c9b2e8c90 100644 --- a/paimon-format/src/main/java/org/apache/parquet/filter2/predicate/ParquetFilters.java +++ b/paimon-format/src/main/java/org/apache/parquet/filter2/predicate/ParquetFilters.java @@ -25,6 +25,7 @@ import org.apache.paimon.predicate.FieldRef; import org.apache.paimon.predicate.FunctionVisitor; import org.apache.paimon.predicate.LeafPredicate; +import org.apache.paimon.predicate.NestedFieldTransform; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.types.ArrayType; import org.apache.paimon.types.BigIntType; @@ -56,6 +57,7 @@ import org.apache.parquet.filter2.predicate.Operators.DoubleColumn; import org.apache.parquet.filter2.predicate.Operators.FloatColumn; import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.GroupType; import org.apache.parquet.schema.LogicalTypeAnnotation; import org.apache.parquet.schema.LogicalTypeAnnotation.DecimalLogicalTypeAnnotation; import org.apache.parquet.schema.LogicalTypeAnnotation.TimestampLogicalTypeAnnotation; @@ -77,6 +79,9 @@ /** Convert {@link Predicate} to {@link FilterCompat.Filter}. */ public class ParquetFilters { + /** Columns here are named, never indexed, so a nested field's index is left unset. */ + private static final int UNUSED_INDEX = -1; + private ParquetFilters() {} public static FilterCompat.Filter convert( @@ -273,9 +278,19 @@ public FilterPredicate visitNotIn(FieldRef fieldRef, List literals) { throw new UnsupportedOperationException(); } + /** + * A nested field carries no index into the file, only a path, so it is re-dispatched under + * a {@link FieldRef} naming that path. Every other transform - casts, string functions - + * has no column of its own to filter on and is given up here. + */ @Override public FilterPredicate visitNonFieldLeaf(LeafPredicate predicate) { - throw new UnsupportedOperationException(); + if (!(predicate.transform() instanceof NestedFieldTransform)) { + throw new UnsupportedOperationException(); + } + NestedFieldTransform nested = (NestedFieldTransform) predicate.transform(); + FieldRef pathRef = new FieldRef(UNUSED_INDEX, nested.fieldName(), nested.outputType()); + return predicate.function().visit(this, pathRef, predicate.literals()); } private Set convertSets(List values, Class kclass, FieldRef fieldRef) { @@ -511,30 +526,89 @@ private static PrimitiveType timestampPrimitiveType( private static PrimitiveType primitiveType( FieldRef fieldRef, MessageType fileSchema, boolean caseSensitive) { - PrimitiveType matched = findPrimitiveType(fieldRef, fileSchema, caseSensitive); + FileColumn matched = findFileColumn(fieldRef, fileSchema, caseSensitive); if (matched == null) { throw new UnsupportedOperationException(); } - return matched; + return matched.type; + } + + /** A column the file actually holds: its own spelling of the path, and its physical type. */ + private static class FileColumn { + + private final String path; + private final PrimitiveType type; + + private FileColumn(String path, PrimitiveType type) { + this.path = path; + this.type = type; + } } /** * The file's column for {@code fieldRef}, or null when the file has no such column. A column * that exists but is not primitive cannot carry a predicate at all, so it is rejected outright. + * + *

{@code fieldRef} names a nested field with dots ({@code addr.city}), which is resolved by + * descending the file's groups. A top-level column matching the whole name wins over that walk, + * keeping flat columns spelled with dots resolving as they always did. parquet-mr identifies + * columns by dot-joined path too, so it cannot tell the two apart either way. */ @Nullable - private static PrimitiveType findPrimitiveType( + private static FileColumn findFileColumn( FieldRef fieldRef, MessageType fileSchema, boolean caseSensitive) { - // Paimon predicates currently reference top-level fields only. Nested field - // predicates are rejected before reaching the format reader. - for (Type field : fileSchema.getFields()) { + Type matched = findChild(fileSchema, fieldRef.name(), caseSensitive); + if (matched != null) { + return toFileColumn(matched.getName(), matched); + } + + String[] parts = fieldRef.name().split("\\."); + if (parts.length < 2) { + return null; + } + + StringBuilder resolved = new StringBuilder(); + GroupType parent = fileSchema; + for (int i = 0; i < parts.length; i++) { + Type child = findChild(parent, parts[i], caseSensitive); + if (child == null) { + return null; + } + if (child.getRepetition() == Type.Repetition.REPEATED) { + // A column under repetition has no one value per row, and parquet-mr refuses a + // predicate on it outright. + throw new UnsupportedOperationException(); + } + if (i > 0) { + resolved.append('.'); + } + resolved.append(child.getName()); + + if (i == parts.length - 1) { + return toFileColumn(resolved.toString(), child); + } + if (child.isPrimitive()) { + return null; + } + parent = child.asGroupType(); + } + return null; + } + + private static FileColumn toFileColumn(String path, Type field) { + if (!field.isPrimitive()) { + throw new UnsupportedOperationException(); + } + return new FileColumn(path, field.asPrimitiveType()); + } + + @Nullable + private static Type findChild(GroupType parent, String name, boolean caseSensitive) { + for (Type field : parent.getFields()) { if (caseSensitive - ? field.getName().equals(fieldRef.name()) - : field.getName().equalsIgnoreCase(fieldRef.name())) { - if (!field.isPrimitive()) { - throw new UnsupportedOperationException(); - } - return field.asPrimitiveType(); + ? field.getName().equals(name) + : field.getName().equalsIgnoreCase(name)) { + return field; } } return null; @@ -557,10 +631,11 @@ private static PrimitiveType findPrimitiveType( private static PushdownTarget pushdownTarget( FieldRef fieldRef, MessageType fileSchema, boolean caseSensitive) { PrimitiveType.PrimitiveTypeName[] acceptable = acceptableTypes(fieldRef.type()); - PrimitiveType fileType = findPrimitiveType(fieldRef, fileSchema, caseSensitive); - if (fileType == null) { + FileColumn fileColumn = findFileColumn(fieldRef, fileSchema, caseSensitive); + if (fileColumn == null) { return new PushdownTarget(fieldRef.name(), acceptable[0]); } + PrimitiveType fileType = fileColumn.type; validateBigIntCompatibility(fieldRef, fileType); @@ -574,7 +649,7 @@ private static PushdownTarget pushdownTarget( for (PrimitiveType.PrimitiveTypeName candidate : acceptable) { if (fileType.getPrimitiveTypeName() == candidate) { - return new PushdownTarget(fileType.getName(), candidate); + return new PushdownTarget(fileColumn.path, candidate); } } throw new UnsupportedOperationException(); diff --git a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java index e6f4c4426046..29e819a8d9db 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java @@ -18,13 +18,18 @@ package org.apache.paimon.format.parquet; +import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.Decimal; import org.apache.paimon.data.Timestamp; +import org.apache.paimon.predicate.FieldRef; +import org.apache.paimon.predicate.NestedFieldTransform; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.types.ArrayType; import org.apache.paimon.types.BigIntType; import org.apache.paimon.types.BooleanType; import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataType; import org.apache.paimon.types.DecimalType; import org.apache.paimon.types.DoubleType; import org.apache.paimon.types.FloatType; @@ -1147,6 +1152,133 @@ private void test( } } + // --------------------------------------------------------------------------------------- + // nested fields + // --------------------------------------------------------------------------------------- + + private static final RowType ADDR_TYPE = + RowType.of( + new DataType[] {new VarCharType(), new VarCharType()}, + new String[] {"city", "zip"}); + + private static RowType nestedRowType() { + return RowType.of( + new DataType[] { + new BigIntType(), + RowType.of( + new DataType[] {new BigIntType(), ADDR_TYPE}, + new String[] {"id", "addr"}), + new ArrayType(ADDR_TYPE) + }, + new String[] {"pk", "user", "addrs"}); + } + + private static Predicate nestedPredicate(RowType rowType, String column, int... path) { + DataField field = rowType.getFields().get(rowType.getFieldIndex(column)); + FieldRef ref = new FieldRef(rowType.getFieldIndex(column), field.name(), field.type()); + List positions = IntStream.of(path).boxed().collect(Collectors.toList()); + return new PredicateBuilder(rowType) + .equal( + new NestedFieldTransform(ref, positions), + BinaryString.fromString("Beijing")); + } + + @Test + public void testNestedField() { + RowType rowType = nestedRowType(); + MessageType schema = ParquetSchemaConverter.convertToParquetMessageType(rowType); + + // user.addr.city + test( + schema, + nestedPredicate(rowType, "user", 1, 0), + "eq(user.addr.city, Binary{\"Beijing\"})", + true); + } + + /** A nested field is dispatched through the same visitors as a top-level one. */ + @Test + public void testNestedFieldSupportsEveryPushableFunction() { + RowType rowType = nestedRowType(); + MessageType schema = ParquetSchemaConverter.convertToParquetMessageType(rowType); + // user.id, a BIGINT one level down + FieldRef ref = new FieldRef(1, "user", rowType.getTypeAt(1)); + NestedFieldTransform id = new NestedFieldTransform(ref, Collections.singletonList(0)); + PredicateBuilder builder = new PredicateBuilder(rowType); + + test(schema, builder.isNull(id), "eq(user.id, null)", true); + test(schema, builder.isNotNull(id), "noteq(user.id, null)", true); + test(schema, builder.equal(id, 5L), "eq(user.id, 5)", true); + test(schema, builder.notEqual(id, 5L), "noteq(user.id, 5)", true); + test(schema, builder.lessThan(id, 5L), "lt(user.id, 5)", true); + test(schema, builder.lessOrEqual(id, 5L), "lteq(user.id, 5)", true); + test(schema, builder.greaterThan(id, 5L), "gt(user.id, 5)", true); + test(schema, builder.greaterOrEqual(id, 5L), "gteq(user.id, 5)", true); + test(schema, builder.between(id, 1L, 3L), "and(gteq(user.id, 1), lteq(user.id, 3))", true); + test( + schema, + builder.in(id, Arrays.asList(1L, 2L)), + "or(eq(user.id, 1), eq(user.id, 2))", + true); + test( + schema, + builder.notIn(id, Arrays.asList(1L, 2L)), + "and(noteq(user.id, 1), noteq(user.id, 2))", + true); + + // AND/OR mixing a nested field with a top-level one + test( + schema, + PredicateBuilder.and(builder.greaterThan(id, 5L), builder.lessThan(0, 100L)), + "and(gt(user.id, 5), lt(pk, 100))", + true); + + // string functions have no parquet equivalent, for nested and top-level alike + test(schema, builder.startsWith(id, BinaryString.fromString("x")), (String) null, false); + } + + /** + * A field under a repeated group has no single value per row, and parquet-mr rejects a + * predicate on one outright. A table declaring a struct over a file that repeats it - a format + * table reading files someone else wrote - must give up rather than hand one over. + */ + @Test + public void testNestedFieldUnderRepeatedGroupIsNotPushedDown() { + RowType rowType = nestedRowType(); + MessageType schema = + new MessageType( + "paimon_schema", + Types.repeatedGroup() + .addField(Types.required(PrimitiveTypeName.INT64).named("id")) + .addField( + Types.requiredGroup() + .addField( + Types.required(PrimitiveTypeName.BINARY) + .as( + LogicalTypeAnnotation + .stringType()) + .named("city")) + .named("addr")) + .named("user")); + + test(schema, nestedPredicate(rowType, "user", 1, 0), (String) null, false); + } + + /** A nested column the file does not hold still prunes: parquet-mr reads it as all-null. */ + @Test + public void testNestedFieldMissingFromFile() { + RowType rowType = nestedRowType(); + MessageType schema = + ParquetSchemaConverter.convertToParquetMessageType( + RowType.of(new DataType[] {new BigIntType()}, new String[] {"pk"})); + + test( + schema, + nestedPredicate(rowType, "user", 1, 0), + "eq(user.addr.city, Binary{\"Beijing\"})", + true); + } + private FilterPredicate convert(MessageType schema, Predicate predicate) { FilterCompat.Filter filter = ParquetFilters.convert(PredicateBuilder.splitAnd(predicate), schema, true); diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala index 00898c98ca53..90550f0be821 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala @@ -61,7 +61,7 @@ object SparkExpressionConverter { } exp match { - case n: NamedReference => Some(new FieldTransform(toPaimonFieldRef(n, rowType))) + case n: NamedReference => toPaimonFieldTransform(n, rowType) case s: GeneralScalarExpression => s.name() match { case CONCAT => convertChildren(s.children()).map(i => new ConcatTransform(i)) @@ -149,6 +149,41 @@ object SparkExpressionConverter { } } + /** + * A reference is either a top-level column or a path down into row-typed ones. Anything the path + * cannot descend - a field inside an array or a map, a name the schema does not hold - yields + * None, leaving the predicate for Spark to evaluate after the scan. + */ + private def toPaimonFieldTransform(ref: NamedReference, rowType: RowType): Option[Transform] = { + val parts = ref.fieldNames() + val index = rowType.getFieldIndex(parts.head) + if (index == -1) { + return None + } + val root = rowType.getField(parts.head) + val rootRef = new FieldRef(index, root.name(), root.`type`()) + if (parts.length == 1) { + return Some(new FieldTransform(rootRef)) + } + + val path = new java.util.ArrayList[Integer](parts.length - 1) + var current = root.`type`() + parts.tail.foreach { + part => + current match { + case nested: RowType => + val position = nested.getFieldIndex(part) + if (position == -1) { + return None + } + path.add(position) + current = nested.getTypeAt(position) + case _ => return None + } + } + Some(new NestedFieldTransform(rootRef, path)) + } + private def toPaimonFieldRef(ref: NamedReference, rowType: RowType): FieldRef = { val fieldName = toFieldName(ref) val f = rowType.getField(fieldName) diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/SparkV2FilterConverterTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/SparkV2FilterConverterTestBase.scala index ea80fe476d5d..7b2e53a06a3f 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/SparkV2FilterConverterTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/SparkV2FilterConverterTestBase.scala @@ -19,7 +19,7 @@ package org.apache.paimon.spark.sql import org.apache.paimon.data.{BinaryString, Decimal, Timestamp} -import org.apache.paimon.predicate.PredicateBuilder +import org.apache.paimon.predicate.{FieldTransform, LeafPredicate, NestedFieldTransform, PredicateBuilder} import org.apache.paimon.spark.{PaimonSparkTestBase, SparkV2FilterConverter} import org.apache.paimon.spark.util.shim.TypeUtils.treatPaimonTimestampTypeAsSparkTimestampType import org.apache.paimon.table.source.DataSplit @@ -539,6 +539,63 @@ abstract class SparkV2FilterConverterTestBase extends PaimonSparkTestBase { assert(filesScanned == 4, s"Expected 4 files but scanned $filesScanned files") } + test("V2Filter: nested field") { + withTable("nested_tbl") { + sql(""" + |CREATE TABLE nested_tbl ( + | id INT, + | info STRUCT> + |) USING paimon + |""".stripMargin) + sql("INSERT INTO nested_tbl VALUES (1, struct(10, struct('Beijing', '100080')))") + sql("INSERT INTO nested_tbl VALUES (2, struct(20, struct('Shanghai', '200000')))") + + val nestedConverter = SparkV2FilterConverter(loadTable("nested_tbl").rowType()) + + Seq("info.uid = 10" -> "info.uid", "info.addr.city = 'Beijing'" -> "info.addr.city") + .foreach { + case (filter, expectedName) => + val predicate = + nestedConverter + .convert(v2Filter(filter, "nested_tbl")) + .get + .asInstanceOf[LeafPredicate] + val transform = predicate.transform().asInstanceOf[NestedFieldTransform] + assert(transform.fieldName() == expectedName) + // no FieldRef is handed out, so nothing mistakes this for a top-level column + assert(!predicate.fieldRefOptional().isPresent) + // the enclosing column is what field-name based rewrites see + assert(predicate.fieldNames().asScala == Seq("info")) + + checkAnswer(sql(s"SELECT id FROM nested_tbl WHERE $filter"), Seq(Row(1))) + assert( + getPaimonScan(s"SELECT * FROM nested_tbl WHERE $filter").pushedDataFilters + .exists(_.toString.contains(expectedName))) + } + + // a nested field still reads correctly alongside a projection of a sibling field + checkAnswer( + sql("SELECT info.addr.zip FROM nested_tbl WHERE info.addr.city = 'Shanghai'"), + Seq(Row("200000"))) + } + } + + test("V2Filter: a top-level column whose name contains a dot") { + withTable("dotted_tbl") { + sql("CREATE TABLE dotted_tbl (id INT, `a.b` STRING) USING paimon") + + val dottedConverter = SparkV2FilterConverter(loadTable("dotted_tbl").rowType()) + val predicate = dottedConverter + .convert(v2Filter("`a.b` = 'x'", "dotted_tbl")) + .get + .asInstanceOf[LeafPredicate] + + // resolves as the flat column it is, not as a path into a struct named "a" + assert(predicate.transform().isInstanceOf[FieldTransform]) + assert(predicate.fieldNames().asScala == Seq("a.b")) + } + } + private def v2Filter(str: String, tableName: String = "test_tbl"): SparkPredicate = { val condition = sql(s"SELECT * FROM $tableName WHERE $str").queryExecution.optimizedPlan .collectFirst { case f: Filter => f }