You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/04/17 13:36:06 UTC
(paimon) branch master updated: [core] Do not pushdown limit when non partition filter is present (#7665)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 6f6dcc9739 [core] Do not pushdown limit when non partition filter is present (#7665)
6f6dcc9739 is described below
commit 6f6dcc9739d3758d453065dab5df20feb7f4e914
Author: XiaoHongbo <xi...@alibaba-inc.com>
AuthorDate: Fri Apr 17 21:36:00 2026 +0800
[core] Do not pushdown limit when non partition filter is present (#7665)
If limit pushdown is applied before non-partition filters are evaluated,
scanning may stop too early and return fewer rows than requested. To fix
this, limit pushdown is disabled when non-partition filters exist.
---
.../paimon/operation/AppendOnlyFileStoreScan.java | 6 +-
.../paimon/operation/KeyValueFileStoreScan.java | 6 +-
.../paimon/table/source/DataTableBatchScan.java | 2 +-
.../table/source/snapshot/SnapshotReader.java | 3 +
.../table/source/snapshot/SnapshotReaderImpl.java | 7 ++
.../apache/paimon/table/system/AuditLogTable.java | 5 ++
.../operation/KeyValueFileStoreScanTest.java | 67 ++++++++++++++--
.../apache/paimon/table/source/TableScanTest.java | 88 +++++++++++++++++++++-
8 files changed, 169 insertions(+), 15 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java
index 2714e773a2..7a0cdec312 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java
@@ -99,7 +99,11 @@ public class AppendOnlyFileStoreScan extends AbstractFileStoreScan {
public Iterator<ManifestEntry> readManifestEntries(
List<ManifestFileMeta> manifestFiles, boolean useSequential) {
Iterator<ManifestEntry> result = super.readManifestEntries(manifestFiles, useSequential);
- if (limit == null || limit <= 0 || deletionVectorsEnabled || dataEvolutionEnabled) {
+ if (limit == null
+ || limit <= 0
+ || deletionVectorsEnabled
+ || dataEvolutionEnabled
+ || inputFilter != null) {
return result;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreScan.java
index 8595753cf0..a3623c1b90 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreScan.java
@@ -225,7 +225,11 @@ public class KeyValueFileStoreScan extends AbstractFileStoreScan {
return false;
}
- return mergeEngine != PARTIAL_UPDATE && mergeEngine != AGGREGATE && !deletionVectorsEnabled;
+ return mergeEngine != PARTIAL_UPDATE
+ && mergeEngine != AGGREGATE
+ && !deletionVectorsEnabled
+ && valueFilter == null
+ && keyFilter == null;
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java
index 4e0416f86e..255ff5f672 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java
@@ -126,7 +126,7 @@ public class DataTableBatchScan extends AbstractDataTableScan {
}
private Optional<StartingScanner.Result> applyPushDownLimit() {
- if (pushDownLimit == null) {
+ if (pushDownLimit == null || snapshotReader.hasNonPartitionFilter()) {
return Optional.empty();
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
index f837daaaec..1ea0854eea 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
@@ -121,6 +121,9 @@ public interface SnapshotReader {
SnapshotReader withLimit(int limit);
+ /** Whether the pushed filter still contains non-partition predicates. */
+ boolean hasNonPartitionFilter();
+
/** Get splits plan from snapshot. */
Plan read();
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
index 67d0b9a566..a61f43b4eb 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
@@ -102,6 +102,7 @@ public class SnapshotReaderImpl implements SnapshotReader {
@Nullable private final DVMetaCache dvMetaCache;
private ScanMode scanMode = ScanMode.ALL;
+ private boolean hasNonPartitionFilter;
private RecordComparator lazyPartitionComparator;
private CacheMetrics dvMetaCacheMetrics;
@@ -239,6 +240,7 @@ public class SnapshotReaderImpl implements SnapshotReader {
scan.withPartitionFilter(pair.getLeft().get());
}
if (!pair.getRight().isEmpty()) {
+ this.hasNonPartitionFilter = true;
nonPartitionFilterConsumer.accept(scan, PredicateBuilder.and(pair.getRight()));
}
scan.withCompleteFilter(predicate);
@@ -338,6 +340,11 @@ public class SnapshotReaderImpl implements SnapshotReader {
return this;
}
+ @Override
+ public boolean hasNonPartitionFilter() {
+ return hasNonPartitionFilter;
+ }
+
@Override
public SnapshotReader dropStats() {
scan.dropStats();
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
index 980183d801..1e912d4068 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
@@ -477,6 +477,11 @@ public class AuditLogTable implements DataTable, ReadonlyTable {
return this;
}
+ @Override
+ public boolean hasNonPartitionFilter() {
+ return wrapped.hasNonPartitionFilter();
+ }
+
@Override
public Plan read() {
return wrapped.read();
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/KeyValueFileStoreScanTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/KeyValueFileStoreScanTest.java
index 9a9c290304..ac6825ed8c 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/KeyValueFileStoreScanTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/KeyValueFileStoreScanTest.java
@@ -28,6 +28,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.mergetree.compact.DeduplicateMergeFunction;
+import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
@@ -331,20 +332,28 @@ public class KeyValueFileStoreScanTest {
@Test
public void testLimitPushdownWithKeyFilter() throws Exception {
- // Write data with different shop IDs
List<KeyValue> data = generateData(200);
Snapshot snapshot = writeData(data);
- // With keyFilter, limit pushdown should still work (keyFilter doesn't affect limit
- // pushdown)
+ Predicate keyPredicate =
+ new PredicateBuilder(RowType.of(new IntType(false)))
+ .equal(0, data.get(0).key().getInt(0));
+
+ // baseline: keyFilter without limit
+ KeyValueFileStoreScan scanFilterOnly = store.newScan();
+ scanFilterOnly.withSnapshot(snapshot.id());
+ scanFilterOnly.withKeyFilter(keyPredicate);
+ int filteredFiles = scanFilterOnly.plan().files().size();
+ assertThat(filteredFiles).isGreaterThan(0);
+
+ // keyFilter + limit: early-stop by rowCount() is unsafe, should be disabled
KeyValueFileStoreScan scan = store.newScan();
scan.withSnapshot(snapshot.id());
- scan.withKeyFilter(
- new PredicateBuilder(RowType.of(new IntType(false)))
- .equal(0, data.get(0).key().getInt(0)));
+ scan.withKeyFilter(keyPredicate);
scan.withLimit(5);
- List<ManifestEntry> files = scan.plan().files();
- assertThat(files.size()).isGreaterThan(0);
+
+ assertThat(scan.limitPushdownEnabled()).isFalse();
+ assertThat(scan.plan().files().size()).isEqualTo(filteredFiles);
}
@Test
@@ -569,6 +578,48 @@ public class KeyValueFileStoreScanTest {
return store.toKvMap(actualKvs);
}
+ @Test
+ void testLimitPushdownWithFilter() throws Exception {
+ int numFiles = 10;
+ int rowsPerFile = 100;
+
+ Snapshot snapshot = null;
+ for (int bucket = 0; bucket < numFiles; bucket++) {
+ List<KeyValue> data = new ArrayList<>();
+ for (int i = 0; i < rowsPerFile; i++) {
+ data.add(gen.nextInsert("", 0, (long) i, null, null));
+ }
+ snapshot = writeData(data, bucket);
+ }
+
+ KeyValueFileStoreScan scanAll = store.newScan();
+ scanAll.withSnapshot(snapshot.id());
+ List<ManifestEntry> allFiles = scanAll.plan().files();
+ assertThat(allFiles.size()).isEqualTo(numFiles);
+
+ KeyValueFileStoreScan scanFilterOnly = store.newScan();
+ scanFilterOnly.withSnapshot(snapshot.id());
+ scanFilterOnly.withValueFilter(
+ new PredicateBuilder(TestKeyValueGenerator.DEFAULT_ROW_TYPE).equal(4, 50L));
+ List<ManifestEntry> filteredFiles = scanFilterOnly.plan().files();
+ assertThat(filteredFiles.size()).isEqualTo(numFiles); // no file eliminated by stats
+
+ KeyValueFileStoreScan scanWithLimit = store.newScan();
+ scanWithLimit.withSnapshot(snapshot.id());
+ scanWithLimit.withValueFilter(
+ new PredicateBuilder(TestKeyValueGenerator.DEFAULT_ROW_TYPE).equal(4, 50L));
+ scanWithLimit.withLimit(5);
+
+ assertThat(scanWithLimit.limitPushdownEnabled()).isFalse();
+
+ List<ManifestEntry> limitedFiles = scanWithLimit.plan().files();
+
+ assertThat(limitedFiles.size())
+ .as(
+ "When filter is present, limit pushdown should be disabled, returning all files")
+ .isEqualTo(numFiles);
+ }
+
private List<KeyValue> generateData(int numRecords) {
List<KeyValue> data = new ArrayList<>();
for (int i = 0; i < numRecords; i++) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java
index 32f6041307..90cfb7be1a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java
@@ -146,10 +146,90 @@ public class TableScanTest extends ScannerTestBase {
table.newScan().withFilter(filter).withLimit(10).plan();
int splitsWithFilterAndLimit = planWithFilterAndLimit.splits().size();
- // Should read exactly 10 files (from index 25 to 34) to get 10 rows
- assertThat(splitsWithFilterAndLimit).isLessThanOrEqualTo(10);
- assertThat(splitsWithFilterAndLimit).isGreaterThan(0);
- assertThat(splitsWithFilterAndLimit).isLessThan(totalSplits);
+ // With non-partition filter, limit pushdown should be disabled to avoid
+ // returning insufficient rows. All 25 matching files should be returned.
+ assertThat(splitsWithFilterAndLimit).isEqualTo(25);
+
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ void testLimitPushdownWithNonPartitionFilter() throws Exception {
+ createAppendOnlyTable();
+
+ StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ int filesCount = 10;
+ int rowsPerFile = 100;
+ int filterValue = 50;
+
+ for (int fileIdx = 0; fileIdx < filesCount; fileIdx++) {
+ for (int i = 0; i < rowsPerFile; i++) {
+ write.write(rowData(fileIdx, i, (long) (fileIdx * rowsPerFile + i)));
+ }
+ commit.commit(fileIdx, write.prepareCommit(true, fileIdx));
+ }
+
+ TableScan.Plan planAll = table.newScan().plan();
+ assertThat(planAll.splits().size()).isEqualTo(filesCount);
+
+ Predicate filter =
+ new PredicateBuilder(table.schema().logicalRowType()).equal(1, filterValue);
+ TableScan.Plan planFilterOnly = table.newScan().withFilter(filter).plan();
+ assertThat(planFilterOnly.splits().size()).isEqualTo(filesCount);
+
+ List<String> allRows = getResult(table.newRead(), planFilterOnly.splits());
+ long totalMatchingRows =
+ allRows.stream().filter(r -> r.contains("|" + filterValue + "|")).count();
+ assertThat(totalMatchingRows).isEqualTo(filesCount);
+
+ int limit = 5;
+ TableScan.Plan planWithFilterAndLimit =
+ table.newScan().withFilter(filter).withLimit(limit).plan();
+
+ List<String> limitedAllRows = getResult(table.newRead(), planWithFilterAndLimit.splits());
+ long limitedMatchingRows =
+ limitedAllRows.stream().filter(r -> r.contains("|" + filterValue + "|")).count();
+
+ assertThat(limitedMatchingRows)
+ .as(
+ "Filter+limit bug: scan returned %d splits, but only %d rows match "
+ + "filter (expected >= %d). Total matching = %d",
+ planWithFilterAndLimit.splits().size(),
+ limitedMatchingRows,
+ limit,
+ totalMatchingRows)
+ .isGreaterThanOrEqualTo(limit);
+
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ void testLimitPushdownWithPartitionFilter() throws Exception {
+ createAppendOnlyTable();
+
+ StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+
+ for (int i = 0; i < 10; i++) {
+ write.write(rowData(i, i, (long) i * 100));
+ commit.commit(i, write.prepareCommit(true, i));
+ }
+
+ Predicate partitionFilter =
+ new PredicateBuilder(table.schema().logicalRowType()).lessOrEqual(0, 4);
+
+ TableScan.Plan planNoLimit = table.newScan().withFilter(partitionFilter).plan();
+ assertThat(planNoLimit.splits().size()).isEqualTo(5);
+
+ TableScan.Plan plan = table.newScan().withFilter(partitionFilter).withLimit(2).plan();
+
+ assertThat(plan.splits().size())
+ .as("Partition filter + limit: limit pushdown should not be disabled")
+ .isEqualTo(2);
write.close();
commit.close();