twalthr commented on code in PR #27913:
URL: https://github.com/apache/flink/pull/27913#discussion_r3116083778
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoSourceScanRuleBase.java:
##########
@@ -102,17 +107,122 @@ protected RexNode createRemainingCondition(
return new Tuple2<>(result, newTableSourceTable);
}
- /**
- * Determines wether we can pushdown the filter into the source. we can
not push filter twice,
- * make sure FilterPushDownSpec has not been assigned as a capability.
- *
- * @param tableSourceTable Table scan to attempt to push into
- * @return Whether we can push or not
- */
+ /** Whether filter push-down is possible and not already assigned. */
protected boolean canPushdownFilter(TableSourceTable tableSourceTable) {
return tableSourceTable != null
&& tableSourceTable.tableSource() instanceof
SupportsFilterPushDown
&& Arrays.stream(tableSourceTable.abilitySpecs())
.noneMatch(spec -> spec instanceof FilterPushDownSpec);
}
+
+ /** Whether metadata filter push-down is possible and not already
assigned. */
+ protected boolean canPushdownMetadataFilter(TableSourceTable
tableSourceTable) {
+ if (tableSourceTable == null) {
+ return false;
+ }
+ DynamicTableSource source = tableSourceTable.tableSource();
+ if (!(source instanceof SupportsReadingMetadata)) {
+ return false;
+ }
+ if (!((SupportsReadingMetadata)
source).supportsMetadataFilterPushDown()) {
+ return false;
+ }
+ return Arrays.stream(tableSourceTable.abilitySpecs())
+ .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];
+ }
+
+ /** 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);
+ }
+ }
+ return mapping;
+ }
+
+ /** Resolves metadata filters and creates a new {@link TableSourceTable}.
*/
+ protected Tuple2<MetadataFilterResult, TableSourceTable>
+ resolveMetadataFiltersAndCreateTableSourceTable(
+ RexNode[] metadataPredicates,
+ TableSourceTable oldTableSourceTable,
+ TableScan scan,
+ RelBuilder relBuilder) {
+ DynamicTableSource newTableSource =
oldTableSourceTable.tableSource().copy();
+ Map<String, String> columnToMetadataKey =
buildColumnToMetadataKeyMap(oldTableSourceTable);
+ SourceAbilityContext abilityContext = SourceAbilityContext.from(scan);
+
+ // Pre-translate the RowType to use metadata key names instead of SQL
aliases.
+ RowType sourceRowType = abilityContext.getSourceRowType();
+ RowType translatedRowType = translateRowType(sourceRowType,
columnToMetadataKey);
+
+ MetadataFilterResult result =
+ MetadataFilterPushDownSpec.applyMetadataFilters(
+ Arrays.asList(metadataPredicates),
+ translatedRowType,
+ newTableSource,
+ abilityContext);
+
+ int acceptedCount = result.getAcceptedFilters().size();
+ List<RexNode> acceptedOriginalPredicates = new ArrayList<>();
+ for (int i = 0; i < acceptedCount; i++) {
+ acceptedOriginalPredicates.add(metadataPredicates[i]);
+ }
+ MetadataFilterPushDownSpec metadataSpec =
+ new MetadataFilterPushDownSpec(acceptedOriginalPredicates,
translatedRowType);
+
+ TableSourceTable newTableSourceTable =
+ oldTableSourceTable.copy(
+ newTableSource,
+ oldTableSourceTable.getStatistic(),
+ new SourceAbilitySpec[] {metadataSpec});
+
+ return new Tuple2<>(result, newTableSourceTable);
+ }
+
+ /** Replaces SQL alias names with metadata key names in the RowType. */
Review Comment:
Is there a change that fields can collide?
`CREATE TABLE t (offset INT, msg_offset INT METADATA FROM 'offset')`
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/serde/DynamicTableSourceSpecSerdeTest.java:
##########
@@ -385,6 +386,52 @@ void
testDynamicTableSourceSpecSerdeWithEnrichmentOptions() throws Exception {
.isEqualTo(",");
}
+ @Test
+ void testMetadataFilterPushDownSpecSerde() throws IOException {
+ FlinkTypeFactory factory =
+ new FlinkTypeFactory(
+ Thread.currentThread().getContextClassLoader(),
FlinkTypeSystem.INSTANCE);
+ RexBuilder rexBuilder = new RexBuilder(factory);
+
+ // RowType uses metadata key names (already translated from SQL
aliases)
+ RowType predicateRowType =
+ RowType.of(
+ new LogicalType[] {new TimestampType(3), new
BigIntType()},
+ new String[] {"timestamp", "offset"});
+
+ MetadataFilterPushDownSpec original =
+ new MetadataFilterPushDownSpec(
+ Arrays.asList(
+ // timestamp > '2024-01-01'
+ rexBuilder.makeCall(
+ SqlStdOperatorTable.GREATER_THAN,
+ rexBuilder.makeInputRef(
+
factory.createSqlType(SqlTypeName.TIMESTAMP, 3), 0),
+ rexBuilder.makeTimestampLiteral(
+ new
org.apache.calcite.util.TimestampString(
Review Comment:
```suggestion
new TimestampString(
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/MetadataFilterPushDownSpec.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.flink.table.planner.plan.abilities.source;
+
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import
org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata;
+import
org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata.MetadataFilterResult;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.expressions.resolver.ExpressionResolver;
+import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
+import org.apache.flink.table.planner.plan.utils.RexNodeToExpressionConverter;
+import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
+import org.apache.flink.table.types.logical.RowType;
+
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
+
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import scala.Option;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Serializes metadata filter predicates and replays them during compiled plan
restoration.
+ *
+ * <p>Predicates are stored with a {@code predicateRowType} that already uses
metadata key names
+ * (not SQL aliases). The alias-to-key translation happens once at
optimization time, so no
+ * column-to-key mapping needs to be persisted.
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonTypeName("MetadataFilterPushDown")
+public final class MetadataFilterPushDownSpec extends SourceAbilitySpecBase {
+
+ public static final String FIELD_NAME_PREDICATES = "predicates";
+ public static final String FIELD_NAME_PREDICATE_ROW_TYPE =
"predicateRowType";
+
+ @JsonProperty(FIELD_NAME_PREDICATES)
+ private final List<RexNode> predicates;
+
+ /**
+ * Row type snapshot using metadata key names. Stored because
ProjectPushDownSpec may narrow the
+ * context's row type during restore.
+ */
+ @JsonProperty(FIELD_NAME_PREDICATE_ROW_TYPE)
+ private final RowType predicateRowType;
+
+ @JsonCreator
+ public MetadataFilterPushDownSpec(
+ @JsonProperty(FIELD_NAME_PREDICATES) List<RexNode> predicates,
+ @JsonProperty(FIELD_NAME_PREDICATE_ROW_TYPE) RowType
predicateRowType) {
+ this.predicates = new ArrayList<>(checkNotNull(predicates));
+ this.predicateRowType = checkNotNull(predicateRowType);
+ }
+
+ public List<RexNode> getPredicates() {
+ return predicates;
+ }
+
+ @Override
+ public void apply(DynamicTableSource tableSource, SourceAbilityContext
context) {
+ // Use stored predicateRowType; context's row type may be narrowed by
ProjectPushDownSpec.
+ MetadataFilterResult result =
+ applyMetadataFilters(predicates, predicateRowType,
tableSource, context);
+ if (result.getAcceptedFilters().size() != predicates.size()) {
+ throw new TableException("All metadata predicates should be
accepted here.");
+ }
+ }
+
+ /**
+ * Converts RexNode predicates to ResolvedExpressions using the given row
type and calls
+ * applyMetadataFilters on the source. The row type must already use
metadata key names.
+ */
+ public static MetadataFilterResult applyMetadataFilters(
+ List<RexNode> predicates,
+ RowType metadataKeyRowType,
+ DynamicTableSource tableSource,
+ SourceAbilityContext context) {
+ if (!(tableSource instanceof SupportsReadingMetadata)) {
+ throw new TableException(
+ String.format(
+ "%s does not support SupportsReadingMetadata.",
+ tableSource.getClass().getName()));
+ }
Review Comment:
we should also check whether the source still returns
`supportsMetadataFilterPushDown = true`.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoSourceScanRuleBase.java:
##########
@@ -102,17 +107,122 @@ protected RexNode createRemainingCondition(
return new Tuple2<>(result, newTableSourceTable);
}
- /**
- * Determines wether we can pushdown the filter into the source. we can
not push filter twice,
- * make sure FilterPushDownSpec has not been assigned as a capability.
- *
- * @param tableSourceTable Table scan to attempt to push into
- * @return Whether we can push or not
- */
+ /** Whether filter push-down is possible and not already assigned. */
protected boolean canPushdownFilter(TableSourceTable tableSourceTable) {
return tableSourceTable != null
&& tableSourceTable.tableSource() instanceof
SupportsFilterPushDown
&& Arrays.stream(tableSourceTable.abilitySpecs())
.noneMatch(spec -> spec instanceof FilterPushDownSpec);
}
+
+ /** Whether metadata filter push-down is possible and not already
assigned. */
+ protected boolean canPushdownMetadataFilter(TableSourceTable
tableSourceTable) {
+ if (tableSourceTable == null) {
+ return false;
+ }
+ DynamicTableSource source = tableSourceTable.tableSource();
+ if (!(source instanceof SupportsReadingMetadata)) {
+ return false;
+ }
+ if (!((SupportsReadingMetadata)
source).supportsMetadataFilterPushDown()) {
+ return false;
+ }
+ return Arrays.stream(tableSourceTable.abilitySpecs())
+ .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];
+ }
+
+ /** 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);
+ }
+ }
+ return mapping;
+ }
+
+ /** Resolves metadata filters and creates a new {@link TableSourceTable}.
*/
+ protected Tuple2<MetadataFilterResult, TableSourceTable>
+ resolveMetadataFiltersAndCreateTableSourceTable(
+ RexNode[] metadataPredicates,
+ TableSourceTable oldTableSourceTable,
+ TableScan scan,
+ RelBuilder relBuilder) {
+ DynamicTableSource newTableSource =
oldTableSourceTable.tableSource().copy();
+ Map<String, String> columnToMetadataKey =
buildColumnToMetadataKeyMap(oldTableSourceTable);
+ SourceAbilityContext abilityContext = SourceAbilityContext.from(scan);
+
+ // Pre-translate the RowType to use metadata key names instead of SQL
aliases.
+ RowType sourceRowType = abilityContext.getSourceRowType();
+ RowType translatedRowType = translateRowType(sourceRowType,
columnToMetadataKey);
+
+ MetadataFilterResult result =
+ MetadataFilterPushDownSpec.applyMetadataFilters(
+ Arrays.asList(metadataPredicates),
+ translatedRowType,
+ newTableSource,
+ abilityContext);
+
+ int acceptedCount = result.getAcceptedFilters().size();
+ List<RexNode> acceptedOriginalPredicates = new ArrayList<>();
+ for (int i = 0; i < acceptedCount; i++) {
+ acceptedOriginalPredicates.add(metadataPredicates[i]);
+ }
+ MetadataFilterPushDownSpec metadataSpec =
+ new MetadataFilterPushDownSpec(acceptedOriginalPredicates,
translatedRowType);
+
+ TableSourceTable newTableSourceTable =
+ oldTableSourceTable.copy(
+ newTableSource,
+ oldTableSourceTable.getStatistic(),
+ new SourceAbilitySpec[] {metadataSpec});
+
+ return new Tuple2<>(result, newTableSourceTable);
+ }
+
+ /** Replaces SQL alias names with metadata key names in the RowType. */
Review Comment:
We should most likely only store metadata columns in this row, not combined
with physical.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/abilities/source/MetadataFilterPushDownSpec.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.flink.table.planner.plan.abilities.source;
+
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import
org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata;
+import
org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata.MetadataFilterResult;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.expressions.resolver.ExpressionResolver;
+import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
+import org.apache.flink.table.planner.plan.utils.RexNodeToExpressionConverter;
+import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
+import org.apache.flink.table.types.logical.RowType;
+
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
+
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import scala.Option;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Serializes metadata filter predicates and replays them during compiled plan
restoration.
+ *
+ * <p>Predicates are stored with a {@code predicateRowType} that already uses
metadata key names
+ * (not SQL aliases). The alias-to-key translation happens once at
optimization time, so no
+ * column-to-key mapping needs to be persisted.
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonTypeName("MetadataFilterPushDown")
+public final class MetadataFilterPushDownSpec extends SourceAbilitySpecBase {
+
+ public static final String FIELD_NAME_PREDICATES = "predicates";
+ public static final String FIELD_NAME_PREDICATE_ROW_TYPE =
"predicateRowType";
+
+ @JsonProperty(FIELD_NAME_PREDICATES)
+ private final List<RexNode> predicates;
+
+ /**
+ * Row type snapshot using metadata key names. Stored because
ProjectPushDownSpec may narrow the
+ * context's row type during restore.
+ */
+ @JsonProperty(FIELD_NAME_PREDICATE_ROW_TYPE)
+ private final RowType predicateRowType;
+
+ @JsonCreator
+ public MetadataFilterPushDownSpec(
+ @JsonProperty(FIELD_NAME_PREDICATES) List<RexNode> predicates,
+ @JsonProperty(FIELD_NAME_PREDICATE_ROW_TYPE) RowType
predicateRowType) {
+ this.predicates = new ArrayList<>(checkNotNull(predicates));
+ this.predicateRowType = checkNotNull(predicateRowType);
+ }
+
+ public List<RexNode> getPredicates() {
+ return predicates;
+ }
+
+ @Override
+ public void apply(DynamicTableSource tableSource, SourceAbilityContext
context) {
+ // Use stored predicateRowType; context's row type may be narrowed by
ProjectPushDownSpec.
+ MetadataFilterResult result =
+ applyMetadataFilters(predicates, predicateRowType,
tableSource, context);
+ if (result.getAcceptedFilters().size() != predicates.size()) {
+ throw new TableException("All metadata predicates should be
accepted here.");
+ }
+ }
+
+ /**
+ * Converts RexNode predicates to ResolvedExpressions using the given row
type and calls
+ * applyMetadataFilters on the source. The row type must already use
metadata key names.
+ */
+ public static MetadataFilterResult applyMetadataFilters(
+ List<RexNode> predicates,
+ RowType metadataKeyRowType,
+ DynamicTableSource tableSource,
+ SourceAbilityContext context) {
+ if (!(tableSource instanceof SupportsReadingMetadata)) {
+ throw new TableException(
+ String.format(
+ "%s does not support SupportsReadingMetadata.",
+ tableSource.getClass().getName()));
+ }
+
+ String[] fieldNames = metadataKeyRowType.getFieldNames().toArray(new
String[0]);
+
+ RexNodeToExpressionConverter converter =
+ new RexNodeToExpressionConverter(
+ new RexBuilder(context.getTypeFactory()),
+ fieldNames,
+ context.getFunctionCatalog(),
+ context.getCatalogManager(),
+ Option.apply(
+
context.getTypeFactory().buildRelNodeRowType(metadataKeyRowType)));
+
+ List<Expression> filters =
Review Comment:
to avoid code deduplication, feel free to create a package visible helper
method in FilterPushDownSpec and reuse it here.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/logical/MetadataFilterInReadingMetadataTest.java:
##########
@@ -0,0 +1,383 @@
+/*
+ * 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.flink.table.planner.plan.rules.logical;
+
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableDescriptor;
+import
org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown;
+import
org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata;
+import
org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata.MetadataFilterResult;
+import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.planner.calcite.CalciteConfig;
+import org.apache.flink.table.planner.factories.TableFactoryHarness;
+import
org.apache.flink.table.planner.plan.optimize.program.BatchOptimizeContext;
+import org.apache.flink.table.planner.plan.optimize.program.FlinkBatchProgram;
+import
org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgramBuilder;
+import
org.apache.flink.table.planner.plan.optimize.program.HEP_RULES_EXECUTION_TYPE;
+import org.apache.flink.table.planner.utils.BatchTableTestUtil;
+import org.apache.flink.table.planner.utils.TableConfigUtils;
+import org.apache.flink.table.planner.utils.TableTestBase;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.testutils.junit.SharedObjectsExtension;
+import org.apache.flink.testutils.junit.SharedReference;
+
+import org.apache.calcite.plan.hep.HepMatchOrder;
+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.api.extension.RegisterExtension;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.table.api.DataTypes.INT;
+import static org.apache.flink.table.api.DataTypes.STRING;
+import static org.apache.flink.table.api.DataTypes.TIMESTAMP;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for metadata filter push-down through {@link
SupportsReadingMetadata}. */
+class MetadataFilterInReadingMetadataTest extends TableTestBase {
+
+ @RegisterExtension
+ private final SharedObjectsExtension sharedObjects =
SharedObjectsExtension.create();
+
+ private BatchTableTestUtil util;
+
+ @BeforeEach
+ void setup() {
+ util = batchTestUtil(TableConfig.getDefault());
+ util.buildBatchProgram(FlinkBatchProgram.DEFAULT_REWRITE());
+ CalciteConfig calciteConfig =
+ TableConfigUtils.getCalciteConfig(util.tableEnv().getConfig());
+ calciteConfig
+ .getBatchProgram()
+ .get()
+ .addLast(
+ "rules",
+
FlinkHepRuleSetProgramBuilder.<BatchOptimizeContext>newBuilder()
+ .setHepRulesExecutionType(
+
HEP_RULES_EXECUTION_TYPE.RULE_COLLECTION())
+ .setHepMatchOrder(HepMatchOrder.BOTTOM_UP)
+ .add(
+ RuleSets.ofList(
+
PushFilterIntoTableSourceScanRule.INSTANCE,
+
CoreRules.FILTER_PROJECT_TRANSPOSE))
+ .build());
+ }
+
+ @Test
+ void testMetadataFilterPushDown() {
+ SharedReference<List<ResolvedExpression>> receivedFilters =
+ sharedObjects.add(new ArrayList<>());
+ TableDescriptor descriptor =
+ TableFactoryHarness.newBuilder()
+ .schema(MetadataFilterSource.SCHEMA)
+ .source(new MetadataFilterSource(true,
receivedFilters))
+ .build();
+ util.tableEnv().createTable("T1", descriptor);
+
+ util.verifyRelPlan("SELECT id FROM T1 WHERE event_time > TIMESTAMP
'2024-01-01 00:00:00'");
+
+ assertThat(receivedFilters.get()).isNotEmpty();
+ }
+
+ @Test
+ void testMetadataFilterNotPushedWhenNotSupported() {
+ SharedReference<List<ResolvedExpression>> receivedFilters =
+ sharedObjects.add(new ArrayList<>());
+ TableDescriptor descriptor =
+ TableFactoryHarness.newBuilder()
+ .schema(MetadataFilterSource.SCHEMA)
+ .source(new MetadataFilterSource(false,
receivedFilters))
+ .build();
+ util.tableEnv().createTable("T2", descriptor);
+
+ util.verifyRelPlan("SELECT id FROM T2 WHERE event_time > TIMESTAMP
'2024-01-01 00:00:00'");
+
+ // No metadata filters should have been pushed
+ assertThat(receivedFilters.get()).isEmpty();
+ }
+
+ @Test
+ void testAliasedMetadataColumnFilter() {
+ SharedReference<List<ResolvedExpression>> receivedFilters =
+ sharedObjects.add(new ArrayList<>());
+ TableDescriptor descriptor =
+ TableFactoryHarness.newBuilder()
+ .schema(RenamedMetadataFilterSource.SCHEMA)
+ .source(new
RenamedMetadataFilterSource(receivedFilters))
+ .build();
+ util.tableEnv().createTable("T3", descriptor);
+
+ // 'event_ts' is the SQL alias for metadata key 'timestamp'
+ util.verifyRelPlan("SELECT id FROM T3 WHERE event_ts > TIMESTAMP
'2024-01-01 00:00:00'");
+
+ // The source should receive the filter with metadata key 'timestamp',
not 'event_ts'
+ assertThat(receivedFilters.get()).isNotEmpty();
+
assertThat(receivedFilters.get().get(0).toString()).contains("timestamp");
+ }
+
+ @Test
+ void testMixedPhysicalAndMetadataFilters() {
+ SharedReference<List<ResolvedExpression>> metadataFilters =
+ sharedObjects.add(new ArrayList<>());
+ SharedReference<List<ResolvedExpression>> physicalFilters =
+ sharedObjects.add(new ArrayList<>());
+ TableDescriptor descriptor =
+ TableFactoryHarness.newBuilder()
+ .schema(MixedFilterSource.SCHEMA)
+ .source(new MixedFilterSource(metadataFilters,
physicalFilters))
+ .build();
+ util.tableEnv().createTable("T4", descriptor);
+
+ util.verifyRelPlan(
+ "SELECT id FROM T4 WHERE id > 10 AND event_time > TIMESTAMP
'2024-01-01 00:00:00'");
+
+ // Both paths should receive their respective filters
+ assertThat(physicalFilters.get()).isNotEmpty();
+ assertThat(metadataFilters.get()).isNotEmpty();
+ }
+
+ @Test
+ void testPartialMetadataFilterAcceptance() {
+ SharedReference<List<ResolvedExpression>> receivedFilters =
+ sharedObjects.add(new ArrayList<>());
+ TableDescriptor descriptor =
+ TableFactoryHarness.newBuilder()
+ .schema(PartialMetadataFilterSource.SCHEMA)
+ .source(new
PartialMetadataFilterSource(receivedFilters))
+ .build();
+ util.tableEnv().createTable("T6", descriptor);
+
+ // Two metadata filters: the source accepts only the first one
+ util.verifyRelPlan(
+ "SELECT id FROM T6 WHERE event_time > TIMESTAMP '2024-01-01
00:00:00'"
+ + " AND priority > 5");
+
+ // Source receives both filters but only accepts the first
+ assertThat(receivedFilters.get()).hasSize(2);
+ }
+
+ @Test
+ void testMetadataFilterWithProjection() {
+ SharedReference<List<ResolvedExpression>> receivedFilters =
+ sharedObjects.add(new ArrayList<>());
+ TableDescriptor descriptor =
+ TableFactoryHarness.newBuilder()
+ .schema(MetadataFilterSource.SCHEMA)
+ .source(new MetadataFilterSource(true,
receivedFilters))
+ .build();
+ util.tableEnv().createTable("T5", descriptor);
+
+ util.verifyRelPlan(
+ "SELECT id, name FROM T5 WHERE event_time > TIMESTAMP
'2024-01-01 00:00:00'");
+
+ assertThat(receivedFilters.get()).isNotEmpty();
+ }
+
+ //
-----------------------------------------------------------------------------------------
+ // Test sources
+ //
-----------------------------------------------------------------------------------------
+
+ /** Supports metadata filter push-down. */
+ private static class MetadataFilterSource extends
TableFactoryHarness.ScanSourceBase
+ implements SupportsReadingMetadata {
+
+ public static final Schema SCHEMA =
+ Schema.newBuilder()
+ .column("id", INT())
+ .column("name", STRING())
+ .columnByMetadata("event_time", TIMESTAMP(3))
+ .build();
+
+ private final boolean supportsMetadataFilter;
+ private final SharedReference<List<ResolvedExpression>>
receivedMetadataFilters;
+
+ MetadataFilterSource(
+ boolean supportsMetadataFilter,
+ SharedReference<List<ResolvedExpression>>
receivedMetadataFilters) {
+ this.supportsMetadataFilter = supportsMetadataFilter;
+ this.receivedMetadataFilters = receivedMetadataFilters;
+ }
+
+ @Override
+ public Map<String, DataType> listReadableMetadata() {
+ Map<String, DataType> metadata = new HashMap<>();
+ metadata.put("event_time",
org.apache.flink.table.api.DataTypes.TIMESTAMP(3));
+ return metadata;
+ }
+
+ @Override
+ public void applyReadableMetadata(List<String> metadataKeys, DataType
producedDataType) {}
+
+ @Override
+ public boolean supportsMetadataFilterPushDown() {
+ return supportsMetadataFilter;
+ }
+
+ @Override
+ public MetadataFilterResult
applyMetadataFilters(List<ResolvedExpression> metadataFilters) {
+ receivedMetadataFilters.get().addAll(metadataFilters);
+ return MetadataFilterResult.of(metadataFilters,
Collections.emptyList());
+ }
+ }
+
+ /** Tests key translation when SQL alias differs from metadata key. */
+ private static class RenamedMetadataFilterSource extends
TableFactoryHarness.ScanSourceBase
+ implements SupportsReadingMetadata {
+
+ public static final Schema SCHEMA =
+ Schema.newBuilder()
+ .column("id", INT())
+ .columnByMetadata("event_ts", TIMESTAMP(3),
"timestamp")
+ .build();
+
+ private final SharedReference<List<ResolvedExpression>>
receivedMetadataFilters;
+
+ RenamedMetadataFilterSource(
+ SharedReference<List<ResolvedExpression>>
receivedMetadataFilters) {
+ this.receivedMetadataFilters = receivedMetadataFilters;
+ }
+
+ @Override
+ public Map<String, DataType> listReadableMetadata() {
+ Map<String, DataType> metadata = new HashMap<>();
+ metadata.put("timestamp",
org.apache.flink.table.api.DataTypes.TIMESTAMP(3));
+ return metadata;
+ }
+
+ @Override
+ public void applyReadableMetadata(List<String> metadataKeys, DataType
producedDataType) {}
+
+ @Override
+ public boolean supportsMetadataFilterPushDown() {
+ return true;
+ }
+
+ @Override
+ public MetadataFilterResult
applyMetadataFilters(List<ResolvedExpression> metadataFilters) {
+ receivedMetadataFilters.get().addAll(metadataFilters);
+ return MetadataFilterResult.of(metadataFilters,
Collections.emptyList());
+ }
+ }
+
+ /** Accepts only the first metadata filter; rejected filters remain in
plan. */
+ private static class PartialMetadataFilterSource extends
TableFactoryHarness.ScanSourceBase
+ implements SupportsReadingMetadata {
+
+ public static final Schema SCHEMA =
+ Schema.newBuilder()
+ .column("id", INT())
+ .columnByMetadata("event_time", TIMESTAMP(3))
+ .columnByMetadata("priority", INT())
+ .build();
+
+ private final SharedReference<List<ResolvedExpression>>
receivedMetadataFilters;
+
+ PartialMetadataFilterSource(
+ SharedReference<List<ResolvedExpression>>
receivedMetadataFilters) {
+ this.receivedMetadataFilters = receivedMetadataFilters;
+ }
+
+ @Override
+ public Map<String, DataType> listReadableMetadata() {
+ Map<String, DataType> metadata = new HashMap<>();
+ metadata.put("event_time",
org.apache.flink.table.api.DataTypes.TIMESTAMP(3));
Review Comment:
```suggestion
metadata.put("event_time", DataTypes.TIMESTAMP(3));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
