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 f90d8c9ed3e7..78f2f9532e8a 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 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 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 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 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 boolean equals(Object o) { 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 6c394f3b8f81..05ad36da72f3 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 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 TableRead newRead() { } protected RecordReader 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 readFilters = excludePredicateWithFields( @@ -176,14 +176,30 @@ protected RecordReader createReader(FormatDataSplit dataSplit) thro Pair partitionMapping = PartitionUtils.getPartitionMapping( table.partitionKeys(), readType().getFields(), table.partitionType()); + + BinaryRow partition = dataSplit.partition(); + List> suppliers = new ArrayList<>(); + for (FormatDataSplit.FileMeta file : dataSplit.files()) { + suppliers.add(() -> createFileReader(file, partition, readerFactory, partitionMapping)); + } + return ConcatRecordReader.create(suppliers); + } + + private RecordReader createFileReader( + FormatDataSplit.FileMeta file, + @Nullable BinaryRow partition, + FormatReaderFactory readerFactory, + Pair partitionMapping) + throws IOException { + FormatReaderContext formatReaderContext = + new FormatReaderContext(table.fileIO(), file.filePath(), file.fileSize(), null); try { FileRecordReader 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 @@ protected RecordReader createReader(FormatDataSplit dataSplit) thro 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 9bbd64ccdf9c..0b63e7131859 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.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 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 FormatTableScan( this.partitionFilter = partitionFilter; this.limit = limit; this.targetSplitSize = coreOptions.splitTargetSize(); + this.openFileCost = coreOptions.splitOpenFileCost(); this.format = table.format(); } @@ -267,37 +272,44 @@ protected static Pair computeScanPathAndLevel( private List createSplits(FileIO fileIO, Path path, BinaryRow partition) throws IOException { - List splits = new ArrayList<>(); + List 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 fileSplits = tryToSplitLargeFile(file, partition); - splits.addAll(fileSplits); + segments.addAll(toSegments(file)); } } + + List splits = new ArrayList<>(); + for (List bin : + BinPacking.packForOrdered( + segments, + file -> Math.max(file.readSize(), openFileCost), + targetSplitSize)) { + splits.add(new FormatDataSplit(bin, partition)); + } return splits; } - private List tryToSplitLargeFile(FileStatus file, BinaryRow partition) { + private List toSegments(FileStatus file) { if (!preferToSplitFile(file)) { return Collections.singletonList( - new FormatDataSplit(file.getPath(), file.getLen(), partition)); + new FormatDataSplit.FileMeta(file.getPath(), file.getLen())); } - List splits = new ArrayList<>(); + List 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 601941063d70..73a1bf6f2232 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 7b441e691d6b..19bc1d74fc3f 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 void testCreateReaderWithCsvSplit() throws IOException { 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 fullReader = readBuilder.createReader(fullSplit); List fullResult = readAllRows(fullReader, rowType); @@ -204,7 +206,10 @@ public void testCreateReaderWithCsvSplit() throws IOException { // 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 partialReader = readBuilder.createReader(partialSplit); List 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 cda08b61e725..936b3d6bd180 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 void testCreateSplitsWithParquetFile() throws IOException { // 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 f2d06b45eaad..f27af93cf604 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 038f3ae30725..c485fe4da345 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 bbd39ae70cda..adac70df2e30 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) + } + } + } }