This is an automated email from the ASF dual-hosted git repository.
AHeise pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 69b9cbb14c7 [FLINK-40909][table-planner] Support filter push down for
element access on ARRAY, MAP and VARIANT
69b9cbb14c7 is described below
commit 69b9cbb14c7243f95918e690c801e20d856bc015
Author: Arvid Heise <[email protected]>
AuthorDate: Mon Oct 5 13:56:19 2026 +0200
[FLINK-40909][table-planner] Support filter push down for element access on
ARRAY, MAP and VARIANT
SQL element access (`a[1]`, `m['k']`, `v['k']`) is Calcite's ITEM operator.
RexNodeToExpressionConverter resolves operators it does not map explicitly by
name, finds no built-in function called ITEM, and classifies the whole conjunct
as unconvertible, so such predicates never reach
SupportsFilterPushDown#applyFilters.
ITEM on an ARRAY, MAP or VARIANT operand now converts to
BuiltInFunctionDefinitions.AT, the reverse of the existing AT -> ITEM mapping
in DirectConvertRule. ROW and structured operands, which AT does not accept,
stay unconvertible. The values test source learns to evaluate AT, and CAST or
TRY_CAST around AT for VARIANT elements, so the push down on all three types is
covered by plan and restore tests.
Generated-by: Claude Code (Claude Opus 5.5)
---
.../planner/plan/utils/RexNodeExtractor.scala | 8 +-
.../planner/factories/TestValuesTableFactory.java | 9 +-
.../exec/common/TableSourceScanTestPrograms.java | 52 ++++++
.../exec/stream/TableSourceScanRestoreTest.java | 9 +
.../PushFilterIntoTableSourceScanRuleTest.java | 34 ++++
.../flink/table/planner/utils/FilterUtils.java | 202 ++++++++++++++++++++-
.../PushFilterIntoTableSourceScanRuleTest.xml | 73 ++++++++
...can-filter-pushdown-on-collection-elements.json | 163 +++++++++++++++++
.../savepoint/_metadata | Bin 0 -> 5889 bytes
.../planner/plan/utils/RexNodeExtractorTest.scala | 68 ++++++-
10 files changed, 603 insertions(+), 15 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala
index 8ebefbbdc95..24a4cf72a00 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala
@@ -25,7 +25,7 @@ import
org.apache.flink.table.data.util.DataFormatConverters.{LocalDateConverter
import org.apache.flink.table.expressions._
import org.apache.flink.table.expressions.ApiExpressionUtils._
import org.apache.flink.table.functions.{BuiltInFunctionDefinition,
FunctionIdentifier}
-import org.apache.flink.table.functions.BuiltInFunctionDefinitions.{AND, CAST,
OR, TRY_CAST}
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions.{AND, AT,
CAST, OR, TRY_CAST}
import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.codegen.CodeGenException
import org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable
@@ -533,6 +533,12 @@ class RexNodeToExpressionConverter(
Option(
CallExpression
.permanent(TRY_CAST, Seq(operands.head,
typeLiteral(outputType)).asJava, outputType))
+ case FlinkSqlOperatorTable.ITEM =>
+ operands.head.getOutputDataType.getLogicalType.getTypeRoot match {
+ case ARRAY | MAP | VARIANT =>
+ Option(CallExpression.permanent(AT, operands.asJava, outputType))
+ case _ => None
+ }
case _: SqlFunction | _: SqlPostfixOperator =>
val names = new
util.ArrayList[String](rexCall.getOperator.getNameAsId.names)
names.set(names.size() - 1, replace(names.get(names.size() - 1)))
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java
index 7049fc1c8c1..746f68ab9ca 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java
@@ -1290,22 +1290,21 @@ public final class TestValuesTableFactory
return Result.of(acceptedFilters, remainingFilters);
}
- private Function<String, Comparable<?>> getValueGetter(Row row) {
+ private Function<String, Object> getValueGetter(Row row) {
final List<String> fieldNames =
DataTypeUtils.flattenToNames(producedDataType);
return fieldName -> {
int idx = fieldNames.indexOf(fieldName);
- return (Comparable<?>) row.getField(idx);
+ return row.getField(idx);
};
}
- private Function<int[], Comparable<?>> getNestedValueGetter(Row row) {
+ private Function<int[], Object> getNestedValueGetter(Row row) {
return fieldIndices -> {
Object current = row;
for (int i = 0; i < fieldIndices.length - 1; i++) {
current = ((Row) current).getField(fieldIndices[i]);
}
- return (Comparable<?>)
- ((Row)
current).getField(fieldIndices[fieldIndices.length - 1]);
+ return ((Row)
current).getField(fieldIndices[fieldIndices.length - 1]);
};
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/TableSourceScanTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/TableSourceScanTestPrograms.java
index 13e2816bb85..5ff4e1df74e 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/TableSourceScanTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/TableSourceScanTestPrograms.java
@@ -23,10 +23,16 @@ import org.apache.flink.table.test.program.SourceTestStep;
import org.apache.flink.table.test.program.TableTestProgram;
import org.apache.flink.table.utils.DateTimeUtils;
import org.apache.flink.types.Row;
+import org.apache.flink.types.variant.Variant;
+import org.apache.flink.types.variant.VariantBuilder;
+
+import java.util.Map;
/** {@link TableTestProgram} definitions for testing {@link
StreamExecTableSourceScan}. */
public class TableSourceScanTestPrograms {
+ private static final VariantBuilder VARIANT_BUILDER = Variant.newBuilder();
+
static final Row[] BEFORE_DATA = {
Row.of(1, 1L, "hi", DateTimeUtils.toLocalDateTime(1586937601000L)),
Row.of(2, 2L, "hello", DateTimeUtils.toLocalDateTime(1586937602000L)),
@@ -105,6 +111,44 @@ public class TableSourceScanTestPrograms {
.runSql("INSERT INTO sink_t SELECT * FROM source_t WHERE a
> 1")
.build();
+ public static final TableTestProgram
FILTER_PUSHDOWN_ON_COLLECTION_ELEMENTS =
+ TableTestProgram.of(
+
"table-source-scan-filter-pushdown-on-collection-elements",
+ "validates table source scan with pushed filters
on array, map and"
+ + " variant elements")
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+ .addSchema(
+ "a INT",
+ "arr ARRAY<INT>",
+ "m MAP<STRING, INT>",
+ "v VARIANT")
+ .addOption("filterable-fields", "arr;m;v")
+ .producedBeforeRestore(
+ collectionRow(1, 2, 2, 2),
+ collectionRow(2, 1, 2, 2),
+ collectionRow(3, 2, 1, 2),
+ collectionRow(4, 2, 2, 1))
+ .producedAfterRestore(
+ collectionRow(5, 3, 3, 3),
+ // empty containers: every element
access is NULL
+ Row.of(
+ 6,
+ new Integer[0],
+ Map.of(),
+
VARIANT_BUILDER.object().build()))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema("a INT")
+ .consumedBeforeRestore("+I[1]")
+ .consumedAfterRestore("+I[5]")
+ .build())
+ .runSql(
+ "INSERT INTO sink_t SELECT a FROM source_t"
+ + " WHERE arr[1] > 1 AND m['k'] > 1 AND
CAST(v['k'] AS INT) > 1")
+ .build();
+
public static final TableTestProgram LIMIT_PUSHDOWN =
TableTestProgram.of(
"table-source-scan-limit-pushdown",
@@ -260,4 +304,12 @@ public class TableSourceScanTestPrograms {
"INSERT INTO sink_one_t SELECT a, c FROM source_t",
"INSERT INTO sink_two_t SELECT a, b FROM source_t")
.build();
+
+ private static Row collectionRow(int a, int arrayElement, int mapValue,
int variantField) {
+ return Row.of(
+ a,
+ new Integer[] {arrayElement},
+ Map.of("k", mapValue),
+ VARIANT_BUILDER.object().add("k",
VARIANT_BUILDER.of(variantField)).build());
+ }
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSourceScanRestoreTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSourceScanRestoreTest.java
index ccc5133cabb..858accafe10 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSourceScanRestoreTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSourceScanRestoreTest.java
@@ -24,6 +24,7 @@ import org.apache.flink.table.test.program.TableTestProgram;
import java.util.Arrays;
import java.util.List;
+import java.util.Map;
/** Restore tests for {@link StreamExecTableSourceScan}. */
public class TableSourceScanRestoreTest extends RestoreTestBase {
@@ -38,6 +39,7 @@ public class TableSourceScanRestoreTest extends
RestoreTestBase {
TableSourceScanTestPrograms.PROJECT_PUSHDOWN,
TableSourceScanTestPrograms.PROJECT_PUSHDOWN_DISABLED,
TableSourceScanTestPrograms.FILTER_PUSHDOWN,
+
TableSourceScanTestPrograms.FILTER_PUSHDOWN_ON_COLLECTION_ELEMENTS,
TableSourceScanTestPrograms.LIMIT_PUSHDOWN,
TableSourceScanTestPrograms.PARTITION_PUSHDOWN,
TableSourceScanTestPrograms.READING_METADATA,
@@ -45,4 +47,11 @@ public class TableSourceScanRestoreTest extends
RestoreTestBase {
TableSourceScanTestPrograms.SOURCE_WATERMARK,
TableSourceScanTestPrograms.REUSE_SOURCE);
}
+
+ @Override
+ protected Map<Integer, List<TableTestProgram>> programsToIgnore() {
+ return Map.of(
+ // plans of this program exist only for version 2
+ 1,
List.of(TableSourceScanTestPrograms.FILTER_PUSHDOWN_ON_COLLECTION_ELEMENTS));
+ }
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.java
index da21dbc5b33..b294db59e89 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.java
@@ -120,6 +120,20 @@ class PushFilterIntoTableSourceScanRuleTest extends
PushFilterIntoTableSourceSca
+ " 'bounded' = 'true'\n"
+ ")";
util.tableEnv().executeSql(ddl4);
+
+ String ddl5 =
+ "CREATE TABLE ItemTable (\n"
+ + " id INT,\n"
+ + " arr ARRAY<INT>,\n"
+ + " m MAP<STRING, INT>,\n"
+ + " v VARIANT,\n"
+ + " nested ROW<arr ARRAY<INT>>\n"
+ + ") WITH (\n"
+ + " 'connector' = 'values',\n"
+ + " 'filterable-fields' = 'arr;m;v;`nested.arr`',\n"
+ + " 'bounded' = 'true'\n"
+ + ")";
+ util.tableEnv().executeSql(ddl5);
}
@Test
@@ -204,4 +218,24 @@ class PushFilterIntoTableSourceScanRuleTest extends
PushFilterIntoTableSourceSca
util.verifyRelPlan(
"SELECT * FROM NestedItemTable WHERE
`Result`.`Mid`.data_arr[2].`value` = 3");
}
+
+ @Test
+ void testFilterOnArrayElement() {
+ util.verifyRelPlan("SELECT * FROM ItemTable WHERE arr[1] = 3");
+ }
+
+ @Test
+ void testFilterOnMapValue() {
+ util.verifyRelPlan("SELECT * FROM ItemTable WHERE m['k'] > 2 AND id >
1");
+ }
+
+ @Test
+ void testFilterOnVariantField() {
+ util.verifyRelPlan("SELECT * FROM ItemTable WHERE CAST(v['k'] AS INT)
> 2");
+ }
+
+ @Test
+ void testFilterOnNestedArrayElement() {
+ util.verifyRelPlan("SELECT * FROM ItemTable WHERE nested.arr[1] = 3");
+ }
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/FilterUtils.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/FilterUtils.java
index 32d3afa233a..6965d55cef7 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/FilterUtils.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/utils/FilterUtils.java
@@ -27,21 +27,89 @@ 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.table.functions.FunctionDefinition;
+import org.apache.flink.table.runtime.functions.VariantCastUtils;
+import org.apache.flink.table.types.logical.CharType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.types.variant.Variant;
import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
+import java.lang.reflect.Array;
+import java.util.EnumMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.TimeZone;
+import java.util.function.BiFunction;
import java.util.function.Function;
+import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.AT;
+import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.CAST;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.LOWER;
+import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.TRY_CAST;
import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.UPPER;
/** Utils for catalog and source to filter partition or row. */
public class FilterUtils {
+ private static final TimeZone UTC = TimeZone.getTimeZone("UTC");
+
+ /**
+ * Casts of a VARIANT element that can be evaluated, by target type, using
the same {@link
+ * VariantCastUtils} helpers as the generated code of {@code
VariantToPrimitiveCastRule}.
+ */
+ private static final Map<LogicalTypeRoot, BiFunction<Variant, LogicalType,
Object>>
+ VARIANT_CASTS = new EnumMap<>(LogicalTypeRoot.class);
+
+ static {
+ VARIANT_CASTS.put(LogicalTypeRoot.BOOLEAN, (variant, type) ->
variant.getBoolean());
+ VARIANT_CASTS.put(
+ LogicalTypeRoot.TINYINT,
+ (variant, type) ->
+ (byte)
+ VariantCastUtils.toIntegral(
+ variant, Byte.MIN_VALUE,
Byte.MAX_VALUE, "TINYINT"));
+ VARIANT_CASTS.put(
+ LogicalTypeRoot.SMALLINT,
+ (variant, type) ->
+ (short)
+ VariantCastUtils.toIntegral(
+ variant, Short.MIN_VALUE,
Short.MAX_VALUE, "SMALLINT"));
+ VARIANT_CASTS.put(
+ LogicalTypeRoot.INTEGER,
+ (variant, type) ->
+ (int)
+ VariantCastUtils.toIntegral(
+ variant, Integer.MIN_VALUE,
Integer.MAX_VALUE, "INT"));
+ VARIANT_CASTS.put(
+ LogicalTypeRoot.BIGINT,
+ (variant, type) ->
+ VariantCastUtils.toIntegral(
+ variant, Long.MIN_VALUE, Long.MAX_VALUE,
"BIGINT"));
+ VARIANT_CASTS.put(
+ LogicalTypeRoot.FLOAT, (variant, type) ->
VariantCastUtils.toFloat(variant));
+ VARIANT_CASTS.put(
+ LogicalTypeRoot.DOUBLE, (variant, type) ->
VariantCastUtils.toDouble(variant));
+ VARIANT_CASTS.put(
+ LogicalTypeRoot.CHAR,
+ (variant, type) ->
+ VariantCastUtils.toStringValue(
+ variant, UTC, ((CharType)
type).getLength(), true)
+ .toString());
+ VARIANT_CASTS.put(
+ LogicalTypeRoot.VARCHAR,
+ (variant, type) ->
+ VariantCastUtils.toStringValue(
+ variant, UTC, ((VarCharType)
type).getLength(), false)
+ .toString());
+ }
+
public static boolean shouldPushDown(ResolvedExpression expr, Set<String>
filterableFields) {
if (expr instanceof CallExpression && expr.getChildren().size() == 2) {
return shouldPushDownUnaryExpression(
@@ -54,8 +122,8 @@ public class FilterUtils {
public static boolean isRetainedAfterApplyingFilterPredicates(
List<ResolvedExpression> predicates,
- Function<String, Comparable<?>> getter,
- @Nullable Function<int[], Comparable<?>> nestedFieldGetter) {
+ Function<String, ?> getter,
+ @Nullable Function<int[], ?> nestedFieldGetter) {
for (ResolvedExpression predicate : predicates) {
if (predicate instanceof CallExpression) {
FunctionDefinition definition =
@@ -115,6 +183,19 @@ public class FilterUtils {
return true;
}
+ if (isElementAccess(expr, filterableFields)) {
+ return true;
+ }
+
+ // A cast makes an element of a VARIANT comparable, e.g. CAST(v['k']
AS INT).
+ if (isCall(expr, CAST) || isCall(expr, TRY_CAST)) {
+ final ResolvedExpression element =
expr.getResolvedChildren().get(0);
+ return VARIANT_CASTS.containsKey(
+
expr.getOutputDataType().getLogicalType().getTypeRoot())
+ &&
element.getOutputDataType().getLogicalType().is(LogicalTypeRoot.VARIANT)
+ && isElementAccess(element, filterableFields);
+ }
+
if (expr instanceof CallExpression && expr.getChildren().size() == 1) {
if (((CallExpression) expr).getFunctionDefinition().equals(UPPER)
|| ((CallExpression)
expr).getFunctionDefinition().equals(LOWER)) {
@@ -126,16 +207,60 @@ public class FilterUtils {
return false;
}
+ private static boolean isElementAccess(ResolvedExpression expr,
Set<String> filterableFields) {
+ if (!isCall(expr, AT)) {
+ return false;
+ }
+ final ResolvedExpression container = expr.getResolvedChildren().get(0);
+ final ResolvedExpression key = expr.getResolvedChildren().get(1);
+ return isFilterableField(container, filterableFields)
+ && key instanceof ValueLiteralExpression
+ && isEvaluableKey(
+ container.getOutputDataType().getLogicalType(),
+ key.getOutputDataType().getLogicalType());
+ }
+
+ /** A map lookup only finds the key if the literal has the key's Java
type. */
+ private static boolean isEvaluableKey(LogicalType containerType,
LogicalType keyType) {
+ if (!containerType.is(LogicalTypeRoot.MAP)) {
+ return true;
+ }
+ final LogicalType mapKeyType = ((MapType) containerType).getKeyType();
+ return mapKeyType.is(keyType.getTypeRoot())
+ || (mapKeyType.is(LogicalTypeFamily.CHARACTER_STRING)
+ && keyType.is(LogicalTypeFamily.CHARACTER_STRING));
+ }
+
+ private static boolean isCall(Expression expr, FunctionDefinition
definition) {
+ return expr instanceof CallExpression
+ && ((CallExpression)
expr).getFunctionDefinition().equals(definition);
+ }
+
+ private static boolean isFilterableField(
+ ResolvedExpression expr, Set<String> filterableFields) {
+ if (expr instanceof FieldReferenceExpression) {
+ return filterableFields.contains(((FieldReferenceExpression)
expr).getName());
+ }
+ if (expr instanceof NestedFieldReferenceExpression) {
+ return filterableFields.contains(((NestedFieldReferenceExpression)
expr).getName());
+ }
+ return false;
+ }
+
@SuppressWarnings({"unchecked", "rawtypes"})
private static boolean binaryFilterApplies(
CallExpression binExpr,
- Function<String, Comparable<?>> getter,
- Function<int[], Comparable<?>> nestedFieldGetter) {
+ Function<String, ?> getter,
+ Function<int[], ?> nestedFieldGetter) {
List<Expression> children = binExpr.getChildren();
Preconditions.checkArgument(children.size() == 2);
Comparable lhsValue = getValue(children.get(0), getter,
nestedFieldGetter);
Comparable rhsValue = getValue(children.get(1), getter,
nestedFieldGetter);
+ if (lhsValue == null || rhsValue == null) {
+ // A comparison with NULL is never true, e.g. for an array index
out of bounds.
+ return false;
+ }
FunctionDefinition functionDefinition =
binExpr.getFunctionDefinition();
if
(BuiltInFunctionDefinitions.GREATER_THAN.equals(functionDefinition)) {
return lhsValue.compareTo(rhsValue) > 0;
@@ -154,14 +279,44 @@ public class FilterUtils {
}
}
+ /** Returns NULL for a missing field, an out-of-range index or a container
of the wrong kind. */
+ private static @Nullable Variant getVariantElement(Variant variant, Object
key) {
+ if (key instanceof String) {
+ return variant.isObject() ? variant.getField((String) key) : null;
+ }
+ final int index = ((Number) key).intValue();
+ return variant.isArray() && 1 <= index && index <=
variant.getArraySize()
+ ? variant.getElement(index - 1)
+ : null;
+ }
+
+ /** Casts a VARIANT element like the generated code: an error yields NULL
for TRY_CAST. */
+ private static @Nullable Object castVariant(
+ @Nullable Variant variant, LogicalType targetType, boolean
tryCast) {
+ if (variant == null || variant.isNull()) {
+ return null;
+ }
+ try {
+ return VARIANT_CASTS.get(targetType.getTypeRoot()).apply(variant,
targetType);
+ } catch (RuntimeException e) {
+ if (tryCast) {
+ return null;
+ }
+ throw e;
+ }
+ }
+
private static boolean isComparable(Class<?> clazz) {
return Comparable.class.isAssignableFrom(clazz);
}
private static Comparable<?> getValue(
- Expression expr,
- Function<String, Comparable<?>> getter,
- Function<int[], Comparable<?>> nestedFieldGetter) {
+ Expression expr, Function<String, ?> getter, Function<int[], ?>
nestedFieldGetter) {
+ return (Comparable<?>) getRawValue(expr, getter, nestedFieldGetter);
+ }
+
+ private static Object getRawValue(
+ Expression expr, Function<String, ?> getter, Function<int[], ?>
nestedFieldGetter) {
if (expr instanceof ValueLiteralExpression) {
Optional<?> value =
((ValueLiteralExpression) expr)
@@ -169,7 +324,7 @@ public class FilterUtils {
((ValueLiteralExpression) expr)
.getOutputDataType()
.getConversionClass());
- return (Comparable<?>) value.orElse(null);
+ return value.orElse(null);
}
if (expr instanceof FieldReferenceExpression) {
@@ -185,6 +340,37 @@ public class FilterUtils {
}
}
+ if (isCall(expr, CAST) || isCall(expr, TRY_CAST)) {
+ return castVariant(
+ (Variant) getRawValue(expr.getChildren().get(0), getter,
nestedFieldGetter),
+ ((CallExpression)
expr).getOutputDataType().getLogicalType(),
+ isCall(expr, TRY_CAST));
+ }
+
+ if (isCall(expr, AT)) {
+ final Object container =
+ getRawValue(expr.getChildren().get(0), getter,
nestedFieldGetter);
+ final Object key = getRawValue(expr.getChildren().get(1), getter,
nestedFieldGetter);
+ if (container == null || key == null) {
+ return null;
+ }
+ if (container instanceof Map) {
+ return ((Map<?, ?>) container).get(key);
+ }
+ if (container.getClass().isArray()) {
+ // SQL array indices are 1-based.
+ final int index = ((Number) key).intValue();
+ return 1 <= index && index <= Array.getLength(container)
+ ? Array.get(container, index - 1)
+ : null;
+ }
+ if (container instanceof Variant) {
+ return getVariantElement((Variant) container, key);
+ }
+ throw new UnsupportedOperationException(
+ String.format("Unsupported container for %s: %s.", expr,
container.getClass()));
+ }
+
if (expr instanceof CallExpression && expr.getChildren().size() == 1) {
Object child = getValue(expr.getChildren().get(0), getter,
nestedFieldGetter);
FunctionDefinition functionDefinition = ((CallExpression)
expr).getFunctionDefinition();
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.xml
index c28a91adaa8..72f4d2ca52b 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRuleTest.xml
@@ -205,6 +205,79 @@ LogicalProject(name=[$0], id=[$1], amount=[$2],
virtualField=[$3], price=[$4])
+- LogicalProject(name=[$0], id=[$1], amount=[$2], virtualField=[+($2, 1)],
price=[$3])
+- LogicalFilter(condition=[>($3, 10)])
+- LogicalTableScan(table=[[default_catalog, default_database,
VirtualTable, filter=[]]])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testFilterOnArrayElement">
+ <Resource name="sql">
+ <![CDATA[SELECT * FROM ItemTable WHERE arr[1] = 3]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(id=[$0], arr=[$1], m=[$2], v=[$3], nested=[$4])
++- LogicalFilter(condition=[=(ITEM($1, 1), 3)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, ItemTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+LogicalProject(id=[$0], arr=[$1], m=[$2], v=[$3], nested=[$4])
++- LogicalTableScan(table=[[default_catalog, default_database, ItemTable,
filter=[=(ITEM(arr, 1), 3)]]])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testFilterOnMapValue">
+ <Resource name="sql">
+ <![CDATA[SELECT * FROM ItemTable WHERE m['k'] > 2 AND id > 1]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(id=[$0], arr=[$1], m=[$2], v=[$3], nested=[$4])
++- LogicalFilter(condition=[AND(>(ITEM($2, _UTF-16LE'k'), 2), >($0, 1))])
+ +- LogicalTableScan(table=[[default_catalog, default_database, ItemTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+LogicalProject(id=[$0], arr=[$1], m=[$2], v=[$3], nested=[$4])
++- LogicalFilter(condition=[>($0, 1)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, ItemTable,
filter=[>(ITEM(m, _UTF-16LE'k'), 2)]]])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testFilterOnNestedArrayElement">
+ <Resource name="sql">
+ <![CDATA[SELECT * FROM ItemTable WHERE nested.arr[1] = 3]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(id=[$0], arr=[$1], m=[$2], v=[$3], nested=[$4])
++- LogicalFilter(condition=[=(ITEM($4.arr, 1), 3)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, ItemTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+LogicalProject(id=[$0], arr=[$1], m=[$2], v=[$3], nested=[$4])
++- LogicalTableScan(table=[[default_catalog, default_database, ItemTable,
filter=[=(ITEM(nested.arr, 1), 3)]]])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testFilterOnVariantField">
+ <Resource name="sql">
+ <![CDATA[SELECT * FROM ItemTable WHERE CAST(v['k'] AS INT) > 2]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(id=[$0], arr=[$1], m=[$2], v=[$3], nested=[$4])
++- LogicalFilter(condition=[>(CAST(ITEM($3, _UTF-16LE'k')):INTEGER, 2)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, ItemTable]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+LogicalProject(id=[$0], arr=[$1], m=[$2], v=[$3], nested=[$4])
++- LogicalTableScan(table=[[default_catalog, default_database, ItemTable,
filter=[>(CAST(ITEM(v, _UTF-16LE'k') AS INTEGER), 2)]]])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-table-source-scan_2/table-source-scan-filter-pushdown-on-collection-elements/plan/table-source-scan-filter-pushdown-on-collection-elements.json
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-table-source-scan_2/table-source-scan-filter-pushdown-on-collection-elements/plan/table-source-scan-filter-pushdown-on-collection-elements.json
new file mode 100644
index 00000000000..2adaa8164c4
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-table-source-scan_2/table-source-scan-filter-pushdown-on-collection-elements/plan/table-source-scan-filter-pushdown-on-collection-elements.json
@@ -0,0 +1,163 @@
+{
+ "flinkVersion" : "2.4",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "stream-exec-table-source-scan_2",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`source_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ }, {
+ "name" : "arr",
+ "dataType" : "ARRAY<INT>"
+ }, {
+ "name" : "m",
+ "dataType" : "MAP<VARCHAR(2147483647), INT>"
+ }, {
+ "name" : "v",
+ "dataType" : "VARIANT"
+ } ]
+ }
+ }
+ },
+ "abilities" : [ {
+ "type" : "FilterPushDown",
+ "predicates" : [ {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$ITEM$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "ARRAY<INT>"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 1,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 1,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$ITEM$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "MAP<VARCHAR(2147483647), INT>"
+ }, {
+ "kind" : "LITERAL",
+ "value" : "k",
+ "type" : "CHAR(1) NOT NULL"
+ } ],
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 1,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ }, {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$>$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$CAST$1",
+ "operands" : [ {
+ "kind" : "CALL",
+ "syntax" : "SPECIAL",
+ "internalName" : "$ITEM$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "VARIANT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : "k",
+ "type" : "CHAR(1) NOT NULL"
+ } ],
+ "type" : "VARIANT"
+ } ],
+ "type" : "INT"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 1,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN"
+ } ]
+ }, {
+ "type" : "ProjectPushDown",
+ "projectedFields" : [ [ 0 ] ],
+ "producedType" : "ROW<`a` INT> NOT NULL"
+ }, {
+ "type" : "ReadingMetadata",
+ "metadataKeys" : [ ],
+ "producedType" : "ROW<`a` INT> NOT NULL"
+ } ]
+ },
+ "outputType" : "ROW<`a` INT>",
+ "description" : "TableSourceScan(table=[[default_catalog,
default_database, source_t, filter=[AND(AND(>(ITEM(arr, 1), 1), >(ITEM(m,
_UTF-16LE'k'), 1)), >(CAST(ITEM(v, _UTF-16LE'k') AS INTEGER), 1))],
project=[a], metadata=[]]], fields=[a])"
+ }, {
+ "id" : 2,
+ "type" : "stream-exec-sink_2",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "INT"
+ } ]
+ }
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT" ],
+ "upsertMaterializeStrategy" : "ADAPTIVE",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` INT>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t],
fields=[a])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git
a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-table-source-scan_2/table-source-scan-filter-pushdown-on-collection-elements/savepoint/_metadata
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-table-source-scan_2/table-source-scan-filter-pushdown-on-collection-elements/savepoint/_metadata
new file mode 100644
index 00000000000..080b403ca35
Binary files /dev/null and
b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-table-source-scan_2/table-source-scan-filter-pushdown-on-collection-elements/savepoint/_metadata
differ
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala
index 4af1f7be9a6..6792afc81b4 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala
@@ -31,6 +31,8 @@ import
org.apache.flink.table.planner.functions.sql.FlinkSqlOperatorTable
import org.apache.flink.table.planner.functions.utils.ScalarSqlFunction
import org.apache.flink.table.planner.utils.{DateTimeTestUtil,
IntSumAggFunction}
import org.apache.flink.table.resource.ResourceManager
+import org.apache.flink.table.runtime.types.LogicalTypeDataTypeConverter
+import org.apache.flink.table.types.logical.IntType
import org.apache.flink.table.utils.CatalogManagerMocks
import org.apache.calcite.avatica.util.ByteString
@@ -38,7 +40,7 @@ import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rex.{RexBuilder, RexNode}
import org.apache.calcite.sql.`type`.SqlTypeName
import org.apache.calcite.sql.SqlPostfixOperator
-import org.apache.calcite.sql.fun.{SqlStdOperatorTable, SqlTrimFunction}
+import org.apache.calcite.sql.fun.{SqlLibraryOperators, SqlStdOperatorTable,
SqlTrimFunction}
import org.apache.calcite.util.{DateString, TimestampString, TimeString}
import org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.Test
@@ -498,6 +500,70 @@ class RexNodeExtractorTest extends RexNodeTestBase {
assertExpressionArrayEquals(Array($"amount" <= $"id"),
Array(convertedExpressions(1)))
}
+ @Test
+ def testExtractItem(): Unit = {
+ val fieldNames = List("arr", "map", "variant", "row").asJava
+ val fieldTypes = List(
+ DataTypes.ARRAY(DataTypes.INT()),
+ DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()),
+ DataTypes.VARIANT(),
+ DataTypes.ROW(DataTypes.FIELD("f", DataTypes.INT()))
+ )
+ .map(LogicalTypeDataTypeConverter.fromDataTypeToLogicalType)
+ .map(typeFactory.createFieldTypeFromLogicalType)
+ val Seq(arr, map, variant, row) =
+ fieldTypes.zipWithIndex.map { case (t, i) => rexBuilder.makeInputRef(t,
i) }
+ val three = rexBuilder.makeExactLiteral(BigDecimal.valueOf(3))
+ val intType = typeFactory.createFieldTypeFromLogicalType(new IntType())
+
+ // arr[1] = 3
+ val onArray = rexBuilder.makeCall(
+ SqlStdOperatorTable.EQUALS,
+ rexBuilder.makeCall(
+ SqlStdOperatorTable.ITEM,
+ arr,
+ rexBuilder.makeExactLiteral(BigDecimal.ONE)),
+ three)
+ // map['k'] = 3
+ val onMap = rexBuilder.makeCall(
+ SqlStdOperatorTable.EQUALS,
+ rexBuilder.makeCall(SqlStdOperatorTable.ITEM, map,
rexBuilder.makeLiteral("k")),
+ three)
+ // CAST(variant['k'] AS INT) = 3
+ val onVariant = rexBuilder.makeCall(
+ SqlStdOperatorTable.EQUALS,
+ rexBuilder.makeCast(
+ intType,
+ rexBuilder.makeCall(SqlStdOperatorTable.ITEM, variant,
rexBuilder.makeLiteral("k"))),
+ three)
+ // arr[OFFSET(0)] = 3, a 0-based variant of ITEM that AT does not cover
+ val onOffset = rexBuilder.makeCall(
+ SqlStdOperatorTable.EQUALS,
+ rexBuilder.makeCall(
+ SqlLibraryOperators.OFFSET,
+ arr,
+ rexBuilder.makeExactLiteral(BigDecimal.ZERO)),
+ three)
+ // row['f'] = 3, which AT does not support
+ val onRow = rexBuilder.makeCall(
+ SqlStdOperatorTable.EQUALS,
+ rexBuilder.makeCall(SqlStdOperatorTable.ITEM, row,
rexBuilder.makeLiteral("f")),
+ three)
+
+ val (convertedExpressions, unconvertedRexNodes) =
extractConjunctiveConditions(
+ rexBuilder.makeCall(SqlStdOperatorTable.AND, onArray, onMap, onVariant,
onOffset, onRow),
+ fieldNames,
+ rexBuilder,
+ functionCatalog)
+
+ assertThat(convertedExpressions.map(_.toString))
+ .containsExactly(
+ "equals(at(arr, 1), 3)",
+ "equals(at(map, 'k'), 3)",
+ "equals(cast(at(variant, 'k'), INT), 3)")
+ assertThat(unconvertedRexNodes).containsExactly(onOffset, onRow)
+ }
+
@Test
def testExtractWithTrim(): Unit = {
// name
