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..761de46119cf --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/predicate/NestedFieldTransform.java @@ -0,0 +1,196 @@ +/* + * 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; + + /** + * Names of the fields to descend into, relative to {@code fieldRef}'s row type. Never empty. + * + *

Deliberately names rather than positions: {@link #copyWithNewInputs} may be handed a + * structurally different row type — column masking and row filters remap that way — and a bare + * position would stay in range while silently addressing whatever now sits there. Names are + * re-resolved on every remap, so a reference either finds the same field or fails. + */ + private final List path; + + /** {@link #path} resolved to positions against {@code fieldRef}'s row type. */ + private final int[] positions; + + 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)); + this.positions = new int[this.path.size()]; + + StringBuilder nameBuilder = new StringBuilder(fieldRef.name()); + DataType current = fieldRef.type(); + for (int i = 0; i < this.path.size(); i++) { + checkArgument( + current instanceof RowType, + "Nested field path of '%s' descends into a non-row type %s.", + fieldRef.name(), + current); + RowType rowType = (RowType) current; + String component = this.path.get(i); + int position = rowType.getFieldIndex(component); + checkArgument( + position >= 0, + "Nested field '%s' does not contain a field named '%s'.", + nameBuilder, + component); + positions[i] = position; + nameBuilder.append('.').append(component); + 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 < positions.length - 1; i++) { + position = positions[i]; + if (current.isNullAt(position)) { + return null; + } + RowType nextType = (RowType) currentType.getTypeAt(position); + current = current.getRow(position, nextType.getFieldCount()); + currentType = nextType; + } + + int leaf = positions[positions.length - 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..91fe19297b03 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/predicate/NestedFieldTransformTest.java @@ -0,0 +1,274 @@ +/* + * 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("id")); + + 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("addr", "city")); + + 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")); + } + + /** The descent loop is recursive; three levels must read as well as two. */ + @Test + public void testReadThreeLevels() { + RowType level3 = + RowType.of( + new org.apache.paimon.types.DataType[] {DataTypes.BIGINT()}, + new String[] {"d"}); + RowType level2 = + RowType.of(new org.apache.paimon.types.DataType[] {level3}, new String[] {"c"}); + RowType level1 = + RowType.of(new org.apache.paimon.types.DataType[] {level2}, new String[] {"b"}); + FieldRef ref = new FieldRef(0, "a", level1); + + NestedFieldTransform transform = + new NestedFieldTransform(ref, Arrays.asList("b", "c", "d")); + assertThat(transform.fieldName()).isEqualTo("a.b.c.d"); + assertThat(transform.outputType()).isEqualTo(DataTypes.BIGINT()); + + GenericRow row = GenericRow.of(GenericRow.of(GenericRow.of(GenericRow.of(42L)))); + assertThat(transform.transform(row)).isEqualTo(42L); + + // a null two levels down still yields null + GenericRow withNull = GenericRow.of(GenericRow.of(GenericRow.of((Object) null))); + assertThat(transform.transform(withNull)).isNull(); + } + + @Test + public void testNullAnywhereOnThePathYieldsNull() { + NestedFieldTransform transform = + new NestedFieldTransform(USER_REF, Arrays.asList("addr", "city")); + + // 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("addr", "city")), + 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("id")), + 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("id")), + 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("addr", "city")), + 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("addr", "city"); + assertThat(transform.fieldName()).isEqualTo("user.addr.city"); + } + + @Test + public void testJsonRoundTrip() { + Predicate predicate = + new PredicateBuilder(ROW_TYPE) + .equal( + new NestedFieldTransform(USER_REF, Arrays.asList("addr", "city")), + 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("x"))) + .isInstanceOf(IllegalArgumentException.class); + + assertThatThrownBy(() -> new NestedFieldTransform(USER_REF, Collections.emptyList())) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy( + () -> new NestedFieldTransform(USER_REF, Collections.singletonList("nope"))) + .isInstanceOf(IllegalArgumentException.class); + } + + /** + * Remapping must not let a nested reference drift onto a different field. Column pruning can + * hand {@code copyWithNewInputs} a structurally different row type — a bare position stays in + * range and silently addresses whatever now sits there. Row filters and column masks are + * remapped this way, so drifting has to fail closed rather than resolve elsewhere. + */ + @Test + public void testRemapOntoAPrunedRowTypeDoesNotDrift() { + RowType full = + RowType.of( + new org.apache.paimon.types.DataType[] { + DataTypes.STRING(), DataTypes.STRING() + }, + new String[] {"secret", "region"}); + FieldRef infoRef = new FieldRef(0, "info", full); + NestedFieldTransform onSecret = + new NestedFieldTransform(infoRef, Collections.singletonList("secret")); + assertThat(onSecret.fieldName()).isEqualTo("info.secret"); + + // "secret" was pruned away; position 0 is now "region" + RowType pruned = + RowType.of( + new org.apache.paimon.types.DataType[] {DataTypes.STRING()}, + new String[] {"region"}); + FieldRef prunedRef = new FieldRef(0, "info", pruned); + + assertThatThrownBy(() -> onSecret.copyWithNewInputs(Collections.singletonList(prunedRef))) + .isInstanceOf(IllegalArgumentException.class); + } + + /** Remapping onto a reordered row type must keep addressing the same field. */ + @Test + public void testRemapFollowsTheFieldWhenPositionsShift() { + RowType full = + RowType.of( + new org.apache.paimon.types.DataType[] { + DataTypes.STRING(), DataTypes.STRING() + }, + new String[] {"secret", "region"}); + NestedFieldTransform onSecret = + new NestedFieldTransform( + new FieldRef(0, "info", full), Collections.singletonList("secret")); + + RowType reordered = + RowType.of( + new org.apache.paimon.types.DataType[] { + DataTypes.STRING(), DataTypes.STRING() + }, + new String[] {"region", "secret"}); + Transform remapped = + onSecret.copyWithNewInputs( + Collections.singletonList(new FieldRef(0, "info", reordered))); + + assertThat(((NestedFieldTransform) remapped).fieldName()).isEqualTo("info.secret"); + } +} 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 index 5b7f64fe8a59..71bd9af1ae22 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/TableQueryAuthResultTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/TableQueryAuthResultTest.java @@ -24,7 +24,10 @@ import org.apache.paimon.predicate.FieldRef; import org.apache.paimon.predicate.FieldTransform; import org.apache.paimon.predicate.LeafPredicate; +import org.apache.paimon.predicate.NestedFieldTransform; import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.JsonSerdeUtil; @@ -162,4 +165,49 @@ public void testHasRules() { Map masking = Collections.singletonMap("display", maskJson()); assertThat(new TableQueryAuthResult(null, masking).hasRules()).isTrue(); } + + private static RowType infoRowType(String... nestedFields) { + DataType[] types = new DataType[nestedFields.length]; + for (int i = 0; i < types.length; i++) { + types[i] = DataTypes.STRING(); + } + return RowType.of( + new DataType[] {DataTypes.INT(), RowType.of(types, nestedFields)}, + new String[] {"pk", "info"}); + } + + private static Predicate rowFilterOnInfoSecret(RowType rowType) { + RowType info = (RowType) rowType.getTypeAt(1); + return new PredicateBuilder(rowType) + .equal( + new NestedFieldTransform( + new FieldRef(1, "info", info), Collections.singletonList("secret")), + org.apache.paimon.data.BinaryString.fromString("x")); + } + + /** + * A row filter on a nested field must not silently follow column pruning onto a different + * field. Remapping resolves the components by name, so a pruned-away leaf fails closed rather + * than letting the policy address whatever now sits at that position. + */ + @Test + void testNestedRowFilterDoesNotDriftWhenTheLeafIsPruned() { + Predicate filter = rowFilterOnInfoSecret(infoRowType("secret", "region")); + + // the projection kept "info" but dropped "info.secret" + RowType pruned = infoRowType("region"); + assertThatThrownBy(() -> TableQueryAuthResult.remapPredicate(filter, pruned)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("secret"); + } + + /** Remapping onto a reordered row type must keep addressing the same nested field. */ + @Test + void testNestedRowFilterFollowsTheFieldWhenPositionsShift() { + Predicate filter = rowFilterOnInfoSecret(infoRowType("secret", "region")); + + Predicate remapped = + TableQueryAuthResult.remapPredicate(filter, infoRowType("region", "secret")); + assertThat(remapped.toString()).contains("info.secret"); + } } 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..dee9d44df89c 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,31 @@ 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(); + // The path reaches parquet-mr as a dot-joined string, which it splits back into + // components. A component that itself contains a dot does not survive that round trip: + // the filter would address a column the file does not hold, and a missing column reads + // as all-null, pruning row groups that actually match. Give up the pruning instead. + if (nested.fieldRef().name().indexOf('.') >= 0) { + throw new UnsupportedOperationException(); + } + for (String component : nested.path()) { + if (component.indexOf('.') >= 0) { + throw new UnsupportedOperationException(); + } + } + 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) { @@ -306,7 +333,7 @@ private Comparable toParquetObject(Object value, FieldRef fieldRef) { DecimalType decimalType = (DecimalType) fieldRef.type(); Decimal decimal = normalizeDecimal((Decimal) value, decimalType); PrimitiveType primitiveType = - decimalPrimitiveType(fieldRef, fileSchema, caseSensitive); + decimalColumn(fieldRef, fileSchema, caseSensitive).type; switch (primitiveType.getPrimitiveTypeName()) { case INT32: long intValue = toUnscaledLong(decimal); @@ -327,7 +354,7 @@ private Comparable toParquetObject(Object value, FieldRef fieldRef) { if (value instanceof Timestamp) { Timestamp timestamp = (Timestamp) value; - timestampPrimitiveType(fieldRef, fileSchema, caseSensitive); + timestampColumn(fieldRef, fileSchema, caseSensitive); int precision = getTimestampPrecision(type); if (precision <= 3) { // milliseconds @@ -464,9 +491,10 @@ private Binary decimalToBinary(Decimal decimal, int numBytes) { } } - private static PrimitiveType decimalPrimitiveType( + private static FileColumn decimalColumn( FieldRef fieldRef, MessageType fileSchema, boolean caseSensitive) { - PrimitiveType primitiveType = primitiveType(fieldRef, fileSchema, caseSensitive); + FileColumn column = fileColumn(fieldRef, fileSchema, caseSensitive); + PrimitiveType primitiveType = column.type; LogicalTypeAnnotation logicalType = primitiveType.getLogicalTypeAnnotation(); if (!(logicalType instanceof DecimalLogicalTypeAnnotation)) { throw new UnsupportedOperationException(); @@ -477,19 +505,20 @@ private static PrimitiveType decimalPrimitiveType( if (decimalLogicalType.getScale() != ((DecimalType) fieldRef.type()).getScale()) { throw new UnsupportedOperationException(); } - return primitiveType; + return column; } - private static PrimitiveType timestampPrimitiveType( + private static FileColumn timestampColumn( FieldRef fieldRef, MessageType fileSchema, boolean caseSensitive) { - PrimitiveType primitiveType = primitiveType(fieldRef, fileSchema, caseSensitive); + FileColumn column = fileColumn(fieldRef, fileSchema, caseSensitive); + PrimitiveType primitiveType = column.type; if (primitiveType.getPrimitiveTypeName() != PrimitiveType.PrimitiveTypeName.INT64) { throw new UnsupportedOperationException(); } LogicalTypeAnnotation logicalType = primitiveType.getLogicalTypeAnnotation(); if (logicalType == null) { - return primitiveType; + return column; } if (!(logicalType instanceof TimestampLogicalTypeAnnotation)) { throw new UnsupportedOperationException(); @@ -506,35 +535,100 @@ private static PrimitiveType timestampPrimitiveType( || timestampType.isAdjustedToUTC() != expectedAdjustedToUtc) { throw new UnsupportedOperationException(); } - return primitiveType; + return column; } - private static PrimitiveType primitiveType( + /** + * The column the file holds for {@code fieldRef}, with the file's own spelling of the path. + * Callers that build a parquet column must use {@link FileColumn#path}: a {@link PrimitiveType} + * only knows its own leaf name, so rebuilding the column from it drops the enclosing path and + * addresses a column the file does not have. + */ + private static FileColumn fileColumn( 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; } + /** 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 +651,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 +669,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(); @@ -800,15 +895,16 @@ public Operators.Column visit(TimeType timeType) { @Override public Operators.Column visit(DecimalType decimalType) { - PrimitiveType primitiveType = decimalPrimitiveType(fieldRef, fileSchema, caseSensitive); + FileColumn column = decimalColumn(fieldRef, fileSchema, caseSensitive); + PrimitiveType primitiveType = column.type; switch (primitiveType.getPrimitiveTypeName()) { case INT32: - return FilterApi.intColumn(primitiveType.getName()); + return FilterApi.intColumn(column.path); case INT64: - return FilterApi.longColumn(primitiveType.getName()); + return FilterApi.longColumn(column.path); case BINARY: case FIXED_LEN_BYTE_ARRAY: - return FilterApi.binaryColumn(primitiveType.getName()); + return FilterApi.binaryColumn(column.path); default: throw new UnsupportedOperationException(); } @@ -819,7 +915,7 @@ public Operators.Column visit(TimestampType timestampType) { int precision = timestampType.getPrecision(); if (precision <= 6) { return FilterApi.longColumn( - timestampPrimitiveType(fieldRef, fileSchema, caseSensitive).getName()); + timestampColumn(fieldRef, fileSchema, caseSensitive).path); } // precision > 6 uses INT96, not supported for filter pushdown throw new UnsupportedOperationException(); @@ -830,7 +926,7 @@ public Operators.Column visit(LocalZonedTimestampType localZonedTimestampType int precision = localZonedTimestampType.getPrecision(); if (precision <= 6) { return FilterApi.longColumn( - timestampPrimitiveType(fieldRef, fileSchema, caseSensitive).getName()); + timestampColumn(fieldRef, fileSchema, caseSensitive).path); } // precision > 6 uses INT96, not supported for filter pushdown 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..65b936a185c6 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,387 @@ 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, String... path) { + DataField field = rowType.getFields().get(rowType.getFieldIndex(column)); + FieldRef ref = new FieldRef(rowType.getFieldIndex(column), field.name(), field.type()); + return new PredicateBuilder(rowType) + .equal( + new NestedFieldTransform(ref, Arrays.asList(path)), + BinaryString.fromString("Beijing")); + } + + @Test + public void testNestedField() { + RowType rowType = nestedRowType(); + MessageType schema = ParquetSchemaConverter.convertToParquetMessageType(rowType); + + // user.addr.city + test( + schema, + nestedPredicate(rowType, "user", "addr", "city"), + "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("id")); + 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", "addr", "city"), (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", "addr", "city"), + "eq(user.addr.city, Binary{\"Beijing\"})", + true); + } + + private static RowType payloadRowType() { + return RowType.of( + new DataType[] { + new BigIntType(), + RowType.of( + new DataType[] { + new DecimalType(10, 2), + new TimestampType(3), + new LocalZonedTimestampType(3), + new BigIntType() + }, + new String[] {"amount", "ts", "ltz", "qty"}), + new DecimalType(10, 2) + }, + new String[] {"pk", "payload", "amt_top"}); + } + + private static NestedFieldTransform payloadLeaf(RowType rowType, String leaf) { + RowType payload = (RowType) rowType.getTypeAt(1); + return new NestedFieldTransform( + new FieldRef(1, "payload", payload), Collections.singletonList(leaf)); + } + + /** Control: the two shapes that already worked must keep working. */ + @Test + public void testTopLevelDecimalAndNestedBigIntAreUnaffected() { + RowType rowType = payloadRowType(); + MessageType schema = ParquetSchemaConverter.convertToParquetMessageType(rowType); + PredicateBuilder builder = new PredicateBuilder(rowType); + Decimal amount = Decimal.fromBigDecimal(new BigDecimal("12.34"), 10, 2); + + test(schema, builder.equal(2, amount), "eq(amt_top, 1234)", true); + test( + schema, + builder.greaterThan(payloadLeaf(rowType, "qty"), 5L), + "gt(payload.qty, 5)", + true); + } + + /** + * A nested DECIMAL must be filtered on its full path. The physical type is resolved by walking + * the path, but the column handed to parquet-mr used to be rebuilt from the leaf {@code + * PrimitiveType}, which only knows its own name — parquet-mr then saw a missing top-level + * column and could drop every row group. + */ + @Test + public void testNestedDecimalKeepsTheFullPath() { + RowType rowType = payloadRowType(); + MessageType schema = ParquetSchemaConverter.convertToParquetMessageType(rowType); + PredicateBuilder builder = new PredicateBuilder(rowType); + NestedFieldTransform amount = payloadLeaf(rowType, "amount"); + Decimal value = Decimal.fromBigDecimal(new BigDecimal("12.34"), 10, 2); + + test(schema, builder.equal(amount, value), "eq(payload.amount, 1234)", true); + test(schema, builder.lessThan(amount, value), "lt(payload.amount, 1234)", true); + } + + /** Same as {@link #testNestedDecimalKeepsTheFullPath()} for TIMESTAMP. */ + @Test + public void testNestedTimestampKeepsTheFullPath() { + RowType rowType = payloadRowType(); + MessageType schema = ParquetSchemaConverter.convertToParquetMessageType(rowType); + PredicateBuilder builder = new PredicateBuilder(rowType); + NestedFieldTransform ts = payloadLeaf(rowType, "ts"); + Timestamp value = Timestamp.fromEpochMillis(1704067200000L); + long millis = value.getMillisecond(); + + test(schema, builder.equal(ts, value), "eq(payload.ts, " + millis + ")", true); + test(schema, builder.greaterThan(ts, value), "gt(payload.ts, " + millis + ")", true); + } + + /** Same as {@link #testNestedDecimalKeepsTheFullPath()} for LOCAL ZONED TIMESTAMP. */ + @Test + public void testNestedLocalZonedTimestampKeepsTheFullPath() { + RowType rowType = payloadRowType(); + MessageType schema = ParquetSchemaConverter.convertToParquetMessageType(rowType); + PredicateBuilder builder = new PredicateBuilder(rowType); + NestedFieldTransform ltz = payloadLeaf(rowType, "ltz"); + Timestamp value = Timestamp.fromEpochMillis(1704067200000L); + long millis = value.getMillisecond(); + + test(schema, builder.equal(ltz, value), "eq(payload.ltz, " + millis + ")", true); + } + + /** + * A nested decimal schema with an explicit physical type, the way a Format Table may hold it. + */ + private static MessageType nestedDecimalSchema( + PrimitiveTypeName physicalType, int fixedLength, int precision, int scale) { + Types.PrimitiveBuilder builder = Types.optional(physicalType); + if (physicalType == PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY) { + builder.length(fixedLength); + } + PrimitiveType amount = + builder.as(LogicalTypeAnnotation.decimalType(scale, precision)).named("amount"); + return new MessageType( + "paimon_schema", + Arrays.asList( + Types.optional(PrimitiveTypeName.INT64).named("pk"), + Types.optionalGroup().addField(amount).named("payload"))); + } + + private void testNestedDecimalPhysicalType( + PrimitiveTypeName physicalType, int fixedLength, int precision, String literal) { + int scale = 2; + RowType payload = + RowType.of( + new DataType[] {new DecimalType(precision, scale)}, + new String[] {"amount"}); + RowType rowType = + RowType.of( + new DataType[] {new BigIntType(), payload}, new String[] {"pk", "payload"}); + MessageType schema = nestedDecimalSchema(physicalType, fixedLength, precision, scale); + NestedFieldTransform amount = + new NestedFieldTransform( + new FieldRef(1, "payload", payload), Collections.singletonList("amount")); + Decimal value = Decimal.fromBigDecimal(new BigDecimal(literal), precision, scale); + + FilterPredicate filter = + convert(schema, new PredicateBuilder(rowType).equal(amount, value)); + // whatever the physical type, the column must be the full path + assertThat(filter.toString()).startsWith("eq(payload.amount, "); + } + + /** The decimal visitor builds a column per physical type; every branch must keep the path. */ + @Test + public void testNestedDecimalKeepsTheFullPathForEveryPhysicalType() { + // precision <= 9 -> INT32 + testNestedDecimalPhysicalType(PrimitiveTypeName.INT32, 0, 8, "12.34"); + // precision <= 18 -> INT64 + testNestedDecimalPhysicalType(PrimitiveTypeName.INT64, 0, 15, "12.34"); + // larger -> FIXED_LEN_BYTE_ARRAY + testNestedDecimalPhysicalType(PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY, 16, 30, "12.34"); + // a Format Table may hold a decimal as BINARY + testNestedDecimalPhysicalType(PrimitiveTypeName.BINARY, 0, 30, "12.34"); + } + + /** Micros-precision timestamps take a different literal path than millis. */ + @Test + public void testNestedTimestampMicrosKeepsTheFullPath() { + RowType payload = + RowType.of( + new DataType[] {new TimestampType(6), new LocalZonedTimestampType(6)}, + new String[] {"ts", "ltz"}); + RowType rowType = + RowType.of( + new DataType[] {new BigIntType(), payload}, new String[] {"pk", "payload"}); + MessageType schema = ParquetSchemaConverter.convertToParquetMessageType(rowType); + PredicateBuilder builder = new PredicateBuilder(rowType); + FieldRef payloadRef = new FieldRef(1, "payload", payload); + Timestamp value = Timestamp.fromEpochMillis(1704067200000L); + long micros = value.toMicros(); + + test( + schema, + builder.equal( + new NestedFieldTransform(payloadRef, Collections.singletonList("ts")), + value), + "eq(payload.ts, " + micros + ")", + true); + test( + schema, + builder.greaterThan( + new NestedFieldTransform(payloadRef, Collections.singletonList("ltz")), + value), + "gt(payload.ltz, " + micros + ")", + true); + } + + /** IN and NOT IN build the column through the same visitor; a nested decimal must keep it. */ + @Test + public void testNestedDecimalInAndNotInKeepTheFullPath() { + RowType rowType = payloadRowType(); + MessageType schema = ParquetSchemaConverter.convertToParquetMessageType(rowType); + PredicateBuilder builder = new PredicateBuilder(rowType); + NestedFieldTransform amount = payloadLeaf(rowType, "amount"); + Decimal one = Decimal.fromBigDecimal(new BigDecimal("1.00"), 10, 2); + Decimal two = Decimal.fromBigDecimal(new BigDecimal("2.00"), 10, 2); + + test( + schema, + builder.in(amount, Arrays.asList(one, two)), + "or(eq(payload.amount, 100), eq(payload.amount, 200))", + true); + test( + schema, + builder.notIn(amount, Arrays.asList(one, two)), + "and(noteq(payload.amount, 100), noteq(payload.amount, 200))", + true); + } + + /** The path walk is recursive; three levels must resolve as well as two. */ + @Test + public void testDeeplyNestedFieldKeepsTheFullPath() { + RowType level3 = RowType.of(new DataType[] {new BigIntType()}, new String[] {"d"}); + RowType level2 = RowType.of(new DataType[] {level3}, new String[] {"c"}); + RowType level1 = RowType.of(new DataType[] {level2}, new String[] {"b"}); + RowType rowType = + RowType.of(new DataType[] {new BigIntType(), level1}, new String[] {"pk", "a"}); + MessageType schema = ParquetSchemaConverter.convertToParquetMessageType(rowType); + + NestedFieldTransform deep = + new NestedFieldTransform( + new FieldRef(1, "a", level1), Arrays.asList("b", "c", "d")); + test(schema, new PredicateBuilder(rowType).equal(deep, 7L), "eq(a.b.c.d, 7)", true); + } + + /** + * A nested component whose own name contains a dot cannot be expressed as a dot-joined path: + * parquet-mr would split {@code s.a.b} into three components and miss the real two-component + * column, treating it as all-null and pruning matching row groups. Refuse the pushdown. + */ + @Test + public void testNestedComponentContainingADotIsNotPushedDown() { + RowType inner = RowType.of(new DataType[] {new BigIntType()}, new String[] {"a.b"}); + RowType rowType = + RowType.of(new DataType[] {new BigIntType(), inner}, new String[] {"pk", "s"}); + MessageType schema = ParquetSchemaConverter.convertToParquetMessageType(rowType); + PredicateBuilder builder = new PredicateBuilder(rowType); + + // the file really holds s -> "a.b"; a dot-joined "s.a.b" does not address it + assertThat(schema.getType("s").asGroupType().containsField("a.b")).isTrue(); + + NestedFieldTransform dotted = + new NestedFieldTransform( + new FieldRef(1, "s", inner), Collections.singletonList("a.b")); + test(schema, builder.equal(dotted, 7L), (String) null, false); + } + + /** + * The dot may also sit in the top-level column's own name. The joined path then splits into + * components the file does not have — and, worse, could collide with a genuinely nested column + * of the same spelling. Refuse the pushdown here too. + */ + @Test + public void testNestedFieldUnderATopLevelNameContainingADotIsNotPushedDown() { + RowType inner = RowType.of(new DataType[] {new VarCharType()}, new String[] {"city"}); + RowType rowType = + RowType.of(new DataType[] {new BigIntType(), inner}, new String[] {"pk", "a.b"}); + MessageType schema = ParquetSchemaConverter.convertToParquetMessageType(rowType); + + // the file holds ["a.b", "city"]; the joined name "a.b.city" splits into [a, b, city] + assertThat(schema.containsField("a.b")).isTrue(); + assertThat(schema.containsField("a")).isFalse(); + + NestedFieldTransform nested = + new NestedFieldTransform( + new FieldRef(1, "a.b", inner), Collections.singletonList("city")); + test( + schema, + new PredicateBuilder(rowType).equal(nested, BinaryString.fromString("Beijing")), + (String) null, + false); + } + private FilterPredicate convert(MessageType schema, Predicate predicate) { FilterCompat.Filter filter = ParquetFilters.convert(PredicateBuilder.splitAnd(predicate), schema, true); diff --git a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java index 2784f157a113..d00dd68d9c84 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFormatReadWriteTest.java @@ -19,6 +19,7 @@ package org.apache.paimon.format.parquet; import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.Decimal; import org.apache.paimon.data.GenericArray; import org.apache.paimon.data.GenericMap; import org.apache.paimon.data.GenericRow; @@ -36,7 +37,12 @@ import org.apache.paimon.format.SupportsWriterMetadata; import org.apache.paimon.fs.PositionOutputStream; import org.apache.paimon.options.Options; +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.reader.RecordReader; +import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; @@ -51,7 +57,11 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import java.io.IOException; +import java.math.BigDecimal; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; @@ -292,4 +302,181 @@ public void testColumnCompressionCodec() throws Exception { .containsEntry("name", CompressionCodecName.UNCOMPRESSED); } } + + // ----------------------------------------------------------------------------------------- + // end-to-end: a nested predicate must not silently drop rows + // ----------------------------------------------------------------------------------------- + + private RowType nestedPayloadType() { + return RowType.of( + new DataType[] { + DataTypes.BIGINT(), + RowType.of( + new DataType[] { + DataTypes.DECIMAL(10, 2), + DataTypes.TIMESTAMP(3), + DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(3), + DataTypes.BIGINT() + }, + new String[] {"amount", "ts", "ltz", "qty"}) + }, + new String[] {"pk", "payload"}); + } + + /** + * Reads with {@code predicate} pushed down and returns the primary keys that survived. Parquet + * filtering is row-group granular, so a matching row may come back alongside non-matching ones + * — what must never happen is the matching row disappearing. + */ + private List readPks(RowType rowType, Predicate predicate) throws IOException { + List filters = new ArrayList<>(); + filters.add(predicate); + List pks = new ArrayList<>(); + try (RecordReader reader = + fileFormat() + .createReaderFactory(rowType, rowType, filters) + .createReader( + new FormatReaderContext( + fileIO, file, fileIO.getFileSize(file), null, null))) { + RecordReader.RecordIterator batch; + while ((batch = reader.readBatch()) != null) { + InternalRow row; + while ((row = batch.next()) != null) { + pks.add(row.getLong(0)); + } + batch.releaseBatch(); + } + } + return pks; + } + + private void writeTwoPayloadRows(RowType rowType) throws IOException { + Decimal match = Decimal.fromBigDecimal(new BigDecimal("12.34"), 10, 2); + Decimal other = Decimal.fromBigDecimal(new BigDecimal("99.99"), 10, 2); + org.apache.paimon.data.Timestamp early = + org.apache.paimon.data.Timestamp.fromEpochMillis(1704067200000L); + org.apache.paimon.data.Timestamp late = + org.apache.paimon.data.Timestamp.fromEpochMillis(1704067200000L + 60_000L); + write( + fileFormat().createWriterFactory(rowType), + file, + GenericRow.of(1L, GenericRow.of(match, early, early, 7L)), + GenericRow.of(2L, GenericRow.of(other, late, late, 8L))); + } + + private NestedFieldTransform payloadLeaf(RowType rowType, String leaf) { + RowType payload = (RowType) rowType.getTypeAt(1); + return new NestedFieldTransform( + new FieldRef(1, "payload", payload), Collections.singletonList(leaf)); + } + + /** Control: a nested BIGINT predicate already carried its full path and kept its row. */ + @Test + public void testNestedBigIntPredicateKeepsMatchingRows() throws IOException { + RowType rowType = nestedPayloadType(); + writeTwoPayloadRows(rowType); + Predicate onQty = new PredicateBuilder(rowType).equal(payloadLeaf(rowType, "qty"), 7L); + Assertions.assertThat(readPks(rowType, onQty)) + .as("control: the row whose payload.qty equals 7 must survive the filter") + .contains(1L); + } + + /** + * Reading with a predicate on a nested DECIMAL must still return the matching row. The column + * handed to parquet-mr used to carry only the leaf name, so parquet-mr saw a missing top-level + * column, treated it as all-null and pruned the row group holding the match. + */ + @Test + public void testNestedDecimalPredicateKeepsMatchingRows() throws IOException { + RowType rowType = nestedPayloadType(); + writeTwoPayloadRows(rowType); + Decimal match = Decimal.fromBigDecimal(new BigDecimal("12.34"), 10, 2); + Predicate onAmount = + new PredicateBuilder(rowType).equal(payloadLeaf(rowType, "amount"), match); + Assertions.assertThat(readPks(rowType, onAmount)) + .as("the row whose payload.amount equals the literal must survive the filter") + .contains(1L); + } + + /** Same as {@link #testNestedDecimalPredicateKeepsMatchingRows()} for LOCAL ZONED TIMESTAMP. */ + @Test + public void testNestedLocalZonedTimestampPredicateKeepsMatchingRows() throws IOException { + RowType rowType = nestedPayloadType(); + writeTwoPayloadRows(rowType); + org.apache.paimon.data.Timestamp early = + org.apache.paimon.data.Timestamp.fromEpochMillis(1704067200000L); + Predicate onLtz = new PredicateBuilder(rowType).equal(payloadLeaf(rowType, "ltz"), early); + Assertions.assertThat(readPks(rowType, onLtz)) + .as("the row whose payload.ltz equals the literal must survive the filter") + .contains(1L); + } + + /** Same as {@link #testNestedDecimalPredicateKeepsMatchingRows()} for TIMESTAMP. */ + @Test + public void testNestedTimestampPredicateKeepsMatchingRows() throws IOException { + RowType rowType = nestedPayloadType(); + writeTwoPayloadRows(rowType); + org.apache.paimon.data.Timestamp early = + org.apache.paimon.data.Timestamp.fromEpochMillis(1704067200000L); + Predicate onTs = new PredicateBuilder(rowType).equal(payloadLeaf(rowType, "ts"), early); + Assertions.assertThat(readPks(rowType, onTs)) + .as("the row whose payload.ts equals the literal must survive the filter") + .contains(1L); + } + + /** + * A nested component whose own name contains a dot cannot be addressed by a dot-joined path. + * Resolution fails, the filter is built against a path the file does not hold, and every row + * group is pruned — so the matching row disappears. + */ + @Test + public void testNestedComponentContainingADotKeepsMatchingRows() throws IOException { + RowType inner = RowType.of(new DataType[] {DataTypes.BIGINT()}, new String[] {"a.b"}); + RowType rowType = + RowType.of(new DataType[] {DataTypes.BIGINT(), inner}, new String[] {"pk", "s"}); + + write( + fileFormat().createWriterFactory(rowType), + file, + GenericRow.of(1L, GenericRow.of(7L)), + GenericRow.of(2L, GenericRow.of(8L))); + + Predicate onDotted = + new PredicateBuilder(rowType) + .equal( + new NestedFieldTransform( + new FieldRef(1, "s", inner), + Collections.singletonList("a.b")), + 7L); + + Assertions.assertThat(readPks(rowType, onDotted)) + .as("the row whose s.`a.b` equals 7 must survive the filter") + .contains(1L); + } + + /** The dot may sit in the top-level column's own name; the matching row must still survive. */ + @Test + public void testTopLevelNameContainingADotKeepsMatchingRows() throws IOException { + RowType inner = RowType.of(new DataType[] {DataTypes.BIGINT()}, new String[] {"city"}); + RowType rowType = + RowType.of(new DataType[] {DataTypes.BIGINT(), inner}, new String[] {"pk", "a.b"}); + + write( + fileFormat().createWriterFactory(rowType), + file, + GenericRow.of(1L, GenericRow.of(7L)), + GenericRow.of(2L, GenericRow.of(8L))); + + Predicate onNested = + new PredicateBuilder(rowType) + .equal( + new NestedFieldTransform( + new FieldRef(1, "a.b", inner), + Collections.singletonList("city")), + 7L); + + Assertions.assertThat(readPks(rowType, onNested)) + .as("the row whose `a.b`.city equals 7 must survive the filter") + .contains(1L); + } } 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..9d4fae2396d6 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,43 @@ 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)) + } + + // Keep the components Spark gave us: they are the transform's identity, and joining them + // would lose the boundaries of a name that itself contains a dot. + val path = new java.util.ArrayList[String](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(part) + 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 }