This is an automated email from the ASF dual-hosted git repository.
luoyuxia 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 71e6bb6f0 [lake/iceberg] Enhance IcebergSplitPlanner to only include
filter column stats (#3237)
71e6bb6f0 is described below
commit 71e6bb6f0f7120a2b6092a5aac2c947086f94857
Author: Junfan Zhang <[email protected]>
AuthorDate: Sat May 9 15:00:19 2026 +0800
[lake/iceberg] Enhance IcebergSplitPlanner to only include filter column
stats (#3237)
---
.../lake/iceberg/source/IcebergSplitPlanner.java | 60 ++++++++++++++++++-
.../iceberg/source/IcebergSplitPlannerTest.java | 70 ++++++++++++++++++++++
2 files changed, 129 insertions(+), 1 deletion(-)
diff --git
a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlanner.java
b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlanner.java
index 2c308088c..a397a5af0 100644
---
a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlanner.java
+++
b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlanner.java
@@ -18,6 +18,7 @@
package org.apache.fluss.lake.iceberg.source;
+import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.lake.iceberg.utils.IcebergCatalogUtils;
import org.apache.fluss.lake.source.Planner;
@@ -29,7 +30,10 @@ import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableScan;
import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.expressions.BoundPredicate;
import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.ExpressionVisitors;
+import org.apache.iceberg.expressions.UnboundPredicate;
import org.apache.iceberg.io.CloseableIterable;
import javax.annotation.Nullable;
@@ -37,7 +41,9 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.LinkedHashSet;
import java.util.List;
+import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -70,7 +76,11 @@ public class IcebergSplitPlanner implements
Planner<IcebergSplit> {
Function<FileScanTask, Integer> bucketExtractor =
createBucketExtractor(table);
TableScan tableScan = table.newScan().useSnapshot(snapshotId);
if (filter != null) {
- tableScan = tableScan.includeColumnStats().filter(filter);
+ Set<String> filterColumns = referencedColumns(filter);
+ if (!filterColumns.isEmpty()) {
+ tableScan = tableScan.includeColumnStats(filterColumns);
+ }
+ tableScan = tableScan.filter(filter);
}
try (CloseableIterable<FileScanTask> tasks = tableScan.planFiles()) {
tasks.forEach(
@@ -84,6 +94,54 @@ public class IcebergSplitPlanner implements
Planner<IcebergSplit> {
return splits;
}
+ @VisibleForTesting
+ Set<String> referencedColumns(Expression expression) {
+ return ExpressionVisitors.visit(
+ expression,
+ new ExpressionVisitors.ExpressionVisitor<Set<String>>() {
+ @Override
+ public Set<String> alwaysTrue() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<String> alwaysFalse() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<String> not(Set<String> result) {
+ return result;
+ }
+
+ @Override
+ public Set<String> and(Set<String> leftResult, Set<String>
rightResult) {
+ return union(leftResult, rightResult);
+ }
+
+ @Override
+ public Set<String> or(Set<String> leftResult, Set<String>
rightResult) {
+ return union(leftResult, rightResult);
+ }
+
+ @Override
+ public <T> Set<String> predicate(BoundPredicate<T> pred) {
+ return Collections.singleton(pred.ref().name());
+ }
+
+ @Override
+ public <T> Set<String> predicate(UnboundPredicate<T> pred)
{
+ return Collections.singleton(pred.ref().name());
+ }
+ });
+ }
+
+ private Set<String> union(Set<String> left, Set<String> right) {
+ Set<String> result = new LinkedHashSet<>(left);
+ result.addAll(right);
+ return result;
+ }
+
private Function<FileScanTask, Integer> createBucketExtractor(Table table)
{
PartitionSpec partitionSpec = table.spec();
List<PartitionField> partitionFields = partitionSpec.fields();
diff --git
a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlannerTest.java
b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlannerTest.java
index 2697339e6..a083b63f2 100644
---
a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlannerTest.java
+++
b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergSplitPlannerTest.java
@@ -18,15 +18,23 @@
package org.apache.fluss.lake.iceberg.source;
+import org.apache.fluss.config.Configuration;
import org.apache.fluss.lake.source.LakeSource;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.predicate.PredicateBuilder;
+import org.apache.fluss.types.IntType;
+import org.apache.fluss.types.RowType;
+import org.apache.fluss.types.StringType;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -34,6 +42,7 @@ import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.Collections;
import java.util.List;
+import java.util.Set;
import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
@@ -144,4 +153,65 @@ class IcebergSplitPlannerTest extends
IcebergSourceTestBase {
Collections.singletonList("a"),
Collections.singletonList("b"));
}
}
+
+ @Test
+ void testOnlyIncludesFilterColumnStats() throws Exception {
+ TablePath tablePath = TablePath.of(DEFAULT_DB,
"test_filter_column_stats");
+ Schema schema =
+ new Schema(
+ optional(1, "c1", Types.IntegerType.get()),
+ optional(2, "c2", Types.StringType.get()),
+ optional(3, "c3", Types.StringType.get()));
+ PartitionSpec partitionSpec =
PartitionSpec.builderFor(schema).bucket("c1", 2).build();
+ createTable(tablePath, schema, partitionSpec);
+
+ Table table = getTable(tablePath);
+ GenericRecord record = createIcebergRecord(schema, 12, "a", "A");
+ writeRecord(table, Collections.singletonList(record), null, 0);
+ table.refresh();
+ Snapshot snapshot = table.currentSnapshot();
+
+ PredicateBuilder predicateBuilder =
+ new PredicateBuilder(RowType.of(new IntType(), new
StringType(), new StringType()));
+ LakeSource<IcebergSplit> lakeSource =
lakeStorage.createLakeSource(tablePath);
+
lakeSource.withFilters(Collections.singletonList(predicateBuilder.greaterOrEqual(0,
10)));
+ List<IcebergSplit> icebergSplits =
lakeSource.createPlanner(snapshot::snapshotId).plan();
+
+ assertThat(icebergSplits).hasSize(1);
+
assertThat(icebergSplits.get(0).fileScanTask().file().lowerBounds().keySet())
+ .containsOnly(1);
+ }
+
+ @Test
+ void testReferencedColumns() {
+ // Unbound predicates expose the referenced column through their named
reference.
+ assertThat(referencedColumns(Expressions.greaterThanOrEqual("c1",
10))).containsOnly("c1");
+
+ // Compound expressions merge referenced columns and deduplicate
repeated columns.
+ Expression compoundExpression =
+ Expressions.and(
+ Expressions.or(
+ Expressions.greaterThanOrEqual("c1", 10),
+ Expressions.equal("c2", "a")),
+ Expressions.not(Expressions.lessThan("c1", 20)));
+ assertThat(referencedColumns(compoundExpression)).containsOnly("c1",
"c2");
+
+ // Bound predicates expose the resolved field through their bound
reference.
+ Schema schema =
+ new Schema(
+ optional(1, "c1", Types.IntegerType.get()),
+ optional(2, "c2", Types.StringType.get()));
+ Expression boundExpression =
+ Expressions.greaterThanOrEqual("c2",
"a").bind(schema.asStruct(), true);
+ assertThat(referencedColumns(boundExpression)).containsOnly("c2");
+ }
+
+ private Set<String> referencedColumns(Expression expression) {
+ return new IcebergSplitPlanner(
+ new Configuration(),
+ TablePath.of(DEFAULT_DB, "test_referenced_columns"),
+ 0,
+ null)
+ .referencedColumns(expression);
+ }
}
