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 8b589d42d0 [core] Pack multiple files into one split for format table
reads (#8317)
8b589d42d0 is described below
commit 8b589d42d0697448cf08996ecbdb14b7cc168464
Author: Zouxxyy <[email protected]>
AuthorDate: Mon Jun 22 17:23:51 2026 +0800
[core] Pack multiple files into one split for format table reads (#8317)
Format tables (`format-table.implementation = paimon`) currently
generate one split per data file, which produces a large number of
splits/tasks when a directory contains many small files.
This change lets a single split carry multiple files, packed in the core
scan by `source.split.target-size` (with `source.split.open-file-cost`
as a per-file weight floor, mirroring `AppendOnlySplitGenerator`).
Because packing happens in `FormatTableScan`, it is engine-agnostic; the
reader concatenates a split's files via `ConcatRecordReader`.
---
.../paimon/table/format/FormatDataSplit.java | 134 ++++++++++++++-------
.../paimon/table/format/FormatReadBuilder.java | 34 ++++--
.../paimon/table/format/FormatTableScan.java | 36 ++++--
.../paimon/table/format/FormatDataSplitTest.java | 53 ++++----
.../paimon/table/format/FormatReadBuilderTest.java | 9 +-
.../paimon/table/format/FormatTableScanTest.java | 5 +-
.../paimon/spark/read/BinPackingSplits.scala | 3 +-
.../org/apache/paimon/spark/util/SplitUtils.scala | 4 +-
.../paimon/spark/table/PaimonFormatTableTest.scala | 49 ++++++++
9 files changed, 225 insertions(+), 102 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatDataSplit.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatDataSplit.java
index f90d8c9ed3..78f2f9532e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatDataSplit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatDataSplit.java
@@ -24,61 +24,44 @@ import org.apache.paimon.table.source.Split;
import javax.annotation.Nullable;
+import java.io.Serializable;
+import java.util.List;
import java.util.Objects;
import java.util.OptionalLong;
-/** {@link FormatDataSplit} for format table. */
+/**
+ * {@link Split} for format table. A split may contain multiple files packed
by {@code
+ * source.split.target-size}, so a single reader task can read several files
sequentially.
+ */
public class FormatDataSplit implements Split {
- private static final long serialVersionUID = 2L;
+ private static final long serialVersionUID = 3L;
- private final Path filePath;
- private final long fileSize;
- private final long offset;
- // If null, means reading the whole file.
- @Nullable private final Long length;
+ private final List<FileMeta> files;
@Nullable private final BinaryRow partition;
- public FormatDataSplit(
- Path filePath,
- long fileSize,
- long offset,
- @Nullable Long length,
- @Nullable BinaryRow partition) {
- this.filePath = filePath;
- this.fileSize = fileSize;
- this.offset = offset;
- this.length = length;
+ public FormatDataSplit(List<FileMeta> files, @Nullable BinaryRow
partition) {
+ this.files = files;
this.partition = partition;
}
- public FormatDataSplit(Path filePath, long fileSize, @Nullable BinaryRow
partition) {
- this(filePath, fileSize, 0L, null, partition);
- }
-
- public Path filePath() {
- return this.filePath;
- }
-
- public Path dataPath() {
- return this.filePath;
+ public List<FileMeta> files() {
+ return files;
}
- public long fileSize() {
- return this.fileSize;
- }
-
- public long offset() {
- return offset;
+ @Nullable
+ public BinaryRow partition() {
+ return partition;
}
- @Nullable
- public Long length() {
- return length;
+ /** Total bytes to read for this split, i.e. the sum of {@link
FileMeta#readSize()}. */
+ public long totalSize() {
+ return files.stream().mapToLong(FileMeta::readSize).sum();
}
- public BinaryRow partition() {
- return partition;
+ /** Number of files (or file ranges) in this split. */
+ public int fileCount() {
+ return files.size();
}
@Override
@@ -100,15 +83,78 @@ public class FormatDataSplit implements Split {
return false;
}
FormatDataSplit that = (FormatDataSplit) o;
- return offset == that.offset
- && fileSize == that.fileSize
- && Objects.equals(length, that.length)
- && Objects.equals(filePath, that.filePath)
- && Objects.equals(partition, that.partition);
+ return Objects.equals(files, that.files) && Objects.equals(partition,
that.partition);
}
@Override
public int hashCode() {
- return Objects.hash(filePath, fileSize, offset, length, partition);
+ return Objects.hash(files, partition);
+ }
+
+ /**
+ * A single file (or one offset range of a splittable file) inside a
{@link FormatDataSplit}.
+ */
+ public static class FileMeta implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Path filePath;
+ private final long fileSize;
+ private final long offset;
+ // If null, means reading the whole file.
+ @Nullable private final Long length;
+
+ public FileMeta(Path filePath, long fileSize, long offset, @Nullable
Long length) {
+ this.filePath = filePath;
+ this.fileSize = fileSize;
+ this.offset = offset;
+ this.length = length;
+ }
+
+ public FileMeta(Path filePath, long fileSize) {
+ this(filePath, fileSize, 0L, null);
+ }
+
+ public Path filePath() {
+ return filePath;
+ }
+
+ public long fileSize() {
+ return fileSize;
+ }
+
+ public long offset() {
+ return offset;
+ }
+
+ @Nullable
+ public Long length() {
+ return length;
+ }
+
+ /** Bytes this segment actually reads: range length when sliced,
otherwise whole file. */
+ public long readSize() {
+ return length != null ? length : fileSize;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ FileMeta that = (FileMeta) o;
+ return fileSize == that.fileSize
+ && offset == that.offset
+ && Objects.equals(length, that.length)
+ && Objects.equals(filePath, that.filePath);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(filePath, fileSize, offset, length);
+ }
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
index 6c394f3b8f..05ad36da72 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
@@ -19,18 +19,20 @@
package org.apache.paimon.table.format;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.FileFormatDiscover;
import org.apache.paimon.format.FormatReaderContext;
import org.apache.paimon.format.FormatReaderFactory;
-import org.apache.paimon.fs.Path;
import org.apache.paimon.io.DataFileRecordReader;
+import org.apache.paimon.mergetree.compact.ConcatRecordReader;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.partition.PartitionUtils;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.predicate.TopN;
import org.apache.paimon.reader.FileRecordReader;
+import org.apache.paimon.reader.ReaderSupplier;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.FormatTable;
import org.apache.paimon.table.source.ReadBuilder;
@@ -47,6 +49,7 @@ import org.apache.paimon.utils.RowRangeIndex;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
@@ -159,9 +162,6 @@ public class FormatReadBuilder implements ReadBuilder {
}
protected RecordReader<InternalRow> createReader(FormatDataSplit
dataSplit) throws IOException {
- Path filePath = dataSplit.dataPath();
- FormatReaderContext formatReaderContext =
- new FormatReaderContext(table.fileIO(), filePath,
dataSplit.fileSize(), null);
// Skip pushing down partition filters to reader.
List<Predicate> readFilters =
excludePredicateWithFields(
@@ -176,14 +176,30 @@ public class FormatReadBuilder implements ReadBuilder {
Pair<int[], RowType> partitionMapping =
PartitionUtils.getPartitionMapping(
table.partitionKeys(), readType().getFields(),
table.partitionType());
+
+ BinaryRow partition = dataSplit.partition();
+ List<ReaderSupplier<InternalRow>> suppliers = new ArrayList<>();
+ for (FormatDataSplit.FileMeta file : dataSplit.files()) {
+ suppliers.add(() -> createFileReader(file, partition,
readerFactory, partitionMapping));
+ }
+ return ConcatRecordReader.create(suppliers);
+ }
+
+ private RecordReader<InternalRow> createFileReader(
+ FormatDataSplit.FileMeta file,
+ @Nullable BinaryRow partition,
+ FormatReaderFactory readerFactory,
+ Pair<int[], RowType> partitionMapping)
+ throws IOException {
+ FormatReaderContext formatReaderContext =
+ new FormatReaderContext(table.fileIO(), file.filePath(),
file.fileSize(), null);
try {
FileRecordReader<InternalRow> reader;
- Long length = dataSplit.length();
+ Long length = file.length();
if (length != null) {
- reader =
- readerFactory.createReader(formatReaderContext,
dataSplit.offset(), length);
+ reader = readerFactory.createReader(formatReaderContext,
file.offset(), length);
} else {
- checkArgument(dataSplit.offset() == 0, "Offset must be 0.");
+ checkArgument(file.offset() == 0, "Offset must be 0.");
reader = readerFactory.createReader(formatReaderContext);
}
return new DataFileRecordReader(
@@ -193,7 +209,7 @@ public class FormatReadBuilder implements ReadBuilder {
options.scanIgnoreLostFile(),
null,
null,
- PartitionUtils.create(partitionMapping,
dataSplit.partition()),
+ PartitionUtils.create(partitionMapping, partition),
false,
null,
0,
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 9bbd64ccdf..0b63e71318 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
@@ -49,6 +49,7 @@ import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.BinPacking;
import org.apache.paimon.utils.InternalRowPartitionComputer;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.PartitionPathUtils;
@@ -57,7 +58,9 @@ import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
@@ -78,6 +81,7 @@ public class FormatTableScan implements InnerTableScan {
@Nullable private PartitionPredicate partitionFilter;
@Nullable private final Integer limit;
private final long targetSplitSize;
+ private final long openFileCost;
private final FormatTable.Format format;
public FormatTableScan(
@@ -89,6 +93,7 @@ public class FormatTableScan implements InnerTableScan {
this.partitionFilter = partitionFilter;
this.limit = limit;
this.targetSplitSize = coreOptions.splitTargetSize();
+ this.openFileCost = coreOptions.splitOpenFileCost();
this.format = table.format();
}
@@ -267,37 +272,44 @@ public class FormatTableScan implements InnerTableScan {
private List<Split> createSplits(FileIO fileIO, Path path, BinaryRow
partition)
throws IOException {
- List<Split> splits = new ArrayList<>();
+ List<FormatDataSplit.FileMeta> segments = new ArrayList<>();
FileStatus[] files = fileIO.listFiles(path, true);
+ Arrays.sort(files, Comparator.comparing(file ->
file.getPath().toString()));
for (FileStatus file : files) {
if (isDataFileName(file.getPath().getName())) {
- List<FormatDataSplit> fileSplits = tryToSplitLargeFile(file,
partition);
- splits.addAll(fileSplits);
+ segments.addAll(toSegments(file));
}
}
+
+ List<Split> splits = new ArrayList<>();
+ for (List<FormatDataSplit.FileMeta> bin :
+ BinPacking.packForOrdered(
+ segments,
+ file -> Math.max(file.readSize(), openFileCost),
+ targetSplitSize)) {
+ splits.add(new FormatDataSplit(bin, partition));
+ }
return splits;
}
- private List<FormatDataSplit> tryToSplitLargeFile(FileStatus file,
BinaryRow partition) {
+ private List<FormatDataSplit.FileMeta> toSegments(FileStatus file) {
if (!preferToSplitFile(file)) {
return Collections.singletonList(
- new FormatDataSplit(file.getPath(), file.getLen(),
partition));
+ new FormatDataSplit.FileMeta(file.getPath(),
file.getLen()));
}
- List<FormatDataSplit> splits = new ArrayList<>();
+ List<FormatDataSplit.FileMeta> segments = new ArrayList<>();
long remainingBytes = file.getLen();
long currentStart = 0;
while (remainingBytes > 0) {
long splitSize = Math.min(targetSplitSize, remainingBytes);
-
- FormatDataSplit split =
- new FormatDataSplit(
- file.getPath(), file.getLen(), currentStart,
splitSize, partition);
- splits.add(split);
+ segments.add(
+ new FormatDataSplit.FileMeta(
+ file.getPath(), file.getLen(), currentStart,
splitSize));
currentStart += splitSize;
remainingBytes -= splitSize;
}
- return splits;
+ return segments;
}
private boolean preferToSplitFile(FileStatus file) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatDataSplitTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatDataSplitTest.java
index 601941063d..73a1bf6f22 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatDataSplitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatDataSplitTest.java
@@ -19,15 +19,13 @@
package org.apache.paimon.table.format;
import org.apache.paimon.fs.Path;
-import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.predicate.PredicateBuilder;
-import org.apache.paimon.types.IntType;
-import org.apache.paimon.types.RowType;
+import org.apache.paimon.table.format.FormatDataSplit.FileMeta;
import org.apache.paimon.utils.InstantiationUtil;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.util.Arrays;
import static org.assertj.core.api.Assertions.assertThat;
@@ -36,38 +34,33 @@ public class FormatDataSplitTest {
@Test
public void testSerializeAndDeserialize() throws IOException,
ClassNotFoundException {
- // Create test data
- Path filePath = new Path("/test/path/file.parquet");
- RowType rowType = RowType.builder().field("id", new IntType()).build();
- long modificationTime = System.currentTimeMillis();
+ // A split packing one whole file and one offset range of another file.
+ FileMeta wholeFile = new FileMeta(new
Path("/test/path/file1.parquet"), 1024L);
+ FileMeta rangeFile = new FileMeta(new Path("/test/path/file2.csv"),
2048L, 100L, 512L);
+ FormatDataSplit split = new FormatDataSplit(Arrays.asList(wholeFile,
rangeFile), null);
- // Create a predicate for testing
- PredicateBuilder builder = new PredicateBuilder(rowType);
- Predicate predicate = builder.equal(0, 5);
-
- // Create FormatDataSplit
- FormatDataSplit split = new FormatDataSplit(filePath, 1024L, null);
-
- // Test Java serialization
byte[] serialized = InstantiationUtil.serializeObject(split);
FormatDataSplit deserialized =
InstantiationUtil.deserializeObject(serialized,
getClass().getClassLoader());
- // Verify the deserialized object
- assertThat(deserialized.filePath()).isEqualTo(split.filePath());
- assertThat(deserialized.offset()).isEqualTo(split.offset());
- assertThat(deserialized.fileSize()).isEqualTo(split.fileSize());
- assertThat(deserialized.length()).isEqualTo(split.length());
-
- split = new FormatDataSplit(filePath, 1024L, 100L, 512L, null);
+ assertThat(deserialized).isEqualTo(split);
+ assertThat(deserialized.files()).isEqualTo(split.files());
+ assertThat(deserialized.partition()).isEqualTo(split.partition());
+ assertThat(deserialized.fileCount()).isEqualTo(2);
+ // readSize: whole file -> fileSize (1024), range -> length (512).
+ assertThat(deserialized.totalSize()).isEqualTo(1024L + 512L);
- serialized = InstantiationUtil.serializeObject(split);
- deserialized = InstantiationUtil.deserializeObject(serialized,
getClass().getClassLoader());
+ FileMeta f0 = deserialized.files().get(0);
+ assertThat(f0.filePath()).isEqualTo(wholeFile.filePath());
+ assertThat(f0.fileSize()).isEqualTo(1024L);
+ assertThat(f0.offset()).isEqualTo(0L);
+ assertThat(f0.length()).isNull();
+ assertThat(f0.readSize()).isEqualTo(1024L);
- // Verify the deserialized object
- assertThat(deserialized.filePath()).isEqualTo(split.filePath());
- assertThat(deserialized.offset()).isEqualTo(split.offset());
- assertThat(deserialized.fileSize()).isEqualTo(split.fileSize());
- assertThat(deserialized.length()).isEqualTo(split.length());
+ FileMeta f1 = deserialized.files().get(1);
+ assertThat(f1.filePath()).isEqualTo(rangeFile.filePath());
+ assertThat(f1.offset()).isEqualTo(100L);
+ assertThat(f1.length()).isEqualTo(512L);
+ assertThat(f1.readSize()).isEqualTo(512L);
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatReadBuilderTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatReadBuilderTest.java
index 7b441e691d..19bc1d74fc 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatReadBuilderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatReadBuilderTest.java
@@ -185,7 +185,9 @@ public class FormatReadBuilderTest {
long fileSize = fileIO.getFileSize(csvFile);
// Test 1: Read entire CSV file (offset = 0, length = fileSize)
- FormatDataSplit fullSplit = new FormatDataSplit(csvFile, fileSize,
null);
+ FormatDataSplit fullSplit =
+ new FormatDataSplit(
+ Arrays.asList(new FormatDataSplit.FileMeta(csvFile,
fileSize)), null);
RecordReader<InternalRow> fullReader =
readBuilder.createReader(fullSplit);
List<InternalRow> fullResult = readAllRows(fullReader, rowType);
@@ -204,7 +206,10 @@ public class FormatReadBuilderTest {
// Read from offset 0 with a limited length (first 2 lines
approximately)
long partialLength = fileSize / 2;
FormatDataSplit partialSplit =
- new FormatDataSplit(csvFile, fileSize, 0, partialLength, null);
+ new FormatDataSplit(
+ Arrays.asList(
+ new FormatDataSplit.FileMeta(csvFile,
fileSize, 0, partialLength)),
+ null);
RecordReader<InternalRow> partialReader =
readBuilder.createReader(partialSplit);
List<InternalRow> partialResult = readAllRows(partialReader, rowType);
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 cda08b61e7..936b3d6bd1 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
@@ -785,8 +785,9 @@ public class FormatTableScanTest {
// Parquet files should NOT be split, should be a single split
assertThat(splits).hasSize(1);
FormatDataSplit split = (FormatDataSplit) splits.get(0);
- assertThat(split.filePath()).isEqualTo(parquetFile);
- assertThat(split.offset()).isEqualTo(0);
+ assertThat(split.files()).hasSize(1);
+ assertThat(split.files().get(0).filePath()).isEqualTo(parquetFile);
+ assertThat(split.files().get(0).offset()).isEqualTo(0);
}
@TestTemplate
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BinPackingSplits.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BinPackingSplits.scala
index f2d06b45ea..f27af93cf6 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BinPackingSplits.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BinPackingSplits.scala
@@ -79,7 +79,8 @@ case class BinPackingSplits(coreOptions: CoreOptions,
readRowSizeRatio: Double =
val (toReshuffle, reserved) = splits.partition {
case _: FallbackSplit => false
case split: DataSplit => split.rawConvertible() ||
coreOptions.dataEvolutionEnabled()
- // Currently, format table reader only supports reading one file.
+ // FormatDataSplit is already packed with multiple files by target size
in the core scan,
+ // so each split maps directly to one input partition here.
case _: FormatDataSplit => false
case _ => false
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SplitUtils.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SplitUtils.scala
index 038f3ae307..c485fe4da3 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SplitUtils.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SplitUtils.scala
@@ -32,7 +32,7 @@ object SplitUtils {
case ds: DataSplit =>
ds.dataFiles().asScala.map(_.fileSize).sum
case fs: FormatDataSplit =>
- if (fs.length() == null) fs.fileSize() else fs.length().longValue()
+ fs.totalSize()
case _ => 0
}
}
@@ -42,7 +42,7 @@ object SplitUtils {
def dataFileCount(split: Split): Long = {
split match {
case ds: DataSplit => ds.dataFiles().size()
- case _: FormatDataSplit => 1
+ case fs: FormatDataSplit => fs.fileCount()
case _ => 0
}
}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
index bbd39ae70c..adac70df2e 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/table/PaimonFormatTableTest.scala
@@ -22,6 +22,7 @@ import org.apache.paimon.catalog.Identifier
import org.apache.paimon.fs.Path
import org.apache.paimon.spark.PaimonSparkTestWithRestCatalogBase
import org.apache.paimon.table.FormatTable
+import org.apache.paimon.table.format.FormatDataSplit
import org.apache.spark.sql.Row
import org.apache.spark.sql.connector.catalog.TableCapability
@@ -447,4 +448,52 @@ class PaimonFormatTableTest extends
PaimonSparkTestWithRestCatalogBase {
checkAnswer(sql("SHOW PARTITIONS t PARTITION (p1=2, p2='2')"),
Seq(Row("p1=2/p2=2")))
}
}
+
+ test("PaimonFormatTable: pack multiple files into one split by
source.split.target-size") {
+ val tableName = "paimon_format_multifile_split"
+ withTable(tableName) {
+ sql(
+ s"CREATE TABLE $tableName (f0 INT, f1 STRING) USING CSV TBLPROPERTIES
(" +
+ "'seq'='|', 'lineSep'='\n', 'file.compression'='none', " +
+ "'format-table.implementation'='paimon')")
+ val table =
+ paimonCatalog.getTable(Identifier.create("test_db",
tableName)).asInstanceOf[FormatTable]
+
+ // Three data files of equal byte size, each with two distinct rows.
+ val contents = Seq("1|aaa\n2|bbb", "3|ccc\n4|ddd", "5|eee\n6|fff")
+ contents.zipWithIndex.foreach {
+ case (content, i) =>
+ table.fileIO().writeFile(new Path(table.location(),
s"part-0000$i.csv"), content, false)
+ }
+ val fileSize = contents.head.getBytes("UTF-8").length
+
+ val expected = Seq(
+ Row(1, "aaa"),
+ Row(2, "bbb"),
+ Row(3, "ccc"),
+ Row(4, "ddd"),
+ Row(5, "eee"),
+ Row(6, "fff"))
+
+ // Default target size (128MB): all three files are packed into a single
split.
+ val combined = getFormatTableScan(s"SELECT * FROM
$tableName").inputSplits
+ assert(combined.length == 1, s"Expected 1 packed split but got
${combined.length}")
+ assert(
+ combined.head.asInstanceOf[FormatDataSplit].files().size() == 3,
+ "The single split should contain all 3 files")
+ checkAnswer(sql(s"SELECT * FROM $tableName ORDER BY f0"), expected)
+
+ // Target size = one file size: each file becomes its own split (no
slicing since len <= target).
+ withSparkSQLConf("spark.paimon.source.split.target-size" ->
s"${fileSize}b") {
+ val perFile = getFormatTableScan(s"SELECT * FROM
$tableName").inputSplits
+ assert(perFile.length == 3, s"Expected 3 splits but got
${perFile.length}")
+ perFile.foreach(
+ s =>
+ assert(
+ s.asInstanceOf[FormatDataSplit].files().size() == 1,
+ "Each split should contain exactly 1 file"))
+ checkAnswer(sql(s"SELECT * FROM $tableName ORDER BY f0"), expected)
+ }
+ }
+ }
}
