This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new cde07773c6 [core] Preserve merge correctness when pushing filters into
chain scans (#9979)
cde07773c6 is described below
commit cde07773c6866920a346b93cf12b743e848e1ace
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Sun Sep 20 13:46:05 2026 +0800
[core] Preserve merge correctness when pushing filters into chain scans
(#9979)
---
.../apache/paimon/table/ChainGroupReadTable.java | 25 +-
.../paimon/table/ChainTableFileStoreTableTest.java | 429 +++++++++++++++++++++
2 files changed, 448 insertions(+), 6 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
index c141d9b8e2..5bfeec5129 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
@@ -225,7 +225,7 @@ public class ChainGroupReadTable extends
FallbackReadFileStoreTable {
private final ChainGroupReadTable chainGroupReadTable;
private final RecordComparator chainPartitionComparator;
private final ChainPartitionProjector partitionProjector;
- private Predicate dataPredicate;
+ private Predicate keyPredicate;
private Filter<Integer> bucketFilter;
protected boolean preloadTargetSnapshot = true;
@@ -267,15 +267,28 @@ public class ChainGroupReadTable extends
FallbackReadFileStoreTable {
public ChainTableBatchScan withFilter(Predicate predicate) {
super.withFilter(predicate);
if (predicate == null) {
- dataPredicate = null;
+ keyPredicate = null;
} else {
Pair<Optional<PartitionPredicate>, List<Predicate>> pair =
PartitionPredicate.splitPartitionPredicatesAndDataPredicates(
predicate,
tableSchema.logicalRowType(),
tableSchema.partitionKeys());
- dataPredicate =
- pair.getRight().isEmpty() ? null :
PredicateBuilder.and(pair.getRight());
+ List<String> fieldNames = tableSchema.fieldNames();
+ List<String> primaryKeys = tableSchema.trimmedPrimaryKeys();
+ int[] keyMapping = new int[fieldNames.size()];
+ for (int i = 0; i < keyMapping.length; i++) {
+ keyMapping[i] = primaryKeys.contains(fieldNames.get(i)) ?
i : -1;
+ }
+ // Branch scans are incomplete merge inputs. A value filter
could remove an
+ // update or delete needed to suppress an older matching row
in another branch.
+ // Keep only an inclusive key predicate, retaining the
original row indices.
+ keyPredicate =
+ pair.getRight().isEmpty()
+ ? null
+ : PredicateBuilder.transformFieldMapping(
+
PredicateBuilder.and(pair.getRight()), keyMapping)
+ .orElse(null);
}
return this;
}
@@ -540,8 +553,8 @@ public class ChainGroupReadTable extends
FallbackReadFileStoreTable {
?
chainGroupReadTable.newSnapshotScan(scanCreator)
:
chainGroupReadTable.newDeltaScan(scanCreator))
.withoutAuthPartitionPushdown();
- if (dataPredicate != null) {
- scan.withFilter(dataPredicate);
+ if (keyPredicate != null) {
+ scan.withFilter(keyPredicate);
}
if (bucketFilter != null) {
scan.withBucketFilter(bucketFilter);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
index b65b2e6ec0..a651cfc2c9 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
@@ -29,7 +29,10 @@ import
org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.disk.IOManagerImpl;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.CastTransform;
import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.FieldTransform;
import org.apache.paimon.predicate.Predicate;
@@ -45,6 +48,7 @@ import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.InnerTableWrite;
import org.apache.paimon.table.sink.StreamTableCommit;
import org.apache.paimon.table.source.ChainSplit;
+import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.DataTableScan;
import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.table.source.InnerTableRead;
@@ -65,15 +69,18 @@ import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mockito;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.Random;
import java.util.UUID;
import java.util.function.Consumer;
import java.util.function.Supplier;
@@ -113,6 +120,428 @@ public class ChainTableFileStoreTableTest {
tableName = "chain_t_" + uuid.replace("-", "");
}
+ private static Stream<Arguments> filterPushdownParams() {
+ return Stream.of("parquet", "orc")
+ .flatMap(
+ format ->
+ Stream.of(false, true)
+ .flatMap(
+ dv ->
+ Stream.of(false, true)
+ .map(
+
keyRange ->
+
Arguments.of(
+
format, dv,
+
keyRange))));
+ }
+
+ @ParameterizedTest
+ @MethodSource("filterPushdownParams")
+ public void testFilterPushdownAcrossChainMerge(
+ String format, boolean deletionVectors, boolean keyRangeSplit)
throws Exception {
+ createChainTable(
+ options -> {
+ options.set(CoreOptions.FILE_FORMAT, format);
+ options.set(DELETION_VECTORS_ENABLED, deletionVectors);
+
options.set(CoreOptions.CHAIN_TABLE_KEY_RANGE_SPLIT_ENABLED, keyRangeSplit);
+ options.set(CoreOptions.SOURCE_SPLIT_TARGET_SIZE, new
MemorySize(1));
+ });
+ FileStoreTable table = loadTable();
+ FileStoreTable snapshot = table.switchToBranch(SNAPSHOT_BRANCH);
+ FileStoreTable delta = table.switchToBranch(DELTA_BRANCH);
+ // Separate files let branch-local stats discard individual versions
independently.
+ writeWithCommit(snapshot, row(1L, 1L, "old", "CN", "20250810", "20"));
+ writeWithCommit(snapshot, row(2L, 1L, "new", "CN", "20250810", "20"));
+ if (!deletionVectors) {
+ writeWithCommit(snapshot, row(3L, 1L, "old", "CN", "20250810",
"20"));
+ }
+ writeWithCommit(snapshot, row(4L, 1L, "old", "CN", "20250810", "20"));
+ writeWithCommit(snapshot, row(5L, 1L, null, "CN", "20250810", "20"));
+ writeWithCommit(snapshot, row(6L, 1L, "old", "CN", "20250810", "20"));
+ writeWithCommit(snapshot, row(1L, 1L, "other-group", "US", "20250810",
"20"));
+ writeWithCommit(delta, row(1L, 2L, "new", "CN", "20250810", "21"));
+ writeWithCommit(delta, row(2L, 2L, "old", "CN", "20250810", "21"));
+ if (!deletionVectors) {
+ writeWithCommit(delta, row(RowKind.DELETE, 3L, 2L, "new", "CN",
"20250810", "21"));
+ }
+ writeWithCommit(delta, row(4L, 2L, null, "CN", "20250810", "21"));
+ writeWithCommit(delta, row(5L, 2L, "old", "CN", "20250810", "21"));
+ writeWithCommit(delta, row(6L, 2L, "new", "CN", "20250810", "21"));
+ writeWithCommit(delta, row(7L, 1L, "old", "CN", "20250810", "21"));
+ writeWithCommit(delta, row(6L, 3L, "latest", "CN", "20250810", "22"));
+ writeWithCommit(delta, row(1L, 2L, "other-group-new", "US",
"20250810", "22"));
+ table = loadTable();
+ Map<String, String> partition =
+ ImmutableMap.of("region", "CN", "dt", "20250810", "hour",
"22");
+ List<GenericRow> baseline = getResult(table, partition);
+ assertThat(baseline)
+ .containsExactlyInAnyOrder(
+ row(1L, 2L, "new", "CN", "20250810", "22"),
+ row(2L, 2L, "old", "CN", "20250810", "22"),
+ row(4L, 2L, null, "CN", "20250810", "22"),
+ row(5L, 2L, "old", "CN", "20250810", "22"),
+ row(6L, 3L, "latest", "CN", "20250810", "22"),
+ row(7L, 1L, "old", "CN", "20250810", "22"));
+ PredicateBuilder builder = new PredicateBuilder(table.rowType());
+ Predicate old = builder.equal(2, BinaryString.fromString("old"));
+ Predicate newer = builder.equal(2, BinaryString.fromString("new"));
+ for (Predicate filter :
+ Arrays.asList(
+ old,
+ newer,
+ builder.isNull(2),
+ builder.isNotNull(2),
+ old.negate().get(),
+ builder.greaterThan(1, 1L),
+ builder.equal(0, 1L),
+ builder.equal(
+ new CastTransform(
+ new FieldRef(0, "k",
DataTypes.BIGINT()),
+ DataTypes.STRING()),
+ BinaryString.fromString("1")),
+ PredicateBuilder.and(builder.equal(0, 1L), old),
+ PredicateBuilder.or(builder.equal(0, 1L), old),
+ PredicateBuilder.or(builder.equal(0, 1L),
builder.equal(0, 6L)),
+ PredicateBuilder.or(
+ PredicateBuilder.and(builder.equal(0, 1L),
old),
+ PredicateBuilder.and(builder.equal(0, 6L),
newer)),
+ builder.equal(2, BinaryString.fromString("absent")),
+ PredicateBuilder.alwaysTrue(),
+ PredicateBuilder.alwaysFalse())) {
+ assertFilterMatchesMergedRows(table, partition, filter, baseline);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"parquet", "orc"})
+ public void testPartialUpdateFilterPushdownAcrossChainMerge(String format)
throws Exception {
+ createPartialUpdateChainTable(
+ options -> {
+ options.set(CoreOptions.FILE_FORMAT, format);
+ options.set(PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE, true);
+ });
+ FileStoreTable table = loadTable();
+ FileStoreTable snapshot = table.switchToBranch(SNAPSHOT_BRANCH);
+ FileStoreTable delta = table.switchToBranch(DELTA_BRANCH);
+ writeWithCommit(snapshot, row(1L, 1L, "a", null, "20250810"));
+ writeWithCommit(snapshot, row(2L, 1L, "a", "old", "20250810"));
+ writeWithCommit(delta, row(1L, 2L, null, "b", "20250811"));
+ writeWithCommit(delta, row(RowKind.DELETE, 2L, 2L, null, null,
"20250811"));
+ table = loadTable();
+ Map<String, String> partition = ImmutableMap.of("dt", "20250811");
+ List<GenericRow> baseline = getResult(table, partition);
+ assertThat(baseline).containsExactly(row(1L, 2L, "a", "b",
"20250811"));
+ PredicateBuilder builder = new PredicateBuilder(table.rowType());
+ Predicate a = builder.equal(2, BinaryString.fromString("a"));
+ Predicate b = builder.equal(3, BinaryString.fromString("b"));
+ for (Predicate filter :
+ Arrays.asList(
+ a,
+ b,
+ PredicateBuilder.and(a, b),
+ PredicateBuilder.or(a, b),
+ builder.isNull(3),
+ builder.equal(0, 2L),
+ PredicateBuilder.and(builder.equal(0, 1L), a))) {
+ assertFilterMatchesMergedRows(table, partition, filter, baseline);
+ }
+ }
+
+ @Test
+ public void testChainFilterPreservesKeyAndCompleteSnapshotPruning() throws
Exception {
+ createChainTable(options -> options.set(BUCKET, 2));
+ FileStoreTable table = loadTable();
+ FileStoreTable snapshot = table.switchToBranch(SNAPSHOT_BRANCH);
+ FileStoreTable delta = table.switchToBranch(DELTA_BRANCH);
+ for (long key = 1; key <= 4; key++) {
+ writeWithCommit(snapshot, row(key, 1L, "old", "CN", "20250810",
"20"));
+ writeWithCommit(delta, row(key, 2L, "new", "CN", "20250810",
"21"));
+ }
+ table = loadTable();
+ Map<String, String> partition =
+ ImmutableMap.of("region", "CN", "dt", "20250810", "hour",
"21");
+ PredicateBuilder builder = new PredicateBuilder(table.rowType());
+ Predicate key = builder.equal(0, 1L);
+ Predicate old = builder.equal(2, BinaryString.fromString("old"));
+ long allBytes = plannedFileBytes(table, partition,
PredicateBuilder.alwaysTrue());
+ long keyBytes = plannedFileBytes(table, partition, key);
+ assertThat(keyBytes).isPositive().isLessThan(allBytes);
+ assertThat(plannedFileBytes(table, partition,
PredicateBuilder.and(key, old)))
+ .isEqualTo(keyBytes);
+ Predicate keyOr = PredicateBuilder.or(key, builder.equal(0, 2L));
+ Predicate mixedOr =
+ PredicateBuilder.or(
+ PredicateBuilder.and(key, old),
+ PredicateBuilder.and(builder.equal(0, 2L),
builder.isNull(2)));
+ assertThat(plannedFileBytes(table, partition, mixedOr))
+ .isEqualTo(plannedFileBytes(table, partition, keyOr));
+ assertThat(plannedFileBytes(table, partition,
mixedOr)).isLessThan(allBytes);
+ assertThat(plannedFileBytes(table, partition, PredicateBuilder.or(key,
old)))
+ .isEqualTo(allBytes);
+ Map<String, String> complete =
+ ImmutableMap.of("region", "CN", "dt", "20250810", "hour",
"20");
+ assertThat(plannedFileBytes(table, complete, old)).isPositive();
+ assertThat(
+ plannedFileBytes(
+ table,
+ complete,
+ builder.equal(2,
BinaryString.fromString("absent"))))
+ .isZero();
+ assertThat(
+ table.newReadBuilder()
+ .withPartitionFilter(partition)
+ .withFilter(PredicateBuilder.and(key, old))
+ .withBucketFilter(bucket -> false)
+ .newScan()
+ .plan()
+ .splits())
+ .isEmpty();
+ long bucketBytes = 0;
+ for (int bucket = 0; bucket < 2; bucket++) {
+ final int selectedBucket = bucket;
+ TableScan.Plan plan =
+ table.newReadBuilder()
+ .withPartitionFilter(partition)
+ .withFilter(PredicateBuilder.alwaysTrue())
+ .withBucketFilter(candidate -> candidate ==
selectedBucket)
+ .newScan()
+ .plan();
+ long bytes =
+ plan.splits().stream()
+ .flatMap(this::plannedDataFiles)
+ .mapToLong(file -> file.fileSize())
+ .sum();
+ assertThat(bytes).isPositive().isLessThan(allBytes);
+ bucketBytes += bytes;
+ }
+ assertThat(bucketBytes).isEqualTo(allBytes);
+ }
+
+ @Test
+ public void testChainKeyFilterAfterSchemaReorder() throws Exception {
+ createChainTable(options -> {});
+ writeWithCommit(
+ loadTable().switchToBranch(SNAPSHOT_BRANCH),
+ row(9L, 1L, "old", "CN", "20250810", "20"));
+ Path path = new Path(tempDir.toUri().toString(), tableName);
+ for (String branch : Arrays.asList(DEFAULT_MAIN_BRANCH,
SNAPSHOT_BRANCH, DELTA_BRANCH)) {
+ new FileSystemSchemaManager(LocalFileIO.create(), path, branch)
+
.commitChanges(SchemaChange.updateColumnPosition(SchemaChange.Move.last("k")));
+ }
+ FileStoreTable table = loadTable();
+ writeWithCommit(
+ table.switchToBranch(DELTA_BRANCH), row(2L, "new", "CN",
"20250810", "21", 9L));
+ table = loadTable();
+ assertThat(table.rowType().getFieldIndex("k")).isEqualTo(5);
+ Map<String, String> partition =
+ ImmutableMap.of("region", "CN", "dt", "20250810", "hour",
"21");
+ List<GenericRow> baseline = getResult(table, partition);
+ assertThat(baseline).containsExactly(row(2L, "new", "CN", "20250810",
"21", 9L));
+ PredicateBuilder builder = new PredicateBuilder(table.rowType());
+ assertFilterMatchesMergedRows(table, partition, builder.equal(5, 9L),
baseline);
+ assertFilterMatchesMergedRows(
+ table,
+ partition,
+ PredicateBuilder.and(
+ builder.equal(5, 9L), builder.equal(1,
BinaryString.fromString("old"))),
+ baseline);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"parquet", "orc"})
+ public void testFilterPushdownWithoutSnapshotAnchor(String format) throws
Exception {
+ createChainTable(options -> options.set(CoreOptions.FILE_FORMAT,
format));
+ FileStoreTable delta = loadTable().switchToBranch(DELTA_BRANCH);
+ writeWithCommit(delta, row(1L, 1L, "old", "CN", "20250810", "20"));
+ writeWithCommit(delta, row(2L, 1L, "old", "CN", "20250810", "20"));
+ writeWithCommit(delta, row(1L, 2L, "new", "CN", "20250810", "21"));
+ writeWithCommit(delta, row(RowKind.DELETE, 2L, 2L, "new", "CN",
"20250810", "21"));
+ FileStoreTable table = loadTable();
+ Map<String, String> partition =
+ ImmutableMap.of("region", "CN", "dt", "20250810", "hour",
"21");
+ List<GenericRow> baseline = getResult(table, partition);
+ assertThat(baseline).containsExactly(row(1L, 2L, "new", "CN",
"20250810", "21"));
+ PredicateBuilder builder = new PredicateBuilder(table.rowType());
+ for (Predicate filter :
+ Arrays.asList(
+ builder.equal(2, BinaryString.fromString("old")),
+ builder.equal(2, BinaryString.fromString("new")),
+ builder.equal(0, 2L),
+ PredicateBuilder.and(builder.equal(0, 1L),
builder.greaterThan(1, 1L)))) {
+ assertFilterMatchesMergedRows(table, partition, filter, baseline);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"parquet", "orc"})
+ public void testFilterPushdownAgainstVersionHistory(String format) throws
Exception {
+ createChainTable(options -> options.set(CoreOptions.FILE_FORMAT,
format));
+ FileStoreTable table = loadTable();
+ Map<Long, GenericRow> latest = new HashMap<>();
+ for (long key = 0; key < 24; key++) {
+ String value = key % 4 == 0 ? null : "v" + key % 3;
+ writeWithCommit(
+ table.switchToBranch(SNAPSHOT_BRANCH),
+ row(key, 0L, value, "CN", "20250810", "20"));
+ latest.put(key, row(key, 0L, value, "CN", "20250810", "23"));
+ }
+ Random random = new Random(20260919L);
+ FileStoreTable delta = table.switchToBranch(DELTA_BRANCH);
+ for (int version = 1; version <= 3; version++) {
+ List<GenericRow> updates = new ArrayList<>();
+ for (long key = 0; key < 28; key++) {
+ if (random.nextInt(3) == 0) {
+ continue;
+ }
+ boolean delete = random.nextInt(5) == 0;
+ String value = random.nextInt(4) == 0 ? null : "v" +
random.nextInt(3);
+ updates.add(
+ row(
+ delete ? RowKind.DELETE : RowKind.INSERT,
+ key,
+ (long) version,
+ value,
+ "CN",
+ "20250810",
+ Integer.toString(20 + version)));
+ if (delete) {
+ latest.remove(key);
+ } else {
+ latest.put(key, row(key, (long) version, value, "CN",
"20250810", "23"));
+ }
+ }
+ writeWithCommit(delta, updates.toArray(new GenericRow[0]));
+ }
+ table = loadTable();
+ Map<String, String> partition =
+ ImmutableMap.of("region", "CN", "dt", "20250810", "hour",
"23");
+ List<GenericRow> expectedRows = new ArrayList<>(latest.values());
+ assertThat(getResult(table,
partition)).containsExactlyInAnyOrderElementsOf(expectedRows);
+ PredicateBuilder builder = new PredicateBuilder(table.rowType());
+ for (int i = 0; i < 24; i++) {
+ Predicate key = builder.greaterOrEqual(0, (long)
random.nextInt(28));
+ Predicate value = builder.equal(2, BinaryString.fromString("v" +
random.nextInt(3)));
+ Predicate predicate;
+ switch (i % 4) {
+ case 0:
+ predicate = PredicateBuilder.and(key, value);
+ break;
+ case 1:
+ predicate = PredicateBuilder.or(key, value);
+ break;
+ case 2:
+ predicate =
+ PredicateBuilder.or(
+ PredicateBuilder.and(key,
builder.isNull(2)),
+ PredicateBuilder.and(builder.lessThan(0,
12L), value));
+ break;
+ default:
+ predicate = PredicateBuilder.and(key,
value.negate().get());
+ }
+ assertFilterMatchesMergedRows(table, partition, predicate,
expectedRows);
+ }
+ }
+
+ @Test
+ public void testFilterCombiningLogicalPartitionsAndValues() throws
Exception {
+ createChainTable(options -> {});
+ FileStoreTable table = loadTable();
+ FileStoreTable snapshot = table.switchToBranch(SNAPSHOT_BRANCH);
+ FileStoreTable delta = table.switchToBranch(DELTA_BRANCH);
+ writeWithCommit(snapshot, row(1L, 1L, "old", "CN", "20250810", "20"));
+ writeWithCommit(snapshot, row(2L, 1L, "old", "CN", "20250810", "20"));
+ writeWithCommit(delta, row(1L, 2L, "new", "CN", "20250810", "21"));
+ writeWithCommit(delta, row(1L, 3L, "latest", "CN", "20250810", "22"));
+ table = loadTable();
+ List<GenericRow> baseline = getResult(table, null);
+ assertThat(baseline)
+ .containsExactlyInAnyOrder(
+ row(1L, 1L, "old", "CN", "20250810", "20"),
+ row(2L, 1L, "old", "CN", "20250810", "20"),
+ row(1L, 2L, "new", "CN", "20250810", "21"),
+ row(2L, 1L, "old", "CN", "20250810", "21"),
+ row(1L, 3L, "latest", "CN", "20250810", "22"),
+ row(2L, 1L, "old", "CN", "20250810", "22"));
+ PredicateBuilder builder = new PredicateBuilder(table.rowType());
+ Predicate filter =
+ PredicateBuilder.or(
+ PredicateBuilder.and(
+ builder.equal(5,
BinaryString.fromString("22")),
+ builder.equal(0, 1L)),
+ PredicateBuilder.and(
+ builder.equal(5,
BinaryString.fromString("21")),
+ builder.equal(2,
BinaryString.fromString("old"))));
+ assertFilterMatchesMergedRows(table, null, filter, baseline);
+ }
+
+ @Test
+ public void testKeyFilterRespectsQueryAuthMask() throws Exception {
+ createChainTableWithQueryAuth();
+ Transform mask = new FieldTransform(new FieldRef(1, "seq",
DataTypes.BIGINT()));
+ FileStoreTable table =
+ loadTable(
+ queryAuthEnvironment(
+ () ->
+ new TableQueryAuthResult(
+ null,
+ Collections.singletonMap(
+ "k",
JsonSerdeUtil.toFlatJson(mask)))));
+ List<GenericRow> baseline = getResult(table, QUERIED_PARTITION);
+ assertThat(baseline)
+ .containsExactlyInAnyOrder(
+ row(2L, 2L, "1-1", "CN", "20250810", "21"),
+ row(1L, 1L, "2", "CN", "20250810", "21"),
+ row(1L, 1L, "4", "CN", "20250810", "21"));
+ PredicateBuilder builder = new PredicateBuilder(table.rowType());
+ assertFilterMatchesMergedRows(table, QUERIED_PARTITION,
builder.equal(0, 1L), baseline);
+ }
+
+ private void assertFilterMatchesMergedRows(
+ FileStoreTable table,
+ Map<String, String> partition,
+ Predicate filter,
+ List<GenericRow> mergedRows)
+ throws Exception {
+ List<GenericRow> expected =
+
mergedRows.stream().filter(filter::test).collect(Collectors.toList());
+ ReadBuilder readBuilder =
+
table.newReadBuilder().withPartitionFilter(partition).withFilter(filter);
+ List<GenericRow> actual = new ArrayList<>();
+ InternalRowSerializer serializer = new
InternalRowSerializer(table.rowType());
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(
+ record -> {
+ // Keep the residual: extra rows are allowed by
pushdown, missing versions
+ // are not.
+ if (filter.test(record)) {
+ actual.add((GenericRow) serializer.copy(record));
+ }
+ });
+ }
+ assertThat(actual)
+ .as("Pushed predicate %s", filter)
+ .containsExactlyInAnyOrderElementsOf(expected);
+ }
+
+ private Stream<DataFileMeta> plannedDataFiles(Split split) {
+ if (split instanceof FallbackReadFileStoreTable.FallbackSplit) {
+ return
plannedDataFiles(((FallbackReadFileStoreTable.FallbackSplit) split).wrapped());
+ }
+ return split instanceof ChainSplit
+ ? ((ChainSplit) split).dataFiles().stream()
+ : ((DataSplit) split).dataFiles().stream();
+ }
+
+ private long plannedFileBytes(
+ FileStoreTable table, Map<String, String> partition, Predicate
filter) {
+ return
table.newReadBuilder().withPartitionFilter(partition).withFilter(filter).newScan()
+ .plan().splits().stream()
+ .flatMap(this::plannedDataFiles)
+ .mapToLong(file -> file.fileSize())
+ .sum();
+ }
+
@Test
public void testChainTableRejectsAggregate() throws Exception {
assertThatThrownBy(
