github-actions[bot] commented on code in PR #66818:
URL: https://github.com/apache/doris/pull/66818#discussion_r3800994568
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java:
##########
@@ -664,6 +671,15 @@ private List<ConnectorScanRange> planScanInternal(
if (projected.length > 0) {
readBuilder.withProjection(projected);
}
+ if (limit > 0 && limit <= Integer.MAX_VALUE
Review Comment:
[P1] Do not prune before split-type filtering
The limit is applied before the existing `ignoreJni`/`ignoreNative` routing
below. For a supported append-only table written in mixed formats with
deterministic partition sorting, one-row `p=1` Parquet (native) and `p=2` Avro
(JNI) splits let `LIMIT 1` retain only the Parquet split; `IGNORE_NATIVE` then
drops it, so Doris returns zero even though the non-ignored JNI split has a
row. Without this pushdown both splits are planned and the JNI row satisfies
the limit. Please suppress `withLimit` while either behavior-bearing ignore
mode is active (or account only splits that routing will retain), and add a
mixed native/JNI regression.
##########
fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java:
##########
@@ -270,6 +284,359 @@ public void
planScanEnumeratesSplitsInsideAuthScope(@TempDir Path warehouse) thr
}
}
+ @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]);
Review Comment:
[P2] Make the empty-file ordering deterministic
This relies on the `000-`/`999-` names making the empty file Paimon's first
split, but pinned 1.3.1 preserves `FileIO.listFiles` order and `LocalFileIO`
iterates unsorted `java.io.File.list()` results. If the live file is returned
first, the unsafe format-table `withLimit(1)` behavior still reads `7` and this
regression passes. Please inject/control the listing order (or otherwise assert
the first planned split is the empty file) so removing the format-table
exclusion deterministically fails.
##########
fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java:
##########
@@ -270,6 +284,359 @@ public void
planScanEnumeratesSplitsInsideAuthScope(@TempDir Path warehouse) thr
}
}
+ @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(LocalFileIO.create())
+ .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();
+ 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")
+ .primaryKey("id", "pt")
Review Comment:
[P2] Make these regressions reach their safety guards
This fixture declares a primary key, and so does the direct/decorated
fallback fixture below. Because `hasTrustworthyLimitAccounting` rejects both
before the new filter or fallback condition matters, the tests still pass if
`filter.isEmpty()` or `!usesFallbackRead(...)` is removed. Please use otherwise
limit-eligible append-only fixtures and make/assert the filtered split order,
so each regression turns red when its specific guard regresses.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
