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 34ef05103e [flink] Push down predicates on fields nested inside a row
(#10144)
34ef05103e is described below
commit 34ef05103e9199ce18bca5e38b0e3069c91bd762
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Thu Sep 24 17:56:26 2026 +0800
[flink] Push down predicates on fields nested inside a row (#10144)
---
.../apache/paimon/predicate/PredicateBuilder.java | 4 +
.../apache/paimon/flink/NestedFieldReferences.java | 85 ++++
.../apache/paimon/flink/PredicateConverter.java | 307 +++++++++----
.../paimon/flink/source/FlinkTableSource.java | 6 +-
.../paimon/flink/NestedFieldReferencesTest.java | 109 +++++
.../paimon/flink/NestedPredicateConverterTest.java | 499 +++++++++++++++++++++
.../source/NestedFieldFilterPushDownITCase.java | 392 ++++++++++++++++
7 files changed, 1319 insertions(+), 83 deletions(-)
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 cd358c8620..2c8ecba4c1 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
@@ -70,6 +70,10 @@ public class PredicateBuilder {
this.fieldNames = rowType.getFieldNames();
}
+ public RowType rowType() {
+ return rowType;
+ }
+
public int indexOf(String field) {
return fieldNames.indexOf(field);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/NestedFieldReferences.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/NestedFieldReferences.java
new file mode 100644
index 0000000000..b42e2f9b29
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/NestedFieldReferences.java
@@ -0,0 +1,85 @@
+/*
+ * 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.flink;
+
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.types.DataType;
+
+/**
+ * Access to Flink's {@code NestedFieldReferenceExpression}, which only exists
from Flink 1.19 on.
+ *
+ * <p>This module is compiled once and bundled into every {@code
paimon-flink-*} distribution,
+ * including those for Flink 1.16 to 1.18 where the class is absent. Every
direct reference to it
+ * therefore lives in {@link Holder}, a separate class file that is only ever
loaded once {@link
+ * #AVAILABLE} has confirmed the expression is on the classpath.
+ */
+public class NestedFieldReferences {
+
+ private static final String CLASS_NAME =
+
"org.apache.flink.table.expressions.NestedFieldReferenceExpression";
+
+ private static final boolean AVAILABLE = isOnClasspath();
+
+ private NestedFieldReferences() {}
+
+ private static boolean isOnClasspath() {
+ try {
+ Class.forName(CLASS_NAME, false,
NestedFieldReferences.class.getClassLoader());
+ return true;
+ } catch (ClassNotFoundException | LinkageError e) {
+ return false;
+ }
+ }
+
+ /** Whether {@code expression} references a field nested inside a row. */
+ public static boolean isNestedFieldReference(Expression expression) {
+ return AVAILABLE && Holder.isInstance(expression);
+ }
+
+ /**
+ * The path to the referenced field, starting with the name of the
top-level field it is nested
+ * in. Callers must check {@link #isNestedFieldReference} first.
+ */
+ public static String[] fieldNames(Expression expression) {
+ return Holder.fieldNames(expression);
+ }
+
+ /** The type of the referenced field. Callers must check {@link
#isNestedFieldReference}. */
+ public static DataType outputDataType(Expression expression) {
+ return Holder.outputDataType(expression);
+ }
+
+ private static class Holder {
+
+ static boolean isInstance(Expression expression) {
+ return expression
+ instanceof
org.apache.flink.table.expressions.NestedFieldReferenceExpression;
+ }
+
+ static String[] fieldNames(Expression expression) {
+ return
((org.apache.flink.table.expressions.NestedFieldReferenceExpression) expression)
+ .getFieldNames();
+ }
+
+ static DataType outputDataType(Expression expression) {
+ return
((org.apache.flink.table.expressions.NestedFieldReferenceExpression) expression)
+ .getOutputDataType();
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PredicateConverter.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PredicateConverter.java
index d14b1ccce1..47105a7258 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PredicateConverter.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PredicateConverter.java
@@ -19,8 +19,11 @@
package org.apache.paimon.flink;
import org.apache.paimon.data.BinaryString;
+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.utils.TypeUtils;
import org.apache.flink.table.data.conversion.DataStructureConverters;
@@ -39,8 +42,11 @@ import
org.apache.flink.table.types.logical.LogicalTypeFamily;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.flink.table.types.logical.RowType;
+import javax.annotation.Nullable;
+
import java.util.ArrayDeque;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Deque;
import java.util.List;
import java.util.Optional;
@@ -61,12 +67,26 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
private final PredicateBuilder builder;
+ /**
+ * The table's own type, used to address fields nested inside a row. A
type round-tripped
+ * through Flink renumbers field ids, and {@link NestedFieldTransform}
carries those ids as its
+ * identity, so on an evolved schema they would no longer match the table.
Null when the caller
+ * only has a Flink type.
+ */
+ @Nullable private final org.apache.paimon.types.RowType tableType;
+
public PredicateConverter(RowType type) {
this(new PredicateBuilder(toDataType(type)));
}
public PredicateConverter(PredicateBuilder builder) {
+ this(builder, null);
+ }
+
+ private PredicateConverter(
+ PredicateBuilder builder, @Nullable
org.apache.paimon.types.RowType tableType) {
this.builder = builder;
+ this.tableType = tableType;
}
/** Accepts simple LIKE patterns like "abc%". */
@@ -96,56 +116,57 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
return visitComparison(
children,
negated,
- builder::notEqual,
- builder::notEqual,
- builder::equal,
- builder::equal);
+ op(builder::notEqual, builder::notEqual),
+ op(builder::notEqual, builder::notEqual),
+ op(builder::equal, builder::equal),
+ op(builder::equal, builder::equal));
} else if (func == BuiltInFunctionDefinitions.NOT_EQUALS) {
return visitComparison(
children,
negated,
- builder::equal,
- builder::equal,
- builder::notEqual,
- builder::notEqual);
+ op(builder::equal, builder::equal),
+ op(builder::equal, builder::equal),
+ op(builder::notEqual, builder::notEqual),
+ op(builder::notEqual, builder::notEqual));
} else if (func == BuiltInFunctionDefinitions.GREATER_THAN) {
return visitComparison(
children,
negated,
- builder::lessOrEqual,
- builder::greaterOrEqual,
- builder::greaterThan,
- builder::lessThan);
+ op(builder::lessOrEqual, builder::lessOrEqual),
+ op(builder::greaterOrEqual, builder::greaterOrEqual),
+ op(builder::greaterThan, builder::greaterThan),
+ op(builder::lessThan, builder::lessThan));
} else if (func == BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL) {
return visitComparison(
children,
negated,
- builder::lessThan,
- builder::greaterThan,
- builder::greaterOrEqual,
- builder::lessOrEqual);
+ op(builder::lessThan, builder::lessThan),
+ op(builder::greaterThan, builder::greaterThan),
+ op(builder::greaterOrEqual, builder::greaterOrEqual),
+ op(builder::lessOrEqual, builder::lessOrEqual));
} else if (func == BuiltInFunctionDefinitions.LESS_THAN) {
return visitComparison(
children,
negated,
- builder::greaterOrEqual,
- builder::lessOrEqual,
- builder::lessThan,
- builder::greaterThan);
+ op(builder::greaterOrEqual, builder::greaterOrEqual),
+ op(builder::lessOrEqual, builder::lessOrEqual),
+ op(builder::lessThan, builder::lessThan),
+ op(builder::greaterThan, builder::greaterThan));
} else if (func == BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL) {
return visitComparison(
children,
negated,
- builder::greaterThan,
- builder::lessThan,
- builder::lessOrEqual,
- builder::greaterOrEqual);
+ op(builder::greaterThan, builder::greaterThan),
+ op(builder::lessThan, builder::lessThan),
+ op(builder::lessOrEqual, builder::lessOrEqual),
+ op(builder::greaterOrEqual, builder::greaterOrEqual));
} else if (func == BuiltInFunctionDefinitions.IN) {
requireAtLeastArity(children, 2);
ResolvedField field = resolveField(children.get(0));
+ rejectNestedFloatingPoint(field);
List<Object> literals = new ArrayList<>();
for (int i = 1; i < children.size(); i++) {
-
literals.add(extractLiteral(field.expression.getOutputDataType(),
children.get(i)));
+ literals.add(extractLiteral(field.type(), children.get(i)));
}
if (negated) {
// SQL WHERE: v NOT IN (..., NULL, ...) is never true
regardless of
@@ -154,27 +175,31 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
if (literals.contains(null)) {
return PredicateBuilder.alwaysFalse();
}
- rejectNegatedFloatingPoint(field.expression);
- return builder.notIn(field.index, literals);
+ rejectNegatedFloatingPoint(field);
+ return build(field, builder::notIn, builder::notIn, literals);
}
- return builder.in(field.index, literals);
+ return build(field, builder::in, builder::in, literals);
} else if (func == BuiltInFunctionDefinitions.IS_NULL) {
requireArity(children, 1);
ResolvedField field = resolveField(children.get(0));
- return negated ? builder.isNotNull(field.index) :
builder.isNull(field.index);
+ return negated ? isNotNull(field) : isNull(field);
} else if (func == BuiltInFunctionDefinitions.IS_NOT_NULL) {
requireArity(children, 1);
ResolvedField field = resolveField(children.get(0));
- return negated ? builder.isNull(field.index) :
builder.isNotNull(field.index);
+ return negated ? isNull(field) : isNotNull(field);
} else if (func == BuiltInFunctionDefinitions.BETWEEN) {
requireArity(children, 3);
ResolvedField field = resolveField(children.get(0));
- DataType fieldType = field.expression.getOutputDataType();
+ rejectNestedFloatingPoint(field);
+ DataType fieldType = field.type();
Object lower = extractLiteral(fieldType, children.get(1));
Object upper = extractLiteral(fieldType, children.get(2));
- Predicate between = builder.between(field.index, lower, upper);
+ Predicate between =
+ field.isNested()
+ ? builder.between(field.transform, lower, upper)
+ : builder.between(field.index, lower, upper);
if (negated) {
- rejectNegatedFloatingPoint(field.expression);
+ rejectNegatedFloatingPoint(field);
// LeafTernaryFunction.test returns false if any literal is
null, but
// 12 NOT BETWEEN 15 AND NULL is TRUE (TRUE OR UNKNOWN). Keep
residual
// so Flink can evaluate the three-valued cases.
@@ -189,22 +214,16 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
throw new UnsupportedExpression();
}
ResolvedField field = resolveField(children.get(0));
- if (field.expression
- .getOutputDataType()
+ if (field.type()
.getLogicalType()
.getTypeRoot()
.getFamilies()
.contains(LogicalTypeFamily.CHARACTER_STRING)) {
- String sqlPattern =
-
extractNonNullLiteral(field.expression.getOutputDataType(), children.get(1))
- .toString();
+ String sqlPattern = extractNonNullLiteral(field.type(),
children.get(1)).toString();
String escape =
children.size() <= 2
? null
- : extractNonNullLiteral(
-
field.expression.getOutputDataType(),
- children.get(2))
- .toString();
+ : extractNonNullLiteral(field.type(),
children.get(2)).toString();
String escapedSqlPattern = sqlPattern;
boolean allowQuick = false;
if (escape == null && !sqlPattern.contains("_")) {
@@ -252,8 +271,11 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
// residual filter evaluated by Flink.
throw new UnsupportedExpression();
}
- return builder.startsWith(
- field.index,
BinaryString.fromString(beginMatcher.group(1)));
+ return build(
+ field,
+ builder::startsWith,
+ builder::startsWith,
+
BinaryString.fromString(beginMatcher.group(1)));
}
}
}
@@ -282,16 +304,25 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
}
private Predicate booleanTest(ResolvedField field, boolean expected,
boolean complement) {
- if (field.expression.getOutputDataType().getLogicalType().getTypeRoot()
- != LogicalTypeRoot.BOOLEAN) {
+ if (field.type().getLogicalType().getTypeRoot() !=
LogicalTypeRoot.BOOLEAN) {
throw new UnsupportedExpression();
}
- Predicate equals = builder.equal(field.index, expected);
+ Predicate equals = build(field, builder::equal, builder::equal,
expected);
if (!complement) {
return equals;
}
return PredicateBuilder.or(
- builder.isNull(field.index), builder.notEqual(field.index,
expected));
+ isNull(field), build(field, builder::notEqual,
builder::notEqual, expected));
+ }
+
+ private Predicate isNull(ResolvedField field) {
+ return field.isNested() ? builder.isNull(field.transform) :
builder.isNull(field.index);
+ }
+
+ private Predicate isNotNull(ResolvedField field) {
+ return field.isNested()
+ ? builder.isNotNull(field.transform)
+ : builder.isNotNull(field.index);
}
private Predicate negate(Predicate predicate) {
@@ -338,10 +369,10 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
private Predicate visitComparison(
List<Expression> children,
boolean negated,
- BiFunction<Integer, Object, Predicate> negatedVisit1,
- BiFunction<Integer, Object, Predicate> negatedVisit2,
- BiFunction<Integer, Object, Predicate> visit1,
- BiFunction<Integer, Object, Predicate> visit2) {
+ LeafFunction negatedVisit1,
+ LeafFunction negatedVisit2,
+ LeafFunction visit1,
+ LeafFunction visit2) {
// Flink FLOAT/DOUBLE comparisons use Java operators; Paimon uses
// Float/Double.compareTo. Negated equality, inequality, IN and BETWEEN
// are therefore not equivalent (NaN identity and signed zeros). Simple
@@ -356,8 +387,20 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
: visitBiFunction(children, visit1, visit2);
}
- private void rejectNegatedFloatingPoint(FieldReferenceExpression field) {
- if (isFloatingPointField(field)) {
+ /**
+ * Flink compares FLOAT/DOUBLE with Java operators, so {@code -0.0 = 0.0}
holds; Paimon's
+ * predicates use {@code compareTo}, which tells the two apart. Pruning a
file on such a
+ * predicate could drop a row Flink would have kept, before Flink's own
filter sees it, so
+ * comparisons, IN and BETWEEN on a nested floating-point field are left
to Flink.
+ */
+ private void rejectNestedFloatingPoint(ResolvedField field) {
+ if (field.isNested() && isFloatingPointType(field.type())) {
+ throw new UnsupportedExpression();
+ }
+ }
+
+ private void rejectNegatedFloatingPoint(ResolvedField field) {
+ if (isFloatingPointType(field.type())) {
throw new UnsupportedExpression();
}
}
@@ -365,46 +408,113 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
private boolean isFloatingPointComparison(List<Expression> children) {
for (Expression child : children) {
Optional<FieldReferenceExpression> field =
extractFieldReference(child);
- if (field.isPresent() && isFloatingPointField(field.get())) {
+ if (field.isPresent() &&
isFloatingPointType(field.get().getOutputDataType())) {
+ return true;
+ }
+ if (NestedFieldReferences.isNestedFieldReference(child)
+ &&
isFloatingPointType(NestedFieldReferences.outputDataType(child))) {
return true;
}
}
return false;
}
- private boolean isFloatingPointField(FieldReferenceExpression field) {
- LogicalTypeRoot root =
field.getOutputDataType().getLogicalType().getTypeRoot();
+ private boolean isFloatingPointType(DataType type) {
+ LogicalTypeRoot root = type.getLogicalType().getTypeRoot();
return root == LogicalTypeRoot.FLOAT || root == LogicalTypeRoot.DOUBLE;
}
private Predicate visitBiFunction(
- List<Expression> children,
- BiFunction<Integer, Object, Predicate> visit1,
- BiFunction<Integer, Object, Predicate> visit2) {
+ List<Expression> children, LeafFunction visit1, LeafFunction
visit2) {
requireArity(children, 2);
- Optional<FieldReferenceExpression> fieldRefExpr =
extractFieldReference(children.get(0));
- if (fieldRefExpr.isPresent()) {
- int fieldIndex = resolveFieldIndex(fieldRefExpr.get());
- Object literal =
- extractLiteral(fieldRefExpr.get().getOutputDataType(),
children.get(1));
- return visit1.apply(fieldIndex, literal);
- } else {
- fieldRefExpr = extractFieldReference(children.get(1));
- if (fieldRefExpr.isPresent()) {
- int fieldIndex = resolveFieldIndex(fieldRefExpr.get());
- Object literal =
- extractLiteral(fieldRefExpr.get().getOutputDataType(),
children.get(0));
- return visit2.apply(fieldIndex, literal);
- }
+ if (isFieldReference(children.get(0))) {
+ ResolvedField field = resolveField(children.get(0));
+ rejectNestedFloatingPoint(field);
+ return visit1.apply(field, extractLiteral(field.type(),
children.get(1)));
+ }
+ if (isFieldReference(children.get(1))) {
+ ResolvedField field = resolveField(children.get(1));
+ rejectNestedFloatingPoint(field);
+ return visit2.apply(field, extractLiteral(field.type(),
children.get(0)));
}
throw new UnsupportedExpression();
}
+ private boolean isFieldReference(Expression expression) {
+ return expression instanceof FieldReferenceExpression
+ || NestedFieldReferences.isNestedFieldReference(expression);
+ }
+
+ /** A {@link PredicateBuilder} method that builds a leaf predicate over a
field. */
+ @FunctionalInterface
+ private interface LeafFunction {
+ Predicate apply(ResolvedField field, Object literal);
+ }
+
+ /**
+ * Pairs the two {@link PredicateBuilder} overloads of one operation, so
that a field can be
+ * addressed either by index or, when it is nested inside a row, by
transform.
+ */
+ private static LeafFunction op(
+ BiFunction<Integer, Object, Predicate> byIndex,
+ BiFunction<Transform, Object, Predicate> byTransform) {
+ return (field, literal) -> build(field, byIndex, byTransform, literal);
+ }
+
private ResolvedField resolveField(Expression expression) {
+ if (NestedFieldReferences.isNestedFieldReference(expression)) {
+ return resolveNestedField(expression);
+ }
FieldReferenceExpression field =
extractFieldReference(expression).orElseThrow(UnsupportedExpression::new);
- return new ResolvedField(field, resolveFieldIndex(field));
+ return ResolvedField.topLevel(field, resolveFieldIndex(field));
+ }
+
+ /**
+ * Resolves a field nested inside a row. Flink hands the path down as the
names of the fields
+ * walked through, starting at the top-level one, which is what {@link
NestedFieldTransform}
+ * addresses the field by as well.
+ */
+ private ResolvedField resolveNestedField(Expression expression) {
+ String[] fieldNames = NestedFieldReferences.fieldNames(expression);
+ if (fieldNames.length < 2) {
+ throw new UnsupportedExpression();
+ }
+
+ int rootIndex = builder.indexOf(fieldNames[0]);
+ if (rootIndex < 0) {
+ throw new UnsupportedExpression();
+ }
+ // Prefer the table's own type: its field ids are the transform's
identity, and the
+ // round-tripped type behind the builder has renumbered them.
+ org.apache.paimon.types.RowType rootSource =
+ tableType != null ? tableType : builder.rowType();
+ FieldRef rootRef = new FieldRef(rootIndex, fieldNames[0],
rootSource.getTypeAt(rootIndex));
+ List<String> path = Arrays.asList(fieldNames).subList(1,
fieldNames.length);
+ try {
+ return ResolvedField.nested(
+ NestedFieldReferences.outputDataType(expression),
+ new NestedFieldTransform(rootRef, path));
+ } catch (IllegalArgumentException e) {
+ // The path does not address a field of this table: the root is
not a row, or a field
+ // along the way was renamed or dropped. Leave the filter for
Flink to evaluate.
+ throw new UnsupportedExpression();
+ }
+ }
+
+ /**
+ * Binds a {@link PredicateBuilder} method to a field, choosing the
overload that addresses it:
+ * by index for a top-level field, by transform for one nested inside a
row.
+ */
+ private static <T> Predicate build(
+ ResolvedField field,
+ BiFunction<Integer, T, Predicate> byIndex,
+ BiFunction<Transform, T, Predicate> byTransform,
+ T argument) {
+ return field.isNested()
+ ? byTransform.apply(field.transform, argument)
+ : byIndex.apply(field.index, argument);
}
private int resolveFieldIndex(FieldReferenceExpression field) {
@@ -504,14 +614,36 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
}
}
+ /**
+ * A field a predicate can be built on: either a top-level field,
addressed by its index, or a
+ * field nested inside a row, addressed by a {@link NestedFieldTransform}.
+ */
private static class ResolvedField {
- private final FieldReferenceExpression expression;
+ private final DataType type;
private final int index;
+ @Nullable private final Transform transform;
- private ResolvedField(FieldReferenceExpression expression, int index) {
- this.expression = expression;
+ private ResolvedField(DataType type, int index, @Nullable Transform
transform) {
+ this.type = type;
this.index = index;
+ this.transform = transform;
+ }
+
+ static ResolvedField topLevel(FieldReferenceExpression expression, int
index) {
+ return new ResolvedField(expression.getOutputDataType(), index,
null);
+ }
+
+ static ResolvedField nested(DataType type, Transform transform) {
+ return new ResolvedField(type, -1, transform);
+ }
+
+ boolean isNested() {
+ return transform != null;
+ }
+
+ DataType type() {
+ return type;
}
}
@@ -542,8 +674,25 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
* @return {@link Predicate} if no {@link UnsupportedExpression} thrown.
*/
public static Optional<Predicate> convert(RowType rowType,
ResolvedExpression filter) {
+ return convert(new PredicateConverter(rowType), filter);
+ }
+
+ /**
+ * Like {@link #convert(RowType, ResolvedExpression)}, for a table whose
Paimon type is at hand.
+ * Predicates on fields nested inside a row are then bound to the table's
own field ids, which
+ * is what reading the table later checks them against.
+ */
+ public static Optional<Predicate> convert(
+ org.apache.paimon.types.RowType tableType, ResolvedExpression
filter) {
+ PredicateBuilder builder =
+ new
PredicateBuilder(toDataType(LogicalTypeConversion.toLogicalType(tableType)));
+ return convert(new PredicateConverter(builder, tableType), filter);
+ }
+
+ private static Optional<Predicate> convert(
+ PredicateConverter converter, ResolvedExpression filter) {
try {
- return Optional.ofNullable(filter.accept(new
PredicateConverter(rowType)));
+ return Optional.ofNullable(filter.accept(converter));
} catch (UnsupportedExpression e) {
return Optional.empty();
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java
index dd4b7439e0..6c2028eea0 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java
@@ -21,7 +21,6 @@ package org.apache.paimon.flink.source;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.flink.FlinkConnectorOptions;
-import org.apache.paimon.flink.LogicalTypeConversion;
import org.apache.paimon.flink.PredicateConverter;
import org.apache.paimon.flink.lookup.DynamicPartitionLoader;
import org.apache.paimon.flink.lookup.PartitionLoader;
@@ -48,7 +47,6 @@ import
org.apache.flink.table.connector.source.abilities.SupportsProjectionPushD
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.plan.stats.TableStats;
import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.logical.RowType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -112,7 +110,6 @@ public abstract class FlinkTableSource
@Override
public Result applyFilters(List<ResolvedExpression> filters) {
List<String> partitionKeys = table.partitionKeys();
- RowType rowType = LogicalTypeConversion.toLogicalType(table.rowType());
// The source must ensure the consumed filters are fully evaluated,
otherwise the result
// of query will be wrong.
@@ -123,7 +120,8 @@ public abstract class FlinkTableSource
new PartitionPredicateVisitor(partitionKeys);
for (ResolvedExpression filter : filters) {
- Optional<Predicate> predicateOptional =
PredicateConverter.convert(rowType, filter);
+ Optional<Predicate> predicateOptional =
+ PredicateConverter.convert(table.rowType(), filter);
if (!predicateOptional.isPresent()) {
unConsumedFilters.add(filter);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/NestedFieldReferencesTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/NestedFieldReferencesTest.java
new file mode 100644
index 0000000000..afbe1df09a
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/NestedFieldReferencesTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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.flink;
+
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Method;
+import java.net.URL;
+import java.net.URLClassLoader;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/** Tests for {@link NestedFieldReferences}. */
+public class NestedFieldReferencesTest {
+
+ private static final String HIDDEN =
+
"org.apache.flink.table.expressions.NestedFieldReferenceExpression";
+
+ /**
+ * Flink only has the nested field reference from 1.19 on, but this module
is compiled once and
+ * bundled for Flink 1.16 to 1.18 as well. Loaded where the class is
absent, asking whether an
+ * expression is a nested reference must answer no rather than fail to
link.
+ */
+ @Test
+ public void testAnswersNoWhenTheExpressionIsNotOnTheClasspath() throws
Exception {
+ try (URLClassLoader withoutNestedReferences = hidingClassLoader()) {
+ Class<?> loaded =
+
withoutNestedReferences.loadClass(NestedFieldReferences.class.getName());
+
assertThat(loaded.getClassLoader()).isSameAs(withoutNestedReferences);
+
+ Method isNestedFieldReference =
+ loaded.getMethod(
+ "isNestedFieldReference",
+ withoutNestedReferences.loadClass(
+
"org.apache.flink.table.expressions.Expression"));
+
+ // A real expression, not null: `null instanceof X` answers
without ever resolving X,
+ // so only a non-null argument exercises the class the guard has
to keep away.
+ Object expression =
+ withoutNestedReferences
+
.loadClass("org.apache.flink.table.expressions.ValueLiteralExpression")
+ .getConstructor(Object.class)
+ .newInstance(1);
+
+ assertThatCode(() -> isNestedFieldReference.invoke(null,
expression))
+ .doesNotThrowAnyException();
+ assertThat(isNestedFieldReference.invoke(null,
expression)).isEqualTo(false);
+ }
+ }
+
+ /** Loads this module's own classes itself, and pretends the nested
reference does not exist. */
+ private static URLClassLoader hidingClassLoader() {
+ URLClassLoader appLoader = (URLClassLoader) buildClassPathLoader();
+ return new URLClassLoader(appLoader.getURLs(), appLoader.getParent()) {
+ @Override
+ protected Class<?> loadClass(String name, boolean resolve)
+ throws ClassNotFoundException {
+ if (name.equals(HIDDEN)) {
+ throw new ClassNotFoundException(name);
+ }
+ if
(name.startsWith("org.apache.paimon.flink.NestedFieldReferences")) {
+ synchronized (getClassLoadingLock(name)) {
+ Class<?> loaded = findLoadedClass(name);
+ if (loaded == null) {
+ loaded = findClass(name);
+ }
+ if (resolve) {
+ resolveClass(loaded);
+ }
+ return loaded;
+ }
+ }
+ return super.loadClass(name, resolve);
+ }
+ };
+ }
+
+ private static ClassLoader buildClassPathLoader() {
+ String[] entries =
System.getProperty("java.class.path").split(java.io.File.pathSeparator);
+ URL[] urls = new URL[entries.length];
+ for (int i = 0; i < entries.length; i++) {
+ try {
+ urls[i] = new java.io.File(entries[i]).toURI().toURL();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ // Not getPlatformClassLoader(): Paimon builds with Java 8, where it
does not exist. The
+ // system loader's parent serves the same purpose here - a parent
without the classpath.
+ return new URLClassLoader(urls,
ClassLoader.getSystemClassLoader().getParent());
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/NestedPredicateConverterTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/NestedPredicateConverterTest.java
new file mode 100644
index 0000000000..2db576d9eb
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/NestedPredicateConverterTest.java
@@ -0,0 +1,499 @@
+/*
+ * 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.flink;
+
+import org.apache.paimon.data.BinaryString;
+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.PredicateRemapper;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.expressions.NestedFieldReferenceExpression;
+import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.table.functions.BuiltInFunctionDefinition;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests that {@link PredicateConverter} converts a predicate on a field
nested inside a row, which
+ * Flink hands down as a {@code NestedFieldReferenceExpression}.
+ */
+public class NestedPredicateConverterTest {
+
+ private static final RowType DEEP =
+ RowType.of(new DataType[] {DataTypes.INT()}, new String[] {"x"});
+
+ private static final RowType NESTED =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT(),
+ DataTypes.STRING(),
+ DataTypes.DOUBLE(),
+ DataTypes.BOOLEAN(),
+ DEEP,
+ DataTypes.FLOAT()
+ },
+ new String[] {"a", "b", "d", "flag", "inner", "f"});
+
+ private static final RowType TABLE =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), NESTED, DataTypes.INT(),
DataTypes.DOUBLE()},
+ new String[] {"pk", "s", "t", "td"});
+
+ /**
+ * The type the converter itself works on. Round-tripping through Flink's
type system is what
+ * {@link PredicateConverter#convert} does, and it assigns its own field
ids, so expected
+ * predicates have to be built from the same type to compare equal.
+ */
+ private static final RowType CONVERTED =
+
LogicalTypeConversion.toDataType(LogicalTypeConversion.toLogicalType(TABLE));
+
+ private static final PredicateBuilder BUILDER = new
PredicateBuilder(CONVERTED);
+
+ private static NestedFieldTransform transform(String... path) {
+ return new NestedFieldTransform(
+ new FieldRef(1, "s", CONVERTED.getTypeAt(1)),
Arrays.asList(path));
+ }
+
+ private static final NestedFieldTransform S_A = transform("a");
+ private static final NestedFieldTransform S_B = transform("b");
+ private static final NestedFieldTransform S_D = transform("d");
+ private static final NestedFieldTransform S_FLAG = transform("flag");
+ private static final NestedFieldTransform S_INNER_X = transform("inner",
"x");
+
+ private static NestedFieldReferenceExpression ref(
+ org.apache.flink.table.types.DataType type, String... path) {
+ int[] indices = new int[path.length];
+ return new NestedFieldReferenceExpression(path, indices, type);
+ }
+
+ private static NestedFieldReferenceExpression intRef(String... path) {
+ return ref(org.apache.flink.table.api.DataTypes.INT(), path);
+ }
+
+ private static ResolvedExpression call(
+ BuiltInFunctionDefinition func, ResolvedExpression... children) {
+ return CallExpression.permanent(
+ func, Arrays.asList(children),
org.apache.flink.table.api.DataTypes.BOOLEAN());
+ }
+
+ private static ResolvedExpression not(ResolvedExpression child) {
+ return call(BuiltInFunctionDefinitions.NOT, child);
+ }
+
+ private static Predicate convert(ResolvedExpression expression) {
+ return
PredicateConverter.convert(LogicalTypeConversion.toLogicalType(TABLE),
expression)
+ .orElse(null);
+ }
+
+ //
------------------------------------------------------------------------------------
+ // comparisons
+ //
------------------------------------------------------------------------------------
+
+ @Test
+ public void testNestedComparisons() {
+ ValueLiteralExpression seven = new ValueLiteralExpression(7);
+
+ assertThat(convert(call(BuiltInFunctionDefinitions.EQUALS, intRef("s",
"a"), seven)))
+ .isEqualTo(BUILDER.equal(S_A, 7));
+ assertThat(convert(call(BuiltInFunctionDefinitions.NOT_EQUALS,
intRef("s", "a"), seven)))
+ .isEqualTo(BUILDER.notEqual(S_A, 7));
+ assertThat(convert(call(BuiltInFunctionDefinitions.GREATER_THAN,
intRef("s", "a"), seven)))
+ .isEqualTo(BUILDER.greaterThan(S_A, 7));
+ assertThat(
+ convert(
+ call(
+
BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL,
+ intRef("s", "a"),
+ seven)))
+ .isEqualTo(BUILDER.greaterOrEqual(S_A, 7));
+ assertThat(convert(call(BuiltInFunctionDefinitions.LESS_THAN,
intRef("s", "a"), seven)))
+ .isEqualTo(BUILDER.lessThan(S_A, 7));
+ assertThat(
+ convert(
+ call(
+
BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL,
+ intRef("s", "a"),
+ seven)))
+ .isEqualTo(BUILDER.lessOrEqual(S_A, 7));
+ }
+
+ /** The field may be on either side of the comparison; the operator flips
with it. */
+ @Test
+ public void testNestedComparisonWithLiteralOnTheLeft() {
+ ValueLiteralExpression seven = new ValueLiteralExpression(7);
+
+ assertThat(convert(call(BuiltInFunctionDefinitions.GREATER_THAN,
seven, intRef("s", "a"))))
+ .isEqualTo(BUILDER.lessThan(S_A, 7));
+ assertThat(convert(call(BuiltInFunctionDefinitions.LESS_THAN, seven,
intRef("s", "a"))))
+ .isEqualTo(BUILDER.greaterThan(S_A, 7));
+ }
+
+ @Test
+ public void testNegatedNestedComparison() {
+ ValueLiteralExpression seven = new ValueLiteralExpression(7);
+
+ assertThat(convert(not(call(BuiltInFunctionDefinitions.EQUALS,
intRef("s", "a"), seven))))
+ .isEqualTo(BUILDER.notEqual(S_A, 7));
+ assertThat(
+ convert(
+ not(
+ call(
+
BuiltInFunctionDefinitions.GREATER_THAN,
+ intRef("s", "a"),
+ seven))))
+ .isEqualTo(BUILDER.lessOrEqual(S_A, 7));
+ }
+
+ //
------------------------------------------------------------------------------------
+ // set, range, null and string predicates
+ //
------------------------------------------------------------------------------------
+
+ @Test
+ public void testNestedInAndNotIn() {
+ ResolvedExpression in =
+ call(
+ BuiltInFunctionDefinitions.IN,
+ intRef("s", "a"),
+ new ValueLiteralExpression(1),
+ new ValueLiteralExpression(2));
+
+ assertThat(convert(in)).isEqualTo(BUILDER.in(S_A, Arrays.asList(1,
2)));
+ assertThat(convert(not(in))).isEqualTo(BUILDER.notIn(S_A,
Arrays.asList(1, 2)));
+ }
+
+ /** {@code v NOT IN (..., NULL, ...)} is never true, whatever the field. */
+ @Test
+ public void testNestedNotInWithNullLiteralIsAlwaysFalse() {
+ ResolvedExpression in =
+ call(
+ BuiltInFunctionDefinitions.IN,
+ intRef("s", "a"),
+ new ValueLiteralExpression(1),
+ new ValueLiteralExpression(
+ null,
org.apache.flink.table.api.DataTypes.INT()));
+
+ assertThat(convert(not(in))).isEqualTo(PredicateBuilder.alwaysFalse());
+ }
+
+ @Test
+ public void testNestedIsNullAndIsNotNull() {
+ assertThat(convert(call(BuiltInFunctionDefinitions.IS_NULL,
intRef("s", "a"))))
+ .isEqualTo(BUILDER.isNull(S_A));
+ assertThat(convert(call(BuiltInFunctionDefinitions.IS_NOT_NULL,
intRef("s", "a"))))
+ .isEqualTo(BUILDER.isNotNull(S_A));
+ assertThat(convert(not(call(BuiltInFunctionDefinitions.IS_NULL,
intRef("s", "a")))))
+ .isEqualTo(BUILDER.isNotNull(S_A));
+ }
+
+ @Test
+ public void testNestedBetweenAndNotBetween() {
+ ResolvedExpression between =
+ call(
+ BuiltInFunctionDefinitions.BETWEEN,
+ intRef("s", "a"),
+ new ValueLiteralExpression(1),
+ new ValueLiteralExpression(3));
+
+ assertThat(convert(between)).isEqualTo(BUILDER.between(S_A, 1, 3));
+ assertThat(convert(not(between)))
+ .isEqualTo(BUILDER.between(S_A, 1,
3).negate().orElseThrow(AssertionError::new));
+ }
+
+ @Test
+ public void testNestedLikePrefixBecomesStartsWith() {
+ ResolvedExpression like =
+ call(
+ BuiltInFunctionDefinitions.LIKE,
+ ref(org.apache.flink.table.api.DataTypes.STRING(),
"s", "b"),
+ new ValueLiteralExpression("ab%"));
+
+ assertThat(convert(like)).isEqualTo(BUILDER.startsWith(S_B,
BinaryString.fromString("ab")));
+ }
+
+ @Test
+ public void testNestedBooleanTests() {
+ NestedFieldReferenceExpression flag =
+ ref(org.apache.flink.table.api.DataTypes.BOOLEAN(), "s",
"flag");
+
+ assertThat(convert(call(BuiltInFunctionDefinitions.IS_TRUE, flag)))
+ .isEqualTo(BUILDER.equal(S_FLAG, true));
+ assertThat(convert(call(BuiltInFunctionDefinitions.IS_FALSE, flag)))
+ .isEqualTo(BUILDER.equal(S_FLAG, false));
+ assertThat(convert(call(BuiltInFunctionDefinitions.IS_NOT_TRUE, flag)))
+ .isEqualTo(
+ PredicateBuilder.or(
+ BUILDER.isNull(S_FLAG),
BUILDER.notEqual(S_FLAG, true)));
+ assertThat(convert(call(BuiltInFunctionDefinitions.IS_NOT_FALSE,
flag)))
+ .isEqualTo(
+ PredicateBuilder.or(
+ BUILDER.isNull(S_FLAG),
BUILDER.notEqual(S_FLAG, false)));
+ }
+
+ //
------------------------------------------------------------------------------------
+ // floating point: comparisons on a nested FLOAT/DOUBLE are left to Flink
+ //
------------------------------------------------------------------------------------
+
+ /**
+ * Flink compares FLOAT/DOUBLE with Java operators ({@code -0.0 = 0.0}
holds), Paimon with
+ * {@code compareTo} (it does not), so no comparison on a nested
floating-point field is
+ * converted: not in either operand order, not negated, not as IN or
BETWEEN.
+ */
+ @Test
+ public void testNestedFloatingPointComparisonsAreNotConverted() {
+ NestedFieldReferenceExpression d =
+ ref(org.apache.flink.table.api.DataTypes.DOUBLE(), "s", "d");
+ NestedFieldReferenceExpression f =
+ ref(org.apache.flink.table.api.DataTypes.FLOAT(), "s", "f");
+ ValueLiteralExpression one = new ValueLiteralExpression(1.0d);
+ ValueLiteralExpression two = new ValueLiteralExpression(2.0d);
+
+ for (BuiltInFunctionDefinition comparison :
+ Arrays.asList(
+ BuiltInFunctionDefinitions.EQUALS,
+ BuiltInFunctionDefinitions.NOT_EQUALS,
+ BuiltInFunctionDefinitions.GREATER_THAN,
+ BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL,
+ BuiltInFunctionDefinitions.LESS_THAN,
+ BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL)) {
+ assertThat(convert(call(comparison, d, one))).as("s.d %s",
comparison).isNull();
+ assertThat(convert(call(comparison, one, d))).as("%s s.d",
comparison).isNull();
+ assertThat(convert(not(call(comparison, d, one))))
+ .as("NOT s.d %s", comparison)
+ .isNull();
+ assertThat(convert(call(comparison, f, new
ValueLiteralExpression(1.0f))))
+ .as("s.f %s", comparison)
+ .isNull();
+ }
+
+ ResolvedExpression in = call(BuiltInFunctionDefinitions.IN, d, one,
two);
+ assertThat(convert(in)).as("IN").isNull();
+ assertThat(convert(not(in))).as("NOT IN").isNull();
+ ResolvedExpression between = call(BuiltInFunctionDefinitions.BETWEEN,
d, one, two);
+ assertThat(convert(between)).as("BETWEEN").isNull();
+ assertThat(convert(not(between))).as("NOT BETWEEN").isNull();
+ }
+
+ /** A null check is not a comparison and has no signed-zero problem, so it
is still pushed. */
+ @Test
+ public void testNestedFloatingPointNullChecksAreStillConverted() {
+ NestedFieldReferenceExpression d =
+ ref(org.apache.flink.table.api.DataTypes.DOUBLE(), "s", "d");
+
+ assertThat(convert(call(BuiltInFunctionDefinitions.IS_NULL, d)))
+ .isEqualTo(BUILDER.isNull(S_D));
+ assertThat(convert(call(BuiltInFunctionDefinitions.IS_NOT_NULL, d)))
+ .isEqualTo(BUILDER.isNotNull(S_D));
+ }
+
+ /** The guard is for nested fields only: a top-level double converts as
before. */
+ @Test
+ public void testTopLevelFloatingPointComparisonIsUnchanged() {
+ ResolvedExpression equals =
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ new FieldReferenceExpression(
+ "td",
org.apache.flink.table.api.DataTypes.DOUBLE(), 0, 3),
+ new ValueLiteralExpression(1.0d));
+
+ assertThat(convert(equals)).isEqualTo(BUILDER.equal(3, 1.0d));
+ }
+
+ //
------------------------------------------------------------------------------------
+ // field ids
+ //
------------------------------------------------------------------------------------
+
+ /**
+ * On an evolved schema the table's field ids have gaps; a type
round-tripped through Flink
+ * numbers them afresh. Given the table's type, the nested transform
carries the table's own
+ * ids, so remapping it onto the table - as a masked read does - keeps its
identity.
+ */
+ @Test
+ public void testTableTypeEntryPointKeepsTheTablesFieldIds() {
+ RowType evolved = evolvedTable();
+ ResolvedExpression onC =
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ intRef("s", "c"),
+ new ValueLiteralExpression(7));
+
+ Predicate predicate =
+ PredicateConverter.convert(evolved,
onC).orElseThrow(AssertionError::new);
+ assertThatCode(() -> PredicateRemapper.remap(predicate, evolved))
+ .doesNotThrowAnyException();
+ }
+
+ /**
+ * Why callers pass the table's type: from a Flink type alone the
converter cannot know the
+ * table's ids, and the transform it builds carries renumbered ones.
+ */
+ @Test
+ public void testFlinkTypeEntryPointCannotKnowTheTablesFieldIds() {
+ RowType evolved = evolvedTable();
+ ResolvedExpression onC =
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ intRef("s", "c"),
+ new ValueLiteralExpression(7));
+
+ Predicate predicate =
+
PredicateConverter.convert(LogicalTypeConversion.toLogicalType(evolved), onC)
+ .orElseThrow(AssertionError::new);
+ assertThatThrownBy(() -> PredicateRemapper.remap(predicate, evolved))
+ .hasMessageContaining("changed identity");
+ }
+
+ /** Only nested fields use the table's type; every top-level predicate
converts as before. */
+ @Test
+ public void testTableTypeEntryPointLeavesTopLevelPredicatesUnchanged() {
+ FieldReferenceExpression pk =
+ new FieldReferenceExpression(
+ "pk", org.apache.flink.table.api.DataTypes.INT(), 0,
0);
+ for (ResolvedExpression topLevel :
+ Arrays.asList(
+ call(BuiltInFunctionDefinitions.EQUALS, pk, new
ValueLiteralExpression(1)),
+ call(
+ BuiltInFunctionDefinitions.GREATER_THAN,
+ pk,
+ new ValueLiteralExpression(1)),
+ call(
+ BuiltInFunctionDefinitions.IN,
+ pk,
+ new ValueLiteralExpression(1),
+ new ValueLiteralExpression(2)),
+ call(BuiltInFunctionDefinitions.IS_NULL, pk))) {
+ assertThat(PredicateConverter.convert(TABLE, topLevel))
+ .isEqualTo(
+ PredicateConverter.convert(
+
LogicalTypeConversion.toLogicalType(TABLE), topLevel));
+ }
+ }
+
+ /** {@code pk#0 s#1{a#2 b#3 c#6} t#4}: s.c was added after another column
was dropped. */
+ private static RowType evolvedTable() {
+ RowType nested =
+ new RowType(
+ Arrays.asList(
+ new DataField(2, "a", DataTypes.INT()),
+ new DataField(3, "b", DataTypes.STRING()),
+ new DataField(6, "c", DataTypes.INT())));
+ return new RowType(
+ Arrays.asList(
+ new DataField(0, "pk", DataTypes.INT()),
+ new DataField(1, "s", nested),
+ new DataField(4, "t", DataTypes.INT())));
+ }
+
+ //
------------------------------------------------------------------------------------
+ // paths
+ //
------------------------------------------------------------------------------------
+
+ @Test
+ public void testPathThroughSeveralRows() {
+ assertThat(
+ convert(
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ intRef("s", "inner", "x"),
+ new ValueLiteralExpression(7))))
+ .isEqualTo(BUILDER.equal(S_INNER_X, 7));
+ }
+
+ @Test
+ public void testNestedCombinesWithTopLevelPredicates() {
+ ResolvedExpression nested =
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ intRef("s", "a"),
+ new ValueLiteralExpression(7));
+ ResolvedExpression topLevel =
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ new FieldReferenceExpression(
+ "pk",
org.apache.flink.table.api.DataTypes.INT(), 0, 0),
+ new ValueLiteralExpression(1));
+
+ assertThat(convert(call(BuiltInFunctionDefinitions.AND, nested,
topLevel)))
+ .isEqualTo(PredicateBuilder.and(BUILDER.equal(S_A, 7),
BUILDER.equal(0, 1)));
+ assertThat(convert(call(BuiltInFunctionDefinitions.OR, nested,
topLevel)))
+ .isEqualTo(PredicateBuilder.or(BUILDER.equal(S_A, 7),
BUILDER.equal(0, 1)));
+ }
+
+ /** A path whose leaf is not a field of the row is left for Flink to
evaluate. */
+ @Test
+ public void testUnknownLeafIsNotConverted() {
+ assertThat(
+ convert(
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ intRef("s", "missing"),
+ new ValueLiteralExpression(7))))
+ .isNull();
+ }
+
+ /** Nor is a path rooted at a field the table does not have. */
+ @Test
+ public void testUnknownRootIsNotConverted() {
+ assertThat(
+ convert(
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ intRef("missing", "a"),
+ new ValueLiteralExpression(7))))
+ .isNull();
+ }
+
+ /** Only rows can be descended into; a path rooted at a non-row field is
not converted. */
+ @Test
+ public void testPathUnderNonRowFieldIsNotConverted() {
+ assertThat(
+ convert(
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ intRef("t", "a"),
+ new ValueLiteralExpression(7))))
+ .isNull();
+ }
+
+ /** A reference that names only a top-level field is not a nested path. */
+ @Test
+ public void testSingleComponentPathIsNotConverted() {
+ assertThat(
+ convert(
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ intRef("pk"),
+ new ValueLiteralExpression(7))))
+ .isNull();
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/NestedFieldFilterPushDownITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/NestedFieldFilterPushDownITCase.java
new file mode 100644
index 0000000000..b01051d235
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/NestedFieldFilterPushDownITCase.java
@@ -0,0 +1,392 @@
+/*
+ * 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.flink.source;
+
+import org.apache.paimon.catalog.TableQueryAuthResult;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.flink.sink.FlinkTableSink;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateRemapper;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.FileStoreTable;
+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.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.transformations.SourceTransformation;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.expressions.CallExpression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.expressions.NestedFieldReferenceExpression;
+import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/**
+ * ITCase for predicates on a field nested inside a row: that pushing them
down leaves every result
+ * unchanged, that a real SQL plan hands them to the scan, and that the scan
reads less for it.
+ *
+ * <p>The scan digest in {@code EXPLAIN} is deliberately not asserted on.
{@link
+ * FlinkTableSource#applyFilters} reports every filter as accepted whether or
not it could be
+ * converted, so {@code filter=[...]} reads the same either way. Instead, the
source a real SQL plan
+ * ends up with is taken out of the translated job, and its own {@link
ReadBuilder} is used to count
+ * what the scan returns.
+ */
+public class NestedFieldFilterPushDownITCase extends CatalogITCaseBase {
+
+ @Override
+ public List<String> ddl() {
+ return Arrays.asList(
+ "CREATE TABLE NT (pk INT, s ROW<a INT, b STRING>, d DOUBLE)",
+ "CREATE TABLE FT (pk INT, s ROW<d DOUBLE, f FLOAT>, d DOUBLE)",
+ "CREATE TABLE PPT (dt STRING, pk INT, s ROW<a INT>,"
+ + " PRIMARY KEY (dt, pk) NOT ENFORCED) PARTITIONED BY
(dt)"
+ + " WITH ('bucket' = '1')");
+ }
+
+ //
------------------------------------------------------------------------------------
+ // results do not change
+ //
------------------------------------------------------------------------------------
+
+ @Test
+ public void testNestedPredicateKeepsMatchingRows() {
+ batchSql(
+ "INSERT INTO NT VALUES (1, ROW(7, 'abc'), 1.5), (2, ROW(8,
'xyz'), 2.5),"
+ + " (3, CAST(NULL AS ROW<a INT, b STRING>), 3.5)");
+
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.a =
7")).containsExactly(Row.of(1));
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.a >
7")).containsExactly(Row.of(2));
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.a IN (7, 8)"))
+ .containsExactlyInAnyOrder(Row.of(1), Row.of(2));
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.b LIKE
'ab%%'")).containsExactly(Row.of(1));
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.a IS
NULL")).containsExactly(Row.of(3));
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.a IS NOT NULL"))
+ .containsExactlyInAnyOrder(Row.of(1), Row.of(2));
+ }
+
+ @Test
+ public void testNegatedNestedPredicateKeepsMatchingRows() {
+ batchSql("INSERT INTO NT VALUES (1, ROW(7, 'abc'), 1.5), (2, ROW(8,
'xyz'), 2.5)");
+
+ assertThat(batchSql("SELECT pk FROM NT WHERE NOT (s.a =
7)")).containsExactly(Row.of(2));
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.a NOT IN
(7)")).containsExactly(Row.of(2));
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.a NOT BETWEEN 1 AND 7"))
+ .containsExactly(Row.of(2));
+ }
+
+ @Test
+ public void testNestedAndTopLevelPredicatesCombine() {
+ batchSql(
+ "INSERT INTO NT VALUES (1, ROW(7, 'abc'), 1.5), (2, ROW(7,
'xyz'), 2.5),"
+ + " (3, ROW(8, 'abc'), 3.5)");
+
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.a = 7 AND pk = 2"))
+ .containsExactly(Row.of(2));
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.a = 8 OR pk = 1"))
+ .containsExactlyInAnyOrder(Row.of(1), Row.of(3));
+ }
+
+ //
------------------------------------------------------------------------------------
+ // a real SQL plan hands the predicate to the scan, and the scan prunes on
it
+ //
------------------------------------------------------------------------------------
+
+ /**
+ * Two files whose {@code s.a} ranges do not overlap. The count is what
the planned source's own
+ * scan returns, with no row-by-row filtering, so it can only drop below
the table's six rows if
+ * the predicate reached the scan and whole row groups were skipped.
+ */
+ @Test
+ public void testRealPlanPushesNestedPredicateToScan() throws Exception {
+ writeTwoFilesWithDisjointNestedValues();
+
+ assertThat(rowsScannedByPlan("SELECT pk FROM NT")).isEqualTo(6);
+ // 3, not 1: the second file's row group is skipped whole and the
first file's three rows
+ // all come back. A 1 would mean rows were filtered individually
rather than pruned.
+ assertThat(rowsScannedByPlan("SELECT pk FROM NT WHERE s.a =
2")).isEqualTo(3);
+ assertThat(rowsScannedByPlan("SELECT pk FROM NT WHERE s.a =
9999")).isZero();
+ assertThat(rowsScannedByPlan("SELECT pk FROM NT WHERE s.a >
0")).isEqualTo(6);
+
+ // and pruning never costs a matching row
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.a =
2")).containsExactly(Row.of(2));
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.a = 9999")).isEmpty();
+ }
+
+ //
------------------------------------------------------------------------------------
+ // floating point: nested comparisons are left to Flink
+ //
------------------------------------------------------------------------------------
+
+ /** Flink SQL treats {@code -0.0 = 0.0} as true; the nested row must not
be pruned away. */
+ @Test
+ public void testNestedDoubleSignedZeroKeepsMatchingRows() throws Exception
{
+ writeNegativeZero();
+ assertThat(batchSql("SELECT pk FROM FT WHERE s.d =
0.0")).containsExactly(Row.of(1));
+ assertThat(batchSql("SELECT pk FROM FT WHERE s.f =
0.0")).containsExactly(Row.of(1));
+ }
+
+ /**
+ * Two files whose nested floating-point values do not overlap, so any of
these predicates would
+ * prune one of them if it reached the scan. None may: the scan must
return every row. {@code IS
+ * NULL} is not a comparison and is still pushed down, so it does prune.
+ */
+ @Test
+ public void testNestedFloatingPointComparisonsAreNotPushedToScan() throws
Exception {
+ batchSql(
+ "INSERT INTO FT VALUES (1, ROW(1.0, CAST(1.0 AS FLOAT)), 1.0),"
+ + " (2, ROW(2.0, CAST(2.0 AS FLOAT)), 2.0)");
+ batchSql(
+ "INSERT INTO FT VALUES (3, ROW(1001.0, CAST(1001.0 AS FLOAT)),
3.0),"
+ + " (4, ROW(1002.0, CAST(1002.0 AS FLOAT)), 4.0)");
+
+ for (String where :
+ new String[] {
+ "s.d = 1.0",
+ "s.d <> 1.0",
+ "s.d < 5.0",
+ "s.d <= 5.0",
+ "s.d > 1000.0",
+ "s.d >= 1000.0",
+ "5.0 > s.d",
+ "s.d IN (1.0, 2.0)",
+ "s.d BETWEEN 0.0 AND 5.0",
+ "s.f = 1.0",
+ "s.f < 5.0"
+ }) {
+ assertThat(rowsScannedByPlan("SELECT pk FROM FT WHERE " + where))
+ .as("%s must be left to Flink, not pushed to the scan",
where)
+ .isEqualTo(4);
+ }
+
+ assertThat(rowsScannedByPlan("SELECT pk FROM FT WHERE s.d IS
NULL")).isZero();
+
+ // and Flink still evaluates them correctly
+ assertThat(batchSql("SELECT pk FROM FT WHERE s.d =
1.0")).containsExactly(Row.of(1));
+ assertThat(batchSql("SELECT pk FROM FT WHERE s.d BETWEEN 0.0 AND 5.0"))
+ .containsExactlyInAnyOrder(Row.of(1), Row.of(2));
+ }
+
+ //
------------------------------------------------------------------------------------
+ // field ids: the predicate carries the table's own ids
+ //
------------------------------------------------------------------------------------
+
+ /**
+ * On an evolved schema, field ids are not the ones a type round-tripped
through Flink would
+ * number: here {@code s.c} is added after a column was dropped. A masked
read remaps the filter
+ * onto the table's read schema ({@code ReadTransform} keeps the conjuncts
on masked columns via
+ * {@code TableQueryAuthResult.retainFields}, then {@code
PredicateRemapper.remap}) and the
+ * nested transform checks the ids it carries against the table's.
+ *
+ * <p>The predicate is taken from {@link FlinkTableSource#applyFilters}
itself, so this covers
+ * how the source builds it, not only the converter.
+ */
+ @Test
+ public void testNestedPredicateKeepsTableFieldIdsOnEvolvedSchema() throws
Exception {
+ evolveNestedRow();
+ FileStoreTable table = paimonTable("NT");
+
+ DataTableSource source =
+ new DataTableSource(
+ ObjectIdentifier.of("PAIMON", "default", "NT"), table,
false, null);
+ source.applyFilters(Collections.singletonList(nestedEqualsInt("s",
"c", 1)));
+ assertThat(source.predicate).as("s.c = 1 must be
converted").isNotNull();
+
+ Predicate onMaskedColumn =
+ TableQueryAuthResult.retainFields(source.predicate,
Collections.singleton("s"));
+ assertThatCode(() -> PredicateRemapper.remap(onMaskedColumn,
table.rowType()))
+ .doesNotThrowAnyException();
+ }
+
+ /** Queries on an evolved nested row keep returning the right rows. */
+ @Test
+ public void testNestedPredicateOnEvolvedSchemaKeepsMatchingRows() {
+ batchSql("INSERT INTO NT VALUES (1, ROW(7, 'x'), 1.0)");
+ evolveNestedRow();
+ batchSql("INSERT INTO NT VALUES (2, ROW(8, 'y', 1), 2.0), (3, ROW(9,
'z', 2), 3.0)");
+
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.c =
1")).containsExactly(Row.of(2));
+ // rows written before s.c existed read it as null
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.c IS
NULL")).containsExactly(Row.of(1));
+ assertThat(batchSql("SELECT pk FROM NT WHERE s.a =
7")).containsExactly(Row.of(1));
+ }
+
+ //
------------------------------------------------------------------------------------
+ // DELETE: a nested predicate is never executed by Paimon as a partition
drop
+ //
------------------------------------------------------------------------------------
+
+ /**
+ * {@code applyDeleteFilters} returning true hands the whole DELETE to
Paimon, which can only
+ * execute it by dropping partitions. A nested predicate must never
qualify, alone or next to a
+ * partition key, or rows it does not match would be deleted with the
partition.
+ *
+ * <p>Called directly rather than through {@code DELETE FROM ... WHERE s.a
= ...}: Flink's own
+ * delete push-down resolves such a filter without a row type and fails in
the planner before
+ * this sink is ever asked.
+ */
+ @Test
+ public void testNestedDeleteFilterIsNeverExecutedByPaimon() throws
Exception {
+ ObjectIdentifier identifier = ObjectIdentifier.of("PAIMON", "default",
"PPT");
+ ResolvedExpression partition =
+ CallExpression.permanent(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(
+ new FieldReferenceExpression("dt",
DataTypes.STRING(), 0, 0),
+ new ValueLiteralExpression("p1")),
+ DataTypes.BOOLEAN());
+ ResolvedExpression nested = nestedEqualsInt("s", "a", 2);
+
+ // control: a partition key alone is handed to Paimon as a partition
drop
+ assertThat(
+ new FlinkTableSink(identifier, paimonTable("PPT"),
null)
+
.applyDeleteFilters(Collections.singletonList(partition)))
+ .isTrue();
+
+ assertThat(
+ new FlinkTableSink(identifier, paimonTable("PPT"),
null)
+
.applyDeleteFilters(Collections.singletonList(nested)))
+ .isFalse();
+ assertThat(
+ new FlinkTableSink(identifier, paimonTable("PPT"),
null)
+ .applyDeleteFilters(Arrays.asList(partition,
nested)))
+ .as("a nested predicate next to a partition key must not drop
the partition")
+ .isFalse();
+ }
+
+ //
------------------------------------------------------------------------------------
+ // helpers
+ //
------------------------------------------------------------------------------------
+
+ private void writeTwoFilesWithDisjointNestedValues() {
+ // two commits, so the rows land in two files whose s.a ranges do not
overlap
+ batchSql(
+ "INSERT INTO NT VALUES (1, ROW(1, 'x'), 1.0), (2, ROW(2, 'x'),
1.0),"
+ + " (3, ROW(3, 'x'), 1.0)");
+ batchSql(
+ "INSERT INTO NT VALUES (4, ROW(1001, 'y'), 2.0), (5, ROW(1002,
'y'), 2.0),"
+ + " (6, ROW(1003, 'y'), 2.0)");
+ }
+
+ private void evolveNestedRow() {
+ // leaves a gap in the field ids, then adds s.c after it
+ batchSql("ALTER TABLE NT ADD extra INT");
+ batchSql("ALTER TABLE NT DROP extra");
+ batchSql("ALTER TABLE NT MODIFY s ROW<a INT, b STRING, c INT>");
+ }
+
+ private static ResolvedExpression nestedEqualsInt(String root, String
leaf, int literal) {
+ return CallExpression.permanent(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(
+ new NestedFieldReferenceExpression(
+ new String[] {root, leaf}, new int[] {1, 2},
DataTypes.INT()),
+ new ValueLiteralExpression(literal)),
+ DataTypes.BOOLEAN());
+ }
+
+ /**
+ * Writes one row whose nested and top-level floating-point values are all
negative zero.
+ * Written through the table API because SQL cannot express it: {@code
CAST('-0.0' AS DOUBLE)}
+ * goes through a decimal, which has no negative zero, and comes back as
{@code 0.0}.
+ */
+ private void writeNegativeZero() throws Exception {
+ FileStoreTable table = paimonTable("FT");
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(GenericRow.of(1, GenericRow.of(-0.0d, -0.0f), -0.0d));
+ commit.commit(write.prepareCommit());
+ }
+
+ // it really is stored as negative zero
+ List<InternalRow> rows = new ArrayList<>();
+ ReadBuilder readBuilder = table.newReadBuilder();
+ readBuilder
+ .newRead()
+ .createReader(readBuilder.newScan().plan())
+ .forEachRemaining(rows::add);
+ assertThat(rows).hasSize(1);
+ InternalRow nested = rows.get(0).getRow(1, 2);
+ assertThat(Double.doubleToRawLongBits(nested.getDouble(0)))
+ .isEqualTo(Double.doubleToRawLongBits(-0.0d));
+ assertThat(Float.floatToRawIntBits(nested.getFloat(1)))
+ .isEqualTo(Float.floatToRawIntBits(-0.0f));
+ }
+
+ /**
+ * Plans {@code sql} the way a real query is planned, takes the source out
of the translated
+ * job, and counts what that source's own scan returns. Nothing filters
row by row here: a count
+ * below the table's size means the predicate reached the scan and pruned.
+ */
+ private int rowsScannedByPlan(String sql) throws Exception {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setRuntimeMode(RuntimeExecutionMode.BATCH);
+ StreamTableEnvironment planner = StreamTableEnvironment.create(env);
+ planner.registerCatalog("PAIMON", tEnv.getCatalog("PAIMON").get());
+ planner.useCatalog("PAIMON");
+
+ FlinkSource source =
+
plannedSource(planner.toDataStream(planner.sqlQuery(sql)).getTransformation());
+ ReadBuilder readBuilder = source.readBuilder;
+ AtomicInteger count = new AtomicInteger();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(row -> count.incrementAndGet());
+ }
+ return count.get();
+ }
+
+ private static FlinkSource plannedSource(Transformation<?> transformation)
throws Exception {
+ if (transformation instanceof SourceTransformation) {
+ Source<?, ?, ?> source = ((SourceTransformation<?, ?, ?>)
transformation).getSource();
+ if (source instanceof PaimonDataStreamSource) {
+ // the wrapper keeps the source it delegates to private
+ Field inner =
PaimonDataStreamSource.class.getDeclaredField("source");
+ inner.setAccessible(true);
+ source = (Source<?, ?, ?>) inner.get(source);
+ }
+ return (FlinkSource) source;
+ }
+ for (Transformation<?> input : transformation.getInputs()) {
+ FlinkSource found = plannedSource(input);
+ if (found != null) {
+ return found;
+ }
+ }
+ return null;
+ }
+}
