This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 406a118fb0 [core][format][spark] Support nested field predicate
pushdown (#9423)
406a118fb0 is described below
commit 406a118fb08aee2fc659472094c0f6a6ff85de55
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Mon Sep 21 11:45:24 2026 +0800
[core][format][spark] Support nested field predicate pushdown (#9423)
---
.../paimon/predicate/NestedFieldTransform.java | 223 +++++++++++
.../apache/paimon/predicate/PredicateBuilder.java | 8 +-
.../org/apache/paimon/predicate/Transform.java | 1 +
.../paimon/predicate/NestedFieldTransformTest.java | 372 ++++++++++++++++++
.../paimon/predicate/PredicateBuilderTest.java | 31 ++
.../paimon/catalog/TableQueryAuthResultTest.java | 57 +++
.../paimon/table/NestedFieldQueryAuthTest.java | 157 ++++++++
.../parquet/filter2/predicate/ParquetFilters.java | 203 ++++++++--
.../paimon/format/parquet/ParquetFiltersTest.java | 434 +++++++++++++++++++++
.../format/parquet/ParquetFormatReadWriteTest.java | 334 ++++++++++++++++
.../spark/util/SparkExpressionConverter.scala | 39 +-
.../spark/sql/SparkV2FilterConverterTestBase.scala | 59 ++-
12 files changed, 1884 insertions(+), 34 deletions(-)
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 0000000000..7bbbac7bf9
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/predicate/NestedFieldTransform.java
@@ -0,0 +1,223 @@
+/*
+ * 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.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.RowType;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+import static org.apache.paimon.utils.InternalRowUtils.get;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * Transform that extracts a field nested inside a row-typed column, for
example {@code addr.city}.
+ *
+ * <p>The transform keeps the enclosing top-level column as its only {@link
#inputs() input}, so
+ * anything that rewrites field indices (schema projection, for instance)
keeps working without
+ * knowing about nesting. The positions below that column are held separately
in {@link #path()}.
+ *
+ * <p>Deliberately <b>not</b> a {@link FieldTransform}: {@link
LeafPredicate#fieldRefOptional()}
+ * returns empty for it, which is what keeps every consumer that equates a
leaf with a top-level
+ * column — min/max pruning, file index lookup, ORC pushdown, schema evolution
— from silently
+ * reading the enclosing column's metadata as if it belonged to the nested
field. Those consumers
+ * give up on this transform instead, which costs pruning but never rows.
+ */
+public class NestedFieldTransform implements Transform {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final String NAME = "NESTED_FIELD_REF";
+
+ public static final String FIELD_FIELD_REF = "fieldRef";
+ public static final String FIELD_PATH = "path";
+
+ /** The top-level row-typed column the nested field lives in. */
+ private final FieldRef fieldRef;
+
+ /**
+ * Names of the fields to descend into, relative to {@code fieldRef}'s row
type. Never empty.
+ *
+ * <p>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<String> path;
+
+ /** {@link #path} resolved to positions against {@code fieldRef}'s row
type. */
+ private final int[] positions;
+
+ /** Stable field ids of every component in {@link #path}. */
+ private final int[] fieldIds;
+
+ private final String name;
+ private final DataType outputType;
+
+ @JsonCreator
+ public NestedFieldTransform(
+ @JsonProperty(FIELD_FIELD_REF) FieldRef fieldRef,
+ @JsonProperty(FIELD_PATH) List<String> path) {
+ this(fieldRef, path, null);
+ }
+
+ private NestedFieldTransform(FieldRef fieldRef, List<String> path, int[]
expectedFieldIds) {
+ checkArgument(path != null && !path.isEmpty(), "Nested field path must
not be empty.");
+ checkArgument(
+ expectedFieldIds == null || expectedFieldIds.length ==
path.size(),
+ "Nested field path and field ids must have the same size.");
+ this.fieldRef = fieldRef;
+ this.path = Collections.unmodifiableList(new ArrayList<>(path));
+ this.positions = new int[this.path.size()];
+ this.fieldIds = 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);
+ DataField field = rowType.getFields().get(position);
+ if (expectedFieldIds != null) {
+ checkArgument(
+ field.id() == expectedFieldIds[i],
+ "Nested field '%s.%s' changed identity from field id
%s to %s.",
+ nameBuilder,
+ component,
+ expectedFieldIds[i],
+ field.id());
+ }
+ positions[i] = position;
+ fieldIds[i] = field.id();
+ nameBuilder.append('.').append(component);
+ current = field.type();
+ }
+ 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<String> path() {
+ return path;
+ }
+
+ /** Dot-separated name from the top-level column down to the nested field,
{@code addr.city}. */
+ @JsonIgnore
+ public String fieldName() {
+ return name;
+ }
+
+ @Override
+ @JsonIgnore
+ public List<Object> inputs() {
+ return Collections.singletonList(fieldRef);
+ }
+
+ @Override
+ @JsonIgnore
+ public DataType outputType() {
+ return outputType;
+ }
+
+ /**
+ * Reads the nested field out of {@code row}, which must match the row
type {@link #fieldRef}
+ * was built against. A null anywhere along the path yields null, matching
SQL semantics for
+ * field access on a null struct.
+ */
+ @Override
+ public Object transform(InternalRow row) {
+ int position = fieldRef.index();
+ if (row.isNullAt(position)) {
+ return null;
+ }
+ RowType currentType = (RowType) fieldRef.type();
+ InternalRow current = row.getRow(position,
currentType.getFieldCount());
+
+ for (int i = 0; i < 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<Object> inputs) {
+ checkArgument(inputs.size() == 1);
+ return new NestedFieldTransform((FieldRef) inputs.get(0), path,
fieldIds);
+ }
+
+ private Object readResolve() {
+ return fieldIds == null ? new NestedFieldTransform(fieldRef, path) :
this;
+ }
+
+ @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 f576e48fc1..cd358c8620 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
@@ -271,7 +271,9 @@ public class PredicateBuilder {
public Predicate in(Transform transform, List<Object> literals) {
// In the IN predicate, 20 literals are critical for performance.
// If there are more than 20 literals, the performance will decrease.
- if (literals.size() > 20) {
+ // An empty list has no equals to OR together, so it must also take
this branch - mirroring
+ // in(int, List) - rather than fall into or(emptyList()), which throws.
+ if (literals.size() > 20 || literals.isEmpty()) {
return LeafPredicate.of(transform, In.INSTANCE, literals);
}
@@ -286,6 +288,10 @@ public class PredicateBuilder {
return in(idx, literals).negate().get();
}
+ public Predicate notIn(Transform transform, List<Object> literals) {
+ return in(transform, literals).negate().get();
+ }
+
public Predicate between(int idx, Object includedLowerBound, Object
includedUpperBound) {
DataField field = rowType.getFields().get(idx);
return new LeafPredicate(
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 f0aefe1b2d..d0178cb398 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 @@ import java.util.List;
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 0000000000..12b93e66af
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/predicate/NestedFieldTransformTest.java
@@ -0,0 +1,372 @@
+/*
+ * 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.InstantiationUtil;
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Base64;
+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 {
+
+ // Serialized by the NestedFieldTransform implementation at PR head
e55ed4a, before fieldIds
+ // existed. Keep this fixture to verify Java serialization compatibility
across the change.
+ private static final String LEGACY_SERIALIZED_TRANSFORM =
+
"rO0ABXNyADBvcmcuYXBhY2hlLnBhaW1vbi5wcmVkaWNhdGUuTmVzdGVkRmllbGRUcmFuc2Zvcm0AAAAAAAAAAQIABUwACGZpZWxk"
+ +
"UmVmdAAmTG9yZy9hcGFjaGUvcGFpbW9uL3ByZWRpY2F0ZS9GaWVsZFJlZjtMAARuYW1ldAASTGphdmEvbGFuZy9TdHJpbmc7TAAK"
+ +
"b3V0cHV0VHlwZXQAIkxvcmcvYXBhY2hlL3BhaW1vbi90eXBlcy9EYXRhVHlwZTtMAARwYXRodAAQTGphdmEvdXRpbC9MaXN0O1sA"
+ +
"CXBvc2l0aW9uc3QAAltJeHBzcgAkb3JnLmFwYWNoZS5wYWltb24ucHJlZGljYXRlLkZpZWxkUmVmAAAAAAAAAAECAANJAAVpbmRl"
+ +
"eEwABG5hbWVxAH4AAkwABHR5cGVxAH4AA3hwAAAAAHQABGluZm9zcgAfb3JnLmFwYWNoZS5wYWltb24udHlwZXMuUm93VHlwZQAA"
+ +
"AAAAAAABAgABTAAGZmllbGRzcQB+AAR4cgAgb3JnLmFwYWNoZS5wYWltb24udHlwZXMuRGF0YVR5cGUAAAAAAAAAAQIAAloACmlz"
+ +
"TnVsbGFibGVMAAh0eXBlUm9vdHQAJkxvcmcvYXBhY2hlL3BhaW1vbi90eXBlcy9EYXRhVHlwZVJvb3Q7eHABfnIAJG9yZy5hcGFj"
+ +
"aGUucGFpbW9uLnR5cGVzLkRhdGFUeXBlUm9vdAAAAAAAAAAAEgAAeHIADmphdmEubGFuZy5FbnVtAAAAAAAAAAASAAB4cHQAA1JP"
+ +
"V3NyACZqYXZhLnV0aWwuQ29sbGVjdGlvbnMkVW5tb2RpZmlhYmxlTGlzdPwPJTG17I4QAgABTAAEbGlzdHEAfgAEeHIALGphdmEu"
+ +
"dXRpbC5Db2xsZWN0aW9ucyRVbm1vZGlmaWFibGVDb2xsZWN0aW9uGUIAgMte9x4CAAFMAAFjdAAWTGphdmEvdXRpbC9Db2xsZWN0"
+ +
"aW9uO3hwc3IAE2phdmEudXRpbC5BcnJheUxpc3R4gdIdmcdhnQMAAUkABHNpemV4cAAAAAJ3BAAAAAJzcgAhb3JnLmFwYWNoZS5w"
+ +
"YWltb24udHlwZXMuRGF0YUZpZWxkAAAAAAAAAAECAAVJAAJpZEwADGRlZmF1bHRWYWx1ZXEAfgACTAALZGVzY3JpcHRpb25xAH4A"
+ +
"AkwABG5hbWVxAH4AAkwABHR5cGVxAH4AA3hwAAAAAnBwdAAGc2VjcmV0c3IAI29yZy5hcGFjaGUucGFpbW9uLnR5cGVzLlZhckNo"
+ +
"YXJUeXBlAAAAAAAAAAECAAFJAAZsZW5ndGh4cQB+AAsBfnEAfgAOdAAHVkFSQ0hBUn////9zcQB+ABgAAAADcHB0AAZyZWdpb25x"
+ +
"AH4AHHhxAH4AF3QAC2luZm8uc2VjcmV0cQB+ABxzcQB+ABJzcQB+ABYAAAABdwQAAAABcQB+ABp4cQB+ACN1cgACW0lNumAmduqy"
+ + "pQIAAHhwAAAAAQAAAAA=";
+
+ // user STRUCT<id BIGINT, addr STRUCT<city STRING, zip STRING>>
+ private static final RowType ADDR_TYPE =
+ RowType.of(
+ new org.apache.paimon.types.DataType[]
{DataTypes.STRING(), DataTypes.STRING()},
+ new String[] {"city", "zip"});
+ private static final RowType USER_TYPE =
+ RowType.of(
+ new org.apache.paimon.types.DataType[]
{DataTypes.BIGINT(), ADDR_TYPE},
+ new String[] {"id", "addr"});
+ private static final RowType ROW_TYPE =
+ RowType.of(
+ new org.apache.paimon.types.DataType[] {DataTypes.INT(),
USER_TYPE},
+ new String[] {"pk", "user"});
+
+ private static final FieldRef USER_REF = new FieldRef(1, "user",
USER_TYPE);
+
+ private static GenericRow row(Object user) {
+ return GenericRow.of(1, user);
+ }
+
+ @Test
+ public void testReadOneLevel() {
+ NestedFieldTransform transform =
+ new NestedFieldTransform(USER_REF,
Collections.singletonList("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<Predicate> projected =
+
predicate.visit(PredicateProjectionConverter.fromProjection(new int[] {1}));
+
+ assertThat(projected).isPresent();
+ NestedFieldTransform transform =
+ (NestedFieldTransform) ((LeafPredicate)
projected.get()).transform();
+ assertThat(transform.fieldRef().index()).isEqualTo(0);
+ assertThat(transform.path()).containsExactly("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 testLegacyJavaRoundTripRestoresNestedIdentity() throws
Exception {
+ NestedFieldTransform legacy =
+ InstantiationUtil.deserializeObject(
+
Base64.getDecoder().decode(LEGACY_SERIALIZED_TRANSFORM),
+ getClass().getClassLoader());
+ assertThat(legacy.fieldName()).isEqualTo("info.secret");
+ assertThat(
+ legacy.transform(
+ GenericRow.of(
+ GenericRow.of(
+ BinaryString.fromString("x"),
+
BinaryString.fromString("US")))))
+ .isEqualTo(BinaryString.fromString("x"));
+
+ RowType reAdded =
+ RowType.of(
+ new org.apache.paimon.types.DataField(7, "secret",
DataTypes.STRING()),
+ new org.apache.paimon.types.DataField(3, "region",
DataTypes.STRING()));
+ assertThatThrownBy(
+ () ->
+ legacy.copyWithNewInputs(
+ Collections.singletonList(
+ new FieldRef(0, "info",
reAdded))))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("changed identity")
+ .hasMessageContaining("2")
+ .hasMessageContaining("7");
+ }
+
+ @Test
+ public void testJavaRoundTripKeepsNestedIdentity() throws Exception {
+ NestedFieldTransform original =
+ new NestedFieldTransform(USER_REF, Arrays.asList("addr",
"city"));
+ NestedFieldTransform copy =
+ InstantiationUtil.deserializeObject(
+ InstantiationUtil.serializeObject(original),
getClass().getClassLoader());
+ RowType reAddedAddr =
+ RowType.of(
+ new org.apache.paimon.types.DataField(7, "city",
DataTypes.STRING()),
+ new org.apache.paimon.types.DataField(1, "zip",
DataTypes.STRING()));
+ RowType reAddedUser =
+ RowType.of(
+ new org.apache.paimon.types.DataField(0, "id",
DataTypes.BIGINT()),
+ new org.apache.paimon.types.DataField(1, "addr",
reAddedAddr));
+
+ assertThatThrownBy(
+ () ->
+ copy.copyWithNewInputs(
+ Collections.singletonList(
+ new FieldRef(1, "user",
reAddedUser))))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("changed identity");
+ }
+
+ @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);
+ }
+
+ @Test
+ public void testRemapRejectsReAddedFieldOfSameName() {
+ RowType original =
+ RowType.of(
+ new org.apache.paimon.types.DataField(2, "secret",
DataTypes.STRING()),
+ new org.apache.paimon.types.DataField(3, "region",
DataTypes.STRING()));
+ NestedFieldTransform transform =
+ new NestedFieldTransform(
+ new FieldRef(0, "info", original),
Collections.singletonList("secret"));
+ RowType reAdded =
+ RowType.of(
+ new org.apache.paimon.types.DataField(7, "secret",
DataTypes.STRING()),
+ new org.apache.paimon.types.DataField(3, "region",
DataTypes.STRING()));
+
+ assertThatThrownBy(
+ () ->
+ transform.copyWithNewInputs(
+ Collections.singletonList(
+ new FieldRef(0, "info",
reAdded))))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("changed identity")
+ .hasMessageContaining("2")
+ .hasMessageContaining("7");
+ }
+
+ /** 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.DataField(2, "secret",
DataTypes.STRING()),
+ new org.apache.paimon.types.DataField(3, "region",
DataTypes.STRING()));
+ NestedFieldTransform onSecret =
+ new NestedFieldTransform(
+ new FieldRef(0, "info", full),
Collections.singletonList("secret"));
+
+ RowType reordered =
+ RowType.of(
+ new org.apache.paimon.types.DataField(3, "region",
DataTypes.STRING()),
+ new org.apache.paimon.types.DataField(2, "secret",
DataTypes.STRING()));
+ Transform remapped =
+ onSecret.copyWithNewInputs(
+ Collections.singletonList(new FieldRef(0, "info",
reordered)));
+
+ assertThat(((NestedFieldTransform)
remapped).fieldName()).isEqualTo("info.secret");
+ }
+}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateBuilderTest.java
b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateBuilderTest.java
index 5ee46d640d..eb23dc3b27 100644
---
a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateBuilderTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateBuilderTest.java
@@ -223,6 +223,37 @@ public class PredicateBuilderTest {
assertThat(predicate.test(GenericRow.of(10))).isEqualTo(false);
}
+ @Test
+ public void testNotIn() {
+ PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
+ Predicate predicate = builder.notIn(0, new ArrayList<>());
+ assertThat(predicate.test(GenericRow.of(1))).isEqualTo(true);
+ predicate = builder.notIn(0, Arrays.asList(1, 2));
+ assertThat(predicate.test(GenericRow.of(1))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(10))).isEqualTo(true);
+ }
+
+ /**
+ * {@link #testIn()} shows {@code in(idx, emptyList())} evaluates to
always-false rather than
+ * throwing - {@link PredicateBuilder#in(int, List)} special-cases an
empty list. The {@link
+ * Transform} overload has no such special case: for an empty list it
falls through to {@code
+ * or(equals)} with an empty {@code equals}, which throws. {@code notIn}
inherits this from
+ * {@code in} through {@code negate()}, so it throws too, unlike {@link
#testNotIn()}'s {@code
+ * idx} counterpart.
+ */
+ @Test
+ public void testInAndNotInTransformWithEmptyLiteralsMatchTheIdxOverloads()
{
+ RowType rowType = RowType.of(new IntType());
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+ FieldTransform transform = new FieldTransform(new FieldRef(0, "f0",
new IntType()));
+
+ Predicate in = builder.in(transform, new ArrayList<>());
+ assertThat(in.test(GenericRow.of(1))).isEqualTo(false);
+
+ Predicate notIn = builder.notIn(transform, new ArrayList<>());
+ assertThat(notIn.test(GenericRow.of(1))).isEqualTo(true);
+ }
+
@Test
public void testArrayContains() {
PredicateBuilder builder =
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 6ff67de361..241c4377cc 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
@@ -25,7 +25,9 @@ import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.FieldTransform;
import org.apache.paimon.predicate.LeafPredicate;
import org.apache.paimon.predicate.LowerTransform;
+import org.apache.paimon.predicate.NestedFieldTransform;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.predicate.UpperTransform;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
@@ -247,4 +249,59 @@ public class TableQueryAuthResultTest {
Equal.INSTANCE,
Collections.singletonList(BinaryString.fromString("y"))));
}
+
+ private static RowType infoRowType(String... nestedFields) {
+ org.apache.paimon.types.DataField[] fields =
+ new org.apache.paimon.types.DataField[nestedFields.length];
+ for (int i = 0; i < fields.length; i++) {
+ int id;
+ if ("secret".equals(nestedFields[i])) {
+ id = 2;
+ } else if ("region".equals(nestedFields[i])) {
+ id = 3;
+ } else {
+ id = i + 2;
+ }
+ fields[i] =
+ new org.apache.paimon.types.DataField(id, nestedFields[i],
DataTypes.STRING());
+ }
+ return RowType.of(
+ new org.apache.paimon.types.DataField(0, "pk",
DataTypes.INT()),
+ new org.apache.paimon.types.DataField(1, "info",
RowType.of(fields)));
+ }
+
+ 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-core/src/test/java/org/apache/paimon/table/NestedFieldQueryAuthTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/NestedFieldQueryAuthTest.java
new file mode 100644
index 0000000000..c64fbe71fa
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/NestedFieldQueryAuthTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.table;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.catalog.TableQueryAuthResult;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+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.schema.FileSystemSchemaManager;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.types.DataField;
+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 org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests nested-field identity in query authorization across schema changes.
*/
+class NestedFieldQueryAuthTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testHistoricalReadRejectsReAddedNestedFieldOfSameName() throws
Exception {
+ Path path = new Path(tempDir.toUri());
+ LocalFileIO fileIO = LocalFileIO.create();
+ FileSystemSchemaManager schemaManager = new
FileSystemSchemaManager(fileIO, path);
+ RowType initialType =
+ RowType.of(
+ new DataField(0, "pk", DataTypes.INT()),
+ new DataField(
+ 1,
+ "info",
+ RowType.of(
+ new DataField(2, "secret",
DataTypes.STRING()),
+ new DataField(3, "region",
DataTypes.STRING()))));
+ schemaManager.createTable(
+ new Schema(
+ initialType.getFields(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.singletonMap("query-auth.enabled", "true"),
+ ""));
+ TableSchema historicalSchema = schemaManager.latest().get();
+ writeInitialRow(FileStoreTableFactory.create(fileIO, path,
historicalSchema));
+
+ schemaManager.commitChanges(SchemaChange.dropColumn(new String[]
{"info", "secret"}));
+ schemaManager.commitChanges(
+ SchemaChange.addColumn(
+ new String[] {"info", "secret"}, DataTypes.STRING(),
null, null));
+ TableSchema latestSchema = schemaManager.latest().get();
+ TableQueryAuthResult authResult =
authResult(latestSchema.logicalRowType());
+ CatalogEnvironment environment = authEnvironment(authResult);
+
+ assertThat(readKeys(FileStoreTableFactory.create(fileIO, path,
latestSchema, environment)))
+ .isEmpty();
+
+ FileStoreTable historicalTable =
+ FileStoreTableFactory.create(fileIO, path, historicalSchema,
environment);
+ assertThatThrownBy(() ->
historicalTable.newReadBuilder().newScan().plan())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("changed identity")
+ .hasMessageContaining("2")
+ .hasMessageContaining("4");
+ }
+
+ private void writeInitialRow(FileStoreTable table) throws Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(
+ GenericRow.of(
+ 1,
+ GenericRow.of(
+ BinaryString.fromString("x"),
BinaryString.fromString("US"))));
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private TableQueryAuthResult authResult(RowType rowType) {
+ RowType info = (RowType) rowType.getField("info").type();
+ Predicate filter =
+ new PredicateBuilder(rowType)
+ .equal(
+ new NestedFieldTransform(
+ new FieldRef(1, "info", info),
+ Collections.singletonList("secret")),
+ BinaryString.fromString("x"));
+ return new TableQueryAuthResult(
+ Collections.singletonList(JsonSerdeUtil.toFlatJson(filter)),
null);
+ }
+
+ private CatalogEnvironment authEnvironment(TableQueryAuthResult
authResult) throws Exception {
+ Catalog catalog = Mockito.mock(Catalog.class);
+ Mockito.when(catalog.authTableQuery(Mockito.any(),
Mockito.any())).thenReturn(authResult);
+ Mockito.when(catalog.loadSnapshot(Mockito.any(Identifier.class)))
+ .thenThrow(new UnsupportedOperationException());
+ return new CatalogEnvironment(
+ Identifier.create("default", "t"),
+ null,
+ () -> catalog,
+ null,
+ null,
+ null,
+ false,
+ false);
+ }
+
+ private List<Integer> readKeys(FileStoreTable table) throws Exception {
+ ReadBuilder builder = table.newReadBuilder();
+ List<Integer> keys = new ArrayList<>();
+ try (RecordReader<org.apache.paimon.data.InternalRow> reader =
+ builder.newRead().createReader(builder.newScan().plan())) {
+ reader.forEachRemaining(row -> keys.add(row.getInt(0)));
+ }
+ return keys;
+ }
+}
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 57d27578a1..c5d369fb30 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.format.parquet.ParquetSchemaConverter;
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.compat.FilterCompat;
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 @@ import java.util.Set;
/** 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(
@@ -235,6 +240,13 @@ public class ParquetFilters {
@Override
public FilterPredicate visitIn(FieldRef fieldRef, List<Object>
literals) {
+ if (literals.isEmpty()) {
+ // An IN predicate builder can legitimately produce an empty
(always-false) leaf,
+ // but parquet-mr's SetColumnFilterPredicate refuses an empty
set outright. Leave
+ // this predicate out of the pushdown; residual evaluation
upstream still applies
+ // the always-false semantics correctly.
+ throw new UnsupportedOperationException();
+ }
Operators.Column<?> column = toParquetColumn(fieldRef);
if (column instanceof Operators.LongColumn) {
return FilterApi.in(
@@ -262,6 +274,11 @@ public class ParquetFilters {
@Override
public FilterPredicate visitNotIn(FieldRef fieldRef, List<Object>
literals) {
+ if (literals.isEmpty()) {
+ // Same as visitIn: an empty NOT IN is a legitimate
always-true leaf, but parquet-mr
+ // refuses an empty set. Leave it out of the pushdown rather
than crash the read.
+ throw new UnsupportedOperationException();
+ }
Operators.Column<?> column = toParquetColumn(fieldRef);
if (column instanceof Operators.LongColumn) {
return FilterApi.notIn(
@@ -287,9 +304,41 @@ public class ParquetFilters {
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();
+ }
+ }
+ // Even with no dot inside any component, the joined path can
still equal the literal
+ // name of an unrelated top-level column (both a top-level "s.a"
and a nested s -> a
+ // are valid siblings). findFileColumn resolves that joined name
against the file by
+ // exact top-level match first, so it would pick that unrelated
column's type - column
+ // identity itself round-trips back to the right path (parquet-mr
re-splits any
+ // dot-joined name it is given), but the type mismatch fails the
read outright. Give up
+ // the pushdown rather than risk it whenever the file actually has
such a column.
+ if (findChild(fileSchema, nested.fieldName(), caseSensitive) !=
null) {
+ throw new UnsupportedOperationException();
+ }
+ FieldRef pathRef = new FieldRef(UNUSED_INDEX, nested.fieldName(),
nested.outputType());
+ return predicate.function().visit(this, pathRef,
predicate.literals());
}
private <T> Set<T> convertSets(List<Object> values, Class<T> kclass,
FieldRef fieldRef) {
@@ -320,7 +369,7 @@ public class ParquetFilters {
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);
@@ -341,7 +390,7 @@ public class ParquetFilters {
if (value instanceof Timestamp) {
Timestamp timestamp = (Timestamp) value;
- timestampPrimitiveType(fieldRef, fileSchema, caseSensitive);
+ timestampColumn(fieldRef, fileSchema, caseSensitive);
int precision = getTimestampPrecision(type);
if (precision <= 3) {
// milliseconds
@@ -492,9 +541,10 @@ public class ParquetFilters {
}
}
- 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();
@@ -505,19 +555,20 @@ public class ParquetFilters {
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();
@@ -534,35 +585,123 @@ public class ParquetFilters {
|| 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.
+ *
+ * <p>A {@code fieldRef} synthesized from a {@link NestedFieldTransform}
(its index left at
+ * {@link #UNUSED_INDEX}) names a nested field with dots ({@code
addr.city}), 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.
+ *
+ * <p>An ordinary {@code fieldRef} - one naming an actual top-level field,
index not {@link
+ * #UNUSED_INDEX} - never resolves through that walk: a Format Table's
declared schema need not
+ * match what a given file holds, so a dotted top-level name with no exact
match is a genuinely
+ * missing column, not license to reinterpret it as a path into an
unrelated group. But the walk
+ * still runs as a collision check: whatever name is ultimately handed to
{@code FilterApi} gets
+ * re-split by parquet-mr the same way, so if the walk would find a real
column there, using the
+ * joined name is not safe either - the pushdown is refused outright
rather than risking it.
*/
@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;
+ }
+
+ FileColumn walked = walkGroups(fileSchema, parts, caseSensitive);
+ if (fieldRef.index() == UNUSED_INDEX) {
+ return walked;
+ }
+ if (walked != null) {
+ throw new UnsupportedOperationException();
+ }
+ return null;
+ }
+
+ /** Descends {@code fileSchema} through {@code parts}, or null if any
segment is missing. */
+ @Nullable
+ private static FileColumn walkGroups(
+ MessageType fileSchema, String[] parts, boolean caseSensitive) {
+ 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;
@@ -585,10 +724,11 @@ public class ParquetFilters {
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);
@@ -602,7 +742,7 @@ public class ParquetFilters {
for (PrimitiveType.PrimitiveTypeName candidate : acceptable) {
if (fileType.getPrimitiveTypeName() == candidate) {
- return new PushdownTarget(fileType.getName(), candidate);
+ return new PushdownTarget(fileColumn.path, candidate);
}
}
throw new UnsupportedOperationException();
@@ -828,15 +968,16 @@ public class ParquetFilters {
@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();
}
@@ -847,7 +988,7 @@ public class ParquetFilters {
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();
@@ -858,7 +999,7 @@ public class ParquetFilters {
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 b1119bc345..c58f7dec01 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
@@ -21,14 +21,19 @@ 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;
+import org.apache.paimon.types.IntType;
import org.apache.paimon.types.LocalZonedTimestampType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.TimestampType;
@@ -1200,6 +1205,435 @@ class ParquetFiltersTest {
}
}
+ //
---------------------------------------------------------------------------------------
+ // 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<PrimitiveType> 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);
+ }
+
+ /**
+ * No component contains a dot here, but the joined path still equals the
literal name of an
+ * unrelated top-level sibling column. findFileColumn resolves a nested
predicate's joined name
+ * against the file by exact top-level match before walking components, so
it would bind this
+ * predicate to that unrelated column's physical type. Refuse the pushdown
instead of risking
+ * it.
+ */
+ @Test
+ public void
testNestedFieldCollidingWithADottedTopLevelSiblingIsNotPushedDown() {
+ RowType inner = RowType.of(new DataType[] {new BigIntType()}, new
String[] {"a"});
+ RowType rowType =
+ RowType.of(
+ new DataType[] {new BigIntType(), new IntType(),
inner},
+ new String[] {"pk", "s.a", "s"});
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+
+ // the file holds both a top-level "s.a" and a nested s -> a; the
joined name collides
+ assertThat(schema.containsField("s.a")).isTrue();
+
assertThat(schema.getType("s").asGroupType().containsField("a")).isTrue();
+
+ NestedFieldTransform nested =
+ new NestedFieldTransform(
+ new FieldRef(2, "s", inner),
Collections.singletonList("a"));
+ test(schema, new PredicateBuilder(rowType).equal(nested, 7L), (String)
null, false);
+ }
+
+ /**
+ * {@link PredicateBuilder#in(Transform, List)} on an empty list now
builds a valid leaf instead
+ * of throwing, but {@code FilterApi}'s set predicates refuse an empty
set. {@code visitIn} and
+ * {@code visitNotIn} must decline the pushdown for an empty literal list
rather than pass an
+ * empty set through to parquet-mr.
+ */
+ @Test
+ public void testNestedInAndNotInWithEmptyLiteralsAreNotPushedDown() {
+ RowType inner = RowType.of(new DataType[] {new BigIntType()}, new
String[] {"a"});
+ RowType rowType =
+ RowType.of(new DataType[] {new BigIntType(), inner}, new
String[] {"pk", "s"});
+ MessageType schema =
ParquetSchemaConverter.convertToParquetMessageType(rowType);
+
+ NestedFieldTransform nested =
+ new NestedFieldTransform(
+ new FieldRef(1, "s", inner),
Collections.singletonList("a"));
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+
+ test(schema, builder.in(nested, Collections.emptyList()), (String)
null, false);
+ test(schema, builder.notIn(nested, Collections.emptyList()), (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 eb393be3c6..632d82c4f2 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,13 @@ import org.apache.paimon.format.SupportsFieldMetadata;
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.predicate.Transform;
import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
@@ -52,8 +59,11 @@ import org.junit.jupiter.api.Test;
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;
@@ -333,4 +343,328 @@ public class ParquetFormatReadWriteTest extends
FormatReadWriteTest {
.containsExactly(GenericRow.of(1.25f, 2.5d),
GenericRow.of(3.75f, 5.0d));
}
}
+
+ //
-----------------------------------------------------------------------------------------
+ // 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<Long> readPks(RowType rowType, Predicate predicate) throws
IOException {
+ List<Predicate> filters = new ArrayList<>();
+ filters.add(predicate);
+ List<Long> pks = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ fileFormat()
+ .createReaderFactory(rowType, rowType, filters)
+ .createReader(
+ new FormatReaderContext(
+ fileIO, file,
fileIO.getFileSize(file), null, null))) {
+ RecordReader.RecordIterator<InternalRow> 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);
+ }
+
+ /**
+ * {@link PredicateBuilder#in(Transform, List)} on an empty literal list
now builds a valid
+ * (always-false) leaf instead of throwing, but {@code FilterApi}'s set
predicates refuse an
+ * empty set outright, so {@link
org.apache.parquet.filter2.predicate.ParquetFilters}'s {@code
+ * visitIn} now declines the pushdown for an empty literal list instead of
handing parquet-mr a
+ * set it will reject. Declining pushdown means this reader layer returns
every row unfiltered;
+ * the always-false semantics of an empty IN are enforced by residual
predicate evaluation
+ * upstream, not by parquet-level pruning, so they are out of scope for
this test - what matters
+ * here is that opening the reader does not throw.
+ */
+ @Test
+ public void testNestedInWithEmptyLiteralsDoesNotCrashTheReader() throws
IOException {
+ RowType rowType = nestedPayloadType();
+ writeTwoPayloadRows(rowType);
+ Predicate emptyIn =
+ new PredicateBuilder(rowType).in(payloadLeaf(rowType, "qty"),
new ArrayList<>());
+ Assertions.assertThat(readPks(rowType, emptyIn))
+ .as("pushdown is declined, so every row must come back
unfiltered - not crash")
+ .containsExactlyInAnyOrder(1L, 2L);
+ }
+
+ /** Same as {@link #testNestedInWithEmptyLiteralsDoesNotCrashTheReader()},
for NOT IN. */
+ @Test
+ public void testNestedNotInWithEmptyLiteralsDoesNotCrashTheReader() throws
IOException {
+ RowType rowType = nestedPayloadType();
+ writeTwoPayloadRows(rowType);
+ Predicate emptyNotIn =
+ new PredicateBuilder(rowType).notIn(payloadLeaf(rowType,
"qty"), new ArrayList<>());
+ Assertions.assertThat(readPks(rowType, emptyNotIn))
+ .as("NOT IN empty-set matches everything, and must not crash
the read")
+ .containsExactlyInAnyOrder(1L, 2L);
+ }
+
+ /**
+ * 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);
+ }
+
+ /**
+ * A nested path that joins cleanly (no component contains a dot) can
still collide with an
+ * unrelated top-level column whose own literal name equals that joined
path. The schema below
+ * has both a top-level column named {@code "s.a"} (declared INT, so its
physical column is
+ * INT32) and a nested {@code s.a} (row {@code s} with BIGINT field {@code
a}, physical INT64).
+ * A predicate on the nested field is re-dispatched as {@code
FieldRef("s.a")}, and {@link
+ * org.apache.parquet.filter2.predicate.ParquetFilters}'s column lookup
checks an exact
+ * top-level name before walking the split components - so it resolves the
predicate's physical
+ * type against the unrelated top-level column instead of {@code s -> a}.
+ *
+ * <p>parquet-mr itself re-splits any dot-joined column name it is given,
so the {@link
+ * org.apache.parquet.filter2.predicate.FilterPredicate} still ends up
addressing the real,
+ * two-segment {@code s -> a} column chunk - but tagged with the wrong
(top-level) column's
+ * physical type. parquet-mr's {@code SchemaCompatibilityValidator}
catches that mismatch at
+ * read time and throws {@link IllegalArgumentException}, so today this
does not silently drop
+ * the row - it fails the read outright for a query that has nothing wrong
with it.
+ */
+ @Test
+ public void
testNestedPathCollidingWithADottedTopLevelNameKeepsMatchingRows()
+ throws IOException {
+ RowType inner = RowType.of(new DataType[] {DataTypes.BIGINT()}, new
String[] {"a"});
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.BIGINT(), DataTypes.INT(),
inner},
+ new String[] {"pk", "s.a", "s"});
+
+ write(
+ fileFormat().createWriterFactory(rowType),
+ file,
+ // top-level `s.a` (INT32) = 999 (does not match), nested s.a
(INT64) = 7 (matches)
+ GenericRow.of(1L, 999, GenericRow.of(7L)));
+
+ Predicate onNestedSA =
+ new PredicateBuilder(rowType)
+ .equal(
+ new NestedFieldTransform(
+ new FieldRef(2, "s", inner),
+ Collections.singletonList("a")),
+ 7L);
+
+ Assertions.assertThat(readPks(rowType, onNestedSA))
+ .as(
+ "the row whose nested s.a equals 7 must survive the
filter, "
+ + "even though the unrelated top-level `s.a`
column does not")
+ .contains(1L);
+ }
+
+ /**
+ * A declared top-level column whose name contains a dot (say {@code
"s.a"}) can be entirely
+ * absent from a file - a Format Table's metastore schema need not match
what a given file
+ * holds. The file below only has a nested {@code s -> a} (INT32); it
holds no top-level {@code
+ * "s.a"} at all. A predicate built against the declared schema names that
missing column as an
+ * ordinary (non-nested) {@code FieldRef("s.a")}, which {@code
findFileColumn}'s fallback walk
+ * still descends into the unrelated nested group instead of reporting the
column missing. Since
+ * a genuinely missing column always reads as null, {@code s.a IS NULL}
must keep every row.
+ */
+ @Test
+ public void
testMissingDottedTopLevelFieldIsTreatedAsNullNotAsANestedPath() throws
IOException {
+ RowType innerActual = RowType.of(new DataType[] {DataTypes.INT()}, new
String[] {"a"});
+ RowType actualFileType =
+ RowType.of(
+ new DataType[] {DataTypes.BIGINT(), innerActual}, new
String[] {"pk", "s"});
+
+ write(
+ fileFormat().createWriterFactory(actualFileType),
+ file,
+ GenericRow.of(1L, GenericRow.of(7)),
+ GenericRow.of(2L, GenericRow.of(8)));
+
+ // the declared/read schema has a top-level "s.a" that the file above
does not hold at all
+ RowType declaredType =
+ RowType.of(
+ new DataType[] {DataTypes.BIGINT(),
DataTypes.BIGINT()},
+ new String[] {"pk", "s.a"});
+ Predicate isNull = new PredicateBuilder(declaredType).isNull(1);
+
+ Assertions.assertThat(readPks(declaredType, isNull))
+ .as(
+ "every row must survive: the declared top-level `s.a`
does not exist in the file")
+ .containsExactlyInAnyOrder(1L, 2L);
+ }
+
+ /**
+ * Companion to {@link
#testMissingDottedTopLevelFieldIsTreatedAsNullNotAsANestedPath()}: when a
+ * declared dotted top-level column is missing from the file AND there is
no group at all under
+ * its first component (so there is no ambiguity to guard against), the
predicate must still be
+ * pushed down and treat the column as null - the fix must not turn every
dotted, missing column
+ * into a rejected pushdown, only the ones that collide with a real nested
column.
+ */
+ @Test
+ public void
testMissingDottedTopLevelFieldWithNoCollisionIsStillPushedDown()
+ throws IOException {
+ RowType actualFileType =
+ RowType.of(new DataType[] {DataTypes.BIGINT()}, new String[]
{"pk"});
+
+ write(
+ fileFormat().createWriterFactory(actualFileType),
+ file,
+ GenericRow.of(1L),
+ GenericRow.of(2L));
+
+ // the declared/read schema has a top-level "x.y" that the file above
has no trace of at
+ // all - no exact match, and no group "x" to even attempt a walk into
+ RowType declaredType =
+ RowType.of(
+ new DataType[] {DataTypes.BIGINT(),
DataTypes.BIGINT()},
+ new String[] {"pk", "x.y"});
+ Predicate isNull = new PredicateBuilder(declaredType).isNull(1);
+
+ Assertions.assertThat(readPks(declaredType, isNull))
+ .as("every row must survive: `x.y` is simply absent, with
nothing to collide with")
+ .containsExactlyInAnyOrder(1L, 2L);
+ }
}
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 95cd0440bc..f0a5453d38 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
@@ -130,7 +130,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))
@@ -298,6 +298,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 6321a9d5d7..e141b3d043 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.{BitLengthTransform, DateAddTransform,
DateDiffTransform, DateTruncTransform, DayOfWeekTransform, DayOfYearTransform,
DayTransform, FieldRef, HourTransform, LengthTransform, MinuteTransform,
MonthTransform, OverlayTransform, PadTransform, PredicateBuilder,
QuarterTransform, SecondTransform, Transform, TranslateTransform,
WeekdayTransform, WeekTransform, YearOfWeekTransform, YearTransform}
+import org.apache.paimon.predicate.{BitLengthTransform, DateAddTransform,
DateDiffTransform, DateTruncTransform, DayOfWeekTransform, DayOfYearTransform,
DayTransform, FieldRef, FieldTransform, HourTransform, LeafPredicate,
LengthTransform, MinuteTransform, MonthTransform, NestedFieldTransform,
OverlayTransform, PadTransform, PredicateBuilder, QuarterTransform,
SecondTransform, Transform, TranslateTransform, WeekdayTransform,
WeekTransform, YearOfWeekTransform, YearTransform}
import org.apache.paimon.spark.{PaimonSparkTestBase, SparkV2FilterConverter}
import
org.apache.paimon.spark.util.shim.TypeUtils.treatPaimonTimestampTypeAsSparkTimestampType
import org.apache.paimon.table.source.DataSplit
@@ -846,6 +846,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<uid: INT, addr: STRUCT<city: STRING, zip: STRING>>
+ |) 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 }
