This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch release-2.4
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.4 by this push:
new b0da6f679e3 [FLINK-40002][table-planner] Keep metadata column
references out of physical filter push-down
b0da6f679e3 is described below
commit b0da6f679e3c0eaa068ef7b4406c2c657f038d28
Author: Jim Hughes <[email protected]>
AuthorDate: Wed Nov 4 02:46:11 2026 -0500
[FLINK-40002][table-planner] Keep metadata column references out of
physical filter push-down
---
.../PushFilterInCalcIntoTableSourceScanRule.java | 24 ++-
.../logical/PushFilterIntoSourceScanRuleBase.java | 187 +++++++++++++--------
.../logical/PushFilterIntoTableSourceScanRule.java | 71 ++------
.../MetadataFilterInReadingMetadataTest.java | 5 +-
.../PushFilterInCalcIntoTableSourceRuleTest.java | 57 +++++++
.../MetadataFilterInReadingMetadataTest.xml | 7 +-
.../PushFilterInCalcIntoTableSourceRuleTest.xml | 60 +++++++
7 files changed, 260 insertions(+), 151 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceScanRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceScanRule.java
index 4dc970af831..fe5a9805201 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceScanRule.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceScanRule.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.planner.plan.rules.logical;
-import
org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown;
import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalCalc;
import
org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalTableSourceScan;
import org.apache.flink.table.planner.plan.schema.FlinkPreparingTableBase;
@@ -63,8 +62,8 @@ public class PushFilterInCalcIntoTableSourceScanRule extends
PushFilterIntoSourc
FlinkLogicalTableSourceScan scan = call.rel(1);
TableSourceTable tableSourceTable =
scan.getTable().unwrap(TableSourceTable.class);
- // we can not push filter twice
- return canPushdownFilter(tableSourceTable);
+ // each push-down ability is applied at most once
+ return canPushdownFilter(tableSourceTable) ||
canPushdownMetadataFilter(tableSourceTable);
}
@Override
@@ -98,29 +97,28 @@ public class PushFilterInCalcIntoTableSourceScanRule
extends PushFilterIntoSourc
return;
}
- Tuple2<SupportsFilterPushDown.Result, TableSourceTable>
pushdownResultWithScan =
- resolveFiltersAndCreateTableSourceTable(
+ FilterPushDownResult pushDownResult =
+ classifyAndPushFilters(
convertiblePredicates,
+ unconvertedPredicates,
relOptTable.unwrap(TableSourceTable.class),
scan,
relBuilder);
-
- SupportsFilterPushDown.Result result = pushdownResultWithScan._1;
- TableSourceTable tableSourceTable = pushdownResultWithScan._2;
+ if (pushDownResult == null) {
+ return;
+ }
FlinkLogicalTableSourceScan newScan =
FlinkLogicalTableSourceScan.create(
- scan.getCluster(), scan.getHints(), tableSourceTable);
+ scan.getCluster(), scan.getHints(),
pushDownResult.newTableSourceTable);
// build new calc program
RexProgramBuilder programBuilder =
RexProgramBuilder.forProgram(originProgram,
call.builder().getRexBuilder(), true);
programBuilder.clearCondition();
- if (!result.getRemainingFilters().isEmpty() ||
unconvertedPredicates.length != 0) {
- RexNode remainingCondition =
- createRemainingCondition(
- relBuilder, result.getRemainingFilters(),
unconvertedPredicates);
+ if (!pushDownResult.remainingRexNodes.isEmpty()) {
+ RexNode remainingCondition =
relBuilder.and(pushDownResult.remainingRexNodes);
RexNode simplifiedRemainingCondition =
FlinkRexUtil.simplify(
relBuilder.getRexBuilder(),
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoSourceScanRuleBase.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoSourceScanRuleBase.java
index ef8bc092f98..cedb883d02c 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoSourceScanRuleBase.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoSourceScanRuleBase.java
@@ -18,8 +18,6 @@
package org.apache.flink.table.planner.plan.rules.logical;
import org.apache.flink.table.api.TableException;
-import org.apache.flink.table.catalog.Column;
-import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.connector.source.DynamicTableSource;
import
org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown;
import
org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata;
@@ -28,8 +26,10 @@ import org.apache.flink.table.expressions.ResolvedExpression;
import
org.apache.flink.table.planner.expressions.converter.ExpressionConverter;
import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec;
import
org.apache.flink.table.planner.plan.abilities.source.MetadataFilterPushDownSpec;
+import
org.apache.flink.table.planner.plan.abilities.source.ReadingMetadataSpec;
import
org.apache.flink.table.planner.plan.abilities.source.SourceAbilityContext;
import org.apache.flink.table.planner.plan.abilities.source.SourceAbilitySpec;
+import org.apache.flink.table.planner.plan.reuse.ScanReuserUtils;
import org.apache.flink.table.planner.plan.schema.TableSourceTable;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.RowType.RowField;
@@ -39,12 +39,15 @@ import
org.apache.flink.shaded.guava33.com.google.common.collect.Sets;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleOperand;
+import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.rel.core.TableScan;
import org.apache.calcite.rex.RexInputRef;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.rex.RexShuttle;
-import org.apache.calcite.rex.RexVisitorImpl;
import org.apache.calcite.tools.RelBuilder;
+import org.apache.calcite.util.ImmutableBitSet;
+
+import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Arrays;
@@ -68,13 +71,80 @@ public abstract class PushFilterIntoSourceScanRuleBase
extends RelOptRule {
return expressions.stream().map(e ->
e.accept(exprConverter)).collect(Collectors.toList());
}
- protected RexNode createRemainingCondition(
- RelBuilder relBuilder,
- List<ResolvedExpression> expressions,
- RexNode[] unconvertedPredicates) {
- List<RexNode> remainingPredicates =
convertExpressionToRexNode(expressions, relBuilder);
- remainingPredicates.addAll(Arrays.asList(unconvertedPredicates));
- return relBuilder.and(remainingPredicates);
+ /**
+ * Pushes physical-only and metadata-only predicates to their own
abilities and keeps mixed ones
+ * as runtime filters, so per FLIP-574 {@code applyFilters} never sees a
metadata column.
+ * Returns {@code null} if nothing can be pushed.
+ */
+ @Nullable
+ protected FilterPushDownResult classifyAndPushFilters(
+ RexNode[] convertiblePredicates,
+ RexNode[] unconvertedPredicates,
+ TableSourceTable tableSourceTable,
+ TableScan scan,
+ RelBuilder relBuilder) {
+ boolean supportsPhysicalFilter = canPushdownFilter(tableSourceTable);
+ boolean supportsMetadataFilter =
canPushdownMetadataFilter(tableSourceTable);
+ List<String> metadataKeys = getReadMetadataKeys(tableSourceTable);
+ int firstMetadataIndex = scan.getRowType().getFieldCount() -
metadataKeys.size();
+
+ List<RexNode> allRemainingRexNodes = new ArrayList<>();
+ List<RexNode> physicalPredicates = new ArrayList<>();
+ List<RexNode> metadataPredicates = new ArrayList<>();
+ for (RexNode predicate : convertiblePredicates) {
+ FilterColumnKind kind = classifyColumns(predicate,
firstMetadataIndex);
+ if (kind == FilterColumnKind.PHYSICAL_ONLY &&
supportsPhysicalFilter) {
+ physicalPredicates.add(predicate);
+ } else if (kind == FilterColumnKind.METADATA_ONLY &&
supportsMetadataFilter) {
+ metadataPredicates.add(predicate);
+ } else {
+ allRemainingRexNodes.add(predicate);
+ }
+ }
+
+ if (physicalPredicates.isEmpty() && metadataPredicates.isEmpty()) {
+ return null;
+ }
+
+ TableSourceTable currentTable = tableSourceTable;
+ if (!physicalPredicates.isEmpty()) {
+ Tuple2<SupportsFilterPushDown.Result, TableSourceTable>
physicalResult =
+ resolveFiltersAndCreateTableSourceTable(
+ physicalPredicates.toArray(new RexNode[0]),
+ currentTable,
+ scan,
+ relBuilder);
+ currentTable = physicalResult._2;
+ List<RexNode> physicalRemaining =
+
convertExpressionToRexNode(physicalResult._1.getRemainingFilters(), relBuilder);
+ allRemainingRexNodes.addAll(physicalRemaining);
+ }
+
+ if (!metadataPredicates.isEmpty()) {
+ FilterPushDownResult metadataResult =
+ resolveMetadataFiltersAndCreateTableSourceTable(
+ metadataPredicates.toArray(new RexNode[0]),
+ currentTable,
+ scan,
+ metadataKeys);
+ currentTable = metadataResult.newTableSourceTable;
+ allRemainingRexNodes.addAll(metadataResult.remainingRexNodes);
+ }
+
+ allRemainingRexNodes.addAll(Arrays.asList(unconvertedPredicates));
+ return new FilterPushDownResult(currentTable, allRemainingRexNodes);
+ }
+
+ /** Updated table and the predicates that must remain as runtime filters.
*/
+ protected static final class FilterPushDownResult {
+ final TableSourceTable newTableSourceTable;
+ final List<RexNode> remainingRexNodes;
+
+ FilterPushDownResult(
+ TableSourceTable newTableSourceTable, List<RexNode>
remainingRexNodes) {
+ this.newTableSourceTable = newTableSourceTable;
+ this.remainingRexNodes = remainingRexNodes;
+ }
}
/**
@@ -137,72 +207,44 @@ public abstract class PushFilterIntoSourceScanRuleBase
extends RelOptRule {
.noneMatch(spec -> spec instanceof MetadataFilterPushDownSpec);
}
- /**
- * True if predicate references metadata columns exclusively (no physical
columns).
- *
- * <p>A predicate like {@code OR(physical_pred, metadata_pred)} returns
false because it
- * references both physical and metadata columns. Mixed predicates remain
as runtime filters.
- */
- protected boolean referencesOnlyMetadataColumns(RexNode predicate, int
physicalColumnCount) {
- boolean[] saw = new boolean[2]; // [0] = sawPhysical, [1] = sawMetadata
- predicate.accept(
- new RexVisitorImpl<Void>(true) {
- @Override
- public Void visitInputRef(RexInputRef inputRef) {
- if (inputRef.getIndex() >= physicalColumnCount) {
- saw[1] = true;
- } else {
- saw[0] = true;
- }
- return null;
- }
- });
- return saw[1] && !saw[0];
+ private enum FilterColumnKind {
+ PHYSICAL_ONLY,
+ METADATA_ONLY,
+ MIXED
}
- /** Number of physical columns in the scan's schema. */
- protected int getPhysicalColumnCount(TableSourceTable tableSourceTable) {
- ResolvedSchema schema =
tableSourceTable.contextResolvedTable().getResolvedSchema();
- return (int)
schema.getColumns().stream().filter(Column::isPhysical).count();
- }
-
- /** Maps SQL column names to metadata keys for metadata columns. */
- protected Map<String, String> buildColumnToMetadataKeyMap(TableSourceTable
tableSourceTable) {
- ResolvedSchema schema =
tableSourceTable.contextResolvedTable().getResolvedSchema();
- Map<String, String> mapping = new HashMap<>();
- for (Column col : schema.getColumns()) {
- if (col instanceof Column.MetadataColumn) {
- Column.MetadataColumn metaCol = (Column.MetadataColumn) col;
- String sqlName = metaCol.getName();
- String metadataKey = metaCol.getMetadataKey().orElse(sqlName);
- mapping.put(sqlName, metadataKey);
- }
+ private static FilterColumnKind classifyColumns(RexNode predicate, int
firstMetadataIndex) {
+ ImmutableBitSet columns = RelOptUtil.InputFinder.bits(predicate);
+ boolean referencesMetadata = columns.nextSetBit(firstMetadataIndex) >=
0;
+ int lowestColumn = columns.nextSetBit(0);
+ boolean referencesPhysical = lowestColumn >= 0 && lowestColumn <
firstMetadataIndex;
+ if (referencesMetadata && referencesPhysical) {
+ return FilterColumnKind.MIXED;
}
- return mapping;
+ return referencesMetadata ? FilterColumnKind.METADATA_ONLY :
FilterColumnKind.PHYSICAL_ONLY;
}
- /** Outcome of metadata filter push-down: updated table and runtime-Calc
predicates. */
- protected static final class MetadataPushDownOutcome {
- final TableSourceTable newTableSourceTable;
- final List<RexNode> remainingInputRexNodes;
-
- MetadataPushDownOutcome(
- TableSourceTable newTableSourceTable, List<RexNode>
remainingInputRexNodes) {
- this.newTableSourceTable = newTableSourceTable;
- this.remainingInputRexNodes = remainingInputRexNodes;
- }
+ /** Keys of the {@link ReadingMetadataSpec}, in the order of the scan's
trailing fields. */
+ private static List<String> getReadMetadataKeys(TableSourceTable
tableSourceTable) {
+ ReadingMetadataSpec spec =
+ ScanReuserUtils.getAbilitySpec(
+ tableSourceTable.abilitySpecs(),
ReadingMetadataSpec.class);
+ return spec == null ? List.of() : spec.getMetadataKeys();
}
/** Resolves metadata filters and creates a new {@link TableSourceTable}.
*/
- protected MetadataPushDownOutcome
resolveMetadataFiltersAndCreateTableSourceTable(
- RexNode[] metadataPredicates, TableSourceTable
oldTableSourceTable, TableScan scan) {
+ private FilterPushDownResult
resolveMetadataFiltersAndCreateTableSourceTable(
+ RexNode[] metadataPredicates,
+ TableSourceTable oldTableSourceTable,
+ TableScan scan,
+ List<String> metadataKeys) {
DynamicTableSource newTableSource =
oldTableSourceTable.tableSource().copy();
SourceAbilityContext abilityContext = SourceAbilityContext.from(scan);
// Field names are metadata keys, not SQL aliases, to avoid
physical/metadata collisions
// (e.g. `offset INT, msg_offset INT METADATA FROM 'offset'`).
MetadataRowInfo metadataRowInfo =
- buildMetadataRowInfo(oldTableSourceTable,
abilityContext.getSourceRowType());
+ buildMetadataRowInfo(abilityContext.getSourceRowType(),
metadataKeys);
RexNode[] remappedPredicates =
remapPredicates(metadataPredicates,
metadataRowInfo.oldIndexToNewIndex);
@@ -263,7 +305,7 @@ public abstract class PushFilterIntoSourceScanRuleBase
extends RelOptRule {
oldTableSourceTable.getStatistic(),
new SourceAbilitySpec[] {metadataSpec});
- return new MetadataPushDownOutcome(newTableSourceTable,
remainingInputRexNodes);
+ return new FilterPushDownResult(newTableSourceTable,
remainingInputRexNodes);
}
/**
@@ -271,18 +313,15 @@ public abstract class PushFilterIntoSourceScanRuleBase
extends RelOptRule {
* with an old-index-to-new-index mapping from the scan's source row type.
Non-metadata
* positions are absent from the mapping.
*/
- private MetadataRowInfo buildMetadataRowInfo(
- TableSourceTable tableSourceTable, RowType sourceRowType) {
- Map<String, String> columnToMetadataKey =
buildColumnToMetadataKeyMap(tableSourceTable);
+ private MetadataRowInfo buildMetadataRowInfo(RowType sourceRowType,
List<String> metadataKeys) {
+ int firstMetadataIndex = sourceRowType.getFieldCount() -
metadataKeys.size();
List<RowField> metadataFields = new ArrayList<>();
Map<Integer, Integer> oldToNewIndex = new HashMap<>();
- for (int i = 0; i < sourceRowType.getFieldCount(); i++) {
- String sqlName = sourceRowType.getFieldNames().get(i);
- String metadataKey = columnToMetadataKey.get(sqlName);
- if (metadataKey != null) {
- oldToNewIndex.put(i, metadataFields.size());
- metadataFields.add(new RowField(metadataKey,
sourceRowType.getTypeAt(i)));
- }
+ for (int i = firstMetadataIndex; i < sourceRowType.getFieldCount();
i++) {
+ oldToNewIndex.put(i, metadataFields.size());
+ metadataFields.add(
+ new RowField(
+ metadataKeys.get(i - firstMetadataIndex),
sourceRowType.getTypeAt(i)));
}
return new MetadataRowInfo(new RowType(false, metadataFields),
oldToNewIndex);
}
@@ -290,7 +329,7 @@ public abstract class PushFilterIntoSourceScanRuleBase
extends RelOptRule {
/**
* Rewrites {@link RexInputRef}s in each predicate using the supplied
old->new index map. Throws
* if a predicate references an index not in the map (would indicate a
non-metadata reference
- * slipped past {@link #referencesOnlyMetadataColumns}).
+ * slipped past {@link #classifyColumns}).
*/
private RexNode[] remapPredicates(
RexNode[] predicates, Map<Integer, Integer> oldIndexToNewIndex) {
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRule.java
index 4e896a30701..28474835583 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRule.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRule.java
@@ -29,10 +29,6 @@ import org.apache.calcite.rel.logical.LogicalTableScan;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.tools.RelBuilder;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
import scala.Tuple2;
/**
@@ -95,68 +91,25 @@ public class PushFilterIntoTableSourceScanRule extends
PushFilterIntoSourceScanR
return;
}
- boolean supportsPhysicalFilter = canPushdownFilter(tableSourceTable);
- boolean supportsMetadataFilter =
canPushdownMetadataFilter(tableSourceTable);
- int physicalColumnCount = getPhysicalColumnCount(tableSourceTable);
-
- List<RexNode> allRemainingRexNodes = new ArrayList<>();
- TableSourceTable currentTable = tableSourceTable;
-
- // Unpushable metadata predicates stay as a runtime Calc, not the
physical path —
- // physical routing produces a FilterPushDownSpec that crashes
compiled-plan restore
- // once ProjectPushDownSpec narrows the scan row type.
- List<RexNode> physicalPredicates = new ArrayList<>();
- List<RexNode> metadataPredicates = new ArrayList<>();
- for (RexNode predicate : convertiblePredicates) {
- if (referencesOnlyMetadataColumns(predicate, physicalColumnCount))
{
- if (supportsMetadataFilter) {
- metadataPredicates.add(predicate);
- } else {
- allRemainingRexNodes.add(predicate);
- }
- } else {
- physicalPredicates.add(predicate);
- }
- }
-
- // Avoid re-firing on shapes we can't transform — saves wasted Hep
iterations.
- boolean nothingToPushPhysically = physicalPredicates.isEmpty() ||
!supportsPhysicalFilter;
- if (nothingToPushPhysically && metadataPredicates.isEmpty()) {
+ FilterPushDownResult pushDownResult =
+ classifyAndPushFilters(
+ convertiblePredicates,
+ unconvertedPredicates,
+ tableSourceTable,
+ scan,
+ relBuilder);
+ if (pushDownResult == null) {
return;
}
- if (!physicalPredicates.isEmpty() && supportsPhysicalFilter) {
- Tuple2<SupportsFilterPushDown.Result, TableSourceTable>
physicalResult =
- resolveFiltersAndCreateTableSourceTable(
- physicalPredicates.toArray(new RexNode[0]),
- currentTable,
- scan,
- relBuilder);
- currentTable = physicalResult._2;
- List<RexNode> physicalRemaining =
-
convertExpressionToRexNode(physicalResult._1.getRemainingFilters(), relBuilder);
- allRemainingRexNodes.addAll(physicalRemaining);
- } else {
- allRemainingRexNodes.addAll(physicalPredicates);
- }
-
- if (!metadataPredicates.isEmpty()) {
- MetadataPushDownOutcome metadataResult =
- resolveMetadataFiltersAndCreateTableSourceTable(
- metadataPredicates.toArray(new RexNode[0]),
currentTable, scan);
- currentTable = metadataResult.newTableSourceTable;
- allRemainingRexNodes.addAll(metadataResult.remainingInputRexNodes);
- }
-
- allRemainingRexNodes.addAll(Arrays.asList(unconvertedPredicates));
-
LogicalTableScan newScan =
- LogicalTableScan.create(scan.getCluster(), currentTable,
scan.getHints());
+ LogicalTableScan.create(
+ scan.getCluster(), pushDownResult.newTableSourceTable,
scan.getHints());
- if (allRemainingRexNodes.isEmpty()) {
+ if (pushDownResult.remainingRexNodes.isEmpty()) {
call.transformTo(newScan);
} else {
- RexNode remainingCondition = relBuilder.and(allRemainingRexNodes);
+ RexNode remainingCondition =
relBuilder.and(pushDownResult.remainingRexNodes);
RexNode simplifiedRemainingCondition =
FlinkRexUtil.simplify(
relBuilder.getRexBuilder(),
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/MetadataFilterInReadingMetadataTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/MetadataFilterInReadingMetadataTest.java
index 97245fbcbf1..5da91c3b5df 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/MetadataFilterInReadingMetadataTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/MetadataFilterInReadingMetadataTest.java
@@ -129,9 +129,10 @@ class MetadataFilterInReadingMetadataTest extends
TableTestBase {
.build();
util.tableEnv().createTable("T4", descriptor);
- // id > 10 → physical path, event_time > ... → metadata path.
+ // id > 10 -> physical path, event_time > ... -> metadata path, mixed
stays above.
util.verifyRelPlan(
- "SELECT id FROM T4 WHERE id > 10 AND event_time > TIMESTAMP
'2024-01-01 00:00:00'");
+ "SELECT id FROM T4 WHERE id > 10 AND event_time > TIMESTAMP
'2024-01-01 00:00:00'"
+ + " AND EXTRACT(HOUR FROM event_time) > id");
}
@Test
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceRuleTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceRuleTest.java
index a01ae2bd7e5..a7b0c83765f 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceRuleTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceRuleTest.java
@@ -36,6 +36,8 @@ import org.apache.calcite.rel.rules.CoreRules;
import org.apache.calcite.tools.RuleSets;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.time.ZoneId;
@@ -117,6 +119,61 @@ class PushFilterInCalcIntoTableSourceRuleTest extends
PushFilterIntoTableSourceS
util.verifyRelPlan("SELECT * FROM WithWatermark WHERE LOWER(name) =
'foo'");
}
+ @ParameterizedTest(name = "virtual={0}")
+ @ValueSource(booleans = {false, true})
+ void testPredicateKindsOnWatermarkedTable(boolean virtual) {
+ // With a watermark, filters are pushed only after projection narrows
the scan.
+ StreamTableTestUtil defaultUtil =
streamTestUtil(TableConfig.getDefault());
+ defaultUtil
+ .tableEnv()
+ .executeSql(
+ "CREATE TABLE T (\n"
+ + " a INT,\n"
+ + " b BIGINT,\n"
+ + " ts TIMESTAMP(3),\n"
+ + " "
+ + (virtual ? "m1 INT METADATA VIRTUAL" : "m1
INT METADATA")
+ + ",\n"
+ + " WATERMARK FOR ts AS ts - INTERVAL '1'
SECOND\n"
+ + ") WITH (\n"
+ + " 'connector' = 'values',\n"
+ + " 'bounded' = 'false',\n"
+ + " 'disable-lookup' = 'true',\n"
+ + " 'readable-metadata' = 'm1:INT',\n"
+ + " 'filterable-fields' = 'a;m1',\n"
+ + " 'enable-metadata-filter-push-down' =
'true',\n"
+ + " 'enable-watermark-push-down' = 'true'\n"
+ + ")");
+ defaultUtil.verifyRelPlan(
+ "SELECT a, m1 FROM T WHERE a > 1 AND m1 > 0 AND m1 > a AND (a
< 5 OR m1 > 3)");
+ }
+
+ @Test
+ void testPredicateKindsOnNestedFieldNamedLikeMetadataColumn() {
+ // Computed columns flatten r.k into the scan as r_k, clashing with
the metadata column.
+ StreamTableTestUtil defaultUtil =
streamTestUtil(TableConfig.getDefault());
+ defaultUtil
+ .tableEnv()
+ .executeSql(
+ "CREATE TABLE TNested (\n"
+ + " r ROW<k INT, t TIMESTAMP(3)>,\n"
+ + " r_k INT METADATA FROM 'm0',\n"
+ + " k AS r.k,\n"
+ + " ts AS r.t,\n"
+ + " WATERMARK FOR ts AS ts - INTERVAL '1'
SECOND\n"
+ + ") WITH (\n"
+ + " 'connector' = 'values',\n"
+ + " 'bounded' = 'false',\n"
+ + " 'disable-lookup' = 'true',\n"
+ + " 'nested-projection-supported' = 'true',\n"
+ + " 'readable-metadata' = 'm0:INT',\n"
+ + " 'filterable-fields' = 'r_k;m0',\n"
+ + " 'enable-metadata-filter-push-down' =
'true',\n"
+ + " 'enable-watermark-push-down' = 'true'\n"
+ + ")");
+ defaultUtil.verifyRelPlan("SELECT k, r_k FROM TNested WHERE k = 0 AND
r_k > 5");
+ }
+
@Test
void testLowerUpperPushdown() {
String ddl =
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/MetadataFilterInReadingMetadataTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/MetadataFilterInReadingMetadataTest.xml
index 899a8640cf9..80feca3d38a 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/MetadataFilterInReadingMetadataTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/MetadataFilterInReadingMetadataTest.xml
@@ -120,12 +120,12 @@ LogicalProject(id=[$0], name=[$1])
</TestCase>
<TestCase name="testMixedPhysicalAndMetadataFilters">
<Resource name="sql">
- <![CDATA[SELECT id FROM T4 WHERE id > 10 AND event_time > TIMESTAMP
'2024-01-01 00:00:00']]>
+ <![CDATA[SELECT id FROM T4 WHERE id > 10 AND event_time > TIMESTAMP
'2024-01-01 00:00:00' AND EXTRACT(HOUR FROM event_time) > id]]>
</Resource>
<Resource name="ast">
<![CDATA[
LogicalProject(id=[$0])
-+- LogicalFilter(condition=[AND(>($0, 10), >($2, 2024-01-01 00:00:00))])
++- LogicalFilter(condition=[AND(>($0, 10), >($2, 2024-01-01 00:00:00),
>(EXTRACT(FLAG(HOUR), $2), $0))])
+- LogicalProject(id=[$0], name=[$1], event_time=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, T4,
metadata=[event_time]]])
]]>
@@ -134,7 +134,8 @@ LogicalProject(id=[$0])
<![CDATA[
LogicalProject(id=[$0])
+- LogicalProject(id=[$0], name=[$1], event_time=[$2])
- +- LogicalTableScan(table=[[default_catalog, default_database, T4,
metadata=[event_time], filter=[>(id, 10)], metadataFilter=[>(event_time,
2024-01-01 00:00:00)]]])
+ +- LogicalFilter(condition=[>(EXTRACT(FLAG(HOUR), $2), $0)])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T4,
metadata=[event_time], filter=[>(id, 10)], metadataFilter=[>(event_time,
2024-01-01 00:00:00)]]])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceRuleTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceRuleTest.xml
index 146c1c55029..7533b0a656a 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceRuleTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/PushFilterInCalcIntoTableSourceRuleTest.xml
@@ -323,6 +323,66 @@ LogicalProject(name=[$0], id=[$1], amount=[$2],
virtualField=[$3], price=[$4])
<![CDATA[
FlinkLogicalCalc(select=[name, id, amount, +(amount, 1) AS virtualField,
price], where=[>(price, 10)])
+- FlinkLogicalTableSourceScan(table=[[default_catalog, default_database,
VirtualTable, filter=[>(amount, 2)]]], fields=[name, id, amount, price])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testPredicateKindsOnNestedFieldNamedLikeMetadataColumn">
+ <Resource name="sql">
+ <![CDATA[SELECT k, r_k FROM TNested WHERE k = 0 AND r_k > 5]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(k=[$2], r_k=[$1])
++- LogicalFilter(condition=[AND(=($2, 0), >($1, 5))])
+ +- LogicalWatermarkAssigner(rowtime=[ts], watermark=[-($3, 1000:INTERVAL
SECOND)])
+ +- LogicalProject(r=[$0], r_k=[$1], k=[$0.k], ts=[$0.t])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
TNested, metadata=[m0]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[CAST(0 AS INTEGER) AS k, r_k_$0 AS r_k])
++- TableSourceScan(table=[[default_catalog, default_database, TNested,
project=[r_k, r_t], metadata=[m0], watermark=[-(r_t, 1000:INTERVAL SECOND)],
watermarkEmitStrategy=[on-periodic], filter=[=(r_k, 0)], metadataFilter=[>(m0,
5)]]], fields=[r_k, r_t, r_k_$0])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testPredicateKindsOnWatermarkedTable[virtual=false]">
+ <Resource name="sql">
+ <![CDATA[SELECT a, m1 FROM T WHERE a > 1 AND m1 > 0 AND m1 > a AND (a <
5 OR m1 > 3)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], m1=[$3])
++- LogicalFilter(condition=[AND(>($0, 1), >($3, 0), >($3, $0), OR(<($0, 5),
>($3, 3)))])
+ +- LogicalWatermarkAssigner(rowtime=[ts], watermark=[-($2, 1000:INTERVAL
SECOND)])
+ +- LogicalProject(a=[$0], b=[$1], ts=[$2], m1=[$3])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T,
metadata=[m1]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a, m1], where=[AND(>(m1, a), OR(<(a, 5), >(m1, 3)))])
++- TableSourceScan(table=[[default_catalog, default_database, T, project=[a,
ts], metadata=[m1], watermark=[-(ts, 1000:INTERVAL SECOND)],
watermarkEmitStrategy=[on-periodic], filter=[>(a, 1)], metadataFilter=[>(m1,
0)]]], fields=[a, ts, m1])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testPredicateKindsOnWatermarkedTable[virtual=true]">
+ <Resource name="sql">
+ <![CDATA[SELECT a, m1 FROM T WHERE a > 1 AND m1 > 0 AND m1 > a AND (a <
5 OR m1 > 3)]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(a=[$0], m1=[$3])
++- LogicalFilter(condition=[AND(>($0, 1), >($3, 0), >($3, $0), OR(<($0, 5),
>($3, 3)))])
+ +- LogicalWatermarkAssigner(rowtime=[ts], watermark=[-($2, 1000:INTERVAL
SECOND)])
+ +- LogicalProject(a=[$0], b=[$1], ts=[$2], m1=[$3])
+ +- LogicalTableScan(table=[[default_catalog, default_database, T,
metadata=[m1]]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[a, m1], where=[AND(>(m1, a), OR(<(a, 5), >(m1, 3)))])
++- TableSourceScan(table=[[default_catalog, default_database, T, project=[a,
ts], metadata=[m1], watermark=[-(ts, 1000:INTERVAL SECOND)],
watermarkEmitStrategy=[on-periodic], filter=[>(a, 1)], metadataFilter=[>(m1,
0)]]], fields=[a, ts, m1])
]]>
</Resource>
</TestCase>
