You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ho...@apache.org on 2026/08/10 11:30:07 UTC
(fluss) branch main updated: [flink] Support projection and filter pushdown for $changelog and $binlog virtual tables (#3759)
This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new ffa665a61f [flink] Support projection and filter pushdown for $changelog and $binlog virtual tables (#3759)
ffa665a61f is described below
commit ffa665a61f62db33ed2fdbece91a61d56d29f145
Author: naivedogger <ya...@gmail.com>
AuthorDate: Mon Aug 10 19:30:02 2026 +0800
[flink] Support projection and filter pushdown for $changelog and $binlog virtual tables (#3759)
---
.../fluss/flink/source/BinlogFlinkTableSource.java | 63 +++--
.../flink/source/ChangelogFlinkTableSource.java | 197 +++++++++++++-
.../org/apache/fluss/flink/source/FlinkSource.java | 160 ++++++++++-
.../fluss/flink/source/FlinkTableSource.java | 94 +------
.../deserializer/BinlogDeserializationSchema.java | 11 +-
.../ChangelogDeserializationSchema.java | 11 +-
.../deserializer/FlussDeserializationSchema.java | 19 +-
.../flink/source/emitter/FlinkRecordEmitter.java | 29 ++
.../flink/source/emitter/RowDataProjection.java | 45 ++++
.../apache/fluss/flink/utils/PushdownUtils.java | 93 +++++++
.../flink/source/BinlogFlinkTableSourceTest.java | 129 +++++++++
.../flink/source/BinlogVirtualTableITCase.java | 67 +++++
.../source/ChangelogFlinkTableSourceTest.java | 296 +++++++++++++++++++++
.../flink/source/ChangelogVirtualTableITCase.java | 141 +++++++++-
.../flink/utils/ChangelogRowConverterTest.java | 130 ++-------
15 files changed, 1230 insertions(+), 255 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 93261e8430..d442dee27a 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,22 +25,26 @@ 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.metadata.TablePath;
-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.types.logical.LogicalType;
-
-import javax.annotation.Nullable;
+import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.types.DataType;
+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,11 +59,8 @@ public class BinlogFlinkTableSource implements ScanTableSource {
private final int splitPerAssignmentBatchSize;
private final Map<String, String> tableOptions;
- // Projection pushdown
- @Nullable private int[] projectedFields;
- private LogicalType producedDataType;
-
- @Nullable private Predicate partitionFilters;
+ // Projection pushdown.
+ private org.apache.flink.table.types.logical.RowType producedDataType;
public BinlogFlinkTableSource(
TablePath tablePath,
@@ -116,11 +117,9 @@ public class BinlogFlinkTableSource implements ScanTableSource {
@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,14 +149,17 @@ public class BinlogFlinkTableSource implements ScanTableSource {
false,
isPartitioned,
flussRowType,
- projectedFields,
+ null,
null,
offsetsInitializer,
scanPartitionDiscoveryIntervalMs,
splitPerAssignmentBatchSize,
new BinlogDeserializationSchema(),
+ FlinkConversions.toFlussRowType(producedDataType),
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);
@@ -177,8 +179,6 @@ public class BinlogFlinkTableSource implements ScanTableSource {
splitPerAssignmentBatchSize,
tableOptions);
copy.producedDataType = producedDataType;
- copy.projectedFields = projectedFields;
- copy.partitionFilters = partitionFilters;
return copy;
}
@@ -187,6 +187,29 @@ public class BinlogFlinkTableSource implements ScanTableSource {
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) {
+ this.producedDataType =
+ (org.apache.flink.table.types.logical.RowType) producedDataType.getLogicalType();
+ }
+
+ @Override
+ public Result applyFilters(List<ResolvedExpression> 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
+ org.apache.flink.table.types.logical.RowType getProducedDataType() {
+ return producedDataType;
+ }
}
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 4304b271b2..fef9f40ebd 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 @@ package org.apache.fluss.flink.source;
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.types.logical.LogicalType;
+import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.types.DataType;
import javax.annotation.Nullable;
+import java.util.ArrayList;
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;
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;
@@ -63,9 +80,16 @@ public class ChangelogFlinkTableSource implements ScanTableSource {
// Projection pushdown
@Nullable private int[] projectedFields;
- private LogicalType producedDataType;
+ private org.apache.flink.table.types.logical.RowType producedDataType;
+ // Data-column indices actually scanned from Fluss (derived from the virtual projection).
+ @Nullable private int[] dataProjection;
@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<String> METADATA_COLUMN_NAMES =
new HashSet<>(
@@ -147,10 +171,8 @@ public class ChangelogFlinkTableSource implements ScanTableSource {
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 +209,13 @@ public class ChangelogFlinkTableSource implements ScanTableSource {
false,
isPartitioned(),
flussRowType,
- projectedFields,
- null,
+ dataProjection,
+ logRecordBatchFilter,
offsetsInitializer,
scanPartitionDiscoveryIntervalMs,
splitPerAssignmentBatchSize,
new ChangelogDeserializationSchema(),
+ FlinkConversions.toFlussRowType(producedDataType),
streaming,
partitionFilters,
LeaseContext.DEFAULT); // Lake source not supported
@@ -215,7 +238,9 @@ public class ChangelogFlinkTableSource implements ScanTableSource {
tableOptions);
copy.producedDataType = producedDataType;
copy.projectedFields = projectedFields;
+ copy.dataProjection = dataProjection;
copy.partitionFilters = partitionFilters;
+ copy.logRecordBatchFilter = logRecordBatchFilter;
return copy;
}
@@ -224,10 +249,160 @@ public class ChangelogFlinkTableSource implements ScanTableSource {
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 =
+ (org.apache.flink.table.types.logical.RowType) producedDataType.getLogicalType();
+
+ // Derive the data-column indices actually scanned from Fluss (first-appearance order).
+ Map<Integer, Integer> dataColumnPositions = new LinkedHashMap<>();
+ for (int virtualIndex : this.projectedFields) {
+ if (virtualIndex >= METADATA_COLUMN_COUNT) {
+ 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 by FlinkSource's output projection.
+ this.dataProjection =
+ dataColumnPositions.isEmpty()
+ ? null
+ : dataColumnPositions.keySet().stream()
+ .mapToInt(Integer::intValue)
+ .toArray();
+ }
+
+ @Override
+ public Result applyFilters(List<ResolvedExpression> filters) {
+ List<ResolvedExpression> acceptedFilters = new ArrayList<>();
+ List<ResolvedExpression> remainingFilters = new ArrayList<>();
+
+ RowType dataFlussRowType = FlinkConversions.toFlussRowType(dataColumnsType);
+
+ if (isPartitioned()) {
+ // apply partition filter pushdown
+ List<Predicate> converted = new ArrayList<>();
+ RowType partitionRowType = dataFlussRowType.project(partitionKeyIndexes);
+ PredicateVisitor<Boolean> checksOnlyPartitionKeys =
+ new PartitionPredicateVisitor(partitionRowType.getFieldNames());
+
+ for (ResolvedExpression filter : filters) {
+ Optional<Predicate> 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<ResolvedExpression> 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<String> availableStatsColumns =
+ PushdownUtils.computeAvailableStatsColumns(dataFlussRowType, tableConfig);
+
+ List<Predicate> pushdownPredicates = new ArrayList<>();
+ List<ResolvedExpression> acceptedFilters = new ArrayList<>();
+ List<ResolvedExpression> remainingFilters = new ArrayList<>();
+
+ for (ResolvedExpression filter : filters) {
+ Optional<Predicate> predicateOpt = convertToFlussPredicate(dataFlussRowType, 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
+ org.apache.flink.table.types.logical.RowType 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/FlinkSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java
index b6cc0049c0..85ace29040 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java
@@ -23,6 +23,7 @@ import org.apache.fluss.flink.FlinkConnectorOptions;
import org.apache.fluss.flink.source.deserializer.DeserializerInitContextImpl;
import org.apache.fluss.flink.source.deserializer.FlussDeserializationSchema;
import org.apache.fluss.flink.source.emitter.FlinkRecordEmitter;
+import org.apache.fluss.flink.source.emitter.RowDataProjection;
import org.apache.fluss.flink.source.enumerator.FlinkSourceEnumerator;
import org.apache.fluss.flink.source.metrics.FlinkSourceReaderMetrics;
import org.apache.fluss.flink.source.reader.FlinkSourceReader;
@@ -32,6 +33,7 @@ import org.apache.fluss.flink.source.split.SourceSplitBase;
import org.apache.fluss.flink.source.split.SourceSplitSerializer;
import org.apache.fluss.flink.source.state.FlussSourceEnumeratorStateSerializer;
import org.apache.fluss.flink.source.state.SourceEnumeratorState;
+import org.apache.fluss.flink.utils.FlinkConversions;
import org.apache.fluss.lake.source.LakeSource;
import org.apache.fluss.lake.source.LakeSplit;
import org.apache.fluss.metadata.TablePath;
@@ -49,6 +51,7 @@ import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import javax.annotation.Nullable;
@@ -67,7 +70,9 @@ public class FlinkSource<OUT>
private final TablePath tablePath;
private final boolean hasPrimaryKey;
private final boolean isPartitioned;
- private final RowType sourceOutputType;
+ private final RowType scanRowType;
+ @Nullable private final RowType producedRowType;
+ @Nullable private final FlinkRecordEmitter.OutputProjection<OUT> outputProjection;
@Nullable private final int[] projectedFields;
protected final OffsetsInitializer offsetsInitializer;
protected final long scanPartitionDiscoveryIntervalMs;
@@ -85,7 +90,7 @@ public class FlinkSource<OUT>
TablePath tablePath,
boolean hasPrimaryKey,
boolean isPartitioned,
- RowType sourceOutputType,
+ RowType scanRowType,
@Nullable int[] projectedFields,
@Nullable Predicate logRecordBatchFilter,
OffsetsInitializer offsetsInitializer,
@@ -99,7 +104,7 @@ public class FlinkSource<OUT>
tablePath,
hasPrimaryKey,
isPartitioned,
- sourceOutputType,
+ scanRowType,
projectedFields,
logRecordBatchFilter,
offsetsInitializer,
@@ -117,7 +122,7 @@ public class FlinkSource<OUT>
TablePath tablePath,
boolean hasPrimaryKey,
boolean isPartitioned,
- RowType sourceOutputType,
+ RowType scanRowType,
@Nullable int[] projectedFields,
@Nullable Predicate logRecordBatchFilter,
OffsetsInitializer offsetsInitializer,
@@ -132,7 +137,7 @@ public class FlinkSource<OUT>
tablePath,
hasPrimaryKey,
isPartitioned,
- sourceOutputType,
+ scanRowType,
projectedFields,
logRecordBatchFilter,
offsetsInitializer,
@@ -150,7 +155,7 @@ public class FlinkSource<OUT>
TablePath tablePath,
boolean hasPrimaryKey,
boolean isPartitioned,
- RowType sourceOutputType,
+ RowType scanRowType,
@Nullable int[] projectedFields,
@Nullable Predicate logRecordBatchFilter,
OffsetsInitializer offsetsInitializer,
@@ -165,7 +170,7 @@ public class FlinkSource<OUT>
tablePath,
hasPrimaryKey,
isPartitioned,
- sourceOutputType,
+ scanRowType,
projectedFields,
logRecordBatchFilter,
offsetsInitializer,
@@ -183,7 +188,7 @@ public class FlinkSource<OUT>
TablePath tablePath,
boolean hasPrimaryKey,
boolean isPartitioned,
- RowType sourceOutputType,
+ RowType scanRowType,
@Nullable int[] projectedFields,
@Nullable Predicate logRecordBatchFilter,
OffsetsInitializer offsetsInitializer,
@@ -194,17 +199,90 @@ public class FlinkSource<OUT>
@Nullable Predicate partitionFilters,
@Nullable LakeSource<LakeSplit> lakeSource,
LeaseContext leaseContext) {
+ this(
+ flussConf,
+ tablePath,
+ hasPrimaryKey,
+ isPartitioned,
+ scanRowType,
+ projectedFields,
+ logRecordBatchFilter,
+ offsetsInitializer,
+ scanPartitionDiscoveryIntervalMs,
+ splitPerAssignmentBatchSize,
+ deserializationSchema,
+ null,
+ streaming,
+ partitionFilters,
+ lakeSource,
+ leaseContext);
+ }
+
+ public FlinkSource(
+ Configuration flussConf,
+ TablePath tablePath,
+ boolean hasPrimaryKey,
+ boolean isPartitioned,
+ RowType scanRowType,
+ @Nullable int[] projectedFields,
+ @Nullable Predicate logRecordBatchFilter,
+ OffsetsInitializer offsetsInitializer,
+ long scanPartitionDiscoveryIntervalMs,
+ int splitPerAssignmentBatchSize,
+ FlussDeserializationSchema<OUT> deserializationSchema,
+ @Nullable RowType producedRowType,
+ boolean streaming,
+ @Nullable Predicate partitionFilters,
+ LeaseContext leaseContext) {
+ this(
+ flussConf,
+ tablePath,
+ hasPrimaryKey,
+ isPartitioned,
+ scanRowType,
+ projectedFields,
+ logRecordBatchFilter,
+ offsetsInitializer,
+ scanPartitionDiscoveryIntervalMs,
+ splitPerAssignmentBatchSize,
+ deserializationSchema,
+ producedRowType,
+ streaming,
+ partitionFilters,
+ null,
+ leaseContext);
+ }
+
+ public FlinkSource(
+ Configuration flussConf,
+ TablePath tablePath,
+ boolean hasPrimaryKey,
+ boolean isPartitioned,
+ RowType scanRowType,
+ @Nullable int[] projectedFields,
+ @Nullable Predicate logRecordBatchFilter,
+ OffsetsInitializer offsetsInitializer,
+ long scanPartitionDiscoveryIntervalMs,
+ int splitPerAssignmentBatchSize,
+ FlussDeserializationSchema<OUT> deserializationSchema,
+ @Nullable RowType producedRowType,
+ boolean streaming,
+ @Nullable Predicate partitionFilters,
+ @Nullable LakeSource<LakeSplit> lakeSource,
+ LeaseContext leaseContext) {
this.flussConf = flussConf;
this.tablePath = tablePath;
this.hasPrimaryKey = hasPrimaryKey;
this.isPartitioned = isPartitioned;
- this.sourceOutputType = sourceOutputType;
+ this.scanRowType = scanRowType;
this.projectedFields = projectedFields;
this.logRecordBatchFilter = logRecordBatchFilter;
this.offsetsInitializer = offsetsInitializer;
this.scanPartitionDiscoveryIntervalMs = scanPartitionDiscoveryIntervalMs;
this.splitPerAssignmentBatchSize = splitPerAssignmentBatchSize;
this.deserializationSchema = deserializationSchema;
+ this.producedRowType = producedRowType;
+ this.outputProjection = createOutputProjection(producedRowType);
this.streaming = streaming;
this.partitionFilters = partitionFilters;
this.lakeSource = lakeSource;
@@ -296,14 +374,15 @@ public class FlinkSource<OUT>
new DeserializerInitContextImpl(
context.metricGroup().addGroup("deserializer"),
context.getUserCodeClassLoader(),
- sourceOutputType));
- FlinkRecordEmitter<OUT> recordEmitter = new FlinkRecordEmitter<>(deserializationSchema);
+ scanRowType));
+ FlinkRecordEmitter<OUT> recordEmitter =
+ new FlinkRecordEmitter<>(deserializationSchema, outputProjection);
return new FlinkSourceReader<>(
elementsQueue,
readerConf,
tablePath,
- sourceOutputType,
+ scanRowType,
context,
projectedFields,
logRecordBatchFilter,
@@ -314,6 +393,61 @@ public class FlinkSource<OUT>
@Override
public TypeInformation<OUT> getProducedType() {
- return deserializationSchema.getProducedType(sourceOutputType);
+ if (producedRowType != null) {
+ return deserializationSchema.getProducedType(scanRowType, producedRowType);
+ }
+
+ // if not specified produce type, inferred by scan row type.
+ return deserializationSchema.getProducedType(scanRowType);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Nullable
+ private FlinkRecordEmitter.OutputProjection<OUT> createOutputProjection(
+ @Nullable RowType producedRowType) {
+ if (producedRowType == null) {
+ return null;
+ }
+
+ TypeInformation<OUT> naturalProducedType =
+ deserializationSchema.getProducedType(scanRowType);
+ if (!(naturalProducedType instanceof InternalTypeInfo)) {
+ throw new IllegalArgumentException(
+ "Produced row type can only be used with RowData deserialization schemas.");
+ }
+ org.apache.flink.table.types.logical.RowType naturalProducedRowType =
+ ((InternalTypeInfo<?>) naturalProducedType).toRowType();
+ org.apache.flink.table.types.logical.RowType flinkProducedRowType =
+ FlinkConversions.toFlinkRowType(producedRowType);
+ int[] projection = createProjection(naturalProducedRowType, flinkProducedRowType);
+ return projection == null
+ ? null
+ : (FlinkRecordEmitter.OutputProjection<OUT>) RowDataProjection.of(projection);
+ }
+
+ @Nullable
+ private int[] createProjection(
+ org.apache.flink.table.types.logical.RowType naturalProducedRowType,
+ org.apache.flink.table.types.logical.RowType producedRowType) {
+ if (naturalProducedRowType.equals(producedRowType)) {
+ return null;
+ }
+
+ int[] projection = new int[producedRowType.getFieldCount()];
+ boolean identityProjection =
+ naturalProducedRowType.getFieldCount() == producedRowType.getFieldCount();
+ for (int i = 0; i < producedRowType.getFieldCount(); i++) {
+ String fieldName = producedRowType.getFieldNames().get(i);
+ int fieldIndex = naturalProducedRowType.getFieldIndex(fieldName);
+ if (fieldIndex < 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Produced field '%s' is missing in natural produced row type %s.",
+ fieldName, naturalProducedRowType));
+ }
+ projection[i] = fieldIndex;
+ identityProjection &= fieldIndex == i;
+ }
+ return identityProjection ? null : projection;
}
}
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 1a48d03c2b..d2d3c7eaee 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.initializer.OffsetsInitializer;
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.DeleteBehavior;
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;
@@ -263,7 +260,8 @@ public class 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
@@ -412,6 +410,8 @@ public class FlinkTableSource
scanPartitionDiscoveryIntervalMs,
splitPerAssignmentBatchSize,
new RowDataDeserializationSchema(),
+ FlinkConversions.toFlussRowType(
+ (org.apache.flink.table.types.logical.RowType) producedDataType),
streaming,
partitionFilters,
enableLakeSource ? lakeSource : null,
@@ -671,7 +671,8 @@ public class FlinkTableSource
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);
}
@@ -724,89 +725,6 @@ public class FlinkTableSource
}
}
- /**
- * 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<String> availableStatsColumns) {
-
- class StatisticsUsageVisitor implements PredicateVisitor<Boolean> {
- @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<String> computeAvailableStatsColumns(RowType flussRowType) {
- StatisticsColumnsConfig statsConfig = tableConfig.getStatisticsColumns();
-
- if (!statsConfig.isEnabled()) {
- LOG.debug("Statistics collection is disabled for the table");
- return Collections.emptySet();
- }
-
- Set<String> 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 1114febb9c..37ae0d98da 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
@@ -77,9 +77,12 @@ public class BinlogDeserializationSchema implements FlussDeserializationSchema<R
*/
@Override
public TypeInformation<RowData> getProducedType(RowType rowSchema) {
- // Build the output type with nested before/after ROW columns
- org.apache.flink.table.types.logical.RowType outputType =
- BinlogRowConverter.buildBinlogRowType(toFlinkRowType(rowSchema));
- return InternalTypeInfo.of(outputType);
+ return InternalTypeInfo.of(
+ BinlogRowConverter.buildBinlogRowType(toFlinkRowType(rowSchema)));
+ }
+
+ @Override
+ public TypeInformation<RowData> getProducedType(RowType scanRowType, RowType expectRowSchema) {
+ return InternalTypeInfo.of(toFlinkRowType(expectRowSchema));
}
}
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 c81141d345..f1868351e0 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
@@ -67,9 +67,12 @@ public class ChangelogDeserializationSchema implements FlussDeserializationSchem
*/
@Override
public TypeInformation<RowData> getProducedType(RowType rowSchema) {
- // Build the output type with metadata columns
- org.apache.flink.table.types.logical.RowType outputType =
- ChangelogRowConverter.buildChangelogRowType(toFlinkRowType(rowSchema));
- return InternalTypeInfo.of(outputType);
+ return InternalTypeInfo.of(
+ ChangelogRowConverter.buildChangelogRowType(toFlinkRowType(rowSchema)));
+ }
+
+ @Override
+ public TypeInformation<RowData> getProducedType(RowType scanRowType, RowType expectRowSchema) {
+ return InternalTypeInfo.of(toFlinkRowType(expectRowSchema));
}
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/FlussDeserializationSchema.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/FlussDeserializationSchema.java
index a3385e239b..699f743b2a 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/FlussDeserializationSchema.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/deserializer/FlussDeserializationSchema.java
@@ -18,11 +18,13 @@
package org.apache.fluss.flink.source.deserializer;
import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.flink.utils.FlinkConversions;
import org.apache.fluss.record.LogRecord;
import org.apache.fluss.types.RowType;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.util.UserCodeClassLoader;
import java.io.Serializable;
@@ -60,10 +62,23 @@ public interface FlussDeserializationSchema<T> extends Serializable {
/**
* Gets the data type (as a {@link TypeInformation}) produced by this deserializer.
*
- * @param rowSchema The schema of the {@link LogRecord#getRow()}.
+ * @param scanRowType The schema of the {@link LogRecord#getRow()}.
* @return The data type produced by this deserializer.
*/
- TypeInformation<T> getProducedType(RowType rowSchema);
+ TypeInformation<T> getProducedType(RowType scanRowType);
+
+ /**
+ * Gets the data type produced by this deserializer when the final produced row type is known.
+ *
+ * @param scanRowType The schema of the {@link LogRecord#getRow()}.
+ * @param producedRowType The expected row type specified by the source.
+ * @return The data type produced by this deserializer.
+ */
+ @SuppressWarnings("unchecked")
+ default TypeInformation<T> getProducedType(RowType scanRowType, RowType producedRowType) {
+ return (TypeInformation<T>)
+ InternalTypeInfo.of(FlinkConversions.toFlinkRowType(producedRowType));
+ }
/**
* A contextual information provided for {@link #open(InitializationContext)} method. It can be
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitter.java
index ba4c9d0131..b659554949 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitter.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitter.java
@@ -30,6 +30,10 @@ import org.apache.flink.connector.base.source.reader.RecordEmitter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+
/**
* The {@link RecordEmitter} implementation for {@link FlinkSourceReader}.
*
@@ -44,10 +48,18 @@ public class FlinkRecordEmitter<OUT> implements RecordEmitter<RecordAndPos, OUT,
private static final Logger LOG = LoggerFactory.getLogger(FlinkRecordEmitter.class);
private final FlussDeserializationSchema<OUT> deserializationSchema;
+ @Nullable private final OutputProjection<OUT> outputProjection;
private LakeRecordRecordEmitter<OUT> lakeRecordRecordEmitter;
public FlinkRecordEmitter(FlussDeserializationSchema<OUT> deserializationSchema) {
+ this(deserializationSchema, null);
+ }
+
+ public FlinkRecordEmitter(
+ FlussDeserializationSchema<OUT> deserializationSchema,
+ @Nullable OutputProjection<OUT> outputProjection) {
this.deserializationSchema = deserializationSchema;
+ this.outputProjection = outputProjection;
}
@Override
@@ -113,6 +125,10 @@ public class FlinkRecordEmitter<OUT> implements RecordEmitter<RecordAndPos, OUT,
e);
}
+ if (record != null && outputProjection != null) {
+ record = outputProjection.project(record);
+ }
+
if (record != null) {
long timestamp = scanRecord.timestamp();
if (timestamp > 0) {
@@ -124,4 +140,17 @@ public class FlinkRecordEmitter<OUT> implements RecordEmitter<RecordAndPos, OUT,
}
return false;
}
+
+ /** Projection applied after deserialization and before the record is emitted. */
+ public interface OutputProjection<OUT> extends Serializable {
+
+ /**
+ * Projects a deserialized record.
+ *
+ * @param record the deserialized record
+ * @return the projected record, or {@code null} to skip emission
+ */
+ @Nullable
+ OUT project(OUT record);
+ }
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/emitter/RowDataProjection.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/emitter/RowDataProjection.java
new file mode 100644
index 0000000000..f753353c4a
--- /dev/null
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/emitter/RowDataProjection.java
@@ -0,0 +1,45 @@
+/*
+ * 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.emitter;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.utils.ProjectedRowData;
+
+import javax.annotation.Nullable;
+
+/** Output projection for {@link RowData} records emitted by {@link FlinkRecordEmitter}. */
+public final class RowDataProjection implements FlinkRecordEmitter.OutputProjection<RowData> {
+ private static final long serialVersionUID = 1L;
+
+ private final int[] projectedFields;
+
+ private RowDataProjection(int[] projectedFields) {
+ this.projectedFields = projectedFields;
+ }
+
+ /** Creates a projection, or returns {@code null} when no projection is requested. */
+ @Nullable
+ public static RowDataProjection of(@Nullable int[] projectedFields) {
+ return projectedFields == null ? null : new RowDataProjection(projectedFields);
+ }
+
+ @Override
+ public RowData project(RowData record) {
+ return ProjectedRowData.from(projectedFields).replaceRow(record);
+ }
+}
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 0c6be61d82..1dde6bf3da 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.Table;
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.PartitionSpec;
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;
@@ -57,6 +64,8 @@ import org.apache.flink.table.functions.LookupFunction;
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;
@@ -67,8 +76,10 @@ import java.time.LocalDateTime;
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;
@@ -79,6 +90,8 @@ import static org.apache.fluss.utils.ExceptionUtils.findThrowable;
/** 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. */
@@ -450,6 +463,86 @@ public class PushdownUtils {
// ------------------------------------------------------------------------------------------
+ /**
+ * 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<String> 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();
+ }
+
+ Set<String> 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) {
+ 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);
+ }
+ }
+ }
+ 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<String> availableStatsColumns) {
+ class StatisticsUsageVisitor implements PredicateVisitor<Boolean> {
+ @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 0000000000..716de2ad9c
--- /dev/null
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogFlinkTableSourceTest.java
@@ -0,0 +1,129 @@
+/*
+ * 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 testApplyProjectionStoresProducedType() {
+ BinlogFlinkTableSource source = createSource();
+ // virtual [_change_type(0), after(4)]
+ source.applyProjection(nested(0, 4), projectedType(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.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);
+ }
+}
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 4787fe162d..fd5f08ba4e 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,73 @@ abstract class BinlogVirtualTableITCase {
.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<Row> rowIter = tEnv.executeSql(query).collect();
+
+ CLOCK.advanceTime(Duration.ofMillis(1000));
+ writeRows(conn, tablePath, Arrays.asList(row(1, "Alice", 100L)), false);
+
+ List<String> 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 testBinlogReorderedTopLevelProjection() throws Exception {
+ // Reordered top-level projection: after before _change_type, i.e. not in the declared
+ // order of the binlog row. This exercises FlinkSource output projection 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<Row> rowIter = tEnv.executeSql(query).collect();
+
+ CLOCK.advanceTime(Duration.ofMillis(100));
+ writeRows(conn, tablePath, Arrays.asList(row(1, "Alice", 100L)), false);
+ List<String> 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<String> 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/ChangelogFlinkTableSourceTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java
new file mode 100644
index 0000000000..0d20f49c6a
--- /dev/null
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogFlinkTableSourceTest.java
@@ -0,0 +1,296 @@
+/*
+ * 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 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;
+
+/** 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<String, String> extraOptions) {
+ FlinkConnectorOptionsUtils.StartupOptions startupOptions =
+ new FlinkConnectorOptionsUtils.StartupOptions();
+ startupOptions.startupMode = ScanStartupMode.EARLIEST;
+ Map<String, String> 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) {
+ 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(func, Arrays.asList(ref, lit), DataTypes.BOOLEAN());
+ }
+
+ @ParameterizedTest
+ @MethodSource("projectionCases")
+ void testApplyProjection(String description, int[] virtual, int[] expectedDataProjection) {
+ ChangelogFlinkTableSource source = createSource(new int[0]);
+ source.applyProjection(nested(virtual), projectedType(virtual));
+
+ assertThat(source.getProjectedFields()).containsExactly(virtual);
+
+ // dataProjection null for metadata-only projection
+ if (expectedDataProjection == null) {
+ assertThat(source.getDataProjection()).isNull();
+ } else {
+ assertThat(source.getDataProjection()).containsExactly(expectedDataProjection);
+ }
+
+ assertThat(source.getProducedDataType()).isEqualTo(projectedType(virtual).getLogicalType());
+ }
+
+ private static Stream<Arguments> projectionCases() {
+ return Stream.of(
+ // description, virtual, expectedDataProjection
+ // metadata only
+ Arguments.of("metadata-only", new int[] {0}, null),
+ // data only
+ Arguments.of("data-only", new int[] {3, 5}, new int[] {0, 2}),
+ // reordered mix
+ Arguments.of("reorderedMix", new int[] {0, 5, 3}, new int[] {2, 0}));
+ }
+
+ @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.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 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());
+ 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<String, String> statsAll() {
+ return statsColumns("*");
+ }
+
+ private Map<String, String> statsColumns(String columns) {
+ Map<String, String> options = new HashMap<>();
+ options.put("table.statistics.columns", 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 bf99111f91..93dc95f019 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,142 @@ abstract class ChangelogVirtualTableITCase {
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 data projection plus FlinkSource output projection 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 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<Row> rowIter = tEnv.executeSql(query).collect();
+
+ CLOCK.advanceTime(Duration.ofMillis(100));
+ writeRows(conn, tablePath, Arrays.asList(row(1, "Item-1", 100L)), false);
+ List<String> 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<String> 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.
+ // 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<Row> 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<String> 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<Row> rowIter = tEnv.executeSql(query).collect();
+ List<String> 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<Row> 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<String> 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
@@ -539,10 +675,11 @@ abstract class ChangelogVirtualTableITCase {
String query = "SELECT _change_type, id, name, region FROM partitioned_test$changelog";
CloseableIterator<Row> 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<String> 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]");
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 55f83a5149..2b0568b7aa 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,7 +32,11 @@ import org.apache.flink.table.data.StringData;
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 static org.assertj.core.api.Assertions.assertThat;
@@ -55,17 +59,19 @@ class ChangelogRowConverterTest {
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 +84,13 @@ class ChangelogRowConverterTest {
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<Arguments> 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, "insert"));
}
private LogRecord createLogRecord(