From eca2e6e3c9769c4a03b270b7758691de8b0e9a16 Mon Sep 17 00:00:00 2001 From: naivedogger Date: Tue, 21 Jul 2026 20:06:31 +0800 Subject: [PATCH 01/11] [flink] Support projection/filter pushdown for changelog and binlog virtual tables Add projection and filter pushdown to the $changelog and $binlog virtual table sources, with dedicated converter/source unit tests and end-to-end ITCases. Fixes two issues found in review: - Metadata-only changelog projection produced an empty (zero-column) scan and yielded no records; normalize the scan projection to null so the full data row is scanned and metadata is selected during deserialization. - Changelog log filter pushdown is now only enabled for the ARROW log format, avoiding an UnsupportedOperationException from TableScan#createLogScanner on INDEXED/COMPACTED tables. --- .../flink/source/BinlogFlinkTableSource.java | 72 +++++- .../source/ChangelogFlinkTableSource.java | 213 +++++++++++++++- .../fluss/flink/source/FlinkTableSource.java | 92 +------ .../BinlogDeserializationSchema.java | 28 +- .../ChangelogDeserializationSchema.java | 30 ++- .../fluss/flink/utils/BinlogRowConverter.java | 53 +++- .../flink/utils/ChangelogRowConverter.java | 55 +++- .../fluss/flink/utils/PushdownUtils.java | 79 ++++++ .../source/BinlogFlinkTableSourceTest.java | 132 ++++++++++ .../source/BinlogVirtualTableITCase.java | 33 +++ .../source/ChangelogFlinkTableSourceTest.java | 241 ++++++++++++++++++ .../source/ChangelogVirtualTableITCase.java | 102 ++++++++ .../flink/utils/BinlogRowConverterTest.java | 65 +++++ .../utils/ChangelogRowConverterTest.java | 34 +++ 14 files changed, 1103 insertions(+), 126 deletions(-) create mode 100644 fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogFlinkTableSourceTest.java create mode 100644 fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java index 93261e8430a..1ffa1de6162 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java @@ -28,19 +28,28 @@ import org.apache.fluss.predicate.Predicate; import org.apache.fluss.types.RowType; +import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.table.connector.ChangelogMode; import org.apache.flink.table.connector.source.DynamicTableSource; import org.apache.flink.table.connector.source.ScanTableSource; import org.apache.flink.table.connector.source.SourceProvider; +import org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown; +import org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown; import org.apache.flink.table.data.RowData; +import org.apache.flink.table.expressions.ResolvedExpression; +import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.LogicalType; import javax.annotation.Nullable; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; import java.util.Map; /** A Flink table source for the $binlog virtual table. */ -public class BinlogFlinkTableSource implements ScanTableSource { +public class BinlogFlinkTableSource + implements ScanTableSource, SupportsProjectionPushDown, SupportsFilterPushDown { private final TablePath tablePath; private final Configuration flussConfig; @@ -55,8 +64,10 @@ public class BinlogFlinkTableSource implements ScanTableSource { private final int splitPerAssignmentBatchSize; private final Map tableOptions; - // Projection pushdown - @Nullable private int[] projectedFields; + // Projection pushdown. Top-level projection over the binlog row [_change_type, _log_offset, + // _commit_timestamp, before, after]; the underlying data scan stays full because before/after + // are whole nested ROWs (nested pruning is out of scope). + @Nullable private int[] projectedTopLevel; private LogicalType producedDataType; @Nullable private Predicate partitionFilters; @@ -116,11 +127,9 @@ public ChangelogMode getChangelogMode() { @Override public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { - // Create the Fluss row type for the data columns (the original table columns) + // Create the Fluss row type for the data columns (the original table columns). The data + // scan always stays full because the projected before/after columns are whole nested ROWs. RowType flussRowType = FlinkConversions.toFlussRowType(dataColumnsType); - if (projectedFields != null) { - flussRowType = flussRowType.project(projectedFields); - } // Determine the offsets initializer based on startup mode OffsetsInitializer offsetsInitializer; @@ -150,12 +159,12 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { false, isPartitioned, flussRowType, - projectedFields, + null, null, offsetsInitializer, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, - new BinlogDeserializationSchema(), + new BinlogDeserializationSchema(projectedTopLevel), streaming, partitionFilters, LeaseContext.DEFAULT); @@ -177,7 +186,7 @@ public DynamicTableSource copy() { splitPerAssignmentBatchSize, tableOptions); copy.producedDataType = producedDataType; - copy.projectedFields = projectedFields; + copy.projectedTopLevel = projectedTopLevel; copy.partitionFilters = partitionFilters; return copy; } @@ -187,6 +196,45 @@ public String asSummaryString() { return "FlussBinlogTableSource"; } - // TODO: Implement projection pushdown handling for nested before/after columns - // TODO: Implement filter pushdown + @Override + public boolean supportsNestedProjection() { + return false; + } + + @Override + public void applyProjection(int[][] projectedFields, DataType producedDataType) { + // Top-level projection over [_change_type, _log_offset, _commit_timestamp, before, after]. + // The data scan stays full (before/after are whole nested ROWs); the deserialization schema + // emits only the projected top-level columns. + this.projectedTopLevel = + Arrays.stream(projectedFields).mapToInt(value -> value[0]).toArray(); + this.producedDataType = producedDataType.getLogicalType(); + } + + @Override + public Result applyFilters(List filters) { + // The $binlog data and partition columns are nested inside the before/after ROW columns, + // and the leading columns are non-pushable metadata (_change_type, _log_offset, + // _commit_timestamp). No top-level filter is convertible to a Fluss predicate, so nothing + // is pushed down; all filters are returned to Flink. Implemented for interface parity with + // the normal table. + return Result.of(Collections.emptyList(), filters); + } + + @VisibleForTesting + @Nullable + int[] getProjectedTopLevel() { + return projectedTopLevel; + } + + @VisibleForTesting + LogicalType getProducedDataType() { + return producedDataType; + } + + @VisibleForTesting + @Nullable + Predicate getPartitionFilters() { + return partitionFilters; + } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java index 4304b271b28..1dde10b8f4f 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java @@ -19,34 +19,51 @@ import org.apache.fluss.client.initializer.OffsetsInitializer; import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.TableConfig; import org.apache.fluss.flink.FlinkConnectorOptions; import org.apache.fluss.flink.source.deserializer.ChangelogDeserializationSchema; import org.apache.fluss.flink.source.reader.LeaseContext; import org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils; import org.apache.fluss.flink.utils.FlinkConversions; +import org.apache.fluss.flink.utils.PushdownUtils; +import org.apache.fluss.metadata.LogFormat; import org.apache.fluss.metadata.TableDescriptor; import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.predicate.PartitionPredicateVisitor; import org.apache.fluss.predicate.Predicate; +import org.apache.fluss.predicate.PredicateBuilder; +import org.apache.fluss.predicate.PredicateVisitor; import org.apache.fluss.types.RowType; +import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.table.connector.ChangelogMode; import org.apache.flink.table.connector.source.DynamicTableSource; import org.apache.flink.table.connector.source.ScanTableSource; import org.apache.flink.table.connector.source.SourceProvider; +import org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown; +import org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown; import org.apache.flink.table.data.RowData; +import org.apache.flink.table.expressions.ResolvedExpression; +import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.LogicalType; import javax.annotation.Nullable; +import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; +import static org.apache.fluss.flink.utils.PredicateConverter.convertToFlussPredicate; + /** A Flink table source for the $changelog virtual table. */ -public class ChangelogFlinkTableSource implements ScanTableSource { +public class ChangelogFlinkTableSource + implements ScanTableSource, SupportsProjectionPushDown, SupportsFilterPushDown { private final TablePath tablePath; private final Configuration flussConfig; @@ -64,8 +81,17 @@ public class ChangelogFlinkTableSource implements ScanTableSource { // Projection pushdown @Nullable private int[] projectedFields; private LogicalType producedDataType; + // Data-column indices actually scanned from Fluss (derived from the virtual projection). + @Nullable private int[] dataProjection; + // Projection over the base changelog row [metadata columns + scanned data columns]. + @Nullable private int[] baseRowProjection; @Nullable private Predicate partitionFilters; + // Filter over the scanned data columns pushed to the Fluss log scan (statistics-based). + @Nullable private Predicate logRecordBatchFilter; + + // Number of leading changelog metadata columns: _change_type, _log_offset, _commit_timestamp. + private static final int METADATA_COLUMN_COUNT = 3; private static final Set METADATA_COLUMN_NAMES = new HashSet<>( @@ -147,10 +173,8 @@ public ChangelogMode getChangelogMode() { public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { // Create the Fluss row type for the data columns (without metadata) RowType flussRowType = FlinkConversions.toFlussRowType(dataColumnsType); - if (projectedFields != null) { - // Adjust projection to account for metadata columns - // TODO: Handle projection properly with metadata columns - flussRowType = flussRowType.project(projectedFields); + if (dataProjection != null) { + flussRowType = flussRowType.project(dataProjection); } // to capture all change types (+I, -U, +U, -D). // FULL mode reads snapshot first (no change types), so we use EARLIEST for log-only @@ -187,12 +211,12 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { false, isPartitioned(), flussRowType, - projectedFields, - null, + dataProjection, + logRecordBatchFilter, offsetsInitializer, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, - new ChangelogDeserializationSchema(), + new ChangelogDeserializationSchema(baseRowProjection), streaming, partitionFilters, LeaseContext.DEFAULT); // Lake source not supported @@ -215,7 +239,10 @@ public DynamicTableSource copy() { tableOptions); copy.producedDataType = producedDataType; copy.projectedFields = projectedFields; + copy.dataProjection = dataProjection; + copy.baseRowProjection = baseRowProjection; copy.partitionFilters = partitionFilters; + copy.logRecordBatchFilter = logRecordBatchFilter; return copy; } @@ -224,10 +251,176 @@ public String asSummaryString() { return "FlussChangelogTableSource"; } - // TODO: Implement projection pushdown handling for metadata columns - // TODO: Implement filter pushdown + @Override + public boolean supportsNestedProjection() { + return false; + } + + @Override + public void applyProjection(int[][] projectedFields, DataType producedDataType) { + this.projectedFields = Arrays.stream(projectedFields).mapToInt(value -> value[0]).toArray(); + this.producedDataType = producedDataType.getLogicalType(); + + // Derive the data-column indices actually scanned from Fluss (first-appearance order), + // and the projection over the base changelog row [metadata columns + scanned data columns]. + List dataColumns = new ArrayList<>(); + for (int virtualIndex : this.projectedFields) { + if (virtualIndex >= METADATA_COLUMN_COUNT) { + int dataIndex = virtualIndex - METADATA_COLUMN_COUNT; + if (!dataColumns.contains(dataIndex)) { + dataColumns.add(dataIndex); + } + } + } + // When no data column is projected (metadata-only projection, e.g. SELECT _change_type), + // normalize the scan projection to null so Fluss reads the full data row. A zero-column + // scan produces no records, so the metadata columns must be emitted from a full data scan + // and selected later during deserialization via baseRowProjection. + this.dataProjection = + dataColumns.isEmpty() + ? null + : dataColumns.stream().mapToInt(Integer::intValue).toArray(); + + int[] base = new int[this.projectedFields.length]; + for (int i = 0; i < this.projectedFields.length; i++) { + int virtualIndex = this.projectedFields[i]; + base[i] = + virtualIndex < METADATA_COLUMN_COUNT + ? virtualIndex + : METADATA_COLUMN_COUNT + + dataColumns.indexOf(virtualIndex - METADATA_COLUMN_COUNT); + } + this.baseRowProjection = base; + } + + @Override + public Result applyFilters(List filters) { + List acceptedFilters = new ArrayList<>(); + List remainingFilters = new ArrayList<>(); + + RowType dataFlussRowType = FlinkConversions.toFlussRowType(dataColumnsType); + + if (isPartitioned()) { + // apply partition filter pushdown + List converted = new ArrayList<>(); + RowType partitionRowType = dataFlussRowType.project(partitionKeyIndexes); + PredicateVisitor checksOnlyPartitionKeys = + new PartitionPredicateVisitor(partitionRowType.getFieldNames()); + + for (ResolvedExpression filter : filters) { + Optional predicateOptional = + convertToFlussPredicate(partitionRowType, filter); + if (predicateOptional.isPresent()) { + Predicate p = predicateOptional.get(); + // partition pushdown can only guarantee to filter out partitions matching the + // predicate, but can't guarantee to filter out all data matching a + // non-partition filter in the partition + if (!p.visit(checksOnlyPartitionKeys)) { + remainingFilters.add(filter); + } else { + acceptedFilters.add(filter); + } + converted.add(p); + } else { + remainingFilters.add(filter); + } + } + partitionFilters = converted.isEmpty() ? null : PredicateBuilder.and(converted); + } + + if (acceptedFilters.isEmpty() && remainingFilters.isEmpty()) { + remainingFilters.addAll(filters); + } + + // Data-column log filter pushdown (statistics-based). Metadata columns (_change_type, + // _log_offset, _commit_timestamp) are resolved by name against the data columns and + // therefore naturally fall out as non-pushable. + Result recordBatchResult = pushdownRecordBatchFilter(remainingFilters, dataFlussRowType); + acceptedFilters.addAll(recordBatchResult.getAcceptedFilters()); + + // We cannot determine whether this source will ultimately be used as a scan source. Always + // return all original filters as remaining so Flink applies them as a safety net. + return Result.of(acceptedFilters, filters); + } + + private Result pushdownRecordBatchFilter( + List filters, RowType dataFlussRowType) { + TableConfig tableConfig = new TableConfig(Configuration.fromMap(tableOptions)); + // Log filter pushdown is only supported for the ARROW log format. For INDEXED/COMPACTED + // formats, TableScan#createLogScanner rejects a record-batch filter at runtime, so we skip + // pushdown here and let Flink apply the filters as a post-scan safety net. + if (tableConfig.getLogFormat() != LogFormat.ARROW) { + this.logRecordBatchFilter = null; + return Result.of(Collections.emptyList(), filters); + } + + Set availableStatsColumns = + PushdownUtils.computeAvailableStatsColumns(dataFlussRowType, tableConfig); + + List pushdownPredicates = new ArrayList<>(); + List acceptedFilters = new ArrayList<>(); + List remainingFilters = new ArrayList<>(); + + for (ResolvedExpression filter : filters) { + Optional predicateOpt = convertToFlussPredicate(dataColumnsType, filter); + if (predicateOpt.isPresent() + && PushdownUtils.canPredicateUseStatistics( + predicateOpt.get(), dataFlussRowType, availableStatsColumns)) { + pushdownPredicates.add(predicateOpt.get()); + acceptedFilters.add(filter); + } + // All filters are kept as remaining so that Flink can still verify the results + // after server-side filtering (safety net). + remainingFilters.add(filter); + } + + if (pushdownPredicates.isEmpty()) { + this.logRecordBatchFilter = null; + } else { + this.logRecordBatchFilter = + pushdownPredicates.size() == 1 + ? pushdownPredicates.get(0) + : PredicateBuilder.and(pushdownPredicates); + } + return Result.of(acceptedFilters, remainingFilters); + } private boolean isPartitioned() { return partitionKeyIndexes.length > 0; } + + @VisibleForTesting + @Nullable + int[] getProjectedFields() { + return projectedFields; + } + + @VisibleForTesting + @Nullable + int[] getDataProjection() { + return dataProjection; + } + + @VisibleForTesting + @Nullable + int[] getBaseRowProjection() { + return baseRowProjection; + } + + @VisibleForTesting + LogicalType getProducedDataType() { + return producedDataType; + } + + @VisibleForTesting + @Nullable + Predicate getLogRecordBatchFilter() { + return logRecordBatchFilter; + } + + @VisibleForTesting + @Nullable + Predicate getPartitionFilters() { + return partitionFilters; + } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java index 24f3d94d670..162993d8f6c 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java @@ -21,7 +21,6 @@ import org.apache.fluss.client.table.getter.PartitionGetter; import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; -import org.apache.fluss.config.StatisticsColumnsConfig; import org.apache.fluss.config.TableConfig; import org.apache.fluss.flink.FlinkConnectorOptions; import org.apache.fluss.flink.row.FlinkAsFlussRow; @@ -42,12 +41,10 @@ import org.apache.fluss.metadata.MergeEngineType; import org.apache.fluss.metadata.PartitionSpec; import org.apache.fluss.metadata.TablePath; -import org.apache.fluss.predicate.CompoundPredicate; import org.apache.fluss.predicate.PartitionPredicateVisitor; import org.apache.fluss.predicate.Predicate; import org.apache.fluss.predicate.PredicateBuilder; import org.apache.fluss.predicate.PredicateVisitor; -import org.apache.fluss.types.DataTypeChecks; import org.apache.fluss.types.RowType; import org.apache.flink.annotation.VisibleForTesting; @@ -266,7 +263,8 @@ public FlinkTableSource( // Pre-compute available statistics columns to avoid repeated calculation RowType flussRowType = FlinkConversions.toFlussRowType(tableOutputType); - this.availableStatsColumns = computeAvailableStatsColumns(flussRowType); + this.availableStatsColumns = + PushdownUtils.computeAvailableStatsColumns(flussRowType, tableConfig); } @Override @@ -678,7 +676,8 @@ private Result pushdownRecordBatchFilter(List filters) { Predicate predicate = predicateOpt.get(); LOG.trace("Converted filter to predicate: {}", predicate); // Check if predicate can benefit from statistics - if (canPredicateUseStatistics(predicate, flussRowType, availableStatsColumns)) { + if (PushdownUtils.canPredicateUseStatistics( + predicate, flussRowType, availableStatsColumns)) { pushdownPredicates.add(predicate); acceptedFilters.add(filter); } @@ -731,89 +730,6 @@ private void pushdownLakeFilters( } } - /** - * Checks if a predicate can benefit from statistics based on the available statistics columns. - * - * @param predicate the predicate to check - * @param rowType the row type - * @param availableStatsColumns the columns that have statistics available - * @return true if the predicate can use statistics - */ - private boolean canPredicateUseStatistics( - Predicate predicate, RowType rowType, Set availableStatsColumns) { - - class StatisticsUsageVisitor implements PredicateVisitor { - @Override - public Boolean visit(org.apache.fluss.predicate.LeafPredicate leaf) { - // Check if the field referenced by this predicate has statistics available - String fieldName = rowType.getFieldNames().get(leaf.index()); - // Check if statistics are available for this column - return availableStatsColumns.contains(fieldName); - } - - @Override - public Boolean visit(CompoundPredicate compound) { - // For compound predicates, all children must be able to use statistics - for (Predicate child : compound.children()) { - if (!child.visit(this)) { - return false; - } - } - return true; - } - } - - return predicate.visit(new StatisticsUsageVisitor()); - } - - /** - * Computes the available statistics columns based on table configuration. This method is called - * once during construction to pre-compute the result. - * - * @param flussRowType the row type - * @return set of column names that have statistics available - */ - private Set computeAvailableStatsColumns(RowType flussRowType) { - StatisticsColumnsConfig statsConfig = tableConfig.getStatisticsColumns(); - - if (!statsConfig.isEnabled()) { - LOG.debug("Statistics collection is disabled for the table"); - return Collections.emptySet(); - } - - Set columns = new HashSet<>(); - if (statsConfig.getMode() == StatisticsColumnsConfig.Mode.ALL) { - // Collect all columns with supported statistics types - for (int i = 0; i < flussRowType.getFieldCount(); i++) { - org.apache.fluss.types.DataType fieldType = flussRowType.getTypeAt(i); - if (DataTypeChecks.isSupportedStatisticsType(fieldType)) { - columns.add(flussRowType.getFieldNames().get(i)); - } - } - } else { - // Use user-specified columns (validate they exist and have supported types) - for (String columnName : statsConfig.getColumns()) { - int columnIndex = flussRowType.getFieldNames().indexOf(columnName); - if (columnIndex >= 0) { - org.apache.fluss.types.DataType fieldType = flussRowType.getTypeAt(columnIndex); - if (DataTypeChecks.isSupportedStatisticsType(fieldType)) { - columns.add(columnName); - } else { - LOG.trace( - "Configured statistics column '{}' has unsupported type and will be ignored", - columnName); - } - } else { - LOG.trace( - "Configured statistics column '{}' does not exist in table schema", - columnName); - } - } - } - - return columns; - } - @Override public RowLevelModificationScanContext applyRowLevelModificationScan( RowLevelModificationType rowLevelModificationType, diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/BinlogDeserializationSchema.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/BinlogDeserializationSchema.java index 1114febb9c9..c06851c9457 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/BinlogDeserializationSchema.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/BinlogDeserializationSchema.java @@ -46,14 +46,31 @@ public class BinlogDeserializationSchema implements FlussDeserializationSchema getProducedType(RowType rowSchema) { - // Build the output type with nested before/after ROW columns + // Build the output type with nested before/after ROW columns, honoring the projection org.apache.flink.table.types.logical.RowType outputType = - BinlogRowConverter.buildBinlogRowType(toFlinkRowType(rowSchema)); + BinlogRowConverter.buildProducedRowType( + toFlinkRowType(rowSchema), projectedTopLevel); return InternalTypeInfo.of(outputType); } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/ChangelogDeserializationSchema.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/ChangelogDeserializationSchema.java index c81141d345c..71c6c7ff6a2 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/ChangelogDeserializationSchema.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/ChangelogDeserializationSchema.java @@ -25,6 +25,8 @@ import org.apache.flink.table.data.RowData; import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import javax.annotation.Nullable; + import static org.apache.fluss.flink.utils.FlinkConversions.toFlinkRowType; /** @@ -39,14 +41,31 @@ public class ChangelogDeserializationSchema implements FlussDeserializationSchem */ private transient ChangelogRowConverter converter; - /** Creates a new ChangelogDeserializationSchema. */ - public ChangelogDeserializationSchema() {} + /** + * Optional projection over the base changelog row {@code [_change_type, _log_offset, + * _commit_timestamp, ]}, or {@code null} for no projection. + */ + @Nullable private final int[] baseRowProjection; + + /** Creates a new ChangelogDeserializationSchema without projection. */ + public ChangelogDeserializationSchema() { + this(null); + } + + /** + * Creates a new ChangelogDeserializationSchema. + * + * @param baseRowProjection projection over the base changelog row, or {@code null} for none + */ + public ChangelogDeserializationSchema(@Nullable int[] baseRowProjection) { + this.baseRowProjection = baseRowProjection; + } /** Initializes the deserialization schema. */ @Override public void open(InitializationContext context) throws Exception { if (converter == null) { - this.converter = new ChangelogRowConverter(context.getRowSchema()); + this.converter = new ChangelogRowConverter(context.getRowSchema(), baseRowProjection); } } @@ -67,9 +86,10 @@ public RowData deserialize(LogRecord record) throws Exception { */ @Override public TypeInformation getProducedType(RowType rowSchema) { - // Build the output type with metadata columns + // Build the output type with metadata columns, honoring the projection when present org.apache.flink.table.types.logical.RowType outputType = - ChangelogRowConverter.buildChangelogRowType(toFlinkRowType(rowSchema)); + ChangelogRowConverter.buildProducedRowType( + toFlinkRowType(rowSchema), baseRowProjection); return InternalTypeInfo.of(outputType); } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/BinlogRowConverter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/BinlogRowConverter.java index 3a9f2c94116..c47fdf7dd2e 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/BinlogRowConverter.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/BinlogRowConverter.java @@ -26,6 +26,7 @@ import org.apache.flink.table.data.RowData; import org.apache.flink.table.data.StringData; import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.data.utils.ProjectedRowData; import org.apache.flink.table.types.logical.BigIntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.VarCharType; @@ -45,16 +46,36 @@ public class BinlogRowConverter implements RecordToFlinkRowConverter { private final FlussRowToFlinkRowConverter baseConverter; private final org.apache.flink.table.types.logical.RowType producedType; + /** + * Optional top-level projection over the binlog row {@code [_change_type, _log_offset, + * _commit_timestamp, before, after]}. When non-null, the converter emits exactly those columns + * in that order; {@code null} means the full binlog row is emitted. Nested pruning inside + * before/after is not supported, so the underlying data scan always reads all columns. + */ + @Nullable private final int[] projectedTopLevel; + /** * Buffer for the UPDATE_BEFORE (-U) record pending merge with the next UPDATE_AFTER (+U) * record. Null when no update is in progress. */ @Nullable private LogRecord pendingUpdateBefore; - /** Creates a new BinlogRowConverter. */ + /** Creates a new BinlogRowConverter without projection. */ public BinlogRowConverter(RowType rowType) { + this(rowType, null); + } + + /** + * Creates a new BinlogRowConverter. + * + * @param rowType the data columns scanned from Fluss (always the full set) + * @param projectedTopLevel top-level projection over the binlog row, or {@code null} for none + */ + public BinlogRowConverter(RowType rowType, @Nullable int[] projectedTopLevel) { this.baseConverter = new FlussRowToFlinkRowConverter(rowType); - this.producedType = buildBinlogRowType(FlinkConversions.toFlinkRowType(rowType)); + this.projectedTopLevel = projectedTopLevel; + this.producedType = + buildProducedRowType(FlinkConversions.toFlinkRowType(rowType), projectedTopLevel); } /** Converts a LogRecord to a binlog RowData with nested before/after structure. */ @@ -137,7 +158,10 @@ private RowData buildBinlogRow( row.setField(3, before); row.setField(4, after); row.setRowKind(RowKind.INSERT); - return row; + if (projectedTopLevel == null) { + return row; + } + return ProjectedRowData.from(projectedTopLevel).replaceRow(row); } /** @@ -171,4 +195,27 @@ public static org.apache.flink.table.types.logical.RowType buildBinlogRowType( return new org.apache.flink.table.types.logical.RowType(fields); } + + /** + * Builds the produced Flink RowType for a binlog scan: the full binlog row type when {@code + * projectedTopLevel} is null, otherwise the projected top-level subset selected in the + * requested order. + * + * @param originalDataType the data columns row type + * @param projectedTopLevel top-level projection over the binlog row, or {@code null} for none + * @return the produced row type + */ + public static org.apache.flink.table.types.logical.RowType buildProducedRowType( + org.apache.flink.table.types.logical.RowType originalDataType, + @Nullable int[] projectedTopLevel) { + org.apache.flink.table.types.logical.RowType full = buildBinlogRowType(originalDataType); + if (projectedTopLevel == null) { + return full; + } + List fields = new ArrayList<>(); + for (int idx : projectedTopLevel) { + fields.add(full.getFields().get(idx)); + } + return new org.apache.flink.table.types.logical.RowType(fields); + } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/ChangelogRowConverter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/ChangelogRowConverter.java index a2b339da5d2..508f4e36de8 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/ChangelogRowConverter.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/ChangelogRowConverter.java @@ -27,11 +27,14 @@ import org.apache.flink.table.data.StringData; import org.apache.flink.table.data.TimestampData; import org.apache.flink.table.data.utils.JoinedRowData; +import org.apache.flink.table.data.utils.ProjectedRowData; import org.apache.flink.table.types.logical.BigIntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.VarCharType; import org.apache.flink.types.RowKind; +import javax.annotation.Nullable; + import java.util.ArrayList; import java.util.List; @@ -44,10 +47,30 @@ public class ChangelogRowConverter implements RecordToFlinkRowConverter { private final FlussRowToFlinkRowConverter baseConverter; private final org.apache.flink.table.types.logical.RowType producedType; - /** Creates a new ChangelogRowConverter. */ + /** + * Optional projection over the base changelog row {@code [_change_type, _log_offset, + * _commit_timestamp, ]}. When non-null, each entry is a position in the + * base row and the converter emits exactly those columns in that order; {@code null} means the + * full changelog row is emitted. + */ + @Nullable private final int[] baseRowProjection; + + /** Creates a new ChangelogRowConverter without projection. */ public ChangelogRowConverter(RowType rowType) { + this(rowType, null); + } + + /** + * Creates a new ChangelogRowConverter. + * + * @param rowType the (possibly projected) data columns scanned from Fluss + * @param baseRowProjection projection over the base changelog row, or {@code null} for none + */ + public ChangelogRowConverter(RowType rowType, @Nullable int[] baseRowProjection) { this.baseConverter = new FlussRowToFlinkRowConverter(rowType); - this.producedType = buildChangelogRowType(FlinkConversions.toFlinkRowType(rowType)); + this.baseRowProjection = baseRowProjection; + this.producedType = + buildProducedRowType(FlinkConversions.toFlinkRowType(rowType), baseRowProjection); } /** Converts a LogRecord to a Flink RowData with metadata columns. */ @@ -71,7 +94,10 @@ public RowData toChangelogRowData(LogRecord record) { JoinedRowData joinedRow = new JoinedRowData(metadataRow, physicalRowData); joinedRow.setRowKind(RowKind.INSERT); - return joinedRow; + if (baseRowProjection == null) { + return joinedRow; + } + return ProjectedRowData.from(baseRowProjection).replaceRow(joinedRow); } @Override @@ -128,4 +154,27 @@ public static org.apache.flink.table.types.logical.RowType buildChangelogRowType return new org.apache.flink.table.types.logical.RowType(fields); } + + /** + * Builds the produced Flink RowType for a changelog scan: the full changelog row type (metadata + * columns prepended to {@code originalDataType}) when {@code baseRowProjection} is null, + * otherwise the projected subset selected in the requested order. + * + * @param originalDataType the (possibly projected) data columns row type + * @param baseRowProjection projection over the base changelog row, or {@code null} for none + * @return the produced row type + */ + public static org.apache.flink.table.types.logical.RowType buildProducedRowType( + org.apache.flink.table.types.logical.RowType originalDataType, + @Nullable int[] baseRowProjection) { + org.apache.flink.table.types.logical.RowType full = buildChangelogRowType(originalDataType); + if (baseRowProjection == null) { + return full; + } + List fields = new ArrayList<>(); + for (int idx : baseRowProjection) { + fields.add(full.getFields().get(idx)); + } + return new org.apache.flink.table.types.logical.RowType(fields); + } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java index 0c6be61d829..4684cd0b5a7 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java @@ -26,6 +26,8 @@ import org.apache.fluss.client.table.scanner.batch.BatchScanner; import org.apache.fluss.client.table.writer.UpsertWriter; import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.StatisticsColumnsConfig; +import org.apache.fluss.config.TableConfig; import org.apache.fluss.exception.FlussRuntimeException; import org.apache.fluss.exception.UnsupportedVersionException; import org.apache.fluss.flink.source.lookup.FlinkLookupFunction; @@ -35,12 +37,17 @@ import org.apache.fluss.metadata.TableInfo; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.metadata.TableStats; +import org.apache.fluss.predicate.CompoundPredicate; +import org.apache.fluss.predicate.LeafPredicate; +import org.apache.fluss.predicate.Predicate; +import org.apache.fluss.predicate.PredicateVisitor; import org.apache.fluss.row.BinaryString; import org.apache.fluss.row.Decimal; import org.apache.fluss.row.GenericRow; import org.apache.fluss.row.InternalRow; import org.apache.fluss.row.TimestampLtz; import org.apache.fluss.row.TimestampNtz; +import org.apache.fluss.types.DataTypeChecks; import org.apache.flink.table.api.TableException; import org.apache.flink.table.data.DecimalData; @@ -67,8 +74,10 @@ import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -450,6 +459,76 @@ private static ListOffsetsResult listOffsets( // ------------------------------------------------------------------------------------------ + /** + * Computes the set of column names that have statistics available for filter pushdown, based on + * the table's statistics configuration. Returns an empty set when statistics collection is + * disabled. + * + * @param flussRowType the row type whose columns are considered + * @param tableConfig the table configuration carrying the statistics settings + * @return the set of column names that have statistics available + */ + public static Set computeAvailableStatsColumns( + org.apache.fluss.types.RowType flussRowType, TableConfig tableConfig) { + StatisticsColumnsConfig statsConfig = tableConfig.getStatisticsColumns(); + if (!statsConfig.isEnabled()) { + return Collections.emptySet(); + } + + Set columns = new HashSet<>(); + if (statsConfig.getMode() == StatisticsColumnsConfig.Mode.ALL) { + for (int i = 0; i < flussRowType.getFieldCount(); i++) { + if (DataTypeChecks.isSupportedStatisticsType(flussRowType.getTypeAt(i))) { + columns.add(flussRowType.getFieldNames().get(i)); + } + } + } else { + for (String columnName : statsConfig.getColumns()) { + int columnIndex = flussRowType.getFieldNames().indexOf(columnName); + if (columnIndex >= 0 + && DataTypeChecks.isSupportedStatisticsType( + flussRowType.getTypeAt(columnIndex))) { + columns.add(columnName); + } + } + } + return columns; + } + + /** + * Checks if a predicate can benefit from statistics based on the available statistics columns. + * A compound predicate is usable only if all of its children are usable. + * + * @param predicate the predicate to check + * @param rowType the row type the predicate indexes into + * @param availableStatsColumns the columns that have statistics available + * @return true if the predicate can use statistics + */ + public static boolean canPredicateUseStatistics( + Predicate predicate, + org.apache.fluss.types.RowType rowType, + Set availableStatsColumns) { + class StatisticsUsageVisitor implements PredicateVisitor { + @Override + public Boolean visit(LeafPredicate leaf) { + String fieldName = rowType.getFieldNames().get(leaf.index()); + return availableStatsColumns.contains(fieldName); + } + + @Override + public Boolean visit(CompoundPredicate compound) { + for (Predicate child : compound.children()) { + if (!child.visit(this)) { + return false; + } + } + return true; + } + } + + return predicate.visit(new StatisticsUsageVisitor()); + } + /** A structure represents a source field equal literal expression. */ public static class FieldEqual implements Serializable { private static final long serialVersionUID = 1L; diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogFlinkTableSourceTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogFlinkTableSourceTest.java new file mode 100644 index 00000000000..9a76ef5eb01 --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogFlinkTableSourceTest.java @@ -0,0 +1,132 @@ +/* + * 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.fluss.flink.source; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.flink.utils.BinlogRowConverter; +import org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils; +import org.apache.fluss.metadata.TablePath; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown; +import org.apache.flink.table.expressions.CallExpression; +import org.apache.flink.table.expressions.FieldReferenceExpression; +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.types.DataType; +import org.apache.flink.table.types.logical.RowType; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; + +import static org.apache.fluss.flink.FlinkConnectorOptions.ScanStartupMode; +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit tests for {@link BinlogFlinkTableSource} projection and filter pushdown. */ +class BinlogFlinkTableSourceTest { + + // Data columns: (id INT, name STRING, amount BIGINT) + private static final RowType DATA_COLUMNS_TYPE = + (RowType) + DataTypes.ROW( + DataTypes.FIELD("id", DataTypes.INT()), + DataTypes.FIELD("name", DataTypes.STRING()), + DataTypes.FIELD("amount", DataTypes.BIGINT())) + .getLogicalType(); + + private BinlogFlinkTableSource createSource() { + FlinkConnectorOptionsUtils.StartupOptions startupOptions = + new FlinkConnectorOptionsUtils.StartupOptions(); + startupOptions.startupMode = ScanStartupMode.EARLIEST; + return new BinlogFlinkTableSource( + TablePath.of("db", "t"), + new Configuration(), + BinlogRowConverter.buildBinlogRowType(DATA_COLUMNS_TYPE), + false, + true, + startupOptions, + 1000L, + Collections.emptyMap()); + } + + private DataType projectedType(int... topLevelIndexes) { + RowType full = BinlogRowConverter.buildBinlogRowType(DATA_COLUMNS_TYPE); + DataTypes.Field[] fields = new DataTypes.Field[topLevelIndexes.length]; + for (int i = 0; i < topLevelIndexes.length; i++) { + RowType.RowField f = full.getFields().get(topLevelIndexes[i]); + fields[i] = + DataTypes.FIELD( + f.getName(), + org.apache.flink.table.types.utils.TypeConversions + .fromLogicalToDataType(f.getType())); + } + return DataTypes.ROW(fields); + } + + private int[][] nested(int... indexes) { + int[][] result = new int[indexes.length][]; + for (int i = 0; i < indexes.length; i++) { + result[i] = new int[] {indexes[i]}; + } + return result; + } + + @Test + void testApplyProjectionStoresTopLevel() { + BinlogFlinkTableSource source = createSource(); + // virtual [_change_type(0), after(4)] + source.applyProjection(nested(0, 4), projectedType(0, 4)); + + assertThat(source.getProjectedTopLevel()).containsExactly(0, 4); + assertThat(source.getProducedDataType()).isEqualTo(projectedType(0, 4).getLogicalType()); + } + + @Test + void testCopyPreservesPushdownState() { + BinlogFlinkTableSource source = createSource(); + source.applyProjection(nested(3, 4), projectedType(3, 4)); + + BinlogFlinkTableSource copy = (BinlogFlinkTableSource) source.copy(); + assertThat(copy.getProjectedTopLevel()).containsExactly(3, 4); + assertThat(copy.getProducedDataType()).isEqualTo(source.getProducedDataType()); + } + + @Test + void testApplyFiltersReturnsAllRemaining() { + BinlogFlinkTableSource source = createSource(); + FieldReferenceExpression ref = + new FieldReferenceExpression("_change_type", DataTypes.STRING(), 0, 0); + ValueLiteralExpression lit = + new ValueLiteralExpression("insert", DataTypes.STRING().notNull()); + ResolvedExpression filter = + CallExpression.permanent( + BuiltInFunctionDefinitions.EQUALS, + Arrays.asList(ref, lit), + DataTypes.BOOLEAN()); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(filter)); + + // Binlog data/partition columns are nested and metadata is non-pushable: nothing accepted. + assertThat(result.getAcceptedFilters()).isEmpty(); + assertThat(result.getRemainingFilters()).containsExactly(filter); + assertThat(source.getPartitionFilters()).isNull(); + } +} diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java index 4787fe162da..675be66d57d 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java @@ -345,6 +345,39 @@ public void testBinlogSelectStar() throws Exception { .isEqualTo("+I[insert, 0, 1970-01-01T00:00:01Z, null, +I[1, Alice]]"); } + @Test + public void testBinlogTopLevelProjection() throws Exception { + // Select only the top-level columns _change_type and after (skip _log_offset, + // _commit_timestamp and before). The data scan stays full, so the nested after ROW still + // carries all original columns. + tEnv.executeSql( + "CREATE TABLE binlog_projection_test (" + + " id INT NOT NULL," + + " name STRING," + + " amount BIGINT," + + " PRIMARY KEY (id) NOT ENFORCED" + + ") WITH ('bucket.num' = '1')"); + + TablePath tablePath = TablePath.of(DEFAULT_DB, "binlog_projection_test"); + + String query = "SELECT _change_type, after FROM binlog_projection_test$binlog"; + + // The top-level projection should be pushed into the source scan (assert the exact + // pushed projection rather than just the presence of "project=["). + assertThat(tEnv.explainSql(query)).contains("project=[_change_type, after]"); + + CloseableIterator rowIter = tEnv.executeSql(query).collect(); + + CLOCK.advanceTime(Duration.ofMillis(1000)); + writeRows(conn, tablePath, Arrays.asList(row(1, "Alice", 100L)), false); + + List results = collectRowsWithTimeout(rowIter, 1, true); + assertThat(results).hasSize(1); + // after ROW retains all data columns even though only two top-level columns are projected. + assertThat(results.get(0)).isEqualTo("+I[insert, +I[1, Alice, 100]]"); + rowIter.close(); + } + @Test public void testBinlogWithPartitionedTable() throws Exception { // Create a partitioned primary key table diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java new file mode 100644 index 00000000000..cbbf390218d --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java @@ -0,0 +1,241 @@ +/* + * 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.fluss.flink.source; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.flink.utils.ChangelogRowConverter; +import org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils; +import org.apache.fluss.metadata.TablePath; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown; +import org.apache.flink.table.expressions.CallExpression; +import org.apache.flink.table.expressions.FieldReferenceExpression; +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.types.DataType; +import org.apache.flink.table.types.logical.RowType; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.apache.fluss.flink.FlinkConnectorOptions.ScanStartupMode; +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit tests for {@link ChangelogFlinkTableSource} projection and filter pushdown. */ +class ChangelogFlinkTableSourceTest { + + // Data columns: (id INT, name STRING, amount BIGINT) + private static final RowType DATA_COLUMNS_TYPE = + (RowType) + DataTypes.ROW( + DataTypes.FIELD("id", DataTypes.INT()), + DataTypes.FIELD("name", DataTypes.STRING()), + DataTypes.FIELD("amount", DataTypes.BIGINT())) + .getLogicalType(); + + private ChangelogFlinkTableSource createSource(int[] partitionKeyIndexes) { + return createSource(partitionKeyIndexes, Collections.emptyMap()); + } + + private ChangelogFlinkTableSource createSource( + int[] partitionKeyIndexes, Map extraOptions) { + FlinkConnectorOptionsUtils.StartupOptions startupOptions = + new FlinkConnectorOptionsUtils.StartupOptions(); + startupOptions.startupMode = ScanStartupMode.EARLIEST; + Map tableOptions = new HashMap<>(extraOptions); + return new ChangelogFlinkTableSource( + TablePath.of("db", "t"), + new Configuration(), + ChangelogRowConverter.buildChangelogRowType(DATA_COLUMNS_TYPE), + partitionKeyIndexes, + true, + startupOptions, + 1000L, + tableOptions); + } + + private DataType projectedType(int... virtualIndexes) { + RowType full = ChangelogRowConverter.buildChangelogRowType(DATA_COLUMNS_TYPE); + DataTypes.Field[] fields = new DataTypes.Field[virtualIndexes.length]; + for (int i = 0; i < virtualIndexes.length; i++) { + RowType.RowField f = full.getFields().get(virtualIndexes[i]); + fields[i] = + DataTypes.FIELD( + f.getName(), + org.apache.flink.table.types.utils.TypeConversions + .fromLogicalToDataType(f.getType())); + } + return DataTypes.ROW(fields); + } + + private int[][] nested(int... indexes) { + int[][] result = new int[indexes.length][]; + for (int i = 0; i < indexes.length; i++) { + result[i] = new int[] {indexes[i]}; + } + return result; + } + + private CallExpression equals(String field, DataType fieldType, Object value) { + FieldReferenceExpression ref = new FieldReferenceExpression(field, fieldType, 0, 0); + ValueLiteralExpression lit = new ValueLiteralExpression(value, fieldType.notNull()); + return CallExpression.permanent( + BuiltInFunctionDefinitions.EQUALS, Arrays.asList(ref, lit), DataTypes.BOOLEAN()); + } + + @Test + void testApplyProjectionMetadataOnly() { + ChangelogFlinkTableSource source = createSource(new int[0]); + source.applyProjection(nested(0), projectedType(0)); + + assertThat(source.getProjectedFields()).containsExactly(0); + assertThat(source.getDataProjection()).isNull(); + assertThat(source.getBaseRowProjection()).containsExactly(0); + assertThat(source.getProducedDataType()).isEqualTo(projectedType(0).getLogicalType()); + } + + @Test + void testApplyProjectionDataOnly() { + ChangelogFlinkTableSource source = createSource(new int[0]); + // virtual [id(3), amount(5)] + source.applyProjection(nested(3, 5), projectedType(3, 5)); + + assertThat(source.getDataProjection()).containsExactly(0, 2); + assertThat(source.getBaseRowProjection()).containsExactly(3, 4); + } + + @Test + void testApplyProjectionReorderedMix() { + ChangelogFlinkTableSource source = createSource(new int[0]); + // virtual [_change_type(0), amount(5), id(3)] + source.applyProjection(nested(0, 5, 3), projectedType(0, 5, 3)); + + // data columns scanned in first-appearance order: amount(2), id(0) + assertThat(source.getDataProjection()).containsExactly(2, 0); + // base row [_change_type, amount, id] over [meta0, meta1, meta2, amount, id] + assertThat(source.getBaseRowProjection()).containsExactly(0, 3, 4); + } + + @Test + void testCopyPreservesPushdownState() { + ChangelogFlinkTableSource source = createSource(new int[0], statsAll()); + source.applyProjection(nested(0, 3), projectedType(0, 3)); + source.applyFilters(Collections.singletonList(equals("amount", DataTypes.BIGINT(), 100L))); + + ChangelogFlinkTableSource copy = (ChangelogFlinkTableSource) source.copy(); + assertThat(copy.getProjectedFields()).containsExactly(0, 3); + assertThat(copy.getDataProjection()).containsExactly(0); + assertThat(copy.getBaseRowProjection()).containsExactly(0, 3); + assertThat(copy.getProducedDataType()).isEqualTo(source.getProducedDataType()); + assertThat(copy.getLogRecordBatchFilter()).isEqualTo(source.getLogRecordBatchFilter()); + } + + @Test + void testApplyFiltersDataColumnPushed() { + ChangelogFlinkTableSource source = createSource(new int[0], statsAll()); + ResolvedExpression filter = equals("amount", DataTypes.BIGINT(), 100L); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(filter)); + + assertThat(result.getAcceptedFilters()).containsExactly(filter); + // All filters are returned as remaining (safety net). + assertThat(result.getRemainingFilters()).containsExactly(filter); + assertThat(source.getLogRecordBatchFilter()).isNotNull(); + } + + @Test + void testApplyFiltersMetadataColumnNotPushed() { + ChangelogFlinkTableSource source = createSource(new int[0], statsAll()); + ResolvedExpression filter = equals("_change_type", DataTypes.STRING(), "insert"); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(filter)); + + assertThat(result.getAcceptedFilters()).isEmpty(); + assertThat(result.getRemainingFilters()).containsExactly(filter); + assertThat(source.getLogRecordBatchFilter()).isNull(); + } + + @Test + void testApplyFiltersPartitionKeyPushed() { + // partition key = data column "name" (index 1) + ChangelogFlinkTableSource source = createSource(new int[] {1}); + ResolvedExpression filter = equals("name", DataTypes.STRING(), "p1"); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(filter)); + + assertThat(result.getAcceptedFilters()).containsExactly(filter); + assertThat(result.getRemainingFilters()).containsExactly(filter); + assertThat(source.getPartitionFilters()).isNotNull(); + } + + @Test + void testApplyFiltersNonPartitionKeyNotPushedToPartitionFilters() { + // partition key = data column "name" (index 1); the filter references the non-partition + // column "amount", so it must not create a partition filter and must stay a post-filter. + ChangelogFlinkTableSource source = createSource(new int[] {1}); + ResolvedExpression filter = equals("amount", DataTypes.BIGINT(), 100L); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(filter)); + + // Statistics are disabled by default, so the filter is not accepted as a batch filter + // either. + assertThat(result.getAcceptedFilters()).isEmpty(); + assertThat(result.getRemainingFilters()).containsExactly(filter); + assertThat(source.getPartitionFilters()).isNull(); + assertThat(source.getLogRecordBatchFilter()).isNull(); + } + + @Test + void testApplyFiltersCompoundPartitionAndNonPartitionNotPushed() { + // partition key = data column "name" (index 1). A compound filter referencing both a + // partition column and a non-partition column cannot be used for partition pruning as a + // whole; it must remain a post-filter so no condition is silently dropped. + ChangelogFlinkTableSource source = createSource(new int[] {1}); + ResolvedExpression compound = + CallExpression.permanent( + BuiltInFunctionDefinitions.AND, + Arrays.asList( + equals("name", DataTypes.STRING(), "p1"), + equals("amount", DataTypes.BIGINT(), 100L)), + DataTypes.BOOLEAN()); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(compound)); + + // Statistics are disabled by default, so nothing is accepted at all. + assertThat(result.getAcceptedFilters()).isEmpty(); + assertThat(result.getRemainingFilters()).containsExactly(compound); + assertThat(source.getPartitionFilters()).isNull(); + assertThat(source.getLogRecordBatchFilter()).isNull(); + } + + private Map statsAll() { + Map options = new HashMap<>(); + options.put("table.statistics.columns", "*"); + return options; + } +} diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java index d9fbf9e5c1d..b9df1a45690 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java @@ -344,6 +344,108 @@ public void testProjectionOnChangelogTable() throws Exception { assertThat(deleteResult.get(0)).isEqualTo("+I[delete, 1, Item-1-Updated]"); } + @Test + public void testChangelogDataColumnFilterAndProjectionPushdown() throws Exception { + // Statistics columns are only supported on log tables (not PK tables), so use a log table. + // A log table's changelog is append-only, which is sufficient to exercise pushdown. + tEnv.executeSql( + "CREATE TABLE data_filter_test (" + + " id INT NOT NULL," + + " name STRING," + + " amount BIGINT" + + ") WITH ('bucket.num' = '1', 'table.statistics.columns' = 'amount')"); + + TablePath tablePath = TablePath.of(DEFAULT_DB, "data_filter_test"); + + String query = + "SELECT _change_type, id, amount FROM data_filter_test$changelog WHERE amount > 150"; + + // Projection and the data-column filter should be pushed into the source scan. Assert the + // exact pushed projection and predicate, because "filter=[" would also match an empty + // "filter=[]" digest when nothing is pushed down. + String plan = tEnv.explainSql(query); + assertThat(plan).contains("project=[_change_type, id, amount]"); + assertThat(plan).contains("filter=[>(amount, 150)]"); + // The filter is still retained in the Calc operator as a safety net (FLINK-38635). + assertThat(plan).contains("where="); + + CloseableIterator rowIter = tEnv.executeSql(query).collect(); + + CLOCK.advanceTime(Duration.ofMillis(100)); + writeRows( + conn, + tablePath, + Arrays.asList( + row(1, "Item-1", 100L), row(2, "Item-2", 200L), row(3, "Item-3", 300L)), + true); + + // Only amount > 150 rows pass the filter (safety net + pushdown produce identical results). + List results = collectRowsWithTimeout(rowIter, 2, true); + assertThat(results).containsExactly("+I[insert, 2, 200]", "+I[insert, 3, 300]"); + rowIter.close(); + } + + @Test + public void testChangelogPartitionFilterPushdown() throws Exception { + tEnv.executeSql( + "CREATE TABLE partition_filter_test (" + + " id INT NOT NULL," + + " name STRING," + + " region STRING NOT NULL," + + " PRIMARY KEY (id, region) NOT ENFORCED" + + ") PARTITIONED BY (region) WITH ('bucket.num' = '1')"); + + CLOCK.advanceTime(Duration.ofMillis(100)); + tEnv.executeSql( + "INSERT INTO partition_filter_test VALUES " + + "(1, 'Item-1', 'us'), " + + "(2, 'Item-2', 'us'), " + + "(3, 'Item-3', 'eu')") + .await(); + + String query = + "SELECT _change_type, id, name, region FROM partition_filter_test$changelog " + + "WHERE region = 'us'"; + + // The partition-key filter should be pushed into the source scan (assert the exact pushed + // predicate), and retained in the Calc operator as a safety net (FLINK-38635). + String plan = tEnv.explainSql(query); + assertThat(plan).contains("filter=[=(region, _UTF-16LE'us'"); + assertThat(plan).contains("where="); + + CloseableIterator rowIter = tEnv.executeSql(query).collect(); + List results = collectRowsWithTimeout(rowIter, 2, true); + assertThat(results) + .containsExactly("+I[insert, 1, Item-1, us]", "+I[insert, 2, Item-2, us]"); + rowIter.close(); + } + + @Test + public void testChangelogMetadataOnlyProjection() throws Exception { + // Projecting only metadata columns leaves no data column to scan. The source normalizes the + // scan projection to null (full data row scan) to avoid a zero-column scan that produces no + // records, then emits only the metadata columns. Verify this runtime path works end to end. + tEnv.executeSql( + "CREATE TABLE metadata_only_test (" + + " id INT NOT NULL," + + " name STRING" + + ") WITH ('bucket.num' = '1')"); + + TablePath tablePath = TablePath.of(DEFAULT_DB, "metadata_only_test"); + + String query = "SELECT _change_type, _log_offset FROM metadata_only_test$changelog"; + assertThat(tEnv.explainSql(query)).contains("project=[_change_type, _log_offset]"); + + CloseableIterator rowIter = tEnv.executeSql(query).collect(); + + CLOCK.advanceTime(Duration.ofMillis(100)); + writeRows(conn, tablePath, Arrays.asList(row(1, "Item-1"), row(2, "Item-2")), true); + + List results = collectRowsWithTimeout(rowIter, 2, true); + assertThat(results).containsExactly("+I[insert, 0]", "+I[insert, 1]"); + rowIter.close(); + } + @Test public void testChangelogScanWithAllChangeTypes() throws Exception { // Create a primary key table with 1 bucket for consistent log_offset numbers diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/BinlogRowConverterTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/BinlogRowConverterTest.java index a09dd7c4199..ee0ab4e67e7 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/BinlogRowConverterTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/BinlogRowConverterTest.java @@ -229,6 +229,71 @@ void testMultipleUpdatesInSequence() throws Exception { assertThat(result2.getLong(1)).isEqualTo(20L); // offset from second -U } + @Test + void testProjectionEmitsSelectedTopLevelColumns() throws Exception { + // Project the binlog row [_change_type(0), _log_offset(1), _commit_timestamp(2), + // before(3), after(4)] down to [_change_type, after]. The data scan stays full, so the + // nested after ROW still contains all columns. + BinlogRowConverter projected = new BinlogRowConverter(testRowType, new int[] {0, 4}); + + org.apache.flink.table.types.logical.RowType producedType = projected.getProducedType(); + assertThat(producedType.getFieldNames()).containsExactly("_change_type", "after"); + assertThat(producedType.getTypeAt(1)) + .isInstanceOf(org.apache.flink.table.types.logical.RowType.class); + + RowData result = + projected.convert( + createLogRecord(ChangeType.INSERT, 100L, 1000L, 1, "Alice", 5000L)); + + assertThat(result.getArity()).isEqualTo(2); + assertThat(result.getRowKind()).isEqualTo(RowKind.INSERT); + assertThat(result.getString(0)).isEqualTo(StringData.fromString("insert")); + RowData afterRow = result.getRow(1, 3); + assertThat(afterRow.getInt(0)).isEqualTo(1); + assertThat(afterRow.getString(1).toString()).isEqualTo("Alice"); + assertThat(afterRow.getLong(2)).isEqualTo(5000L); + } + + @Test + void testProjectionUpdateMerge() throws Exception { + // Projection must not disturb the -U/+U buffering: project [before, after]. + BinlogRowConverter projected = new BinlogRowConverter(testRowType, new int[] {3, 4}); + + assertThat( + projected.convert( + createLogRecord( + ChangeType.UPDATE_BEFORE, 200L, 2000L, 2, "Bob", 3000L))) + .isNull(); + RowData result = + projected.convert( + createLogRecord( + ChangeType.UPDATE_AFTER, 201L, 2000L, 2, "Bob-Updated", 4000L)); + + assertThat(result.getArity()).isEqualTo(2); + RowData beforeRow = result.getRow(0, 3); + assertThat(beforeRow.getString(1).toString()).isEqualTo("Bob"); + RowData afterRow = result.getRow(1, 3); + assertThat(afterRow.getString(1).toString()).isEqualTo("Bob-Updated"); + } + + @Test + void testProjectionDeleteRecord() throws Exception { + // Project [_change_type, before, after]. For a DELETE the after image is null and must + // pass through the projection untouched. + BinlogRowConverter projected = new BinlogRowConverter(testRowType, new int[] {0, 3, 4}); + + RowData result = + projected.convert( + createLogRecord(ChangeType.DELETE, 300L, 3000L, 3, "Charlie", 1000L)); + + assertThat(result.getArity()).isEqualTo(3); + assertThat(result.getString(0)).isEqualTo(StringData.fromString("delete")); + RowData beforeRow = result.getRow(1, 3); + assertThat(beforeRow.getInt(0)).isEqualTo(3); + assertThat(beforeRow.getString(1).toString()).isEqualTo("Charlie"); + assertThat(result.isNullAt(2)).isTrue(); + } + private LogRecord createLogRecord( ChangeType changeType, long offset, long timestamp, int id, String name, long amount) throws Exception { diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java index 55f83a51498..dd828c69817 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java @@ -185,6 +185,40 @@ void testAllChangeTypes() throws Exception { .isEqualTo(StringData.fromString("insert")); } + @Test + void testProjectionEmitsSelectedColumnsInOrder() throws Exception { + // Project the base changelog row + // [_change_type(0), _log_offset(1), _commit_timestamp(2), id(3), name(4), amount(5)] + // down to [_log_offset, amount, id] to verify subsetting + reordering. + ChangelogRowConverter projected = + new ChangelogRowConverter(testRowType, new int[] {1, 5, 3}); + + org.apache.flink.table.types.logical.RowType producedType = projected.getProducedType(); + assertThat(producedType.getFieldNames()).containsExactly("_log_offset", "amount", "id"); + + RowData result = + projected.convert(createLogRecord(ChangeType.INSERT, 100L, 1, "Alice", 5000L)); + + assertThat(result.getArity()).isEqualTo(3); + assertThat(result.getRowKind()).isEqualTo(RowKind.INSERT); + assertThat(result.getLong(0)).isEqualTo(100L); // _log_offset + assertThat(result.getLong(1)).isEqualTo(5000L); // amount + assertThat(result.getInt(2)).isEqualTo(1); // id + } + + @Test + void testProjectionMetadataOnly() throws Exception { + // Project only the _change_type metadata column. + ChangelogRowConverter projected = new ChangelogRowConverter(testRowType, new int[] {0}); + + assertThat(projected.getProducedType().getFieldNames()).containsExactly("_change_type"); + + RowData result = + projected.convert(createLogRecord(ChangeType.DELETE, 300L, 3, "Charlie", 1000L)); + assertThat(result.getArity()).isEqualTo(1); + assertThat(result.getString(0)).isEqualTo(StringData.fromString("delete")); + } + private LogRecord createLogRecord( ChangeType changeType, long offset, int id, String name, long amount) throws Exception { // Create an IndexedRow with test data From f5420ddbd33defdf698d6cccee011230c6f582ab Mon Sep 17 00:00:00 2001 From: naivedogger Date: Wed, 22 Jul 2026 10:29:33 +0800 Subject: [PATCH 02/11] [flink] Add logging to PushdownUtils statistics column resolution Introduce an SLF4J logger in PushdownUtils and emit diagnostics when statistics collection is disabled, and when a configured statistics column is skipped due to an unsupported type or because it is absent from the table schema. --- .../fluss/flink/utils/PushdownUtils.java | 22 +++++++++++++++---- 1 file changed, 18 insertions(+), 4 deletions(-) diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java index 4684cd0b5a7..1dde6bf3da3 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PushdownUtils.java @@ -64,6 +64,8 @@ import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.RowType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import javax.annotation.Nullable; @@ -88,6 +90,8 @@ /** Utilities for pushdown abilities. */ public class PushdownUtils { + private static final Logger LOG = LoggerFactory.getLogger(PushdownUtils.class); + private static final int MAX_LIMIT_PUSHDOWN = 2048; /** Extract field equality information from expressions. */ @@ -472,6 +476,7 @@ public static Set computeAvailableStatsColumns( org.apache.fluss.types.RowType flussRowType, TableConfig tableConfig) { StatisticsColumnsConfig statsConfig = tableConfig.getStatisticsColumns(); if (!statsConfig.isEnabled()) { + LOG.debug("Statistics collection is disabled for the table"); return Collections.emptySet(); } @@ -485,10 +490,19 @@ public static Set computeAvailableStatsColumns( } else { for (String columnName : statsConfig.getColumns()) { int columnIndex = flussRowType.getFieldNames().indexOf(columnName); - if (columnIndex >= 0 - && DataTypeChecks.isSupportedStatisticsType( - flussRowType.getTypeAt(columnIndex))) { - columns.add(columnName); + if (columnIndex >= 0) { + if (DataTypeChecks.isSupportedStatisticsType( + flussRowType.getTypeAt(columnIndex))) { + columns.add(columnName); + } else { + LOG.trace( + "Configured statistics column '{}' has unsupported type and will be ignored", + columnName); + } + } else { + LOG.trace( + "Configured statistics column '{}' does not exist in table schema", + columnName); } } } From 5feaa9f127f52eea5642923c82bfa0ef4fe41bdf Mon Sep 17 00:00:00 2001 From: naivedogger Date: Thu, 23 Jul 2026 20:23:50 +0800 Subject: [PATCH 03/11] nit --- .../fluss/flink/source/BinlogFlinkTableSource.java | 14 +++----------- .../flink/source/BinlogFlinkTableSourceTest.java | 1 - 2 files changed, 3 insertions(+), 12 deletions(-) diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java index 1ffa1de6162..64e59de7725 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java @@ -25,7 +25,6 @@ import org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils; import org.apache.fluss.flink.utils.FlinkConversions; import org.apache.fluss.metadata.TablePath; -import org.apache.fluss.predicate.Predicate; import org.apache.fluss.types.RowType; import org.apache.flink.annotation.VisibleForTesting; @@ -70,8 +69,6 @@ public class BinlogFlinkTableSource @Nullable private int[] projectedTopLevel; private LogicalType producedDataType; - @Nullable private Predicate partitionFilters; - public BinlogFlinkTableSource( TablePath tablePath, Configuration flussConfig, @@ -166,7 +163,9 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { splitPerAssignmentBatchSize, new BinlogDeserializationSchema(projectedTopLevel), streaming, - partitionFilters, + // $binlog data/partition columns are nested inside before/after ROWs, so no + // top-level partition filter is pushable; always scan without one. + null, LeaseContext.DEFAULT); return SourceProvider.of(source); @@ -187,7 +186,6 @@ public DynamicTableSource copy() { tableOptions); copy.producedDataType = producedDataType; copy.projectedTopLevel = projectedTopLevel; - copy.partitionFilters = partitionFilters; return copy; } @@ -231,10 +229,4 @@ int[] getProjectedTopLevel() { LogicalType getProducedDataType() { return producedDataType; } - - @VisibleForTesting - @Nullable - Predicate getPartitionFilters() { - return partitionFilters; - } } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogFlinkTableSourceTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogFlinkTableSourceTest.java index 9a76ef5eb01..b5f8f5e4ab3 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogFlinkTableSourceTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogFlinkTableSourceTest.java @@ -127,6 +127,5 @@ void testApplyFiltersReturnsAllRemaining() { // Binlog data/partition columns are nested and metadata is non-pushable: nothing accepted. assertThat(result.getAcceptedFilters()).isEmpty(); assertThat(result.getRemainingFilters()).containsExactly(filter); - assertThat(source.getPartitionFilters()).isNull(); } } From b30288268b6981ef46bfee966d2b247b64c7cd52 Mon Sep 17 00:00:00 2001 From: naivedogger Date: Fri, 24 Jul 2026 12:13:41 +0800 Subject: [PATCH 04/11] refine tests --- .../source/ChangelogFlinkTableSourceTest.java | 119 ++++++++++++---- .../utils/ChangelogRowConverterTest.java | 130 +++--------------- 2 files changed, 111 insertions(+), 138 deletions(-) diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java index cbbf390218d..1daa50d858a 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java @@ -31,12 +31,17 @@ import org.apache.flink.table.functions.BuiltInFunctionDefinitions; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.RowType; + import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.stream.Stream; import static org.apache.fluss.flink.FlinkConnectorOptions.ScanStartupMode; import static org.assertj.core.api.Assertions.assertThat; @@ -97,43 +102,52 @@ private int[][] nested(int... indexes) { } private CallExpression equals(String field, DataType fieldType, Object value) { + return call(BuiltInFunctionDefinitions.EQUALS, field, fieldType, value); + } + + private CallExpression call( + org.apache.flink.table.functions.BuiltInFunctionDefinition func, + String field, + DataType fieldType, + Object value) { FieldReferenceExpression ref = new FieldReferenceExpression(field, fieldType, 0, 0); ValueLiteralExpression lit = new ValueLiteralExpression(value, fieldType.notNull()); - return CallExpression.permanent( - BuiltInFunctionDefinitions.EQUALS, Arrays.asList(ref, lit), DataTypes.BOOLEAN()); + return CallExpression.permanent(func, Arrays.asList(ref, lit), DataTypes.BOOLEAN()); } - @Test - void testApplyProjectionMetadataOnly() { + @ParameterizedTest + @MethodSource("projectionCases") + void testApplyProjection( + String description, + int[] virtual, + int[] expectedDataProjection, + int[] expectedBaseProjection) { ChangelogFlinkTableSource source = createSource(new int[0]); - source.applyProjection(nested(0), projectedType(0)); + source.applyProjection(nested(virtual), projectedType(virtual)); - assertThat(source.getProjectedFields()).containsExactly(0); - assertThat(source.getDataProjection()).isNull(); - assertThat(source.getBaseRowProjection()).containsExactly(0); - assertThat(source.getProducedDataType()).isEqualTo(projectedType(0).getLogicalType()); - } + assertThat(source.getProjectedFields()).containsExactly(virtual); - @Test - void testApplyProjectionDataOnly() { - ChangelogFlinkTableSource source = createSource(new int[0]); - // virtual [id(3), amount(5)] - source.applyProjection(nested(3, 5), projectedType(3, 5)); + // dataProjection null for metadata-only projection + if (expectedDataProjection == null) { + assertThat(source.getDataProjection()).isNull(); + } else { + assertThat(source.getDataProjection()).containsExactly(expectedDataProjection); + } - assertThat(source.getDataProjection()).containsExactly(0, 2); - assertThat(source.getBaseRowProjection()).containsExactly(3, 4); + assertThat(source.getBaseRowProjection()).containsExactly(expectedBaseProjection); + assertThat(source.getProducedDataType()).isEqualTo(projectedType(virtual).getLogicalType()); } - @Test - void testApplyProjectionReorderedMix() { - ChangelogFlinkTableSource source = createSource(new int[0]); - // virtual [_change_type(0), amount(5), id(3)] - source.applyProjection(nested(0, 5, 3), projectedType(0, 5, 3)); - - // data columns scanned in first-appearance order: amount(2), id(0) - assertThat(source.getDataProjection()).containsExactly(2, 0); - // base row [_change_type, amount, id] over [meta0, meta1, meta2, amount, id] - assertThat(source.getBaseRowProjection()).containsExactly(0, 3, 4); + private static Stream projectionCases() { + return Stream.of( + // description, virtual, expectedDataProjection, expectedBaseProjection + // metadata only + Arguments.of("metadata-only", new int[] {0}, null, new int[] {0}), + // data only + Arguments.of("data-only", new int[] {3, 5}, new int[] {0, 2}, new int[] {3, 4}), + // reordered mix + Arguments.of( + "reorderedMix", new int[] {0, 5, 3}, new int[] {2, 0}, new int[] {0, 3, 4})); } @Test @@ -164,6 +178,51 @@ void testApplyFiltersDataColumnPushed() { assertThat(source.getLogRecordBatchFilter()).isNotNull(); } + @Test + void testApplyFiltersDataColumnNotInStatisticsNotPushed() { + // Statistics are collected only for "amount". A filter on "name" cannot use batch + // statistics, so it must not be pushed as a record-batch filter even though it is a + // convertible data-column predicate. + ChangelogFlinkTableSource source = createSource(new int[0], statsColumns("amount")); + ResolvedExpression filter = equals("name", DataTypes.STRING(), "p1"); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(filter)); + + assertThat(result.getAcceptedFilters()).isEmpty(); + assertThat(result.getRemainingFilters()).containsExactly(filter); + assertThat(source.getLogRecordBatchFilter()).isNull(); + } + + @Test + void testApplyFiltersCompoundDataColumnsPushed() { + // WHERE amount > 100 AND amount < 500: both leaves reference a statistics-backed data + // column, so the whole AND predicate is pushed as a single record-batch filter. + ChangelogFlinkTableSource source = createSource(new int[0], statsAll()); + ResolvedExpression compound = + CallExpression.permanent( + BuiltInFunctionDefinitions.AND, + Arrays.asList( + call( + BuiltInFunctionDefinitions.GREATER_THAN, + "amount", + DataTypes.BIGINT(), + 100L), + call( + BuiltInFunctionDefinitions.LESS_THAN, + "amount", + DataTypes.BIGINT(), + 500L)), + DataTypes.BOOLEAN()); + + SupportsFilterPushDown.Result result = + source.applyFilters(Collections.singletonList(compound)); + + assertThat(result.getAcceptedFilters()).containsExactly(compound); + assertThat(result.getRemainingFilters()).containsExactly(compound); + assertThat(source.getLogRecordBatchFilter()).isNotNull(); + } + @Test void testApplyFiltersMetadataColumnNotPushed() { ChangelogFlinkTableSource source = createSource(new int[0], statsAll()); @@ -234,8 +293,12 @@ void testApplyFiltersCompoundPartitionAndNonPartitionNotPushed() { } private Map statsAll() { + return statsColumns("*"); + } + + private Map statsColumns(String columns) { Map options = new HashMap<>(); - options.put("table.statistics.columns", "*"); + options.put("table.statistics.columns", columns); return options; } } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java index dd828c69817..49502ec97bc 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java @@ -32,6 +32,12 @@ import org.apache.flink.table.data.utils.JoinedRowData; import org.apache.flink.types.RowKind; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import java.util.stream.Stream; + +import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.api.Test; import static org.assertj.core.api.Assertions.assertThat; @@ -55,17 +61,19 @@ void setUp() { converter = new ChangelogRowConverter(testRowType); } - @Test - void testConvertInsertRecord() throws Exception { - LogRecord record = createLogRecord(ChangeType.INSERT, 100L, 1, "Alice", 5000L); + @ParameterizedTest + @MethodSource("changeTypeCases") + void testConvertAllChangeTypes(ChangeType changeType, String expectedChangeType) + throws Exception { + LogRecord record = createLogRecord(changeType, 100L, 1, "Alice", 5000L); RowData result = converter.convert(record); - // Verify row kind + // Virtual table always emits INSERT row kind, regardless of the underlying change type. assertThat(result.getRowKind()).isEqualTo(RowKind.INSERT); // Verify metadata columns - assertThat(result.getString(0)).isEqualTo(StringData.fromString("insert")); + assertThat(result.getString(0)).isEqualTo(StringData.fromString(expectedChangeType)); assertThat(result.getLong(1)).isEqualTo(100L); // log offset assertThat(result.getTimestamp(2, 3)).isNotNull(); // commit timestamp @@ -78,111 +86,13 @@ void testConvertInsertRecord() throws Exception { assertThat(result).isInstanceOf(JoinedRowData.class); } - @Test - void testConvertUpdateBeforeRecord() throws Exception { - LogRecord record = createLogRecord(ChangeType.UPDATE_BEFORE, 200L, 2, "Bob", 3000L); - - RowData result = converter.convert(record); - - // Verify row kind (always INSERT for virtual table) - assertThat(result.getRowKind()).isEqualTo(RowKind.INSERT); - - // Verify change type metadata - assertThat(result.getString(0)).isEqualTo(StringData.fromString("update_before")); - assertThat(result.getLong(1)).isEqualTo(200L); - - // Verify physical columns - assertThat(result.getInt(3)).isEqualTo(2); - assertThat(result.getString(4).toString()).isEqualTo("Bob"); - assertThat(result.getLong(5)).isEqualTo(3000L); - } - - @Test - void testConvertUpdateAfterRecord() throws Exception { - LogRecord record = createLogRecord(ChangeType.UPDATE_AFTER, 201L, 2, "Bob", 4000L); - - RowData result = converter.convert(record); - - assertThat(result.getString(0)).isEqualTo(StringData.fromString("update_after")); - assertThat(result.getLong(1)).isEqualTo(201L); - assertThat(result.getInt(3)).isEqualTo(2); - assertThat(result.getString(4).toString()).isEqualTo("Bob"); - assertThat(result.getLong(5)).isEqualTo(4000L); - } - - @Test - void testConvertDeleteRecord() throws Exception { - LogRecord record = createLogRecord(ChangeType.DELETE, 300L, 3, "Charlie", 1000L); - - RowData result = converter.convert(record); - - assertThat(result.getString(0)).isEqualTo(StringData.fromString("delete")); - assertThat(result.getLong(1)).isEqualTo(300L); - assertThat(result.getInt(3)).isEqualTo(3); - assertThat(result.getString(4).toString()).isEqualTo("Charlie"); - assertThat(result.getLong(5)).isEqualTo(1000L); - } - - @Test - void testProducedTypeHasMetadataColumns() { - org.apache.flink.table.types.logical.RowType producedType = converter.getProducedType(); - - // Should have 3 metadata columns + 3 physical columns - assertThat(producedType.getFieldCount()).isEqualTo(6); - - // Check metadata column names and types - assertThat(producedType.getFieldNames()) - .containsExactly( - "_change_type", "_log_offset", "_commit_timestamp", "id", "name", "amount"); - - // Check metadata column types - assertThat(producedType.getTypeAt(0)) - .isInstanceOf(org.apache.flink.table.types.logical.VarCharType.class); - assertThat(producedType.getTypeAt(1)) - .isInstanceOf(org.apache.flink.table.types.logical.BigIntType.class); - assertThat(producedType.getTypeAt(2)) - .isInstanceOf(org.apache.flink.table.types.logical.LocalZonedTimestampType.class); - } - - @Test - void testAllChangeTypes() throws Exception { - // Test all change type conversions - assertThat( - converter - .convert(createLogRecord(ChangeType.INSERT, 1L, 1, "Test", 100L)) - .getString(0)) - .isEqualTo(StringData.fromString("insert")); - - assertThat( - converter - .convert( - createLogRecord( - ChangeType.UPDATE_BEFORE, 2L, 1, "Test", 100L)) - .getString(0)) - .isEqualTo(StringData.fromString("update_before")); - - assertThat( - converter - .convert( - createLogRecord( - ChangeType.UPDATE_AFTER, 3L, 1, "Test", 100L)) - .getString(0)) - .isEqualTo(StringData.fromString("update_after")); - - assertThat( - converter - .convert(createLogRecord(ChangeType.DELETE, 4L, 1, "Test", 100L)) - .getString(0)) - .isEqualTo(StringData.fromString("delete")); - - // For log tables (append-only) - assertThat( - converter - .convert( - createLogRecord( - ChangeType.APPEND_ONLY, 5L, 1, "Test", 100L)) - .getString(0)) - .isEqualTo(StringData.fromString("insert")); + private static Stream changeTypeCases() { + return Stream.of( + Arguments.of(ChangeType.INSERT, "insert"), + Arguments.of(ChangeType.DELETE, "delete"), + Arguments.of(ChangeType.UPDATE_BEFORE, "update_before"), + Arguments.of(ChangeType.UPDATE_AFTER, "update_after"), + Arguments.of(ChangeType.APPEND_ONLY, "append_only")); } @Test From 1ec333196c1b1eb3d0600562ceab2933917bfa17 Mon Sep 17 00:00:00 2001 From: naivedogger Date: Fri, 24 Jul 2026 12:20:19 +0800 Subject: [PATCH 05/11] spotless apply --- .../source/ChangelogFlinkTableSourceTest.java | 20 ++++++++++--------- .../utils/ChangelogRowConverterTest.java | 9 ++++----- 2 files changed, 15 insertions(+), 14 deletions(-) diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java index 1daa50d858a..a4528ef9d74 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java @@ -31,7 +31,6 @@ import org.apache.flink.table.functions.BuiltInFunctionDefinitions; import org.apache.flink.table.types.DataType; import org.apache.flink.table.types.logical.RowType; - import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; @@ -140,14 +139,17 @@ void testApplyProjection( private static Stream projectionCases() { return Stream.of( - // description, virtual, expectedDataProjection, expectedBaseProjection - // metadata only - Arguments.of("metadata-only", new int[] {0}, null, new int[] {0}), - // data only - Arguments.of("data-only", new int[] {3, 5}, new int[] {0, 2}, new int[] {3, 4}), - // reordered mix - Arguments.of( - "reorderedMix", new int[] {0, 5, 3}, new int[] {2, 0}, new int[] {0, 3, 4})); + // description, virtual, expectedDataProjection, expectedBaseProjection + // metadata only + Arguments.of("metadata-only", new int[] {0}, null, new int[] {0}), + // data only + Arguments.of("data-only", new int[] {3, 5}, new int[] {0, 2}, new int[] {3, 4}), + // reordered mix + Arguments.of( + "reorderedMix", + new int[] {0, 5, 3}, + new int[] {2, 0}, + new int[] {0, 3, 4})); } @Test diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java index 49502ec97bc..42981a94426 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java @@ -32,14 +32,13 @@ import org.apache.flink.table.data.utils.JoinedRowData; import org.apache.flink.types.RowKind; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; import java.util.stream.Stream; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.api.Test; - import static org.assertj.core.api.Assertions.assertThat; /** Unit test for {@link ChangelogRowConverter}. */ @@ -63,8 +62,8 @@ void setUp() { @ParameterizedTest @MethodSource("changeTypeCases") - void testConvertAllChangeTypes(ChangeType changeType, String expectedChangeType) - throws Exception { + void testConvertAllChangeTypes(ChangeType changeType, String expectedChangeType) + throws Exception { LogRecord record = createLogRecord(changeType, 100L, 1, "Alice", 5000L); RowData result = converter.convert(record); From 2123e0f7fbda6527ef405aa6d9de02f3aca13fe8 Mon Sep 17 00:00:00 2001 From: naivedogger Date: Fri, 24 Jul 2026 14:16:18 +0800 Subject: [PATCH 06/11] fix test --- .../org/apache/fluss/flink/utils/ChangelogRowConverterTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java index 42981a94426..5a31fbd591d 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/ChangelogRowConverterTest.java @@ -91,7 +91,7 @@ private static Stream changeTypeCases() { Arguments.of(ChangeType.DELETE, "delete"), Arguments.of(ChangeType.UPDATE_BEFORE, "update_before"), Arguments.of(ChangeType.UPDATE_AFTER, "update_after"), - Arguments.of(ChangeType.APPEND_ONLY, "append_only")); + Arguments.of(ChangeType.APPEND_ONLY, "insert")); } @Test From b2a1eacd1202e52a33288c054c297d4fb4394c43 Mon Sep 17 00:00:00 2001 From: naivedogger Date: Tue, 28 Jul 2026 17:13:59 +0800 Subject: [PATCH 07/11] [flink] Address review comments on changelog pushdown - Use the precomputed Fluss row type for record-batch predicate conversion instead of re-converting the Flink row type per filter. - Replace the O(n^2) contains/indexOf projection derivation with a LinkedHashMap position lookup (first-appearance order preserved). - Only include predicates that pass the partition-key-only check in partitionFilters to avoid pushing down unsafe partition predicates. --- .../source/ChangelogFlinkTableSource.java | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java index 1dde10b8f4f..b1998bac329 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java @@ -53,6 +53,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional; @@ -263,23 +264,24 @@ public void applyProjection(int[][] projectedFields, DataType producedDataType) // Derive the data-column indices actually scanned from Fluss (first-appearance order), // and the projection over the base changelog row [metadata columns + scanned data columns]. - List dataColumns = new ArrayList<>(); + Map dataColumnPositions = new LinkedHashMap<>(); for (int virtualIndex : this.projectedFields) { if (virtualIndex >= METADATA_COLUMN_COUNT) { - int dataIndex = virtualIndex - METADATA_COLUMN_COUNT; - if (!dataColumns.contains(dataIndex)) { - dataColumns.add(dataIndex); - } + dataColumnPositions.putIfAbsent( + virtualIndex - METADATA_COLUMN_COUNT, dataColumnPositions.size()); } } + // When no data column is projected (metadata-only projection, e.g. SELECT _change_type), // normalize the scan projection to null so Fluss reads the full data row. A zero-column // scan produces no records, so the metadata columns must be emitted from a full data scan // and selected later during deserialization via baseRowProjection. this.dataProjection = - dataColumns.isEmpty() + dataColumnPositions.isEmpty() ? null - : dataColumns.stream().mapToInt(Integer::intValue).toArray(); + : dataColumnPositions.keySet().stream() + .mapToInt(Integer::intValue) + .toArray(); int[] base = new int[this.projectedFields.length]; for (int i = 0; i < this.projectedFields.length; i++) { @@ -288,7 +290,7 @@ public void applyProjection(int[][] projectedFields, DataType producedDataType) virtualIndex < METADATA_COLUMN_COUNT ? virtualIndex : METADATA_COLUMN_COUNT - + dataColumns.indexOf(virtualIndex - METADATA_COLUMN_COUNT); + + dataColumnPositions.get(virtualIndex - METADATA_COLUMN_COUNT); } this.baseRowProjection = base; } @@ -319,8 +321,8 @@ public Result applyFilters(List filters) { remainingFilters.add(filter); } else { acceptedFilters.add(filter); + converted.add(p); } - converted.add(p); } else { remainingFilters.add(filter); } @@ -362,7 +364,7 @@ private Result pushdownRecordBatchFilter( List remainingFilters = new ArrayList<>(); for (ResolvedExpression filter : filters) { - Optional predicateOpt = convertToFlussPredicate(dataColumnsType, filter); + Optional predicateOpt = convertToFlussPredicate(dataFlussRowType, filter); if (predicateOpt.isPresent() && PushdownUtils.canPredicateUseStatistics( predicateOpt.get(), dataFlussRowType, availableStatsColumns)) { From 7dfab3e01c27cc580fe0c57ded10232554a10123 Mon Sep 17 00:00:00 2001 From: naivedogger Date: Thu, 30 Jul 2026 16:56:11 +0800 Subject: [PATCH 08/11] [flink] Reuse the produced type computed by the table source instead of re-deriving it in deserialization schemas --- .../flink/source/BinlogFlinkTableSource.java | 4 +++- .../source/ChangelogFlinkTableSource.java | 4 +++- .../BinlogDeserializationSchema.java | 22 +++++++++---------- .../ChangelogDeserializationSchema.java | 22 +++++++++---------- 4 files changed, 26 insertions(+), 26 deletions(-) diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java index 64e59de7725..a4429753482 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java @@ -161,7 +161,9 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { offsetsInitializer, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, - new BinlogDeserializationSchema(projectedTopLevel), + new BinlogDeserializationSchema( + projectedTopLevel, + (org.apache.flink.table.types.logical.RowType) producedDataType), streaming, // $binlog data/partition columns are nested inside before/after ROWs, so no // top-level partition filter is pushable; always scan without one. diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java index b1998bac329..1723057541f 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java @@ -217,7 +217,9 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { offsetsInitializer, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, - new ChangelogDeserializationSchema(baseRowProjection), + new ChangelogDeserializationSchema( + baseRowProjection, + (org.apache.flink.table.types.logical.RowType) producedDataType), streaming, partitionFilters, LeaseContext.DEFAULT); // Lake source not supported diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/BinlogDeserializationSchema.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/BinlogDeserializationSchema.java index c06851c9457..923417251f2 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/BinlogDeserializationSchema.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/BinlogDeserializationSchema.java @@ -27,8 +27,6 @@ import javax.annotation.Nullable; -import static org.apache.fluss.flink.utils.FlinkConversions.toFlinkRowType; - /** * A deserialization schema that converts {@link LogRecord} objects to Flink's {@link RowData} * format with nested before/after row structure for the $binlog virtual table. @@ -52,18 +50,20 @@ public class BinlogDeserializationSchema implements FlussDeserializationSchema getProducedType(RowType rowSchema) { - // Build the output type with nested before/after ROW columns, honoring the projection - org.apache.flink.table.types.logical.RowType outputType = - BinlogRowConverter.buildProducedRowType( - toFlinkRowType(rowSchema), projectedTopLevel); - return InternalTypeInfo.of(outputType); + // The produced type has already been computed by the table source (handed over by the + // Flink planner via applyProjection); no need to derive it again here. + return InternalTypeInfo.of(producedRowType); } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/ChangelogDeserializationSchema.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/ChangelogDeserializationSchema.java index 71c6c7ff6a2..98ff62c5822 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/ChangelogDeserializationSchema.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/ChangelogDeserializationSchema.java @@ -27,8 +27,6 @@ import javax.annotation.Nullable; -import static org.apache.fluss.flink.utils.FlinkConversions.toFlinkRowType; - /** * A deserialization schema that converts {@link LogRecord} objects to Flink's {@link RowData} * format with additional changelog metadata columns. @@ -47,18 +45,20 @@ public class ChangelogDeserializationSchema implements FlussDeserializationSchem */ @Nullable private final int[] baseRowProjection; - /** Creates a new ChangelogDeserializationSchema without projection. */ - public ChangelogDeserializationSchema() { - this(null); - } + /** The produced row type, already computed by the table source honoring the projection. */ + private final org.apache.flink.table.types.logical.RowType producedRowType; /** * Creates a new ChangelogDeserializationSchema. * * @param baseRowProjection projection over the base changelog row, or {@code null} for none + * @param producedRowType the produced row type computed by the table source */ - public ChangelogDeserializationSchema(@Nullable int[] baseRowProjection) { + public ChangelogDeserializationSchema( + @Nullable int[] baseRowProjection, + org.apache.flink.table.types.logical.RowType producedRowType) { this.baseRowProjection = baseRowProjection; + this.producedRowType = producedRowType; } /** Initializes the deserialization schema. */ @@ -86,10 +86,8 @@ public RowData deserialize(LogRecord record) throws Exception { */ @Override public TypeInformation getProducedType(RowType rowSchema) { - // Build the output type with metadata columns, honoring the projection when present - org.apache.flink.table.types.logical.RowType outputType = - ChangelogRowConverter.buildProducedRowType( - toFlinkRowType(rowSchema), baseRowProjection); - return InternalTypeInfo.of(outputType); + // The produced type has already been computed by the table source (handed over by the + // Flink planner via applyProjection); no need to derive it again here. + return InternalTypeInfo.of(producedRowType); } } From c4f764df95d48037270a2d3655b190baef8a5eed Mon Sep 17 00:00:00 2001 From: naivedogger Date: Thu, 30 Jul 2026 17:47:55 +0800 Subject: [PATCH 09/11] [flink] Fix flaky partitioned changelog ITCase by using order-insensitive assertion for cross-partition records --- .../fluss/flink/source/ChangelogVirtualTableITCase.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java index b9df1a45690..aed853d2ae8 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java @@ -641,10 +641,11 @@ public void testChangelogWithPartitionedTable() throws Exception { String query = "SELECT _change_type, id, name, region FROM partitioned_test$changelog"; CloseableIterator rowIter = tEnv.executeSql(query).collect(); - // Collect initial inserts + // Collect initial inserts. Records from different partitions may arrive in any order, + // so the assertion must be order-insensitive. List results = collectRowsWithTimeout(rowIter, 3, false); assertThat(results) - .containsExactly( + .containsExactlyInAnyOrder( "+I[insert, 1, Item-1, us]", "+I[insert, 2, Item-2, us]", "+I[insert, 3, Item-3, eu]"); From 9a7841add460e2f33ff241bef3110e08a3a3885c Mon Sep 17 00:00:00 2001 From: naivedogger Date: Fri, 31 Jul 2026 10:37:09 +0800 Subject: [PATCH 10/11] [flink] Add ITCase for reordered changelog projection --- .../source/ChangelogVirtualTableITCase.java | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java index aed853d2ae8..f5817208f70 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java @@ -344,6 +344,41 @@ public void testProjectionOnChangelogTable() throws Exception { assertThat(deleteResult.get(0)).isEqualTo("+I[delete, 1, Item-1-Updated]"); } + @Test + public void testReorderedProjectionOnChangelogTable() throws Exception { + // Reordered projection: data columns out of their original order with a metadata column + // in between. This exercises the dataProjection/baseRowProjection coordinate translation + // (applyProjection -> ChangelogRowConverter) end to end through the real planner. + tEnv.executeSql( + "CREATE TABLE reordered_projection_test (" + + " id INT NOT NULL," + + " name STRING," + + " amount BIGINT," + + " PRIMARY KEY (id) NOT ENFORCED" + + ") WITH ('bucket.num' = '1')"); + + TablePath tablePath = TablePath.of(DEFAULT_DB, "reordered_projection_test"); + + String query = "SELECT amount, _change_type, id FROM reordered_projection_test$changelog"; + // The exact pushed projection digest may vary across planner versions (the reorder may be + // pushed into the scan or kept in a Calc), so assert result correctness rather than the + // exact plan digest. + assertThat(tEnv.explainSql(query)).contains("project=["); + + CloseableIterator rowIter = tEnv.executeSql(query).collect(); + + CLOCK.advanceTime(Duration.ofMillis(100)); + writeRows(conn, tablePath, Arrays.asList(row(1, "Item-1", 100L)), false); + List insertResult = collectRowsWithTimeout(rowIter, 1, false); + assertThat(insertResult).containsExactly("+I[100, insert, 1]"); + + CLOCK.advanceTime(Duration.ofMillis(100)); + writeRows(conn, tablePath, Arrays.asList(row(1, "Item-1", 250L)), false); + List updateResults = collectRowsWithTimeout(rowIter, 2, true); + assertThat(updateResults) + .containsExactly("+I[100, update_before, 1]", "+I[250, update_after, 1]"); + } + @Test public void testChangelogDataColumnFilterAndProjectionPushdown() throws Exception { // Statistics columns are only supported on log tables (not PK tables), so use a log table. From 842f6ed5cee70a68bb08fcd957c9fab16e66c8b3 Mon Sep 17 00:00:00 2001 From: naivedogger Date: Fri, 31 Jul 2026 11:53:25 +0800 Subject: [PATCH 11/11] [flink] Add binlog reordered projection ITCase and pin exact projection digests --- .../source/BinlogVirtualTableITCase.java | 35 +++++++++++++++++++ .../source/ChangelogVirtualTableITCase.java | 7 ++-- 2 files changed, 38 insertions(+), 4 deletions(-) diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java index 675be66d57d..53f50855bab 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java @@ -378,6 +378,41 @@ public void testBinlogTopLevelProjection() throws Exception { rowIter.close(); } + @Test + public void testBinlogReorderedTopLevelProjection() throws Exception { + // Reordered top-level projection: after before _change_type, i.e. not in the declared + // order of the binlog row. This exercises the projectedTopLevel handling + // (applyProjection -> BinlogRowConverter/ProjectedRowData) end to end through the real + // planner, which no ascending-order projection test covers. + tEnv.executeSql( + "CREATE TABLE binlog_reordered_projection_test (" + + " id INT NOT NULL," + + " name STRING," + + " amount BIGINT," + + " PRIMARY KEY (id) NOT ENFORCED" + + ") WITH ('bucket.num' = '1')"); + + TablePath tablePath = TablePath.of(DEFAULT_DB, "binlog_reordered_projection_test"); + + String query = "SELECT after, _change_type FROM binlog_reordered_projection_test$binlog"; + // The planner pushes the reordered projection into the scan as-is (no Calc reorder), so + // the source receives the out-of-order indices end to end. + assertThat(tEnv.explainSql(query)).contains("project=[after, _change_type]"); + + CloseableIterator rowIter = tEnv.executeSql(query).collect(); + + CLOCK.advanceTime(Duration.ofMillis(100)); + writeRows(conn, tablePath, Arrays.asList(row(1, "Alice", 100L)), false); + List insertResult = collectRowsWithTimeout(rowIter, 1, false); + assertThat(insertResult).containsExactly("+I[+I[1, Alice, 100], insert]"); + + // An update exercises the -U/+U merge together with the reordered projection. + CLOCK.advanceTime(Duration.ofMillis(100)); + writeRows(conn, tablePath, Arrays.asList(row(1, "Alice", 250L)), false); + List updateResult = collectRowsWithTimeout(rowIter, 1, true); + assertThat(updateResult).containsExactly("+I[+I[1, Alice, 250], update]"); + } + @Test public void testBinlogWithPartitionedTable() throws Exception { // Create a partitioned primary key table diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java index f5817208f70..ea5446604c6 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java @@ -360,10 +360,9 @@ public void testReorderedProjectionOnChangelogTable() throws Exception { TablePath tablePath = TablePath.of(DEFAULT_DB, "reordered_projection_test"); String query = "SELECT amount, _change_type, id FROM reordered_projection_test$changelog"; - // The exact pushed projection digest may vary across planner versions (the reorder may be - // pushed into the scan or kept in a Calc), so assert result correctness rather than the - // exact plan digest. - assertThat(tEnv.explainSql(query)).contains("project=["); + // The planner pushes the reordered projection into the scan as-is (no Calc reorder), so + // the source receives the out-of-order indices end to end. + assertThat(tEnv.explainSql(query)).contains("project=[amount, _change_type, id]"); CloseableIterator rowIter = tEnv.executeSql(query).collect();