You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2026/04/12 09:40:21 UTC
(fluss) branch main updated: [flink] Add Flink filter pushdown integration and documentation (#3032)
This is an automated email from the ASF dual-hosted git repository.
jark 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 0fccd3c3f [flink] Add Flink filter pushdown integration and documentation (#3032)
0fccd3c3f is described below
commit 0fccd3c3f6632e845cbc7a49f2d1d07166e94f41
Author: Yang Wang <oc...@alibaba-inc.com>
AuthorDate: Sun Apr 12 17:40:18 2026 +0800
[flink] Add Flink filter pushdown integration and documentation (#3032)
---
.../fluss/client/table/scanner/TableScan.java | 10 +
.../org/apache/fluss/config/FlussConfigUtils.java | 3 +-
.../apache/fluss/config/StatisticsConfigUtils.java | 9 +
.../fluss/flink/catalog/FlinkTableFactory.java | 21 +
.../fluss/flink/source/BinlogFlinkTableSource.java | 1 +
.../flink/source/ChangelogFlinkTableSource.java | 1 +
.../org/apache/fluss/flink/source/FlinkSource.java | 7 +
.../fluss/flink/source/FlinkTableSource.java | 199 +++-
.../org/apache/fluss/flink/source/FlussSource.java | 3 +
.../fluss/flink/source/FlussSourceBuilder.java | 18 +
.../flink/source/reader/FlinkSourceReader.java | 7 +-
.../source/reader/FlinkSourceSplitReader.java | 13 +-
.../fluss/flink/utils/PredicateConverter.java | 6 +-
.../source/FlinkTableSourceFilterPushDownTest.java | 1062 ++++++++++++++++++++
.../fluss/flink/source/FlinkTableSourceITCase.java | 341 ++++++-
.../flink/source/reader/FlinkSourceReaderTest.java | 1 +
.../source/reader/FlinkSourceSplitReaderTest.java | 17 +-
.../server/coordinator/CoordinatorService.java | 4 -
.../server/utils/TableDescriptorValidation.java | 2 +
website/docs/engine-flink/options.md | 1 +
website/docs/engine-flink/reads.md | 78 ++
21 files changed, 1767 insertions(+), 37 deletions(-)
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java
index ef1128e1a..6502b7946 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java
@@ -30,6 +30,7 @@ import org.apache.fluss.client.table.scanner.log.TypedLogScanner;
import org.apache.fluss.client.table.scanner.log.TypedLogScannerImpl;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.metadata.LogFormat;
import org.apache.fluss.metadata.PartitionInfo;
import org.apache.fluss.metadata.SchemaGetter;
import org.apache.fluss.metadata.TableBucket;
@@ -124,6 +125,15 @@ public class TableScan implements Scan {
tableInfo.getTablePath(), limit));
}
+ if (recordBatchFilter != null
+ && tableInfo.getTableConfig().getLogFormat() != LogFormat.ARROW) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Filter pushdown is only supported for ARROW log format. "
+ + "Table: %s, current log format: %s",
+ tableInfo.getTablePath(), tableInfo.getTableConfig().getLogFormat()));
+ }
+
return new LogScannerImpl(
conn.getConfiguration(),
tableInfo,
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java
index bf9b1428e..adbbce0af 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java
@@ -49,7 +49,8 @@ public class FlussConfigUtils {
ConfigOptions.TABLE_DATALAKE_ENABLED.key(),
ConfigOptions.TABLE_DATALAKE_FRESHNESS.key(),
ConfigOptions.TABLE_TIERED_LOG_LOCAL_SEGMENTS.key(),
- ConfigOptions.TABLE_AUTO_PARTITION_NUM_RETENTION.key());
+ ConfigOptions.TABLE_AUTO_PARTITION_NUM_RETENTION.key(),
+ ConfigOptions.TABLE_STATISTICS_COLUMNS.key());
}
public static boolean isTableStorageConfig(String key) {
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/StatisticsConfigUtils.java b/fluss-common/src/main/java/org/apache/fluss/config/StatisticsConfigUtils.java
index 58ebf0e8c..c4235c533 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/StatisticsConfigUtils.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/StatisticsConfigUtils.java
@@ -54,6 +54,15 @@ public class StatisticsConfigUtils {
return;
}
+ // Statistics columns are only supported for log tables (non-PK tables)
+ if (tableDescriptor.hasPrimaryKey()) {
+ throw new InvalidConfigException(
+ "Statistics columns are not supported for primary key tables. "
+ + "Please remove the '"
+ + ConfigOptions.TABLE_STATISTICS_COLUMNS.key()
+ + "' property.");
+ }
+
RowType rowType = tableDescriptor.getSchema().getRowType();
// Wildcard means all supported columns - no validation needed
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java
index 77f7909f2..408a70305 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java
@@ -19,6 +19,7 @@ package org.apache.fluss.flink.catalog;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.TableConfig;
import org.apache.fluss.flink.FlinkConnectorOptions;
import org.apache.fluss.flink.lake.LakeFlinkCatalog;
import org.apache.fluss.flink.lake.LakeTableFactory;
@@ -62,6 +63,7 @@ import java.util.Set;
import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_FORMAT;
import static org.apache.fluss.config.ConfigOptions.TABLE_DELETE_BEHAVIOR;
import static org.apache.fluss.config.FlussConfigUtils.CLIENT_PREFIX;
+import static org.apache.fluss.config.FlussConfigUtils.TABLE_PREFIX;
import static org.apache.fluss.flink.catalog.FlinkCatalog.LAKE_TABLE_SPLITTER;
import static org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils.getBucketKeyIndexes;
import static org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils.getBucketKeys;
@@ -150,6 +152,7 @@ public class FlinkTableFactory implements DynamicTableSourceFactory, DynamicTabl
toFlussTablePath(context.getObjectIdentifier()),
toFlussClientConfig(
context.getCatalogTable().getOptions(), context.getConfiguration()),
+ toFlussTableConfig(tableOptions),
tableOutputType,
primaryKeyIndexes,
bucketKeyIndexes,
@@ -273,6 +276,24 @@ public class FlinkTableFactory implements DynamicTableSourceFactory, DynamicTabl
return flussConfig;
}
+ private static TableConfig toFlussTableConfig(ReadableConfig tableOptions) {
+ Configuration tableConfig = new Configuration();
+
+ // forward all table-level configs by iterating through known table options
+ // this approach is safer than using toMap() which may not exist in all Flink versions
+ for (ConfigOption<?> option : FlinkConnectorOptions.TABLE_OPTIONS) {
+ if (option.key().startsWith(TABLE_PREFIX)) {
+ Object value = tableOptions.getOptional(option).orElse(null);
+ if (value != null) {
+ // convert value to string for configuration storage
+ tableConfig.setString(option.key(), value.toString());
+ }
+ }
+ }
+
+ return new TableConfig(tableConfig);
+ }
+
private static TablePath toFlussTablePath(ObjectIdentifier tablePath) {
return TablePath.of(tablePath.getDatabaseName(), tablePath.getObjectName());
}
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 65083c3be..901995a65 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
@@ -126,6 +126,7 @@ public class BinlogFlinkTableSource implements ScanTableSource {
isPartitioned,
flussRowType,
projectedFields,
+ null,
offsetsInitializer,
scanPartitionDiscoveryIntervalMs,
new BinlogDeserializationSchema(),
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 0b6b4979b..f58fac3df 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
@@ -163,6 +163,7 @@ public class ChangelogFlinkTableSource implements ScanTableSource {
isPartitioned(),
flussRowType,
projectedFields,
+ null,
offsetsInitializer,
scanPartitionDiscoveryIntervalMs,
new ChangelogDeserializationSchema(),
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 0aeb0d59c..efa426e5f 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
@@ -73,6 +73,8 @@ public class FlinkSource<OUT>
@Nullable private final LakeSource<LakeSplit> lakeSource;
private final LeaseContext leaseContext;
+ @Nullable private final Predicate logRecordBatchFilter;
+
public FlinkSource(
Configuration flussConf,
TablePath tablePath,
@@ -80,6 +82,7 @@ public class FlinkSource<OUT>
boolean isPartitioned,
RowType sourceOutputType,
@Nullable int[] projectedFields,
+ @Nullable Predicate logRecordBatchFilter,
OffsetsInitializer offsetsInitializer,
long scanPartitionDiscoveryIntervalMs,
FlussDeserializationSchema<OUT> deserializationSchema,
@@ -93,6 +96,7 @@ public class FlinkSource<OUT>
isPartitioned,
sourceOutputType,
projectedFields,
+ logRecordBatchFilter,
offsetsInitializer,
scanPartitionDiscoveryIntervalMs,
deserializationSchema,
@@ -109,6 +113,7 @@ public class FlinkSource<OUT>
boolean isPartitioned,
RowType sourceOutputType,
@Nullable int[] projectedFields,
+ @Nullable Predicate logRecordBatchFilter,
OffsetsInitializer offsetsInitializer,
long scanPartitionDiscoveryIntervalMs,
FlussDeserializationSchema<OUT> deserializationSchema,
@@ -122,6 +127,7 @@ public class FlinkSource<OUT>
this.isPartitioned = isPartitioned;
this.sourceOutputType = sourceOutputType;
this.projectedFields = projectedFields;
+ this.logRecordBatchFilter = logRecordBatchFilter;
this.offsetsInitializer = offsetsInitializer;
this.scanPartitionDiscoveryIntervalMs = scanPartitionDiscoveryIntervalMs;
this.deserializationSchema = deserializationSchema;
@@ -213,6 +219,7 @@ public class FlinkSource<OUT>
sourceOutputType,
context,
projectedFields,
+ logRecordBatchFilter,
flinkSourceReaderMetrics,
recordEmitter,
lakeSource);
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 83b4dd29e..01cca4ee7 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
@@ -20,6 +20,8 @@ package org.apache.fluss.flink.source;
import org.apache.fluss.client.initializer.OffsetsInitializer;
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.source.deserializer.RowDataDeserializationSchema;
import org.apache.fluss.flink.source.lookup.FlinkAsyncLookupFunction;
@@ -28,6 +30,7 @@ import org.apache.fluss.flink.source.lookup.LookupNormalizer;
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.PredicateConverter;
import org.apache.fluss.flink.utils.PushdownUtils;
import org.apache.fluss.flink.utils.PushdownUtils.FieldEqual;
import org.apache.fluss.lake.source.LakeSource;
@@ -36,10 +39,12 @@ import org.apache.fluss.metadata.ChangelogImage;
import org.apache.fluss.metadata.DeleteBehavior;
import org.apache.fluss.metadata.MergeEngineType;
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;
@@ -89,6 +94,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import static org.apache.fluss.flink.utils.LakeSourceUtils.createLakeSource;
import static org.apache.fluss.flink.utils.PredicateConverter.convertToFlussPredicate;
@@ -133,6 +139,12 @@ public class FlinkTableSource
@Nullable private final MergeEngineType mergeEngineType;
+ // table-level configuration
+ private final TableConfig tableConfig;
+
+ // pre-computed available statistics columns
+ private final Set<String> availableStatsColumns;
+
// output type after projection pushdown
private LogicalType producedDataType;
@@ -154,10 +166,12 @@ public class FlinkTableSource
private final Map<String, String> tableOptions;
@Nullable private LakeSource<LakeSplit> lakeSource;
+ @Nullable private Predicate logRecordBatchFilter;
public FlinkTableSource(
TablePath tablePath,
Configuration flussConfig,
+ TableConfig tableConfig,
org.apache.flink.table.types.logical.RowType tableOutputType,
int[] primaryKeyIndexes,
int[] bucketKeyIndexes,
@@ -197,6 +211,11 @@ public class FlinkTableSource
createLakeSource(tablePath, tableOptions),
"LakeSource must not be null if enable datalake");
}
+ this.tableConfig = checkNotNull(tableConfig, "tableConfig must not be null");
+
+ // Pre-compute available statistics columns to avoid repeated calculation
+ RowType flussRowType = FlinkConversions.toFlussRowType(tableOutputType);
+ this.availableStatsColumns = computeAvailableStatsColumns(flussRowType);
}
@Override
@@ -340,6 +359,7 @@ public class FlinkTableSource
isPartitioned(),
flussRowType,
projectedFields,
+ logRecordBatchFilter,
offsetsInitializer,
scanPartitionDiscoveryIntervalMs,
new RowDataDeserializationSchema(),
@@ -424,6 +444,7 @@ public class FlinkTableSource
new FlinkTableSource(
tablePath,
flussConfig,
+ tableConfig,
tableOutputType,
primaryKeyIndexes,
bucketKeyIndexes,
@@ -444,6 +465,8 @@ public class FlinkTableSource
source.modificationScanType = modificationScanType;
source.partitionFilters = partitionFilters;
source.lakeSource = lakeSource;
+ source.logRecordBatchFilter = logRecordBatchFilter;
+ // Note: availableStatsColumns is already computed in the constructor
return source;
}
@@ -497,16 +520,12 @@ public class FlinkTableSource
lookupRow.setField(keyRowProjection[fieldEqual.fieldIndex], fieldEqual.equalValue);
visitedPkFields.add(fieldEqual.fieldIndex);
}
- // if not all primary key fields are in condition, meaning can not push down single row
- // filter, determine whether to push down partition filter later
+ // if not all primary key fields are in condition, fall through to
+ // try partition filter pushdown for partitioned PK tables
if (visitedPkFields.equals(primaryKeyTypes.keySet())) {
singleRowFilter = lookupRow;
- // FLINK-38635 We cannot determine whether this source will ultimately be used as a
- // scan
- // source or a lookup source. Since fluss lookup sources cannot accept filters yet,
- // to
- // be safe, we return all filters to the Flink planner.
+ // FLINK-38635: return all filters as remaining for scan vs lookup safety net
return Result.of(acceptedFilters, filters);
}
}
@@ -542,6 +561,7 @@ public class FlinkTableSource
}
}
partitionFilters = converted.isEmpty() ? null : PredicateBuilder.and(converted);
+
// lake source is not null
if (lakeSource != null) {
List<Predicate> lakePredicates = new ArrayList<>();
@@ -562,14 +582,146 @@ public class FlinkTableSource
}
}
}
+ }
+
+ if (acceptedFilters.isEmpty() && remainingFilters.isEmpty()) {
+ remainingFilters.addAll(filters);
+ }
+
+ if (!hasPrimaryKey()) {
+ Result recordBatchResult = pushdownRecordBatchFilter(remainingFilters);
+ acceptedFilters.addAll(recordBatchResult.getAcceptedFilters());
+ }
+ // FLINK-38635 We cannot determine whether this source will ultimately be used as a
+ // scan source or a lookup source. If used as a lookup source, the accepted filters
+ // (partition filters, record batch filters) are not enforced in the lookup path.
+ // 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) {
+ // Use pre-computed available statistics columns
+ LOG.trace("Statistics available columns: {}", availableStatsColumns);
+
+ // Convert to fluss row type for predicate operations
+ RowType flussRowType = FlinkConversions.toFlussRowType(tableOutputType);
+
+ List<Predicate> pushdownPredicates = new ArrayList<>();
+ List<ResolvedExpression> acceptedFilters = new ArrayList<>();
+ List<ResolvedExpression> remainingFilters = new ArrayList<>();
+
+ for (ResolvedExpression filter : filters) {
+ Optional<Predicate> predicateOpt =
+ PredicateConverter.convertToFlussPredicate(tableOutputType, filter);
+
+ if (predicateOpt.isPresent()) {
+ Predicate predicate = predicateOpt.get();
+ LOG.trace("Converted filter to predicate: {}", predicate);
+ // Check if predicate can benefit from statistics
+ if (canPredicateUseStatistics(predicate, flussRowType, availableStatsColumns)) {
+ pushdownPredicates.add(predicate);
+ 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);
+ }
- // FLINK-38635 We cannot determine whether this source will ultimately be used as a scan
- // source or a lookup source. Since fluss lookup sources cannot accept filters yet, to
- // be safe, we return all filters to the Flink planner.
- return Result.of(acceptedFilters, filters);
+ if (!pushdownPredicates.isEmpty()) {
+ Predicate merged =
+ pushdownPredicates.size() == 1
+ ? pushdownPredicates.get(0)
+ : PredicateBuilder.and(pushdownPredicates);
+ LOG.info("Accept merged predicate for record batch filter: {}", merged);
+ this.logRecordBatchFilter = merged;
+ } else {
+ this.logRecordBatchFilter = null;
+ }
+ return Result.of(acceptedFilters, remainingFilters);
+ }
+
+ /**
+ * 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 Result.of(Collections.emptyList(), filters);
+ 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
@@ -652,4 +804,27 @@ public class FlinkTableSource
public int[] getBucketKeyIndexes() {
return bucketKeyIndexes;
}
+
+ @VisibleForTesting
+ public int[] getPartitionKeyIndexes() {
+ return partitionKeyIndexes;
+ }
+
+ @VisibleForTesting
+ @Nullable
+ public Predicate getLogRecordBatchFilter() {
+ return logRecordBatchFilter;
+ }
+
+ @VisibleForTesting
+ @Nullable
+ public GenericRowData getSingleRowFilter() {
+ return singleRowFilter;
+ }
+
+ @VisibleForTesting
+ @Nullable
+ public Predicate getPartitionFilters() {
+ return partitionFilters;
+ }
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSource.java
index d00a74a26..53cabcca7 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSource.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSource.java
@@ -23,6 +23,7 @@ import org.apache.fluss.config.Configuration;
import org.apache.fluss.flink.source.deserializer.FlussDeserializationSchema;
import org.apache.fluss.flink.source.reader.LeaseContext;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.predicate.Predicate;
import org.apache.fluss.types.RowType;
import javax.annotation.Nullable;
@@ -65,6 +66,7 @@ public class FlussSource<OUT> extends FlinkSource<OUT> {
boolean isPartitioned,
RowType sourceOutputType,
@Nullable int[] projectedFields,
+ @Nullable Predicate logRecordBatchFilter,
OffsetsInitializer offsetsInitializer,
long scanPartitionDiscoveryIntervalMs,
FlussDeserializationSchema<OUT> deserializationSchema,
@@ -77,6 +79,7 @@ public class FlussSource<OUT> extends FlinkSource<OUT> {
isPartitioned,
sourceOutputType,
projectedFields,
+ logRecordBatchFilter,
offsetsInitializer,
scanPartitionDiscoveryIntervalMs,
deserializationSchema,
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java
index afd955c01..87efc01cd 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java
@@ -27,6 +27,7 @@ import org.apache.fluss.flink.FlinkConnectorOptions;
import org.apache.fluss.flink.source.deserializer.FlussDeserializationSchema;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.predicate.Predicate;
import org.apache.fluss.types.RowType;
import org.slf4j.Logger;
@@ -68,6 +69,7 @@ public class FlussSourceBuilder<OUT> {
private int[] projectedFields;
private String[] projectedFieldNames;
+ private Predicate logRecordBatchFilter;
private Long scanPartitionDiscoveryIntervalMs;
private OffsetsInitializer offsetsInitializer;
private FlussDeserializationSchema<OUT> deserializationSchema;
@@ -174,6 +176,21 @@ public class FlussSourceBuilder<OUT> {
return this;
}
+ /**
+ * Sets the filter predicate for server-side record batch filtering based on column statistics.
+ *
+ * <p>The predicate is evaluated against per-batch column statistics (min/max values) to skip
+ * entire record batches that cannot contain matching rows.
+ *
+ * @param filter the predicate to filter record batches
+ * @return this builder
+ */
+ public FlussSourceBuilder<OUT> setFilter(Predicate filter) {
+ checkNotNull(filter, "filter must not be null");
+ this.logRecordBatchFilter = filter;
+ return this;
+ }
+
/**
* Sets custom Fluss configuration properties for the source connector.
*
@@ -297,6 +314,7 @@ public class FlussSourceBuilder<OUT> {
isPartitioned,
sourceOutputType,
projectedFields,
+ logRecordBatchFilter,
offsetsInitializer,
scanPartitionDiscoveryIntervalMs,
deserializationSchema,
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceReader.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceReader.java
index 9a6c4e8da..68a3489bd 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceReader.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceReader.java
@@ -35,6 +35,7 @@ import org.apache.fluss.lake.source.LakeSource;
import org.apache.fluss.lake.source.LakeSplit;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.predicate.Predicate;
import org.apache.fluss.types.RowType;
import org.apache.flink.api.connector.source.SourceEvent;
@@ -69,6 +70,7 @@ public class FlinkSourceReader<OUT>
RowType sourceOutputType,
SourceReaderContext context,
@Nullable int[] projectedFields,
+ @Nullable Predicate logRecordBatchFilter,
FlinkSourceReaderMetrics flinkSourceReaderMetrics,
FlinkRecordEmitter<OUT> recordEmitter,
LakeSource<LakeSplit> lakeSource) {
@@ -82,8 +84,9 @@ public class FlinkSourceReader<OUT>
tablePath,
sourceOutputType,
projectedFields,
- flinkSourceReaderMetrics,
- lakeSource),
+ logRecordBatchFilter,
+ lakeSource,
+ flinkSourceReaderMetrics),
(ignore) -> {}),
recordEmitter,
context.getConfiguration(),
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java
index bd5305a36..90f2b6e9a 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java
@@ -39,6 +39,7 @@ import org.apache.fluss.lake.source.LakeSource;
import org.apache.fluss.lake.source.LakeSplit;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.predicate.Predicate;
import org.apache.fluss.types.RowType;
import org.apache.fluss.utils.CloseableIterator;
import org.apache.fluss.utils.ExceptionUtils;
@@ -91,6 +92,7 @@ public class FlinkSourceSplitReader implements SplitReader<RecordAndPos, SourceS
private final Map<TableBucket, String> subscribedBuckets;
@Nullable private final int[] projectedFields;
+
private final FlinkSourceReaderMetrics flinkSourceReaderMetrics;
@Nullable private BoundedSplitReader currentBoundedSplitReader;
@@ -120,8 +122,9 @@ public class FlinkSourceSplitReader implements SplitReader<RecordAndPos, SourceS
TablePath tablePath,
RowType sourceOutputType,
@Nullable int[] projectedFields,
- FlinkSourceReaderMetrics flinkSourceReaderMetrics,
- @Nullable LakeSource<LakeSplit> lakeSource) {
+ @Nullable Predicate logRecordBatchFilter,
+ @Nullable LakeSource<LakeSplit> lakeSource,
+ FlinkSourceReaderMetrics flinkSourceReaderMetrics) {
this.flinkMetricRegistry =
new FlinkMetricRegistry(flinkSourceReaderMetrics.getSourceReaderMetricGroup());
this.connection = ConnectionFactory.createConnection(flussConf, flinkMetricRegistry);
@@ -134,7 +137,11 @@ public class FlinkSourceSplitReader implements SplitReader<RecordAndPos, SourceS
this.flinkSourceReaderMetrics = flinkSourceReaderMetrics;
sanityCheck(table.getTableInfo().getRowType(), projectedFields);
- this.logScanner = table.newScan().project(projectedFields).createLogScanner();
+ this.logScanner =
+ table.newScan()
+ .project(projectedFields)
+ .filter(logRecordBatchFilter)
+ .createLogScanner();
this.stoppingOffsets = new HashMap<>();
this.emptyLogSplits = new HashSet<>();
this.lakeSource = lakeSource;
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PredicateConverter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PredicateConverter.java
index 861381c14..7c691cea9 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PredicateConverter.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/PredicateConverter.java
@@ -248,12 +248,12 @@ public class PredicateConverter implements ExpressionVisitor<Predicate> {
throw new UnsupportedExpression();
}
- private static Object fromFlinkObject(Object o, DataType type) {
- if (o == null) {
+ private static Object fromFlinkObject(Object flinkValue, DataType type) {
+ if (flinkValue == null) {
return null;
}
return InternalRow.createFieldGetter(FlinkConversions.toFlussType(type), 0)
- .getFieldOrNull((new FlinkAsFlussRow()).replace(GenericRowData.of(o)));
+ .getFieldOrNull((new FlinkAsFlussRow()).replace(GenericRowData.of(flinkValue)));
}
private boolean supportsPredicate(LogicalType type) {
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceFilterPushDownTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceFilterPushDownTest.java
new file mode 100644
index 000000000..f8d972154
--- /dev/null
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceFilterPushDownTest.java
@@ -0,0 +1,1062 @@
+/*
+ * 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.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.TableConfig;
+import org.apache.fluss.flink.FlinkConnectorOptions;
+import org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.predicate.Predicate;
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.GenericRow;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.Maps;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericRowData;
+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.logical.RowType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Comprehensive unit tests for filter pushdown functionality in FlinkTableSource. Tests cover
+ * various scenarios including streaming/batch modes, different table types, and different filter
+ * pushdown types (primary key, partition key, record batch filter).
+ */
+public class FlinkTableSourceFilterPushDownTest {
+
+ @Nested
+ class NonPartitionedKvTableTests {
+ private FlinkTableSource tableSource;
+
+ @BeforeEach
+ void setUp() {
+ // Create a non-partitioned KV table schema: id (PK), name, value
+ RowType tableOutputType =
+ (RowType)
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.INT()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("value", DataTypes.BIGINT()))
+ .getLogicalType();
+
+ TablePath tablePath = TablePath.of("test_db", "test_kv_table");
+ Configuration flussConfig = new Configuration();
+ flussConfig.setString(FlinkConnectorOptions.BOOTSTRAP_SERVERS.key(), "localhost:9092");
+
+ FlinkConnectorOptionsUtils.StartupOptions startupOptions =
+ new FlinkConnectorOptionsUtils.StartupOptions();
+ startupOptions.startupMode = FlinkConnectorOptions.ScanStartupMode.EARLIEST;
+
+ // Create table config for testing
+ Configuration tableConfig = new Configuration();
+
+ tableConfig.set(ConfigOptions.TABLE_STATISTICS_COLUMNS, "*");
+
+ tableSource =
+ new FlinkTableSource(
+ tablePath,
+ flussConfig,
+ new TableConfig(tableConfig),
+ tableOutputType,
+ new int[] {0}, // primary key indexes (id)
+ new int[] {}, // bucket key indexes
+ new int[] {}, // partition key indexes
+ true, // streaming
+ startupOptions,
+ false, // lookup async
+ false, // insert if not exists
+ null, // cache
+ 1000L, // scan partition discovery interval
+ false, // is data lake enabled
+ null, // merge engine type
+ Maps.newHashMap(),
+ null); // lease context
+ }
+
+ @Test
+ void testStreamingModeNonPrimaryKeyPushdown() {
+ // Test non-primary key filter in streaming mode for KV table
+ // KV tables should not support record batch filter pushdown for non-primary key fields
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("name", DataTypes.STRING(), 0, 1);
+ ValueLiteralExpression literal = new ValueLiteralExpression("test");
+ CallExpression equalCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(fieldRef, literal),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(equalCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // For KV tables, non-primary key filters should not be pushed down
+ assertThat(result.getAcceptedFilters()).isEmpty();
+ assertThat(result.getRemainingFilters()).hasSize(1);
+ assertThat(tableSource.getLogRecordBatchFilter()).isNull();
+ }
+
+ @Test
+ void testStreamingModeMultipleFilters() {
+ // Test multiple filters in streaming mode for KV table
+ // KV tables should not support record batch filter pushdown
+ FieldReferenceExpression idFieldRef =
+ new FieldReferenceExpression("id", DataTypes.INT(), 0, 0);
+ FieldReferenceExpression nameFieldRef =
+ new FieldReferenceExpression("name", DataTypes.STRING(), 0, 1);
+ ValueLiteralExpression idLiteral = new ValueLiteralExpression(5);
+ ValueLiteralExpression nameLiteral = new ValueLiteralExpression("test");
+
+ CallExpression idEqualCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(idFieldRef, idLiteral),
+ DataTypes.BOOLEAN());
+
+ CallExpression nameEqualCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(nameFieldRef, nameLiteral),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(idEqualCall, nameEqualCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // For KV tables in streaming mode, no filters should be pushed down
+ assertThat(result.getAcceptedFilters()).isEmpty();
+ assertThat(result.getRemainingFilters()).hasSize(2);
+ assertThat(tableSource.getLogRecordBatchFilter()).isNull();
+ }
+ }
+
+ @Nested
+ class NonPartitionedLogTableTests {
+ private FlinkTableSource tableSource;
+
+ @BeforeEach
+ void setUp() {
+ // Create a non-partitioned log table schema: id, name, value (no primary key)
+ RowType tableOutputType =
+ (RowType)
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.INT()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("value", DataTypes.BIGINT()),
+ DataTypes.FIELD("region", DataTypes.STRING()))
+ .getLogicalType();
+
+ TablePath tablePath = TablePath.of("test_db", "test_log_table");
+ Configuration flussConfig = new Configuration();
+ flussConfig.setString(FlinkConnectorOptions.BOOTSTRAP_SERVERS.key(), "localhost:9092");
+
+ // Create table config for testing
+ Configuration tableConfig = new Configuration();
+
+ tableConfig.set(ConfigOptions.TABLE_STATISTICS_COLUMNS, "*");
+
+ FlinkConnectorOptionsUtils.StartupOptions startupOptions =
+ new FlinkConnectorOptionsUtils.StartupOptions();
+ startupOptions.startupMode = FlinkConnectorOptions.ScanStartupMode.EARLIEST;
+
+ tableSource =
+ new FlinkTableSource(
+ tablePath,
+ flussConfig,
+ new TableConfig(tableConfig),
+ tableOutputType,
+ new int[] {}, // no primary key indexes
+ new int[] {}, // bucket key indexes
+ new int[] {}, // partition key indexes
+ true, // streaming
+ startupOptions,
+ false, // lookup async
+ false, // insert if not exists
+ null, // cache
+ 1000L, // scan partition discovery interval
+ false, // is data lake enabled
+ null, // merge engine type
+ Maps.newHashMap(),
+ null); // lease context
+ }
+
+ @Test
+ void testLogTableRecordBatchFilterPushdown() {
+ // Test record batch filter pushdown for log table
+ // Log tables should support record batch filter pushdown
+ FieldReferenceExpression valueFieldRef =
+ new FieldReferenceExpression("value", DataTypes.BIGINT(), 0, 2);
+ FieldReferenceExpression regionFieldRef =
+ new FieldReferenceExpression("region", DataTypes.STRING(), 0, 3);
+
+ CallExpression regionEqualCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(regionFieldRef, new ValueLiteralExpression("HangZhou")),
+ DataTypes.BOOLEAN());
+ CallExpression valueRangeCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.GREATER_THAN,
+ Arrays.asList(valueFieldRef, new ValueLiteralExpression(1000L)),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(regionEqualCall, valueRangeCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ assertThat(result.getAcceptedFilters()).hasSize(2);
+ assertThat(result.getRemainingFilters()).hasSize(2);
+ Predicate predicate = tableSource.getLogRecordBatchFilter();
+ assertThat(predicate).isNotNull();
+ assertThat(tableSource.getSingleRowFilter()).isNull();
+
+ // Verify the predicate evaluates correctly against statistics.
+ // Schema: id(INT), name(STRING), value(BIGINT), region(STRING)
+ // Filter: region = 'HangZhou' AND value > 1000
+ Long[] noNulls = new Long[] {0L, 0L, 0L, 0L};
+
+ // Batch with value range [2000, 5000] and region ['HangZhou', 'HangZhou']
+ // → should match
+ assertThat(
+ predicate.test(
+ 100,
+ GenericRow.of(
+ null, null, 2000L, BinaryString.fromString("HangZhou")),
+ GenericRow.of(
+ null, null, 5000L, BinaryString.fromString("HangZhou")),
+ noNulls))
+ .isTrue();
+
+ // Batch with value range [100, 500] → should NOT match (max=500, not > 1000)
+ assertThat(
+ predicate.test(
+ 100,
+ GenericRow.of(
+ null, null, 100L, BinaryString.fromString("HangZhou")),
+ GenericRow.of(
+ null, null, 500L, BinaryString.fromString("HangZhou")),
+ noNulls))
+ .isFalse();
+
+ // Batch with region ['Beijing', 'Beijing'] → should NOT match
+ assertThat(
+ predicate.test(
+ 100,
+ GenericRow.of(
+ null, null, 2000L, BinaryString.fromString("Beijing")),
+ GenericRow.of(
+ null, null, 5000L, BinaryString.fromString("Beijing")),
+ noNulls))
+ .isFalse();
+ }
+
+ @Test
+ void testLogTableRangeFilterPushdown() {
+ // Test range filter pushdown for log table
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("id", DataTypes.INT(), 0, 0);
+ ValueLiteralExpression literal1 = new ValueLiteralExpression(3);
+ ValueLiteralExpression literal2 = new ValueLiteralExpression(10);
+
+ CallExpression greaterThanCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.GREATER_THAN,
+ Arrays.asList(fieldRef, literal1),
+ DataTypes.BOOLEAN());
+
+ CallExpression lessThanCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.LESS_THAN,
+ Arrays.asList(fieldRef, literal2),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(greaterThanCall, lessThanCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // Range filters should be pushed down as record batch filter
+ assertThat(result.getAcceptedFilters()).hasSize(2);
+ assertThat(result.getRemainingFilters()).hasSize(2);
+ Predicate predicate = tableSource.getLogRecordBatchFilter();
+ assertThat(predicate).isNotNull();
+
+ // Verify the predicate actually evaluates correctly against statistics.
+ // Schema: id(INT), name(STRING), value(BIGINT), region(STRING)
+ // Filter: id > 3 AND id < 10
+ Long[] noNulls = new Long[] {0L, 0L, 0L, 0L};
+
+ // Batch with id range [5, 8] → should match (5 > 3 is possible, 8 < 10 is possible)
+ assertThat(
+ predicate.test(
+ 100,
+ GenericRow.of(5, null, null, null),
+ GenericRow.of(8, null, null, null),
+ noNulls))
+ .isTrue();
+
+ // Batch with id range [1, 2] → should NOT match (max=2, not > 3)
+ assertThat(
+ predicate.test(
+ 100,
+ GenericRow.of(1, null, null, null),
+ GenericRow.of(2, null, null, null),
+ noNulls))
+ .isFalse();
+
+ // Batch with id range [11, 20] → should NOT match (min=11, not < 10)
+ assertThat(
+ predicate.test(
+ 100,
+ GenericRow.of(11, null, null, null),
+ GenericRow.of(20, null, null, null),
+ noNulls))
+ .isFalse();
+ }
+
+ @Test
+ void testEmptyFilters() {
+ // Test with empty filters
+ List<ResolvedExpression> filters = Collections.emptyList();
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // No filters should be accepted or remain
+ assertThat(result.getAcceptedFilters()).isEmpty();
+ assertThat(result.getRemainingFilters()).isEmpty();
+ assertThat(tableSource.getLogRecordBatchFilter()).isNull();
+ }
+ }
+
+ @Nested
+ class PartitionedKvTableTests {
+ private FlinkTableSource tableSource;
+
+ @BeforeEach
+ void setUp() {
+ // Create a partitioned KV table schema: id (PK), name, value, region (partition key)
+ RowType tableOutputType =
+ (RowType)
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.INT()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("value", DataTypes.BIGINT()),
+ DataTypes.FIELD("region", DataTypes.STRING()))
+ .getLogicalType();
+
+ TablePath tablePath = TablePath.of("test_db", "test_partitioned_kv_table");
+ Configuration flussConfig = new Configuration();
+ flussConfig.setString(FlinkConnectorOptions.BOOTSTRAP_SERVERS.key(), "localhost:9092");
+
+ // Create table config for testing
+ Configuration tableConfig = new Configuration();
+
+ tableConfig.set(ConfigOptions.TABLE_STATISTICS_COLUMNS, "*");
+
+ FlinkConnectorOptionsUtils.StartupOptions startupOptions =
+ new FlinkConnectorOptionsUtils.StartupOptions();
+ startupOptions.startupMode = FlinkConnectorOptions.ScanStartupMode.EARLIEST;
+
+ tableSource =
+ new FlinkTableSource(
+ tablePath,
+ flussConfig,
+ new TableConfig(tableConfig),
+ tableOutputType,
+ new int[] {0}, // primary key indexes (id)
+ new int[] {}, // bucket key indexes
+ new int[] {3}, // partition key indexes (region)
+ true, // streaming
+ startupOptions,
+ false, // lookup async
+ false, // insert if not exists
+ null, // cache
+ 1000L, // scan partition discovery interval
+ false, // is data lake enabled
+ null, // merge engine type
+ Maps.newHashMap(),
+ null); // lease context
+ }
+
+ @Test
+ void testPartitionKeyPushdown() {
+ // Test partition key pushdown
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("region", DataTypes.STRING(), 0, 3);
+ ValueLiteralExpression literal =
+ new ValueLiteralExpression("us-east", DataTypes.STRING().notNull());
+ CallExpression equalCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(fieldRef, literal),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(equalCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // Partition key filters should be pushed down as partition filters
+ assertThat(result.getAcceptedFilters()).hasSize(1);
+ // FLINK-38635: all filters remain for safety (scan vs lookup ambiguity)
+ assertThat(result.getRemainingFilters()).hasSize(1);
+ assertThat(tableSource.getPartitionFilters()).isNotNull();
+ }
+
+ @Test
+ void testRecordBatchFilterNotPushedDownInKvTable() {
+ // Test partition key pushdown
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("name", DataTypes.STRING(), 0, 1);
+ ValueLiteralExpression literal =
+ new ValueLiteralExpression("bob", DataTypes.STRING().notNull());
+ CallExpression equalCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(fieldRef, literal),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(equalCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ assertThat(result.getAcceptedFilters()).isEmpty();
+ assertThat(result.getRemainingFilters()).hasSize(1);
+ assertThat(tableSource.getLogRecordBatchFilter()).isNull();
+ }
+ }
+
+ @Nested
+ class PartitionedLogTableTests {
+ private FlinkTableSource tableSource;
+
+ @BeforeEach
+ void setUp() {
+ // Create a partitioned log table schema: id, name, value, region (partition key)
+ RowType tableOutputType =
+ (RowType)
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.INT()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("value", DataTypes.BIGINT()),
+ DataTypes.FIELD("region", DataTypes.STRING()),
+ DataTypes.FIELD("attachment", DataTypes.BYTES()))
+ .getLogicalType();
+
+ TablePath tablePath = TablePath.of("test_db", "test_partitioned_log_table");
+ Configuration flussConfig = new Configuration();
+ flussConfig.setString(FlinkConnectorOptions.BOOTSTRAP_SERVERS.key(), "localhost:9092");
+
+ // Create table config for testing
+ Configuration tableConfig = new Configuration();
+
+ tableConfig.set(ConfigOptions.TABLE_STATISTICS_COLUMNS, "*");
+
+ FlinkConnectorOptionsUtils.StartupOptions startupOptions =
+ new FlinkConnectorOptionsUtils.StartupOptions();
+ startupOptions.startupMode = FlinkConnectorOptions.ScanStartupMode.EARLIEST;
+
+ tableSource =
+ new FlinkTableSource(
+ tablePath,
+ flussConfig,
+ new TableConfig(tableConfig),
+ tableOutputType,
+ new int[] {}, // no primary key indexes
+ new int[] {}, // bucket key indexes
+ new int[] {3}, // partition key indexes (region)
+ true, // streaming
+ startupOptions,
+ false, // lookup async
+ false, // insert if not exists
+ null, // cache
+ 1000L, // scan partition discovery interval
+ false, // is data lake enabled
+ null, // merge engine type
+ Maps.newHashMap(),
+ null); // lease context
+ }
+
+ @Test
+ void testPartitionedLogTablePartitionKeyPushdown() {
+ // Test partition key pushdown for partitioned log table
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("region", DataTypes.STRING(), 0, 3);
+ ValueLiteralExpression literal =
+ new ValueLiteralExpression("us-east", DataTypes.STRING().notNull());
+ CallExpression equalCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(fieldRef, literal),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(equalCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // Partition key filters should be pushed down as partition filters
+ assertThat(result.getAcceptedFilters()).hasSize(1);
+ // FLINK-38635: all filters remain for safety (scan vs lookup ambiguity)
+ assertThat(result.getRemainingFilters()).hasSize(1);
+ assertThat(tableSource.getPartitionFilters()).isNotNull();
+ }
+
+ @Test
+ void testPartitionedLogTableRecordBatchFilterPushdown() {
+ // Test record batch filter pushdown for partitioned log table
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("id", DataTypes.INT(), 0, 0);
+ ValueLiteralExpression literal = new ValueLiteralExpression(5);
+ CallExpression equalCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(fieldRef, literal),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(equalCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ assertThat(result.getAcceptedFilters()).hasSize(1);
+ assertThat(result.getRemainingFilters()).hasSize(1);
+ // record batch filter should be successfully pushdown
+ assertThat(tableSource.getLogRecordBatchFilter()).isNotNull();
+ }
+
+ @Test
+ void testCopyPreservesRecordBatchFilters() {
+ // Test that copying preserves all filter states
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("id", DataTypes.INT(), 0, 0);
+ ValueLiteralExpression literal = new ValueLiteralExpression(5);
+ CallExpression equalCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(fieldRef, literal),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(equalCall);
+ tableSource.applyFilters(filters);
+
+ // Verify original has filters
+ Predicate originalFilter = tableSource.getLogRecordBatchFilter();
+ assertThat(originalFilter).isNotNull();
+
+ // Copy the table source
+ FlinkTableSource copiedSource = (FlinkTableSource) tableSource.copy();
+
+ // Verify copied source has the same filters
+ Predicate copiedFilter = copiedSource.getLogRecordBatchFilter();
+ assertThat(copiedFilter).isNotNull();
+ assertThat(copiedFilter).isEqualTo(originalFilter);
+ }
+
+ @Test
+ void testUnsupportedRecordBatchFilterColumnTypes() {
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("attachment", DataTypes.BYTES(), 0, 4);
+ ValueLiteralExpression literal =
+ new ValueLiteralExpression(new byte[] {}, DataTypes.BYTES().notNull());
+ CallExpression equalCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(fieldRef, literal),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(equalCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // Unsupported column should not be pushed down as record batch filter
+ assertThat(result.getAcceptedFilters()).isEmpty();
+ assertThat(result.getRemainingFilters()).hasSize(1);
+ assertThat(tableSource.getLogRecordBatchFilter()).isNull();
+ }
+
+ @Test
+ void testMixedFilterTypesPushdown() {
+ // Test mixed filter types: partition key + regular column
+ FieldReferenceExpression regionFieldRef =
+ new FieldReferenceExpression("region", DataTypes.STRING(), 0, 3);
+ FieldReferenceExpression idFieldRef =
+ new FieldReferenceExpression("id", DataTypes.INT(), 0, 0);
+ FieldReferenceExpression nameFieldRef =
+ new FieldReferenceExpression("name", DataTypes.STRING(), 0, 1);
+ ValueLiteralExpression regionLiteral =
+ new ValueLiteralExpression("us-east", DataTypes.STRING().notNull());
+ ValueLiteralExpression idLiteral = new ValueLiteralExpression(5);
+ ValueLiteralExpression nameLiteral =
+ new ValueLiteralExpression("test", DataTypes.STRING().notNull());
+
+ CallExpression regionEqualCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(regionFieldRef, regionLiteral),
+ DataTypes.BOOLEAN());
+
+ CallExpression idEqualCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(idFieldRef, idLiteral),
+ DataTypes.BOOLEAN());
+
+ CallExpression nameEqualCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(nameFieldRef, nameLiteral),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters =
+ Arrays.asList(regionEqualCall, idEqualCall, nameEqualCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // Partition key should be pushed down as partition filter
+ // Regular columns should be pushed down as record batch filter
+ assertThat(result.getAcceptedFilters()).hasSize(3);
+ // FLINK-38635: all filters remain for safety (scan vs lookup ambiguity)
+ assertThat(result.getRemainingFilters()).hasSize(3);
+ assertThat(tableSource.getPartitionFilters()).isNotNull();
+ assertThat(tableSource.getLogRecordBatchFilter())
+ .isNotNull(); // Record batch filter pushed down for non-PK partitioned log
+ // table
+ }
+ }
+
+ @Nested
+ class PartialStatisticsColumnTests {
+ private FlinkTableSource tableSource;
+
+ @BeforeEach
+ void setUp() {
+ // Create a log table schema: id, name, value, region (no primary key)
+ RowType tableOutputType =
+ (RowType)
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.INT()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("value", DataTypes.BIGINT()),
+ DataTypes.FIELD("region", DataTypes.STRING()),
+ DataTypes.FIELD("attachment", DataTypes.BYTES()))
+ .getLogicalType();
+
+ TablePath tablePath = TablePath.of("test_db", "test_partial_stats_table");
+ Configuration flussConfig = new Configuration();
+ flussConfig.setString(FlinkConnectorOptions.BOOTSTRAP_SERVERS.key(), "localhost:9092");
+
+ // Create table config with only partial columns for statistics
+ Configuration tableConfig = new Configuration();
+ // Only enable statistics for 'id' and 'value' columns, not for 'name', 'region',
+ // 'attachment'
+ tableConfig.set(ConfigOptions.TABLE_STATISTICS_COLUMNS, "id,value");
+
+ FlinkConnectorOptionsUtils.StartupOptions startupOptions =
+ new FlinkConnectorOptionsUtils.StartupOptions();
+ startupOptions.startupMode = FlinkConnectorOptions.ScanStartupMode.EARLIEST;
+
+ tableSource =
+ new FlinkTableSource(
+ tablePath,
+ flussConfig,
+ new TableConfig(tableConfig),
+ tableOutputType,
+ new int[] {}, // no primary key indexes
+ new int[] {}, // bucket key indexes
+ new int[] {}, // partition key indexes
+ true, // streaming
+ startupOptions,
+ false, // lookup async
+ false, // insert if not exists
+ null, // cache
+ 1000L, // scan partition discovery interval
+ false, // is data lake enabled
+ null, // merge engine type
+ Maps.newHashMap(),
+ null); // lease context
+ }
+
+ @Test
+ void testFilterOnColumnWithStatistics() {
+ // Test filter on 'id' column which has statistics enabled
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("id", DataTypes.INT(), 0, 0);
+ ValueLiteralExpression literal = new ValueLiteralExpression(5);
+ CallExpression equalCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(fieldRef, literal),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(equalCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // Filter on 'id' should be pushed down as record batch filter since it has statistics
+ assertThat(result.getAcceptedFilters()).hasSize(1);
+ assertThat(result.getRemainingFilters()).hasSize(1); // Filter remains for execution
+ assertThat(tableSource.getLogRecordBatchFilter()).isNotNull();
+ }
+
+ @Test
+ void testFilterOnColumnWithoutStatistics() {
+ // Test filter on 'name' column which does not have statistics enabled
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("name", DataTypes.STRING(), 0, 1);
+ ValueLiteralExpression literal = new ValueLiteralExpression("test");
+ CallExpression equalCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(fieldRef, literal),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(equalCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // Filter on 'name' should not be pushed down since it doesn't have statistics
+ assertThat(result.getAcceptedFilters()).isEmpty();
+ assertThat(result.getRemainingFilters()).hasSize(1);
+ assertThat(tableSource.getLogRecordBatchFilter()).isNull();
+ }
+
+ @Test
+ void testFilterOnBinaryColumnExcludedFromStatistics() {
+ // Test filter on 'attachment' column which is binary type (excluded from statistics)
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("attachment", DataTypes.BYTES(), 0, 4);
+ ValueLiteralExpression literal =
+ new ValueLiteralExpression(new byte[] {1, 2, 3}, DataTypes.BYTES().notNull());
+ CallExpression equalCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(fieldRef, literal),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(equalCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // Binary columns should not have pushdown even if configured
+ assertThat(result.getAcceptedFilters()).isEmpty();
+ assertThat(result.getRemainingFilters()).hasSize(1);
+ assertThat(tableSource.getLogRecordBatchFilter()).isNull();
+ }
+
+ @Test
+ void testMixedFiltersWithPartialStatistics() {
+ // Test mixed filters: one on column with statistics, one without
+ FieldReferenceExpression idFieldRef =
+ new FieldReferenceExpression("id", DataTypes.INT(), 0, 0);
+ FieldReferenceExpression nameFieldRef =
+ new FieldReferenceExpression("name", DataTypes.STRING(), 0, 1);
+ FieldReferenceExpression valueFieldRef =
+ new FieldReferenceExpression("value", DataTypes.BIGINT(), 0, 2);
+
+ ValueLiteralExpression idLiteral = new ValueLiteralExpression(5);
+ ValueLiteralExpression nameLiteral = new ValueLiteralExpression("test");
+ ValueLiteralExpression valueLiteral = new ValueLiteralExpression(100L);
+
+ CallExpression idEqualCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(idFieldRef, idLiteral),
+ DataTypes.BOOLEAN());
+
+ CallExpression nameEqualCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(nameFieldRef, nameLiteral),
+ DataTypes.BOOLEAN());
+
+ CallExpression valueGreaterCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.GREATER_THAN,
+ Arrays.asList(valueFieldRef, valueLiteral),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters =
+ Arrays.asList(idEqualCall, nameEqualCall, valueGreaterCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // Only filters on 'id' and 'value' should be accepted for pushdown
+ assertThat(result.getAcceptedFilters()).hasSize(2); // id and value filters
+ assertThat(result.getRemainingFilters()).hasSize(3); // all filters remain for execution
+ assertThat(tableSource.getLogRecordBatchFilter())
+ .isNotNull(); // Should have merged predicate for id and value
+ }
+
+ @Test
+ void testRangeFilterOnColumnWithStatistics() {
+ // Test range filter on 'value' column which has statistics enabled
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("value", DataTypes.BIGINT(), 0, 2);
+ ValueLiteralExpression lowerBound = new ValueLiteralExpression(10L);
+ ValueLiteralExpression upperBound = new ValueLiteralExpression(100L);
+
+ CallExpression greaterThanCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.GREATER_THAN,
+ Arrays.asList(fieldRef, lowerBound),
+ DataTypes.BOOLEAN());
+
+ CallExpression lessThanCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.LESS_THAN,
+ Arrays.asList(fieldRef, upperBound),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(greaterThanCall, lessThanCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // Range filters on 'value' should be pushed down since it has statistics
+ assertThat(result.getAcceptedFilters()).hasSize(2);
+ assertThat(result.getRemainingFilters()).hasSize(2); // Filters remain for execution
+ assertThat(tableSource.getLogRecordBatchFilter()).isNotNull();
+ }
+
+ @Test
+ void testEmptyStatisticsConfiguration() {
+ // Test with empty statistics configuration
+ RowType tableOutputType =
+ (RowType)
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.INT()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("value", DataTypes.BIGINT()))
+ .getLogicalType();
+
+ Configuration tableConfig = new Configuration();
+ // No statistics columns configured
+ tableConfig.set(ConfigOptions.TABLE_STATISTICS_COLUMNS, "");
+
+ FlinkTableSource emptyStatsTableSource =
+ new FlinkTableSource(
+ TablePath.of("test_db", "test_empty_stats_table"),
+ new Configuration(),
+ new TableConfig(tableConfig),
+ tableOutputType,
+ new int[] {}, // no primary key indexes
+ new int[] {}, // bucket key indexes
+ new int[] {}, // partition key indexes
+ true, // streaming
+ new FlinkConnectorOptionsUtils.StartupOptions(),
+ false, // lookup async
+ false, // insert if not exists
+ null, // cache
+ 1000L, // scan partition discovery interval
+ false, // is data lake enabled
+ null, // merge engine type
+ Maps.newHashMap(),
+ null); // lease context
+
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("id", DataTypes.INT(), 0, 0);
+ ValueLiteralExpression literal = new ValueLiteralExpression(5);
+ CallExpression equalCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(fieldRef, literal),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(equalCall);
+
+ FlinkTableSource.Result result = emptyStatsTableSource.applyFilters(filters);
+
+ // No filters should be pushed down when statistics are disabled
+ assertThat(result.getAcceptedFilters()).isEmpty();
+ assertThat(result.getRemainingFilters()).hasSize(1);
+ assertThat(emptyStatsTableSource.getLogRecordBatchFilter()).isNull();
+ }
+ }
+
+ @Nested
+ class BatchModePrimaryKeyPushdownTests {
+ private FlinkTableSource tableSource;
+
+ @BeforeEach
+ void setUp() {
+ // Create a KV table schema for batch mode testing
+ RowType tableOutputType =
+ (RowType)
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.INT()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("value", DataTypes.BIGINT()))
+ .getLogicalType();
+
+ TablePath tablePath = TablePath.of("test_db", "test_batch_table");
+ Configuration flussConfig = new Configuration();
+ flussConfig.setString(FlinkConnectorOptions.BOOTSTRAP_SERVERS.key(), "localhost:9092");
+
+ // Create table config for testing
+ Configuration tableConfig = new Configuration();
+
+ tableConfig.set(ConfigOptions.TABLE_STATISTICS_COLUMNS, "*");
+
+ FlinkConnectorOptionsUtils.StartupOptions startupOptions =
+ new FlinkConnectorOptionsUtils.StartupOptions();
+ startupOptions.startupMode = FlinkConnectorOptions.ScanStartupMode.FULL;
+
+ tableSource =
+ new FlinkTableSource(
+ tablePath,
+ flussConfig,
+ new TableConfig(tableConfig),
+ tableOutputType,
+ new int[] {0}, // primary key indexes (id)
+ new int[] {}, // bucket key indexes
+ new int[] {}, // partition key indexes
+ false, // batch mode
+ startupOptions,
+ false, // lookup async
+ false, // insert if not exists
+ null, // cache
+ 1000L, // scan partition discovery interval
+ false, // is data lake enabled
+ null, // merge engine type
+ Maps.newHashMap(),
+ null); // lease context
+ }
+
+ @Test
+ void testBatchModePrimaryKeyPushdown() {
+ // Test primary key pushdown in batch mode with FULL startup mode
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("id", DataTypes.INT(), 0, 0);
+ ValueLiteralExpression literal = new ValueLiteralExpression(5);
+ CallExpression equalCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(fieldRef, literal),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(equalCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // In batch mode with FULL startup and complete primary key filter,
+ // should use single row filter for point lookup
+ assertThat(result.getAcceptedFilters()).hasSize(1);
+ // FLINK-38635: all filters remain for safety (scan vs lookup ambiguity)
+ assertThat(result.getRemainingFilters()).hasSize(1);
+ assertThat(tableSource.getSingleRowFilter()).isNotNull();
+ assertThat(tableSource.getLogRecordBatchFilter()).isNull();
+
+ // Verify the single row filter contains the correct value
+ GenericRowData singleRowFilter = tableSource.getSingleRowFilter();
+ assertThat(singleRowFilter.getInt(0)).isEqualTo(5);
+ }
+
+ @Test
+ void testBatchModeIncompletePrimaryKeyPushdown() {
+ // Test incomplete primary key filter in batch mode (should not push down)
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("name", DataTypes.STRING(), 0, 1);
+ ValueLiteralExpression literal = new ValueLiteralExpression("test");
+ CallExpression equalCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(fieldRef, literal),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(equalCall);
+
+ FlinkTableSource.Result result = tableSource.applyFilters(filters);
+
+ // Non-primary key filters should not be pushed down in batch mode
+ assertThat(result.getAcceptedFilters()).isEmpty();
+ assertThat(result.getRemainingFilters()).hasSize(1);
+ assertThat(tableSource.getSingleRowFilter()).isNull();
+ assertThat(tableSource.getLogRecordBatchFilter()).isNull();
+ }
+
+ @Test
+ void testBatchModeNonFullStartupMode() {
+ // Test batch mode with non-FULL startup mode
+ FlinkConnectorOptionsUtils.StartupOptions startupOptions =
+ new FlinkConnectorOptionsUtils.StartupOptions();
+ startupOptions.startupMode = FlinkConnectorOptions.ScanStartupMode.EARLIEST;
+
+ // Create a new table output type for the non-full startup test
+ RowType nonFullStartupTableOutputType =
+ (RowType)
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.INT()),
+ DataTypes.FIELD("name", DataTypes.STRING()),
+ DataTypes.FIELD("value", DataTypes.BIGINT()))
+ .getLogicalType();
+
+ // Create table config for testing
+ Configuration tableConfig = new Configuration();
+
+ tableConfig.set(ConfigOptions.TABLE_STATISTICS_COLUMNS, "*");
+
+ FlinkTableSource nonFullStartupTableSource =
+ new FlinkTableSource(
+ TablePath.of("test_db", "test_batch_table"),
+ new Configuration(),
+ new TableConfig(tableConfig),
+ nonFullStartupTableOutputType,
+ new int[] {0}, // primary key indexes
+ new int[] {}, // bucket key indexes
+ new int[] {}, // partition key indexes
+ false, // batch mode
+ startupOptions,
+ false, // lookup async
+ false, // insert if not exists
+ null, // cache
+ 1000L, // scan partition discovery interval
+ false, // is data lake enabled
+ null, // merge engine type
+ Maps.newHashMap(),
+ null); // lease context
+
+ FieldReferenceExpression fieldRef =
+ new FieldReferenceExpression("id", DataTypes.INT(), 0, 0);
+ ValueLiteralExpression literal = new ValueLiteralExpression(5);
+ CallExpression equalCall =
+ new CallExpression(
+ BuiltInFunctionDefinitions.EQUALS,
+ Arrays.asList(fieldRef, literal),
+ DataTypes.BOOLEAN());
+
+ List<ResolvedExpression> filters = Arrays.asList(equalCall);
+
+ FlinkTableSource.Result result = nonFullStartupTableSource.applyFilters(filters);
+
+ // With non-FULL startup mode, should not use single row filter
+ assertThat(result.getAcceptedFilters()).isEmpty();
+ assertThat(result.getRemainingFilters()).hasSize(1);
+ assertThat(nonFullStartupTableSource.getSingleRowFilter()).isNull();
+ }
+ }
+}
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceITCase.java
index 224798443..0cb902abe 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceITCase.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceITCase.java
@@ -24,6 +24,7 @@ import org.apache.fluss.client.table.Table;
import org.apache.fluss.client.table.writer.UpsertWriter;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.InvalidConfigException;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.row.GenericRow;
import org.apache.fluss.row.InternalRow;
@@ -71,7 +72,6 @@ import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import static org.apache.fluss.flink.FlinkConnectorOptions.BOOTSTRAP_SERVERS;
import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertQueryResultExactOrder;
import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertResultsIgnoreOrder;
import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.collectRowsWithTimeout;
@@ -130,7 +130,7 @@ abstract class FlinkTableSourceITCase extends AbstractTestBase {
tEnv.executeSql(
String.format(
"create catalog %s with ('type' = 'fluss', '%s' = '%s')",
- CATALOG_NAME, BOOTSTRAP_SERVERS.key(), bootstrapServers));
+ CATALOG_NAME, ConfigOptions.BOOTSTRAP_SERVERS.key(), bootstrapServers));
tEnv.executeSql("use catalog " + CATALOG_NAME);
tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 4);
tEnv.executeSql("create database " + DEFAULT_DB);
@@ -1312,7 +1312,7 @@ abstract class FlinkTableSourceITCase extends AbstractTestBase {
+ "filter=[=(c, _UTF-16LE'2025':VARCHAR(2147483647) CHARACTER SET \"UTF-16LE\")], "
+ "project=[a, c, d]]], fields=[a, c, d])");
- // test column filter、partition filter and flink runtime filter
+ // test column filter, partition filter and flink runtime filter
org.apache.flink.util.CloseableIterator<Row> rowIter =
tEnv.executeSql(
"select a,c,d from combined_filters_table where c ='2025' and d % 200 = 0")
@@ -1329,7 +1329,7 @@ abstract class FlinkTableSourceITCase extends AbstractTestBase {
+ "filter=[=(c, _UTF-16LE'2025':VARCHAR(2147483647) CHARACTER SET \"UTF-16LE\")], "
+ "project=[a, c, d]]], fields=[a, c, d])");
- // test column filter、partition filter and flink runtime filter
+ // test column filter, partition filter and flink runtime filter
rowIter =
tEnv.executeSql(
"select a,c,d from combined_filters_table where c ='2025' and d = 200")
@@ -1356,6 +1356,337 @@ abstract class FlinkTableSourceITCase extends AbstractTestBase {
assertResultsIgnoreOrder(rowIter, expectedRowValues, true);
}
+ @Test
+ void testStreamingReadNonPKTableWithCombinedFilters() throws Exception {
+ tEnv.executeSql(
+ "create table combined_filters_table"
+ + " (a int not null, b varchar, c string, d int) partitioned by (c)"
+ + " with ('table.statistics.columns' = 'd')");
+ TablePath tablePath = TablePath.of(DEFAULT_DB, "combined_filters_table");
+ tEnv.executeSql("alter table combined_filters_table add partition (c=2025)");
+ tEnv.executeSql("alter table combined_filters_table add partition (c=2026)");
+
+ List<InternalRow> rows = new ArrayList<>();
+ List<String> expectedRowValues = new ArrayList<>();
+
+ for (int i = 0; i < 10; i++) {
+ rows.add(row(i, "v" + i, "2025", i * 100));
+ if (i > 2) {
+ expectedRowValues.add(String.format("+I[%d, 2025, %d]", i, i * 100));
+ }
+ }
+ writeRows(conn, tablePath, rows, true);
+
+ List<InternalRow> rows2 = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ rows2.add(row(i, "v" + i, "2026", i * 100));
+ }
+
+ writeRows(conn, tablePath, rows2, true);
+
+ String plan =
+ tEnv.explainSql(
+ "select a,c,d from combined_filters_table where c ='2025' and d > 200");
+
+ // assert both partition filter and record batch filter are pushed down
+ assertThat(plan)
+ .contains(
+ "TableSourceScan(table=[[testcatalog, defaultdb, combined_filters_table, "
+ + "filter=[and(=(c, _UTF-16LE'2025':VARCHAR(2147483647) CHARACTER SET \"UTF-16LE\"), >(d, 200))], "
+ + "project=[a, c, d]]], fields=[a, c, d])");
+ // assert predicates are still retained in the Calc operator (FLINK-38635 safety net)
+ assertThat(plan).contains("where=");
+
+ // test column filter, partition filter and flink runtime filter
+ org.apache.flink.util.CloseableIterator<Row> rowIter =
+ tEnv.executeSql(
+ "select a,c,d from combined_filters_table where c ='2025' and d > 200;")
+ .collect();
+
+ assertResultsIgnoreOrder(rowIter, expectedRowValues, true);
+ }
+
+ /**
+ * Test RecordBatchFilter push down with large dataset and multiple range filters to verify
+ * correctness when log segments have gaps due to filtering.
+ */
+ @Test
+ void testRecordBatchFilterPushDownWithLogGaps() throws Exception {
+ // Create a log table for testing record batch filter push down with specific batch size
+ // configurations
+ // to ensure predictable RecordBatch sizes during writing and reading
+ tEnv.executeSql(
+ "create table record_batch_filter_test "
+ + "(id int, sequence_num int, name varchar, score double) "
+ + "with ("
+ + "'table.log.format' = 'ARROW', "
+ + "'table.statistics.columns' = 'sequence_num', "
+ // Writer configuration: aim for ~500 records per RecordBatch
+ // Each record has: int(4) + int(4) + varchar(~12) + double(8) ≈ 28 bytes
+ // So 500 records ≈ 14KB, set batch size to 32KB to account for overhead
+ + "'client.writer.batch-size' = '32kb', "
+ // Shorter timeout for predictable batching
+ + "'client.writer.batch-timeout' = '10ms', "
+ // limit read batch size to 500 records max
+ + "'client.scanner.log.max-poll-records' = '500', "
+ + "'client.scanner.log.fetch.max-bytes-for-bucket' = '32kb' "
+ + ")");
+
+ TablePath tablePath = TablePath.of(DEFAULT_DB, "record_batch_filter_test");
+
+ // Write 10,000 ordered rows with sequential sequence_num in controlled batches
+ // to ensure predictable RecordBatch sizes
+ int totalRecords = 10000;
+ int batchSize = 500;
+
+ for (int batchStart = 0; batchStart < totalRecords; batchStart += batchSize) {
+ List<InternalRow> batchRows = new ArrayList<>();
+ int batchEnd = Math.min(batchStart + batchSize, totalRecords);
+
+ for (int i = batchStart; i < batchEnd; i++) {
+ batchRows.add(row(i, i, "value_" + i, i * 0.1));
+ }
+
+ // Write each batch separately to ensure proper RecordBatch formation
+ writeRows(conn, tablePath, batchRows, true);
+ }
+
+ // Define three ranges to filter: 5500-5999, 8350-8400, 9400-9999
+ String query =
+ "select id, sequence_num, name from record_batch_filter_test "
+ + "where (sequence_num >= 5500 and sequence_num < 6000) "
+ + "or (sequence_num >= 8350 and sequence_num <= 8400) "
+ + "or (sequence_num >= 9400 and sequence_num < 10000)";
+
+ // Verify that record batch filters are pushed down
+ String plan = tEnv.explainSql(query);
+ assertThat(plan)
+ .contains(
+ "TableSourceScan(table=[[testcatalog, defaultdb, record_batch_filter_test, "
+ + "filter=[OR(OR(AND(>=(sequence_num, 5500), <(sequence_num, 6000)), "
+ + "AND(>=(sequence_num, 8350), <=(sequence_num, 8400))), "
+ + "AND(>=(sequence_num, 9400), <(sequence_num, 10000)))]");
+
+ // Collect results and verify correctness
+ List<String> expectedResults = new ArrayList<>();
+
+ // Range 1: 5500-5999 (500 records)
+ for (int i = 5500; i < 6000; i++) {
+ expectedResults.add(String.format("+I[%d, %d, value_%d]", i, i, i));
+ }
+
+ // Range 2: 8350-8400 (51 records)
+ for (int i = 8350; i <= 8400; i++) {
+ expectedResults.add(String.format("+I[%d, %d, value_%d]", i, i, i));
+ }
+
+ // Range 3: 9400-9999 (600 records)
+ for (int i = 9400; i < 10000; i++) {
+ expectedResults.add(String.format("+I[%d, %d, value_%d]", i, i, i));
+ }
+
+ try (CloseableIterator<Row> rowIter = tEnv.executeSql(query).collect()) {
+ assertResultsIgnoreOrder(rowIter, expectedResults, true);
+ }
+ }
+
+ /**
+ * Tests that ALTER TABLE can dynamically change statistics columns configuration, and filter
+ * pushdown works correctly across different statistics modes: disabled → subset → wildcard.
+ *
+ * <p>Each phase writes data, then verifies filter pushdown behavior. Since {@code writeRows}
+ * re-creates the Table instance each time, the writer picks up the latest TableInfo after ALTER
+ * TABLE, so new batches are written with the updated statistics configuration.
+ */
+ @Test
+ void testAlterTableStatisticsColumnsWithFilterPushDown() throws Exception {
+ // Create table WITHOUT statistics columns (disabled mode)
+ tEnv.executeSql(
+ "create table alter_stats_test "
+ + "(id int, amount bigint, region varchar, score double) "
+ + "with ("
+ + "'table.log.format' = 'ARROW', "
+ + "'client.writer.batch-size' = '16kb', "
+ + "'client.writer.batch-timeout' = '10ms'"
+ + ")");
+ TablePath tablePath = TablePath.of(DEFAULT_DB, "alter_stats_test");
+
+ // ========== Phase 1: Statistics DISABLED ==========
+ // Write data with amount range [0, 900], step=100
+ List<InternalRow> phase1Rows = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ phase1Rows.add(row(i, (long) (i * 100), "HangZhou", i * 1.5));
+ }
+ writeRows(conn, tablePath, phase1Rows, true);
+
+ // Without statistics columns configured, filter is NOT pushed to source for record
+ // batch filtering — it only appears in the Calc operator
+ String query1 = "select id, amount from alter_stats_test where amount > 500";
+ String plan1 = tEnv.explainSql(query1);
+ assertThat(plan1).contains("filter=[]");
+ assertThat(plan1).contains("where=");
+
+ // Verify correctness: amount > 500 means amount in {600, 700, 800, 900}
+ List<String> expected1 = new ArrayList<>();
+ for (int i = 6; i < 10; i++) {
+ expected1.add(String.format("+I[%d, %d]", i, i * 100));
+ }
+ try (CloseableIterator<Row> iter = tEnv.executeSql(query1).collect()) {
+ assertResultsIgnoreOrder(iter, expected1, true);
+ }
+
+ // ========== Phase 2: ALTER TABLE to enable statistics on subset ('amount') ==========
+ tEnv.executeSql("alter table alter_stats_test set ('table.statistics.columns' = 'amount')");
+
+ // Write new data with amount range [1000, 1900], step=100
+ List<InternalRow> phase2Rows = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ phase2Rows.add(row(10 + i, (long) (1000 + i * 100), "Shanghai", (10 + i) * 1.5));
+ }
+ writeRows(conn, tablePath, phase2Rows, true);
+
+ // Filter on 'amount' column — statistics should be collected for new batches
+ String query2 = "select id, amount from alter_stats_test where amount > 1500";
+ String plan2 = tEnv.explainSql(query2);
+ assertThat(plan2).contains("filter=[>(amount, 1500)]");
+
+ // Verify correctness: amount > 1500 means amount in {1600, 1700, 1800, 1900}
+ List<String> expected2 = new ArrayList<>();
+ for (int i = 6; i < 10; i++) {
+ expected2.add(String.format("+I[%d, %d]", 10 + i, 1000 + i * 100));
+ }
+ try (CloseableIterator<Row> iter = tEnv.executeSql(query2).collect()) {
+ assertResultsIgnoreOrder(iter, expected2, true);
+ }
+
+ // Filter on 'score' column — no statistics for 'score' in subset mode,
+ // filter is NOT pushed to source for record batch filtering, only in Calc operator.
+ // Select integer columns to avoid fragile double formatting in assertions.
+ String query2b = "select id, amount from alter_stats_test where score > 20.0";
+ String plan2b = tEnv.explainSql(query2b);
+ assertThat(plan2b).contains("filter=[]");
+ assertThat(plan2b).contains("where=");
+
+ List<String> expected2b = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ double score = (10 + i) * 1.5;
+ if (score > 20.0) {
+ expected2b.add(String.format("+I[%d, %d]", 10 + i, 1000 + i * 100));
+ }
+ }
+ try (CloseableIterator<Row> iter = tEnv.executeSql(query2b).collect()) {
+ assertResultsIgnoreOrder(iter, expected2b, true);
+ }
+
+ // ========== Phase 3: ALTER TABLE to wildcard '*' (all supported columns) ==========
+ tEnv.executeSql("alter table alter_stats_test set ('table.statistics.columns' = '*')");
+
+ // Write new data with amount range [2000, 2900], step=100
+ List<InternalRow> phase3Rows = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ phase3Rows.add(row(20 + i, (long) (2000 + i * 100), "Beijing", (20 + i) * 1.5));
+ }
+ writeRows(conn, tablePath, phase3Rows, true);
+
+ // Filter on 'amount' — statistics available
+ String query3 = "select id, amount, region from alter_stats_test where amount >= 2500";
+ String plan3 = tEnv.explainSql(query3);
+ assertThat(plan3).contains("filter=[>=(amount, 2500)]");
+
+ // Verify correctness: amount >= 2500 means amount in {2500, 2600, 2700, 2800, 2900}
+ List<String> expected3 = new ArrayList<>();
+ for (int i = 5; i < 10; i++) {
+ expected3.add(String.format("+I[%d, %d, Beijing]", 20 + i, 2000 + i * 100));
+ }
+ try (CloseableIterator<Row> iter = tEnv.executeSql(query3).collect()) {
+ assertResultsIgnoreOrder(iter, expected3, true);
+ }
+
+ // Combined filter on 'amount' and 'region' — both have statistics in wildcard mode,
+ // so the filter should be pushed to source (not empty)
+ String query3b =
+ "select id, amount, region from alter_stats_test "
+ + "where amount >= 1000 and region = 'Beijing'";
+ String plan3b = tEnv.explainSql(query3b);
+ assertThat(plan3b)
+ .contains(
+ "filter=[and(>=(amount, 1000), "
+ + "=(region, _UTF-16LE'Beijing':VARCHAR(2147483647) CHARACTER SET \"UTF-16LE\"))]");
+
+ // Verify: amount >= 1000 AND region = 'Beijing' → only phase3 rows
+ // (phase2 rows have region='Shanghai', so they are excluded by the region filter)
+ List<String> expected3b = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ expected3b.add(String.format("+I[%d, %d, Beijing]", 20 + i, 2000 + i * 100));
+ }
+ try (CloseableIterator<Row> iter = tEnv.executeSql(query3b).collect()) {
+ assertResultsIgnoreOrder(iter, expected3b, true);
+ }
+
+ // ========== Phase 4: Cross-phase query — all data, mixed statistics coverage ==========
+ // Query across all phases: phase1 batches have no stats, phase2 has partial (amount only),
+ // phase3 has full stats. The filter should still be pushed down because the latest
+ // table config has statistics enabled, and batches without stats are conservatively
+ // included.
+ String query4 =
+ "select id, amount from alter_stats_test "
+ + "where amount >= 500 and amount < 1500";
+ String plan4 = tEnv.explainSql(query4);
+ assertThat(plan4).contains("filter=[and(>=(amount, 500), <(amount, 1500))]");
+
+ List<String> expected4 = new ArrayList<>();
+ // Phase 1 rows: amount in {500, 600, 700, 800, 900}
+ for (int i = 5; i < 10; i++) {
+ expected4.add(String.format("+I[%d, %d]", i, i * 100));
+ }
+ // Phase 2 rows: amount in {1000, 1100, 1200, 1300, 1400}
+ for (int i = 0; i < 5; i++) {
+ expected4.add(String.format("+I[%d, %d]", 10 + i, 1000 + i * 100));
+ }
+ try (CloseableIterator<Row> iter = tEnv.executeSql(query4).collect()) {
+ assertResultsIgnoreOrder(iter, expected4, true);
+ }
+ }
+
+ /** Tests that ALTER TABLE rejects statistics columns on primary key tables. */
+ @Test
+ void testAlterTableStatisticsColumnsOnPkTableShouldFail() {
+ // Create a PK table without statistics columns
+ tEnv.executeSql(
+ "create table pk_stats_reject_test "
+ + "(id int not null, amount bigint, primary key (id) not enforced)");
+
+ // ALTER TABLE to add statistics columns should fail for PK tables
+ assertThatThrownBy(
+ () ->
+ tEnv.executeSql(
+ "alter table pk_stats_reject_test "
+ + "set ('table.statistics.columns' = 'amount')"))
+ .rootCause()
+ .hasMessageContaining(
+ "Statistics columns are not supported for primary key tables");
+ }
+
+ /** Tests that ALTER TABLE rejects invalid statistics columns on log tables. */
+ @Test
+ void testAlterTableInvalidStatisticsColumnsOnLogTableShouldFail() {
+ // Create a PK table without statistics columns
+ tEnv.executeSql(
+ "create table log_stats_reject_test "
+ + "(id int not null, amount bigint, content bytes)");
+
+ // ALTER TABLE to add statistics columns should fail for PK tables
+ assertThatThrownBy(
+ () ->
+ tEnv.executeSql(
+ "alter table log_stats_reject_test "
+ + "set ('table.statistics.columns' = 'content')"))
+ .rootCause()
+ .isInstanceOf(InvalidConfigException.class)
+ .hasMessageContaining(
+ "Column 'content' of type 'BYTES' is not supported for statistics collection.");
+ }
+
private List<String> writeRowsToTwoPartition(TablePath tablePath, Collection<String> partitions)
throws Exception {
List<InternalRow> rows = new ArrayList<>();
@@ -1455,7 +1786,7 @@ abstract class FlinkTableSourceITCase extends AbstractTestBase {
.contains(
"TableSourceScan(table=[[testcatalog, defaultdb, combined_filters_table_in, filter=[OR(=(c, _UTF-16LE'2025'), =(c, _UTF-16LE'2026'))], project=[a, c, d]]], fields=[a, c, d])");
- // test column filter、partition filter and flink runtime filter
+ // test column filter, partition filter and flink runtime filter
org.apache.flink.util.CloseableIterator<Row> rowIter = tEnv.executeSql(query1).collect();
assertResultsIgnoreOrder(rowIter, expectedRowValues, true);
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceReaderTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceReaderTest.java
index bf9792251..556a81e69 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceReaderTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceReaderTest.java
@@ -183,6 +183,7 @@ class FlinkSourceReaderTest extends FlinkTestBase {
sourceOutputType,
context,
null,
+ null,
new FlinkSourceReaderMetrics(context.metricGroup()),
recordEmitter,
lakeSource);
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java
index 59f1116ed..c17051899 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java
@@ -89,8 +89,9 @@ class FlinkSourceSplitReaderTest extends FlinkTestBase {
"id", DataTypes.BIGINT().copy(false)),
DataTypes.FIELD("name", DataTypes.STRING())),
new int[] {1, 0},
- createMockSourceReaderMetrics(),
- null))
+ null,
+ null,
+ createMockSourceReaderMetrics()))
.isInstanceOf(ValidationException.class)
.hasMessage(
"The Flink query schema is not matched to Fluss table schema. \n"
@@ -106,8 +107,9 @@ class FlinkSourceSplitReaderTest extends FlinkTestBase {
DataTypes.FIELD("name2", DataTypes.STRING()),
DataTypes.FIELD("id", DataTypes.BIGINT())),
null,
- createMockSourceReaderMetrics(),
- null))
+ null,
+ null,
+ createMockSourceReaderMetrics()))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("Field name2 does not exist in the row type.");
@@ -119,8 +121,9 @@ class FlinkSourceSplitReaderTest extends FlinkTestBase {
DataTypes.FIELD("name", DataTypes.STRING()),
DataTypes.FIELD("id", DataTypes.BIGINT().copy(false))),
null,
- createMockSourceReaderMetrics(),
- null);
+ null,
+ null,
+ createMockSourceReaderMetrics());
assertThat(flinkSourceSplitReader.getProjectedFields()).isNull();
}
@@ -427,7 +430,7 @@ class FlinkSourceSplitReaderTest extends FlinkTestBase {
private FlinkSourceSplitReader createSplitReader(TablePath tablePath, RowType rowType) {
return new FlinkSourceSplitReader(
- clientConf, tablePath, rowType, null, createMockSourceReaderMetrics(), null);
+ clientConf, tablePath, rowType, null, null, null, createMockSourceReaderMetrics());
}
private FlinkSourceReaderMetrics createMockSourceReaderMetrics() {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
index aaf2460cf..bc64d8ac4 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
@@ -24,7 +24,6 @@ import org.apache.fluss.cluster.rebalance.GoalType;
import org.apache.fluss.cluster.rebalance.ServerTag;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
-import org.apache.fluss.config.StatisticsConfigUtils;
import org.apache.fluss.config.cluster.AlterConfig;
import org.apache.fluss.config.cluster.AlterConfigOpType;
import org.apache.fluss.exception.ApiException;
@@ -460,9 +459,6 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
tableDescriptor =
applySystemDefaults(tableDescriptor, lakeCatalogContainer.getDataLakeFormat());
- // validate statistics configuration
- StatisticsConfigUtils.validateStatisticsConfig(tableDescriptor);
-
// the distribution and bucket count must be set now
//noinspection OptionalGetWithoutIsPresent
int bucketCount = tableDescriptor.getTableDistribution().get().getBucketCount().get();
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
index 5f0d6416c..93a021577 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java
@@ -53,6 +53,7 @@ import java.util.stream.Collectors;
import static org.apache.fluss.config.FlussConfigUtils.TABLE_OPTIONS;
import static org.apache.fluss.config.FlussConfigUtils.isAlterableTableOption;
import static org.apache.fluss.config.FlussConfigUtils.isTableStorageConfig;
+import static org.apache.fluss.config.StatisticsConfigUtils.validateStatisticsConfig;
import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
import static org.apache.fluss.metadata.TableDescriptor.CHANGE_TYPE_COLUMN;
import static org.apache.fluss.metadata.TableDescriptor.COMMIT_TIMESTAMP_COLUMN;
@@ -118,6 +119,7 @@ public class TableDescriptorValidation {
checkTieredLog(tableConf);
checkPartition(tableConf, tableDescriptor.getPartitionKeys(), schema.getRowType());
checkSystemColumns(schema.getRowType());
+ validateStatisticsConfig(tableDescriptor);
}
public static void validateAlterTableProperties(
diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md
index 6f0425645..af9f97a69 100644
--- a/website/docs/engine-flink/options.md
+++ b/website/docs/engine-flink/options.md
@@ -91,6 +91,7 @@ See more details about [ALTER TABLE ... SET](engine-flink/ddl.md#set-properties)
| table.delete.behavior | Enum | ALLOW | Controls the behavior of delete operations on primary key tables. Three modes are supported: `ALLOW` (default for default merge engine) - allows normal delete operations; `IGNORE` - silently ignores delete requests without errors; `DISABLE` - rejects delete requests and throws explicit errors. This configuration provides system-level guarantees for some downstream pipelines (e.g., Flink Delta Joi [...]
| table.changelog.image | Enum | FULL | Defines the changelog image mode for primary key tables. This configuration is inspired by similar settings in database systems like MySQL's `binlog_row_image` and PostgreSQL's `replica identity`. Two modes are supported: `FULL` (default) - produces both UPDATE_BEFORE and UPDATE_AFTER records for update operations, capturing complete information about updates and allowing tracking of previous val [...]
| table.auto-inc.batch-size | Long | 100000L | The batch size of auto-increment IDs fetched from the distributed counter each time. This value determines the length of the locally cached ID segment. Default: 100000. A larger batch size may cause significant auto-increment ID gaps, especially when unused cached ID segments are discarded due to TabletServer restarts or abnormal terminations. Conversely, a smaller batch size increases the freque [...]
+| table.statistics.columns | String | (None) | Configures column-level statistics collection for the table. By default this option is not set and no column statistics are collected. The value `*` means collect statistics for all supported columns. A comma-separated list of column names means collect statistics only for the specified columns (recommended for minimal overhead). Supported types: BOOLEAN, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT [...]
## Read Options
diff --git a/website/docs/engine-flink/reads.md b/website/docs/engine-flink/reads.md
index aa5c15b0b..57d732366 100644
--- a/website/docs/engine-flink/reads.md
+++ b/website/docs/engine-flink/reads.md
@@ -154,6 +154,84 @@ TableSourceScan(table=[[fluss_catalog, fluss, log_partitioned_table, filter=[=(c
This confirms that only partitions matching `c_nationkey = 'US'` will be scanned.
+### Filter Pushdown
+
+Filter pushdown is a server-side optimization for **Log Tables** (non-primary-key tables). When enabled, the server evaluates filter predicates against per-batch column statistics (min, max, null count) and skips entire record batches that cannot contain matching rows. This reduces network I/O and deserialization cost without changing query semantics — Flink still applies the filter on the client side as a safety net.
+
+:::note
+1. Filter pushdown requires the Arrow log format (`'table.log.format' = 'arrow'`), which is enabled by default.
+2. Column statistics must be explicitly enabled via the `table.statistics.columns` table property. Without this configuration, no filters will be pushed down.
+3. Only data written **after** enabling statistics will contain batch-level statistics. Historical data will not benefit from filter pushdown.
+:::
+
+#### Enabling Column Statistics
+
+Set the `table.statistics.columns` property when creating or altering a table. It is recommended to specify only the columns used in your filter conditions to minimize overhead:
+
+```sql title="Flink SQL"
+-- Recommended: collect statistics only for columns used in filter conditions
+CREATE TABLE sensor_data (
+ sensor_id INT NOT NULL,
+ temperature DOUBLE NOT NULL,
+ humidity DOUBLE NOT NULL,
+ location STRING NOT NULL,
+ ts TIMESTAMP NOT NULL
+) WITH (
+ 'table.statistics.columns' = 'temperature,humidity,location'
+);
+
+-- Or use '*' to collect statistics for all supported columns (higher overhead)
+ALTER TABLE sensor_data SET ('table.statistics.columns' = '*');
+```
+
+Statistics collection supports the following types: `BOOLEAN`, `TINYINT`, `SMALLINT`, `INTEGER`, `BIGINT`, `FLOAT`, `DOUBLE`, `STRING`, `CHAR`, `DECIMAL`, `DATE`, `TIME`, `TIMESTAMP`, `TIMESTAMP_LTZ`. Unsupported types (`BYTES`, `BINARY`, `ARRAY`, `MAP`, `ROW`) are automatically excluded.
+
+#### Supported Filter Operators
+
+The following filter operators can be pushed down to the server:
+
+- `=`, `<>`, `>`, `>=`, `<`, `<=`
+- `IN (...)`
+- `IS NULL`, `IS NOT NULL`
+- `BETWEEN ... AND ...`
+- `LIKE 'abc%'` (prefix), `LIKE '%abc'` (suffix), `LIKE '%abc%'` (contains)
+- `AND` / `OR` conjunctions
+
+All columns referenced in a filter expression must have statistics enabled. If any referenced column lacks statistics, that filter will not be pushed down.
+
+#### Example
+
+**1. Create a table with statistics enabled:**
+```sql title="Flink SQL"
+CREATE TABLE sensor_data (
+ sensor_id INT NOT NULL,
+ temperature DOUBLE NOT NULL,
+ humidity DOUBLE NOT NULL,
+ location STRING NOT NULL,
+ ts TIMESTAMP NOT NULL
+) WITH (
+ 'table.statistics.columns' = 'temperature,location'
+);
+```
+
+**2. Query with filter:**
+```sql title="Flink SQL"
+SELECT * FROM sensor_data WHERE temperature > 30.0 AND location = 'warehouse-A';
+```
+
+**3. Verify with `EXPLAIN`:**
+```sql title="Flink SQL"
+EXPLAIN SELECT * FROM sensor_data WHERE temperature > 30.0 AND location = 'warehouse-A';
+```
+
+If filter pushdown is active, the `TableSourceScan` node in the execution plan will contain a `filter=[...]` clause showing the pushed-down predicates. For example:
+
+```text
+TableSourceScan(table=[[..., sensor_data, filter=[and(>(temperature, 30.0:DOUBLE), =(location, ...))]]], fields=[...])
+```
+
+The server evaluates these predicates against per-batch column statistics and skips entire record batches that cannot contain matching rows. Note that the filter also appears in a `Calc` node above the source — this is expected because Flink retains all filters for client-side verification as a safety net.
+
## Batch Read
### Limit Read