Skip to content

[core] Pack multiple files into one split for format table reads #8317

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 2 commits into from
Jun 22, 2026
Merged

[core] Pack multiple files into one split for format table reads #8317

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
606c9bd
[core] Pack multiple files into one split for format table reads
Zouxxyy Jun 22, 2026
8bae58e
Update
Zouxxyy Jun 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
134 changes: 90 additions & 44 deletions paimon-core/src/main/java/org/apache/paimon/table/format/FormatDataSplit.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<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
Expand All @@ -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);
}
}
}
34 changes: 25 additions & 9 deletions paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -159,9 +162,6 @@ public TableRead newRead() {
}

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(
Expand All @@ -176,14 +176,30 @@ protected RecordReader<InternalRow> createReader(FormatDataSplit dataSplit) thro
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(
Expand All @@ -193,7 +209,7 @@ protected RecordReader<InternalRow> createReader(FormatDataSplit dataSplit) thro
options.scanIgnoreLostFile(),
null,
null,
PartitionUtils.create(partitionMapping, dataSplit.partition()),
PartitionUtils.create(partitionMapping, partition),
false,
null,
0,
Expand Down
36 changes: 24 additions & 12 deletions paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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(
Expand All @@ -89,6 +93,7 @@ public FormatTableScan(
this.partitionFilter = partitionFilter;
this.limit = limit;
this.targetSplitSize = coreOptions.splitTargetSize();
this.openFileCost = coreOptions.splitOpenFileCost();
this.format = table.format();
}

Expand Down Expand Up @@ -267,37 +272,44 @@ protected static Pair<Path, Integer> computeScanPathAndLevel(

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) {
Expand Down
Loading