jnh5y commented on code in PR #27913:
URL: https://github.com/apache/flink/pull/27913#discussion_r3118053005
##########
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:
Good catch. I've addressed this and added a test for it.
##########
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:
Done!
--
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]
