This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 6e887178b99 [fix](paimon) Push limit into split planning (#66818)
6e887178b99 is described below
commit 6e887178b9945b6528ed9f074069f0b485db271e
Author: Gabriel <[email protected]>
AuthorDate: Thu Aug 20 14:16:17 2026 +0800
[fix](paimon) Push limit into split planning (#66818)
### What problem does this PR solve?
Paimon scan planning ignored the limit carried by the Doris connector
request. As a result, even a small unfiltered query such as `SELECT *
FROM t LIMIT 1` could still scan all relevant manifests, enumerate every
file split, and create scan ranges for the full table before Doris
applied the limit during execution.
For tables with many files, this causes unnecessary FE planning latency,
memory usage, and scheduling overhead. However, pushing a limit into
Paimon indiscriminately is unsafe because some planning statistics do
not represent the number of rows that Doris will finally return.
### What is changed and how does it work?
This PR forwards the connector request limit into Paimon scan planning.
For eligible scans, it calls `ReadBuilder.withLimit` before `newScan()`,
allowing Paimon to stop manifest and split planning after it has covered
enough rows.
Limit pushdown is enabled only when all of the following conditions
hold:
- The limit is positive and fits in the integer range supported by the
Paimon API.
- The original request has no filter.
- The scan is an append-only `FileStoreTable` without primary keys.
- The table is not backed by a fallback reader.
- The scan does not use the file-creation-time planning path.
- Neither native nor JNI split routing is configured to be ignored.
The change also retains the resolved source table behind system-table
wrappers so that fallback-backed scans can still be detected after
handles are reloaded or decorated.
### Why are the safety guards required?
- **Filtered scans:** Paimon 1.3.1 can stop planning based on pre-filter
split row counts. A retained split may produce no matching rows while a
later matching split has already been pruned.
- **Primary-key tables:** File metadata can count deleted rows or
tombstones that the execution reader later removes, so planned row
counts may exceed final output rows.
- **Format tables:** Their limit accounting can count files rather than
output rows. An empty first file could satisfy the planning limit and
hide a later non-empty file.
- **Fallback-backed tables:** The Paimon 1.3.1 fallback planner can
derive partition ownership from an already limited main-table plan and
expose stale fallback data.
- **File-creation-time scans:** This path creates a separate snapshot
reader and does not consume the limit-bearing `TableScan`.
- **Ignored split types:** Native/JNI routing happens after planning.
Applying the limit first could retain only a split that Doris
subsequently discards, hiding rows from the other reader path.
Unsafe or uncertain modes therefore keep the complete split plan instead
of trading correctness for pruning.
### User-visible effect
Small, unfiltered LIMIT queries on append-only Paimon FileStore tables
can plan fewer manifests and splits. SQL semantics and the
execution-stage limit are unchanged; this optimization only reduces
planning work where split accounting is known to match final output
rows.
Filtered scans, primary-key tables, format tables, fallback-backed
tables, file-creation-time scans, and split-ignore modes intentionally
do not receive this optimization yet.
### Test coverage
The regression tests cover:
- Limit-aware split pruning for eligible append-only tables.
- Non-positive and oversized limits.
- Primary-key row-accounting safety.
- Format-table file-count versus row-count safety.
- Residual-filter correctness.
- Direct, decorated, and system-hidden fallback readers.
- File-creation-time scan planning.
- Mixed native/JNI split routing and ignore modes.
- Deterministic empty-file and split-order scenarios.
### Check List
- [x] PaimonScanPlanProviderTest
- [x] FE Checkstyle
---
.../connector/paimon/PaimonScanPlanProvider.java | 47 +-
.../paimon/PaimonScanPlanProviderTest.java | 473 +++++++++++++++++++++
2 files changed, 513 insertions(+), 7 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
index 335d01a6178..30c109dfb2f 100644
---
a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
@@ -357,6 +357,9 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
}
try {
Table dataTable =
PaimonTableResolver.resolveSystemSource(catalogOps, handle, context);
+ // System wrappers hide the fallback pair from instanceof checks.
Retain the exact source
+ // resolved here so split-limit safety applies after transient
handles are reloaded too.
+ handle.setSystemTableSource(dataTable);
return PaimonReaderOptions.runtimeSafeSystemTable(
handle.getSysTableName(), systemTable, dataTable,
scanOptions);
} catch (IllegalArgumentException e) {
@@ -475,14 +478,14 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
/**
* The scan entry. Of everything on the request, paimon consumes the
handle, the columns, the filter and
- * the no-grouping {@code COUNT(*)} signal (FIX-COUNT-PUSHDOWN, which lets
a split answer from its
- * precomputed merged row count); the row limit and the pruned partition
set are not consumed by the
- * paimon read path — it is predicate-driven and re-plans through the SDK
from the filter.
+ * the row limit, and the no-grouping {@code COUNT(*)} signal
(FIX-COUNT-PUSHDOWN, which lets a split
+ * answer from its precomputed merged row count); the pruned partition set
is not consumed by the paimon
+ * read path — it is predicate-driven and re-plans through the SDK from
the filter.
*/
@Override
public List<ConnectorScanRange> planScan(ConnectorSession session,
ConnectorScanRequest request) {
return planScanInternal(session, request.getTableHandle(),
request.getColumns(),
- request.getFilter(), request.isCountPushdown());
+ request.getFilter(), request.getLimit(),
request.isCountPushdown());
}
/**
@@ -593,6 +596,7 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
ConnectorTableHandle handle,
List<ConnectorColumnHandle> columns,
Optional<ConnectorExpression> filter,
+ long limit,
boolean countPushdown) {
PaimonTableHandle paimonHandle = (PaimonTableHandle) handle;
@@ -610,6 +614,9 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
return Collections.emptyList();
}
Table table = resolveScanTable(paimonHandle);
+ Optional<Long> fileCreationTime = optionsPin
+ ? PaimonScanParams.getPinnedFileCreationTime(pinnedOptions)
+ : Optional.empty();
// Build predicates from filter expression
RowType rowType = table.rowType();
@@ -664,6 +671,19 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
if (projected.length > 0) {
readBuilder.withProjection(projected);
}
+ if (limit > 0 && limit <= Integer.MAX_VALUE
+ && filter.isEmpty()
+ && fileCreationTime.isEmpty()
+ && hasTrustworthyLimitAccounting(table)
+ && !usesFallbackRead(table, paimonHandle)
+ && !ignoreJni
+ && !ignoreNative) {
+ // Only append-only FileStore manifests count final output rows:
format tables count
+ // files, while primary-key metadata may count deletes that its
reader later removes.
+ // Ignore routing happens after planning, so pruning first could
discard every retained
+ // split and hide rows from the non-ignored reader path.
+ readBuilder.withLimit((int) limit);
+ }
TableScan scan = readBuilder.newScan();
// FIX-SCAN-METRICS: attach a metric registry so scan.plan() records
its ScanMetrics (manifest cache
// hit/miss, scan durations, table files skipped/resulted), then
harvest them below — restores the
@@ -673,9 +693,6 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
if (scan instanceof InnerTableScan) {
scan = ((InnerTableScan) scan).withMetricRegistry(metricRegistry);
}
- Optional<Long> fileCreationTime = optionsPin
- ? PaimonScanParams.getPinnedFileCreationTime(pinnedOptions)
- : Optional.empty();
List<Split> paimonSplits = fileCreationTime.isPresent()
? planFileCreationTimeSplits(table, pinnedOptions, predicates,
fileCreationTime.get())
: planSplits(scan);
@@ -819,6 +836,22 @@ public class PaimonScanPlanProvider implements
ConnectorScanPlanProvider {
return ranges;
}
+ private static boolean usesFallbackRead(Table scanTable, PaimonTableHandle
handle) {
+ return isFallbackFileStoreTable(scanTable)
+ || isFallbackFileStoreTable(handle.getSystemTableSource())
+ || isFallbackFileStoreTable(handle.getSysBaseTable());
+ }
+
+ private static boolean hasTrustworthyLimitAccounting(Table table) {
+ return table instanceof FileStoreTable &&
table.primaryKeys().isEmpty();
+ }
+
+ private static boolean isFallbackFileStoreTable(Table table) {
+ return table instanceof FileStoreTable
+ &&
PaimonTableDecorators.unwrapToFallbackOrBase((FileStoreTable) table)
+ instanceof FallbackReadFileStoreTable;
+ }
+
/**
* Builds the native-reader {@link PaimonScanRange} for one raw
ORC/Parquet file plus its optional
* deletion vector. BOTH the data-file path and the deletion-vector path
are routed through
diff --git
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
index f5f7126a4fc..e66a38c021d 100644
---
a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
@@ -20,8 +20,13 @@ package org.apache.doris.connector.paimon;
import org.apache.doris.connector.spi.ConnectorContext;
import org.apache.doris.connector.spi.ConnectorSession;
import org.apache.doris.connector.spi.ConnectorStorageContext;
+import org.apache.doris.connector.spi.ConnectorType;
import org.apache.doris.connector.spi.DorisConnectorException;
import org.apache.doris.connector.spi.handle.ConnectorColumnHandle;
+import org.apache.doris.connector.spi.pushdown.ConnectorColumnRef;
+import org.apache.doris.connector.spi.pushdown.ConnectorComparison;
+import org.apache.doris.connector.spi.pushdown.ConnectorExpression;
+import org.apache.doris.connector.spi.pushdown.ConnectorLiteral;
import org.apache.doris.connector.spi.scan.ConnectorScanRange;
import org.apache.doris.connector.spi.scan.ConnectorScanRequest;
import org.apache.doris.filesystem.FileSystemType;
@@ -44,13 +49,23 @@ import org.apache.paimon.catalog.FileSystemCatalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataInputViewStreamWrapper;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.privilege.AllGrantedPrivilegeChecker;
+import org.apache.paimon.privilege.PrivilegedFileStoreTable;
+import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FormatTable;
import org.apache.paimon.table.Table;
+import org.apache.paimon.table.format.FormatDataSplit;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
@@ -70,7 +85,9 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.ByteArrayInputStream;
+import java.io.IOException;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
@@ -104,6 +121,18 @@ public class PaimonScanPlanProviderTest {
return builder.build();
}
+ private static final class OrderedLocalFileIO extends LocalFileIO {
+ @Override
+ public FileStatus[] listFiles(org.apache.paimon.fs.Path path, boolean
recursive)
+ throws IOException {
+ FileStatus[] files = super.listFiles(path, recursive);
+ // FormatTableScan preserves FileIO order, so pin it to keep the
empty-file regression
+ // independent of the host filesystem's directory iteration order.
+ Arrays.sort(files, Comparator.comparing(file ->
file.getPath().getName()));
+ return files;
+ }
+ }
+
@Test
public void resolveTableReloadsWhenTransientTableNull() {
RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
@@ -270,6 +299,376 @@ public class PaimonScanPlanProviderTest {
}
}
+ @Test
+ public void planScanPushesLimitIntoPaimonSplitPlanning(@TempDir Path
warehouse) throws Exception {
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "limited");
+ catalog.createTable(id, Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("pt", DataTypes.INT())
+ .partitionKeys("pt")
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .build(), false);
+ Table table = catalog.getTable(id);
+ BatchWriteBuilder wb = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = wb.newWrite()) {
+ write.write(GenericRow.of(1, 1));
+ write.write(GenericRow.of(2, 2));
+ List<CommitMessage> messages = write.prepareCommit();
+ try (BatchTableCommit commit = wb.newCommit()) {
+ commit.commit(messages);
+ }
+ }
+
+ RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
+ ops.table = table;
+ PaimonScanPlanProvider provider = new PaimonScanPlanProvider(
+ PaimonCatalogProperties.of(Collections.emptyMap()), ops);
+ PaimonTableHandle handle = new PaimonTableHandle(
+ "db", "limited", Collections.emptyList(),
Collections.emptyList());
+ ConnectorSession session =
sessionWithProps(Collections.emptyMap());
+
+ List<ConnectorScanRange> unlimited = provider.planScan(session,
+ ConnectorScanRequest.builder(handle,
Collections.emptyList()).build());
+ List<ConnectorScanRange> limited = provider.planScan(session,
+ ConnectorScanRequest.builder(handle,
Collections.emptyList()).limit(1).build());
+ List<ConnectorScanRange> oversized = provider.planScan(session,
+ ConnectorScanRequest.builder(handle,
Collections.emptyList())
+ .limit((long) Integer.MAX_VALUE + 1)
+ .build());
+
+ Assertions.assertTrue(unlimited.size() >= 2,
+ "fixture must plan at least one split for each partition");
+ Assertions.assertEquals(1, limited.size(),
+ "LIMIT 1 must let Paimon stop split planning after enough
rows are covered");
+ Assertions.assertEquals(unlimited.size(), oversized.size(),
+ "a Doris limit wider than Paimon's int must not be
narrowed during split planning");
+ }
+ }
+
+ @Test
+ public void primaryKeyLimitKeepsAllRowsForUnsafeSplitAccounting(@TempDir
Path warehouse)
+ throws Exception {
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "primary_key_limit");
+ catalog.createTable(id, Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("pt", DataTypes.INT())
+ .partitionKeys("pt")
+ .primaryKey("id", "pt")
+ .option("bucket", "1")
+ .build(), false);
+ Table table = catalog.getTable(id);
+ BatchWriteBuilder wb = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = wb.newWrite()) {
+ write.write(GenericRow.of(1, 1));
+ write.write(GenericRow.of(2, 2));
+ List<CommitMessage> messages = write.prepareCommit();
+ try (BatchTableCommit commit = wb.newCommit()) {
+ commit.commit(messages);
+ }
+ }
+
+ RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
+ ops.table = table;
+ PaimonScanPlanProvider provider = new PaimonScanPlanProvider(
+ PaimonCatalogProperties.of(Collections.emptyMap()), ops);
+ PaimonTableHandle handle = new PaimonTableHandle(
+ "db", "primary_key_limit", Collections.emptyList(),
Collections.emptyList());
+ List<ConnectorScanRange> ranges = provider.planScan(
+
sessionWithProps(Collections.singletonMap("force_jni_scanner", "true")),
+ ConnectorScanRequest.builder(handle,
Collections.emptyList())
+ .limit(1)
+ .build());
+
+ RecordReader<InternalRow> reader = table.newReadBuilder()
+ .newRead()
+ .createReader(deserializeJniSplits(ranges));
+ List<Integer> ids = new ArrayList<>();
+ reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+ ids.sort(Integer::compareTo);
+ Assertions.assertEquals(Arrays.asList(1, 2), ids,
+ "primary-key metadata may count deleted rows, so Doris
must retain every split");
+ }
+ }
+
+ @Test
+ public void formatTableLimitDoesNotTreatFilesAsRows(@TempDir Path
warehouse)
+ throws Exception {
+ Path dataDir =
Files.createDirectories(warehouse.resolve("format_data"));
+ Files.write(dataDir.resolve("000-empty.csv"), new byte[0]);
+ Files.write(dataDir.resolve("999-live.csv"),
Collections.singletonList("7"),
+ StandardCharsets.UTF_8);
+ FormatTable table = FormatTable.builder()
+ .fileIO(new OrderedLocalFileIO())
+ .identifier(Identifier.create("db", "format_limit"))
+ .rowType(rowType("id"))
+ .partitionKeys(Collections.emptyList())
+ .location(dataDir.toUri().toString())
+ .format(FormatTable.Format.CSV)
+
.options(Collections.singletonMap(CoreOptions.FILE_FORMAT.key(), "csv"))
+ .build();
+ List<Split> plannedSplits =
table.newReadBuilder().newScan().plan().splits();
+ Assertions.assertTrue(plannedSplits.size() >= 2,
+ "fixture must expose both the empty and live format files");
+ Assertions.assertEquals("000-empty.csv",
+ ((FormatDataSplit) plannedSplits.get(0)).filePath().getName(),
+ "the unsafe file-count limit must encounter the empty file
first");
+ RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
+ ops.table = table;
+ PaimonScanPlanProvider provider = new PaimonScanPlanProvider(
+ PaimonCatalogProperties.of(Collections.emptyMap()), ops);
+ PaimonTableHandle handle = new PaimonTableHandle(
+ "db", "format_limit", Collections.emptyList(),
Collections.emptyList());
+ List<ConnectorScanRange> ranges = provider.planScan(
+ sessionWithProps(Collections.singletonMap("force_jni_scanner",
"true")),
+ ConnectorScanRequest.builder(handle,
Collections.emptyList()).limit(1).build());
+
+ RecordReader<InternalRow> reader = table.newReadBuilder()
+ .newRead()
+ .createReader(deserializeJniSplits(ranges));
+ List<Integer> ids = new ArrayList<>();
+ reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+ Assertions.assertEquals(Collections.singletonList(7), ids,
+ "a LIMIT measured in rows must not stop after an empty format
file");
+ }
+
+ private static List<Split> deserializeJniSplits(List<ConnectorScanRange>
ranges)
+ throws Exception {
+ List<Split> splits = new ArrayList<>();
+ for (ConnectorScanRange range : ranges) {
+ String encoded = range.getProperties().get("paimon.split");
+ Assertions.assertNotNull(encoded, "the result-bearing test
requires JNI splits");
+ splits.add((Split) InstantiationUtil.deserializeObject(
+ Base64.getDecoder().decode(encoded),
+ PaimonScanPlanProviderTest.class.getClassLoader()));
+ }
+ return splits;
+ }
+
+ @Test
+ public void filteredLimitDoesNotDiscardLaterMatchingSplit(@TempDir Path
warehouse)
+ throws Exception {
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "filtered_limit");
+ catalog.createTable(id, Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("pt", DataTypes.INT())
+ .partitionKeys("pt")
+ .option("bucket", "-1")
+ .build(), false);
+ Table table = catalog.getTable(id);
+ BatchWriteBuilder wb = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = wb.newWrite()) {
+ // The first split's [1, 3] min/max admits id=2, but contains
no matching row.
+ write.write(GenericRow.of(1, 2));
+ write.write(GenericRow.of(3, 2));
+ write.write(GenericRow.of(2, 1));
+ List<CommitMessage> messages = write.prepareCommit();
+ try (BatchTableCommit commit = wb.newCommit()) {
+ commit.commit(messages);
+ }
+ }
+
+ RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
+ ops.table = table;
+ PaimonScanPlanProvider provider = new PaimonScanPlanProvider(
+ PaimonCatalogProperties.of(Collections.emptyMap()), ops);
+ PaimonTableHandle handle = new PaimonTableHandle(
+ "db", "filtered_limit", Collections.emptyList(),
Collections.emptyList());
+ ConnectorExpression filter = new ConnectorComparison(
+ ConnectorComparison.Operator.EQ,
+ new ConnectorColumnRef("id", ConnectorType.of("INT")),
+ ConnectorLiteral.ofInt(2));
+ List<ConnectorScanRange> ranges = provider.planScan(
+
sessionWithProps(Collections.singletonMap("force_jni_scanner", "true")),
+ ConnectorScanRequest.builder(handle,
Collections.emptyList())
+ .filter(Optional.of(filter))
+ .limit(1)
+ .build());
+
+ List<Predicate> predicates = new
PaimonPredicateConverter(table.rowType()).convert(filter);
+ List<Split> filteredSplits = table.newReadBuilder()
+ .withFilter(predicates)
+ .newScan()
+ .plan()
+ .splits();
+ Assertions.assertEquals(2, filteredSplits.size(),
+ "both min/max-admitted partitions must remain in the
fixture");
+ Assertions.assertEquals(2,
+ ((DataSplit) filteredSplits.get(0)).partition().getInt(0),
+ "the first split must contain only non-matching ids 1 and
3");
+ Assertions.assertEquals(1,
+ ((DataSplit) filteredSplits.get(1)).partition().getInt(0),
+ "the later split must contain the matching id 2");
+ RecordReader<InternalRow> reader = table.newReadBuilder()
+ .withFilter(predicates)
+ .newRead()
+ .executeFilter()
+ .createReader(deserializeJniSplits(ranges));
+ List<Integer> ids = new ArrayList<>();
+ reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+ Assertions.assertEquals(Collections.singletonList(2), ids,
+ "LIMIT split pruning must not discard a later split
containing the match");
+ }
+ }
+
+ @Test
+ public void fallbackLimitDoesNotExposeStaleFallbackRows(@TempDir Path
warehouse)
+ throws Exception {
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Schema schema = Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("pt", DataTypes.INT())
+ .column("val", DataTypes.INT())
+ .partitionKeys("pt")
+ .option("bucket", "-1")
+ .build();
+ Identifier mainId = Identifier.create("db", "fallback_main");
+ Identifier fallbackId = Identifier.create("db", "fallback_old");
+ catalog.createTable(mainId, schema, false);
+ catalog.createTable(fallbackId, schema, false);
+ FileStoreTable main = (FileStoreTable) catalog.getTable(mainId);
+ FileStoreTable fallback = (FileStoreTable)
catalog.getTable(fallbackId);
+
+ BatchWriteBuilder mainWriteBuilder = main.newBatchWriteBuilder();
+ try (BatchTableWrite write = mainWriteBuilder.newWrite()) {
+ write.write(GenericRow.of(1, 2, 200));
+ write.write(GenericRow.of(2, 1, 100));
+ List<CommitMessage> messages = write.prepareCommit();
+ try (BatchTableCommit commit = mainWriteBuilder.newCommit()) {
+ commit.commit(messages);
+ }
+ }
+ BatchWriteBuilder fallbackWriteBuilder =
fallback.newBatchWriteBuilder();
+ try (BatchTableWrite write = fallbackWriteBuilder.newWrite()) {
+ write.write(GenericRow.of(2, 1, 50));
+ List<CommitMessage> messages = write.prepareCommit();
+ try (BatchTableCommit commit =
fallbackWriteBuilder.newCommit()) {
+ commit.commit(messages);
+ }
+ }
+
+ FallbackReadFileStoreTable pair = new
FallbackReadFileStoreTable(main, fallback);
+ FileStoreTable decorated = PrivilegedFileStoreTable.wrap(
+ pair, new AllGrantedPrivilegeChecker(), mainId);
+ for (Table planningTable : Arrays.asList(pair, decorated)) {
+ RecordingPaimonCatalogOps ops = new
RecordingPaimonCatalogOps();
+ ops.table = planningTable;
+ PaimonScanPlanProvider provider = new PaimonScanPlanProvider(
+ PaimonCatalogProperties.of(Collections.emptyMap()),
ops);
+ PaimonTableHandle handle = new PaimonTableHandle(
+ "db", "fallback_main", Collections.emptyList(),
Collections.emptyList());
+ List<ConnectorScanRange> ranges = provider.planScan(
+
sessionWithProps(Collections.singletonMap("force_jni_scanner", "true")),
+ ConnectorScanRequest.builder(handle,
Collections.emptyList())
+ .limit(1)
+ .build());
+
+ RecordReader<InternalRow> reader = pair.newReadBuilder()
+ .newRead()
+ .createReader(deserializeJniSplits(ranges));
+ List<Integer> values = new ArrayList<>();
+ reader.forEachRemaining(row -> values.add(row.getInt(2)));
+ values.sort(Integer::compareTo);
+ Assertions.assertEquals(Arrays.asList(100, 200), values,
+ "direct and decorated fallback tables must never
expose stale rows");
+ }
+
+ RecordingPaimonCatalogOps systemOps = new
RecordingPaimonCatalogOps();
+ systemOps.table = pair;
+ PaimonScanPlanProvider systemProvider = new PaimonScanPlanProvider(
+ PaimonCatalogProperties.of(Collections.emptyMap()),
systemOps);
+ PaimonTableHandle systemHandle = PaimonTableHandle.forSystemTable(
+ "db", "fallback_main", "ro", false);
+ systemHandle.setPaimonTable(new ReadOptimizedTable(pair));
+ systemHandle.setSysBaseTable(pair);
+ systemHandle.setSystemTableSource(decorated);
+ ConnectorSession forceJni = sessionWithProps(
+ Collections.singletonMap("force_jni_scanner", "true"));
+ List<ConnectorScanRange> unlimitedSystemRanges =
systemProvider.planScan(
+ forceJni,
+ ConnectorScanRequest.builder(systemHandle,
Collections.emptyList()).build());
+ List<ConnectorScanRange> systemRanges = systemProvider.planScan(
+ forceJni,
+ ConnectorScanRequest.builder(systemHandle,
Collections.emptyList())
+ .limit(1)
+ .build());
+ List<String> fallbackFiles = new ArrayList<>();
+ for (Split split :
fallback.newReadBuilder().newScan().plan().splits()) {
+ for (DataFileMeta file : ((DataSplit) split).dataFiles()) {
+ fallbackFiles.add(file.fileName());
+ }
+ }
+ List<Split> systemSplits = deserializeJniSplits(systemRanges);
+ Assertions.assertEquals(unlimitedSystemRanges.size(),
systemSplits.size(),
+ "the system wrapper must not hide fallback ownership from
limit safety");
+ for (Split split : systemSplits) {
+ for (DataFileMeta file : ((DataSplit) split).dataFiles()) {
+
Assertions.assertFalse(fallbackFiles.contains(file.fileName()),
+ "a system wrapper must not hide fallback ownership
from limit safety");
+ }
+ }
+ }
+ }
+
+ @Test
+ public void fileCreationTimeScanDoesNotApplyLimitToDiscardedTableScan(
+ @TempDir Path warehouse) throws Exception {
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "creation_time_limit");
+ catalog.createTable(id, Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("pt", DataTypes.INT())
+ .partitionKeys("pt")
+ .primaryKey("id", "pt")
+ .option("bucket", "1")
+ .build(), false);
+ Table table = catalog.getTable(id);
+ BatchWriteBuilder wb = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = wb.newWrite()) {
+ write.write(GenericRow.of(1, 1));
+ write.write(GenericRow.of(2, 2));
+ List<CommitMessage> messages = write.prepareCommit();
+ try (BatchTableCommit commit = wb.newCommit()) {
+ commit.commit(messages);
+ }
+ }
+
+ RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
+ ops.table = table;
+ PaimonScanPlanProvider provider = new PaimonScanPlanProvider(
+ PaimonCatalogProperties.of(Collections.emptyMap()), ops);
+ PaimonTableHandle handle = new PaimonTableHandle(
+ "db", "creation_time_limit", Collections.emptyList(),
Collections.emptyList());
+ Map<String, String> resolved = PaimonScanParams.markAsOptions(
+ PaimonScanParams.resolveOptions(table,
Collections.singletonMap(
+ CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(),
"0")));
+ PaimonTableHandle pinned = handle.withScanOptions(resolved);
+ ConnectorSession session =
sessionWithProps(Collections.emptyMap());
+
+ List<ConnectorScanRange> unlimited = provider.planScan(session,
+ ConnectorScanRequest.builder(pinned,
Collections.emptyList()).build());
+ List<ConnectorScanRange> limited = provider.planScan(session,
+ ConnectorScanRequest.builder(pinned,
Collections.emptyList()).limit(1).build());
+ Assertions.assertTrue(unlimited.size() >= 2,
+ "fixture must include multiple file-creation-time splits");
+ Assertions.assertEquals(unlimited.size(), limited.size(),
+ "the SnapshotReader path has no safe limit API and must
retain its full plan");
+ }
+ }
+
/** Builds a native-eligible RawFile (parquet suffix). The numeric fields
are irrelevant to the
* native-vs-JNI routing decision under test, only the path suffix
matters. */
private static RawFile parquetRawFile(String path) {
@@ -1617,6 +2016,80 @@ public class PaimonScanPlanProviderTest {
}
}
+ @Test
+ public void ignoreNativeLimitKeepsLaterJniSplit(@TempDir Path warehouse)
throws Exception {
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "mixed_limit");
+ catalog.createTable(id, Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("pt", DataTypes.INT())
+ .partitionKeys("pt")
+ .option("bucket", "-1")
+ .option("file.format", "parquet")
+ .build(), false);
+ Table parquetTable = catalog.getTable(id);
+ BatchWriteBuilder parquetWriteBuilder =
parquetTable.newBatchWriteBuilder();
+ try (BatchTableWrite write = parquetWriteBuilder.newWrite()) {
+ write.write(GenericRow.of(1, 1));
+ List<CommitMessage> messages = write.prepareCommit();
+ try (BatchTableCommit commit =
parquetWriteBuilder.newCommit()) {
+ commit.commit(messages);
+ }
+ }
+
+ catalog.alterTable(id, SchemaChange.setOption("file.format",
"avro"), false);
+ Table mixedTable = catalog.getTable(id);
+ BatchWriteBuilder avroWriteBuilder =
mixedTable.newBatchWriteBuilder();
+ try (BatchTableWrite write = avroWriteBuilder.newWrite()) {
+ write.write(GenericRow.of(2, 2));
+ List<CommitMessage> messages = write.prepareCommit();
+ try (BatchTableCommit commit = avroWriteBuilder.newCommit()) {
+ commit.commit(messages);
+ }
+ }
+
+ List<Split> plannedSplits =
mixedTable.newReadBuilder().newScan().plan().splits();
+ Assertions.assertEquals(2, plannedSplits.size(),
+ "fixture must plan one split for each file format");
+ Assertions.assertEquals("parquet",
firstRawFileFormat(plannedSplits.get(0)),
+ "the native split must be first so an unsafe LIMIT 1 would
retain only it");
+ Assertions.assertEquals("avro",
firstRawFileFormat(plannedSplits.get(1)),
+ "the later split must require the JNI reader");
+
+ RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
+ ops.table = mixedTable;
+ PaimonScanPlanProvider provider = new PaimonScanPlanProvider(
+ PaimonCatalogProperties.of(Collections.emptyMap()), ops);
+ PaimonTableHandle handle = new PaimonTableHandle(
+ "db", "mixed_limit", Collections.emptyList(),
Collections.emptyList());
+ List<ConnectorScanRange> ranges = provider.planScan(
+ sessionWithProps(Collections.singletonMap(
+ "ignore_split_type", "IGNORE_NATIVE")),
+ ConnectorScanRequest.builder(handle,
Collections.emptyList())
+ .limit(1)
+ .build());
+
+ Assertions.assertEquals(1, ranges.size(),
+ "ignoring native splits must retain the later JNI split
despite LIMIT 1");
+ RecordReader<InternalRow> reader = mixedTable.newReadBuilder()
+ .newRead()
+ .createReader(deserializeJniSplits(ranges));
+ List<Integer> ids = new ArrayList<>();
+ reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+ Assertions.assertEquals(Collections.singletonList(2), ids,
+ "routing after split planning must not turn a non-empty
scan into zero rows");
+ }
+ }
+
+ private static String firstRawFileFormat(Split split) {
+ List<RawFile> rawFiles = split.convertToRawFiles().orElseThrow(
+ () -> new AssertionError("fixture split must expose a raw
file"));
+ Assertions.assertFalse(rawFiles.isEmpty(), "fixture split must contain
a raw file");
+ return rawFiles.get(0).format();
+ }
+
@Test
public void ignoreNativeDropsNativeSplit(@TempDir Path warehouse) throws
Exception {
// FIX-L14: an append-only table's DataSplit is native-eligible;
ignore_split_type=IGNORE_NATIVE must
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
