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 f005439ecb [core] Avoid unsafe LIMIT split pruning for format tables
(#8703)
f005439ecb is described below
commit f005439ecb92e921f0790c4781de48ed7f75e9e2
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Fri Jul 17 14:44:16 2026 +0800
[core] Avoid unsafe LIMIT split pruning for format tables (#8703)
Format table scan planning currently truncates planned splits using a
pushed row `LIMIT`. This is unsafe because `FormatDataSplit` has no
row-count information, and an earlier split may produce no rows while a
later split contains valid data.
This change keeps the zero-limit fast path but retains all splits for
positive limits. Reader-side limiting still stops reading after enough
rows are returned. Positive limits may expose more splits to connectors,
which is necessary to avoid skipping valid rows. This is the
format-table counterpart of #7665.
---
.../paimon/table/format/FormatTableScan.java | 10 ++----
.../paimon/table/format/FormatTableScanTest.java | 37 ++++++++++++++++++++++
2 files changed, 40 insertions(+), 7 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
index a75e4d92d0..cb0edefc82 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
@@ -166,13 +166,9 @@ public class FormatTableScan implements InnerTableScan {
} else {
splits.addAll(createSplits(fileIO, new
Path(table.location()), null));
}
- if (limit != null) {
- if (limit <= 0) {
- return new ArrayList<>();
- }
- if (splits.size() > limit) {
- return splits.subList(0, limit);
- }
+ // Keep all splits for a positive limit because
FormatDataSplit has no row count.
+ if (limit != null && limit <= 0) {
+ return new ArrayList<>();
}
} catch (IOException e) {
throw new RuntimeException("Failed to scan files", e);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
index 18f22c8112..3a93233f89 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableScanTest.java
@@ -22,6 +22,7 @@ import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.format.csv.CsvOptions;
import org.apache.paimon.format.json.JsonOptions;
@@ -31,8 +32,11 @@ import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableScan;
import
org.apache.paimon.testutils.junit.parameterized.ParameterizedTestExtension;
import org.apache.paimon.testutils.junit.parameterized.Parameters;
import org.apache.paimon.types.DataTypes;
@@ -45,6 +49,7 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
import java.io.OutputStream;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -54,7 +59,9 @@ import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicInteger;
+import static org.apache.paimon.CoreOptions.FILE_FORMAT;
import static
org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH;
+import static org.apache.paimon.CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST;
import static org.apache.paimon.CoreOptions.SOURCE_SPLIT_TARGET_SIZE;
import static
org.apache.paimon.utils.PartitionPathUtils.searchPartSpecAndPaths;
import static org.assertj.core.api.Assertions.assertThat;
@@ -738,6 +745,36 @@ public class FormatTableScanTest {
}
}
+ @TestTemplate
+ void testPositiveLimitDoesNotPruneUnknownRowCountSplits() throws
IOException {
+ Path tableLocation = new Path(tmpPath.toUri());
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path headerOnlyFile = new Path(tableLocation, "00-header-only.csv");
+ Path dataFile = new Path(tableLocation, "01-data.csv");
+ fileIO.mkdirs(tableLocation);
+ fileIO.writeFile(headerOnlyFile, "id,name\n", false);
+ fileIO.writeFile(dataFile, "id,name\n42,later\n", false);
+
+ Map<String, String> options = new HashMap<>();
+ options.put(FILE_FORMAT.key(), "csv");
+ options.put(CsvOptions.INCLUDE_HEADER.key(), "true");
+ options.put(SOURCE_SPLIT_TARGET_SIZE.key(), "32b");
+ options.put(SOURCE_SPLIT_OPEN_FILE_COST.key(), "32b");
+ FormatTable formatTable =
+ createFormatTableWithOptions(tableLocation,
FormatTable.Format.CSV, options);
+
assertThat(formatTable.newReadBuilder().newScan().plan().splits()).hasSize(2);
+
+ ReadBuilder readBuilder = formatTable.newReadBuilder().withLimit(1);
+ TableScan.Plan plan = readBuilder.newScan().plan();
+ List<String> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(plan)) {
+ reader.forEachRemaining(
+ row -> rows.add(row.getInt(0) + "," +
row.getString(1).toString()));
+ }
+
+ assertThat(rows).containsExactly("42,later");
+ }
+
@TestTemplate
public void testCreateSplitsWhenDefineLineDelimiter() throws IOException {
for (String format : Arrays.asList("csv", "json")) {
