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 b0c64c5d6d [core] Split large normal files in data evolution
compaction (#9709)
b0c64c5d6d is described below
commit b0c64c5d6d46e5b17b077bb7162e5faca0b90ad2
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Sep 10 15:20:21 2026 +0800
[core] Split large normal files in data evolution compaction (#9709)
---
docs/docs/multimodal-table/data-evolution.mdx | 41 ++
docs/generated/core_configuration.html | 14 +-
.../main/java/org/apache/paimon/CoreOptions.java | 41 +-
.../CompactCandidateRangeCollector.java | 24 +-
.../DataEvolutionCompactCoordinator.java | 90 +++-
...DataEvolutionCompactDeletionVectorRewriter.java | 49 +-
.../DataEvolutionCompactRangePlanner.java | 6 +-
.../dataevolution/DataEvolutionCompactTask.java | 2 +-
.../DataEvolutionCompactTaskSerializer.java | 19 +-
.../DataEvolutionNormalCompactTask.java | 127 ++++-
.../CompactCandidateRangeCollectorTest.java | 57 +-
.../DataEvolutionCompactCoordinatorTest.java | 226 +++++++-
.../DataEvolutionCompactRangePlannerTest.java | 1 +
.../DataEvolutionNormalCompactTaskTest.java | 591 ++++++++++++++++++++-
.../table/DataEvolutionDeletionVectorTest.java | 113 ++++
.../paimon/spark/procedure/CompactProcedure.java | 27 +-
.../procedure/DataEvolutionRewriteExecutor.java | 45 +-
.../spark/procedure/CompactProcedureTestBase.scala | 90 +++-
18 files changed, 1486 insertions(+), 77 deletions(-)
diff --git a/docs/docs/multimodal-table/data-evolution.mdx
b/docs/docs/multimodal-table/data-evolution.mdx
index 02b1d27187..c02437b428 100644
--- a/docs/docs/multimodal-table/data-evolution.mdx
+++ b/docs/docs/multimodal-table/data-evolution.mdx
@@ -608,6 +608,47 @@ Ordinary compaction **does not physically remove rows
hidden by deletion
vectors**. Removing their positions would change the alignment with untouched
column files.
+### Splitting Large Files During Compaction
+
+To resize existing large normal data files, enable
+`data-evolution.compaction.split-large-files` (default: `false`) and run
compaction:
+
+```sql
+ALTER TABLE my_table SET TBLPROPERTIES (
+ 'target-file-size' = '128 MB',
+ 'data-evolution.compaction.split-large-files' = 'true',
+ 'data-evolution.compaction.large-file-ratio' = '3.0'
+);
+CALL sys.compact('default.my_table');
+```
+
+The example uses Spark SQL and selects normal files strictly larger than 384
MB.
+`data-evolution.compaction.large-file-ratio` controls the multiplier relative
to
+`target-file-size`; it defaults to `2.0` and accepts finite values of at least
`1.0`,
+including fractional values such as `1.5`. Files strictly exceeding the
threshold
+qualify for compaction below `compaction.min.file-num` when their
dedicated-file
+ranges allow splitting.
+Changing the ratio does not change the output target size. Compaction includes
all
+column updates for the same row-ID range. Before writing, it estimates rows
per output
+from the total normal input file size, the logical row count, and
`target-file-size`.
+It then adjusts the estimated cut points to safe dedicated-file boundaries.
+Actual output sizes can differ from the target because of compression, data
skew,
+overwritten column versions, and dedicated-file ranges; the last file may be
smaller.
+The write-time `target-file-row-num` limit does not apply to compaction.
+
+Row IDs, column updates, and logical deletions are preserved. This option only
rewrites
+normal files: associated BLOB and VECTOR files keep their existing contents
and file names,
+and their sizes do not trigger splitting. Separate dedicated-file compaction
options keep
+their existing behavior. Files referenced by older snapshots or tags remain
until those
+references expire and snapshot expiration removes them.
+
+Every BLOB or VECTOR file must remain fully contained in a single normal file's
+row-ID range. An estimated cut inside a dedicated file moves to the end of its
range,
+including any overlapping ranges from different columns or versions and ranges
produced
+by dedicated compaction in the same batch. Output files may therefore exceed
+`target-file-size`. If these ranges prevent any split, file size alone does
not trigger a compaction task. Normal
+merging based on `compaction.min.file-num` remains available.
+
### Materialize Deletion Vectors
To remove deleted positions from the latest table state, run:
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 6a4955280d..bd231c88c0 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -506,12 +506,24 @@ under the License.
<td>Duration</td>
<td>The TTL in local index for cross partition upsert (primary
keys not contain all partition fields), this can avoid maintaining too many
indexes and lead to worse and worse performance, but please note that this may
also cause data duplication.</td>
</tr>
+ <tr>
+ <td><h5>data-evolution.compaction.large-file-ratio</h5></td>
+ <td style="word-wrap: break-word;">2.0</td>
+ <td>Double</td>
+ <td>Size multiplier relative to target-file-size for selecting
large normal files when data-evolution.compaction.split-large-files is enabled.
An individual file must strictly exceed this threshold. The value must be
finite and at least 1.0. This does not change the target size of compacted
output files.</td>
+ </tr>
<tr>
<td><h5>data-evolution.compaction.rewrite-row-ids</h5></td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
<td>Legacy compatibility option. Setting this option to true
fails. Data-evolution compaction preserves row IDs and logical deletions. Use
the 'materialize_deletion_vectors' procedure to apply deletion vectors to the
latest table state and assign new row IDs. Reclaiming files retained by
historical snapshots or tags requires snapshot expiration.</td>
</tr>
+ <tr>
+ <td><h5>data-evolution.compaction.split-large-files</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether data-evolution compaction selects normal data files
larger than data-evolution.compaction.large-file-ratio times target-file-size,
even below compaction.min.file-num when dedicated-file ranges allow splitting.
Normal output ranges are estimated from input file sizes and row counts toward
target-file-size, then adjusted to avoid cutting through any BLOB or VECTOR
file range. Actual output sizes may differ from the target. Row IDs and logical
deletions are preserved [...]
+ </tr>
<tr>
<td><h5>data-evolution.enabled</h5></td>
<td style="word-wrap: break-word;">false</td>
@@ -1817,7 +1829,7 @@ The size strategy ranges by the data size allocated to
each sorting task, which
<td><h5>target-file-row-num</h5></td>
<td style="word-wrap: break-word;">9223372036854775807</td>
<td>Long</td>
- <td>Target number of rows per newly written data file; a file
rolls when this or target-file-size is reached, whichever comes first. Enforced
at bundle granularity, so a bundled write may exceed it by up to one bundle.
Only constrains files at write time: compaction is size-based and may merge
into larger files, and data-evolution compaction still produces a single file.
Bounds per-file rows for wide columns to avoid data-evolution OOM. PyPaimon
supports this for data-evoluti [...]
+ <td>Target number of rows per newly written data file; a file
rolls when this or target-file-size is reached, whichever comes first. Enforced
at bundle granularity, so a bundled write may exceed it by up to one bundle.
Only constrains files at write time: compaction is size-based and may merge
into larger files, and data-evolution compaction produces a single file unless
data-evolution.compaction.split-large-files is enabled. Bounds per-file rows
for wide columns to avoid dat [...]
</tr>
<tr>
<td><h5>target-file-size</h5></td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 298126f3c6..41b338cb9e 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -865,7 +865,8 @@ public class CoreOptions implements Serializable {
+ "Enforced at bundle granularity, so a
bundled write may exceed it "
+ "by up to one bundle. Only constrains
files at write time: "
+ "compaction is size-based and may merge
into larger files, and "
- + "data-evolution compaction still
produces a single file. Bounds "
+ + "data-evolution compaction produces a
single file unless "
+ +
"data-evolution.compaction.split-large-files is enabled. Bounds "
+ "per-file rows for wide columns to avoid
data-evolution OOM. "
+ "PyPaimon supports this for
data-evolution append tables; its "
+ "primary-key, blob and vector writers
still fail fast when it "
@@ -2633,6 +2634,31 @@ public class CoreOptions implements Serializable {
.withDescription(
"Whether to persist source when process merge into
action on data evolution table.");
+ public static final ConfigOption<Boolean>
DATA_EVOLUTION_COMPACTION_SPLIT_LARGE_FILES =
+ key("data-evolution.compaction.split-large-files")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether data-evolution compaction selects normal
data files larger than "
+ +
"data-evolution.compaction.large-file-ratio times target-file-size, "
+ + "even below compaction.min.file-num when
dedicated-file ranges allow splitting. "
+ + "Normal output ranges are estimated from
input file sizes and row counts "
+ + "toward target-file-size, then adjusted
to avoid cutting through any "
+ + "BLOB or VECTOR file range. Actual
output sizes may differ from the target. "
+ + "Row IDs and logical deletions are
preserved, and associated "
+ + "BLOB and VECTOR files are not rewritten
by this option.");
+
+ public static final ConfigOption<Double>
DATA_EVOLUTION_COMPACTION_LARGE_FILE_RATIO =
+ key("data-evolution.compaction.large-file-ratio")
+ .doubleType()
+ .defaultValue(2.0d)
+ .withDescription(
+ "Size multiplier relative to target-file-size for
selecting large normal "
+ + "files when
data-evolution.compaction.split-large-files is enabled. "
+ + "An individual file must strictly exceed
this threshold. The value "
+ + "must be finite and at least 1.0. This
does not change the target "
+ + "size of compacted output files.");
+
public static final ConfigOption<Boolean>
DATA_EVOLUTION_COMPACTION_REWRITE_ROW_IDS =
key("data-evolution.compaction.rewrite-row-ids")
.booleanType()
@@ -4389,6 +4415,19 @@ public class CoreOptions implements Serializable {
return options.get(DELETION_VECTOR_BITMAP64);
}
+ public boolean dataEvolutionCompactionSplitLargeFiles() {
+ return options.get(DATA_EVOLUTION_COMPACTION_SPLIT_LARGE_FILES);
+ }
+
+ public double dataEvolutionCompactionLargeFileRatio() {
+ double ratio = options.get(DATA_EVOLUTION_COMPACTION_LARGE_FILE_RATIO);
+ checkArgument(
+ Double.isFinite(ratio) && ratio >= 1.0d,
+ "The option %s must be finite and at least 1.0.",
+ DATA_EVOLUTION_COMPACTION_LARGE_FILE_RATIO.key());
+ return ratio;
+ }
+
public boolean dataEvolutionCompactionRewriteRowIds() {
return options.get(DATA_EVOLUTION_COMPACTION_REWRITE_ROW_IDS);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/CompactCandidateRangeCollector.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/CompactCandidateRangeCollector.java
index 947f944182..e991892fd4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/CompactCandidateRangeCollector.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/CompactCandidateRangeCollector.java
@@ -53,6 +53,7 @@ final class CompactCandidateRangeCollector {
private final long blobTargetFileSize;
private final long openFileCost;
private final long compactMinFileNum;
+ private final long largeFileThreshold;
private final List<SortedEntryChunk> sortedChunks = new ArrayList<>();
private long[] words;
private int chunkSize;
@@ -64,7 +65,8 @@ final class CompactCandidateRangeCollector {
long targetFileSize,
long blobTargetFileSize,
long openFileCost,
- long compactMinFileNum) {
+ long compactMinFileNum,
+ long largeFileThreshold) {
checkArgument(expectedFileCount >= 0, "Expected live file count cannot
be negative.");
checkArgument(targetFileSize > 0, "Target file size must be
positive.");
checkArgument(blobTargetFileSize > 0, "Blob target file size must be
positive.");
@@ -75,6 +77,7 @@ final class CompactCandidateRangeCollector {
this.blobTargetFileSize = blobTargetFileSize;
this.openFileCost = openFileCost;
this.compactMinFileNum = compactMinFileNum;
+ this.largeFileThreshold = largeFileThreshold;
int initialEntries = Math.max(16, Math.min(expectedFileCount,
ENTRY_CHUNK_SIZE));
this.words = new long[Math.multiplyExact(initialEntries, ENTRY_WORDS)];
}
@@ -137,6 +140,7 @@ final class CompactCandidateRangeCollector {
blobTargetFileSize,
openFileCost,
compactMinFileNum,
+ largeFileThreshold,
consumer);
if (chunks.size() == 1) {
SortedEntryChunk chunk = chunks.get(0);
@@ -402,6 +406,7 @@ final class CompactCandidateRangeCollector {
private final long blobTargetFileSize;
private final long openFileCost;
private final long compactMinFileNum;
+ private final long largeFileThreshold;
private final CandidateRangeConsumer consumer;
private final CandidateBin bin = new CandidateBin();
private final Map<Integer, BlobFieldAccumulator> blobFields = new
HashMap<>();
@@ -412,6 +417,7 @@ final class CompactCandidateRangeCollector {
private long normalEnd;
private long normalFileCount;
private long normalWeight;
+ private boolean largeFile;
private long vectorFileCount;
private int componentFileCount;
private boolean hasPreviousLogicalRange;
@@ -422,11 +428,13 @@ final class CompactCandidateRangeCollector {
long blobTargetFileSize,
long openFileCost,
long compactMinFileNum,
+ long largeFileThreshold,
CandidateRangeConsumer consumer) {
this.targetFileSize = targetFileSize;
this.blobTargetFileSize = blobTargetFileSize;
this.openFileCost = openFileCost;
this.compactMinFileNum = compactMinFileNum;
+ this.largeFileThreshold = largeFileThreshold;
this.consumer = consumer;
}
@@ -457,6 +465,7 @@ final class CompactCandidateRangeCollector {
normalEnd = end;
normalFileCount = 1L;
normalWeight = Math.max(fileSize, openFileCost);
+ largeFile = fileSize > largeFileThreshold;
vectorFileCount = 0L;
componentFileCount = 1;
blobFields.clear();
@@ -472,6 +481,7 @@ final class CompactCandidateRangeCollector {
checkState(
normalEnd == end,
"Normal files in one overlapping row-id group must
have the same row-id range.");
+ largeFile |= fileSize > largeFileThreshold;
normalFileCount = Math.addExact(normalFileCount, 1L);
normalWeight = Math.addExact(normalWeight, Math.max(fileSize,
openFileCost));
componentFileCount = Math.addExact(componentFileCount, 1);
@@ -525,7 +535,8 @@ final class CompactCandidateRangeCollector {
componentFileCount,
normalFileCount,
normalWeight,
- dedicatedCandidate);
+ dedicatedCandidate,
+ largeFile);
if (normalWeight > targetFileSize) {
flushBin();
emitComponent(component);
@@ -541,7 +552,9 @@ final class CompactCandidateRangeCollector {
}
private void emitComponent(Component component) {
- if (component.normalFileCount >= compactMinFileNum ||
component.dedicatedCandidate) {
+ if (component.normalFileCount >= compactMinFileNum
+ || component.dedicatedCandidate
+ || component.largeFile) {
consumer.accept(component.start, component.end,
component.fileCount);
}
}
@@ -571,6 +584,7 @@ final class CompactCandidateRangeCollector {
private final long normalFileCount;
private final long normalWeight;
private final boolean dedicatedCandidate;
+ private final boolean largeFile;
private Component(
long start,
@@ -578,13 +592,15 @@ final class CompactCandidateRangeCollector {
int fileCount,
long normalFileCount,
long normalWeight,
- boolean dedicatedCandidate) {
+ boolean dedicatedCandidate,
+ boolean largeFile) {
this.start = start;
this.end = end;
this.fileCount = fileCount;
this.normalFileCount = normalFileCount;
this.normalWeight = normalWeight;
this.dedicatedCandidate = dedicatedCandidate;
+ this.largeFile = largeFile;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinator.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinator.java
index 23f15772a0..5d80b8d971 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinator.java
@@ -41,6 +41,7 @@ import org.apache.paimon.utils.RangeHelper;
import javax.annotation.Nullable;
+import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -57,7 +58,9 @@ import static java.util.Comparator.comparingLong;
import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
import static org.apache.paimon.types.BlobType.isBlobFileField;
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+import static
org.apache.paimon.utils.DataEvolutionUtils.checkContiguousRowRange;
import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
/** Compact coordinator to compact data evolution table. */
public class DataEvolutionCompactCoordinator {
@@ -99,6 +102,11 @@ public class DataEvolutionCompactCoordinator {
validateOptions(options);
long targetFileSize = options.targetFileSize(false);
+ long largeFileThreshold =
+ options.dataEvolutionCompactionSplitLargeFiles()
+ ? largeFileThreshold(
+ targetFileSize,
options.dataEvolutionCompactionLargeFileRatio())
+ : Long.MAX_VALUE;
long openFileCost = options.splitOpenFileCost();
long compactMinFileNum = options.compactionMinFileNum();
Set<String> blobInlineFields = options.blobInlineField();
@@ -118,6 +126,7 @@ public class DataEvolutionCompactCoordinator {
new DataEvolutionCompactRangePlanner.CandidateOptions(
compactBlob,
compactVector,
+ largeFileThreshold,
targetFileSize,
options.blobTargetFileSize(),
openFileCost,
@@ -137,6 +146,7 @@ public class DataEvolutionCompactCoordinator {
new CompactPlanner(
compactBlob,
compactVector,
+ largeFileThreshold,
targetFileSize,
options.blobTargetFileSize(),
openFileCost,
@@ -145,6 +155,14 @@ public class DataEvolutionCompactCoordinator {
currentBlobFieldIds);
}
+ static long largeFileThreshold(long targetFileSize, double ratio) {
+ // Preserve decimal boundaries and saturate thresholds beyond the
largest possible file.
+ return BigDecimal.valueOf(targetFileSize)
+ .multiply(BigDecimal.valueOf(ratio))
+ .min(BigDecimal.valueOf(Long.MAX_VALUE))
+ .longValue();
+ }
+
public static void validateOptions(CoreOptions options) {
checkArgument(
!options.dataEvolutionCompactionRewriteRowIds(),
@@ -155,6 +173,12 @@ public class DataEvolutionCompactCoordinator {
CoreOptions.DATA_EVOLUTION_COMPACTION_REWRITE_ROW_IDS.key());
}
+ /** Prevents this run from repeatedly compacting bins containing only its
own normal outputs. */
+ public DataEvolutionCompactCoordinator
withCompletedNormalFiles(Set<String> fileNames) {
+ planner.completedNormalFiles = checkNotNull(fileNames);
+ return this;
+ }
+
public List<DataEvolutionCompactTask> plan() {
// scan files in snapshot
List<ManifestEntry> entries = scanner.scan();
@@ -235,12 +259,14 @@ public class DataEvolutionCompactCoordinator {
private final boolean compactBlob;
private final boolean compactVector;
+ private final long largeFileThreshold;
private final long targetFileSize;
private final long blobTargetFileSize;
private final long openFileCost;
private final long compactMinFileNum;
private final LongFunction<RowType> schemaFetcher;
@Nullable private final Set<Integer> currentBlobFieldIds;
+ private Set<String> completedNormalFiles = Collections.emptySet();
@VisibleForTesting
CompactPlanner(
@@ -252,6 +278,7 @@ public class DataEvolutionCompactCoordinator {
this(
compactBlob,
compactVector,
+ Long.MAX_VALUE,
targetFileSize,
targetFileSize,
openFileCost,
@@ -266,6 +293,7 @@ public class DataEvolutionCompactCoordinator {
CompactPlanner(
boolean compactBlob,
boolean compactVector,
+ long largeFileThreshold,
long targetFileSize,
long blobTargetFileSize,
long openFileCost,
@@ -274,6 +302,7 @@ public class DataEvolutionCompactCoordinator {
@Nullable Set<Integer> currentBlobFieldIds) {
this.compactBlob = compactBlob;
this.compactVector = compactVector;
+ this.largeFileThreshold = largeFileThreshold;
this.targetFileSize = targetFileSize;
this.blobTargetFileSize = blobTargetFileSize;
this.openFileCost = openFileCost;
@@ -314,12 +343,10 @@ public class DataEvolutionCompactCoordinator {
}
}
- if (compactBlob) {
- associateDedicatedFiles(blobFiles, treeMap,
dataFileToBlobFiles);
- }
- if (compactVector) {
- associateDedicatedFiles(vectorStoreFiles, treeMap,
dataFileToVectorStoreFiles);
- }
+ // Retained dedicated files constrain normal output boundaries
even when their
+ // own compaction is disabled.
+ associateDedicatedFiles(blobFiles, treeMap,
dataFileToBlobFiles);
+ associateDedicatedFiles(vectorStoreFiles, treeMap,
dataFileToVectorStoreFiles);
RangeHelper<DataFileMeta> continuousDataRangeHelper =
new RangeHelper<>(
@@ -409,10 +436,32 @@ public class DataEvolutionCompactCoordinator {
List<DataFileMeta> dataFiles = compactBin.files();
List<DataEvolutionCompactTask> tasks = new ArrayList<>();
- boolean triggerNormalFile = dataFiles.size() >= compactMinFileNum;
- if (triggerNormalFile) {
- tasks.add(new DataEvolutionNormalCompactTask(partition,
dataFiles));
+ List<Range> protectedRanges = new ArrayList<>();
+ for (DataFileMeta dataFile : dataFiles) {
+ dataFileToBlobFiles
+ .getOrDefault(dataFile, Collections.emptyList())
+ .forEach(file ->
protectedRanges.add(file.nonNullRowIdRange()));
+ dataFileToVectorStoreFiles
+ .getOrDefault(dataFile, Collections.emptyList())
+ .forEach(file ->
protectedRanges.add(file.nonNullRowIdRange()));
}
+ Range normalRange = checkContiguousRowRange(dataFiles);
+ boolean hasUncompactedFiles =
+ dataFiles.stream()
+ .anyMatch(file ->
!completedNormalFiles.contains(file.fileName()));
+ boolean triggerNormalFile =
+ hasUncompactedFiles
+ && (dataFiles.size() >= compactMinFileNum
+ || (canSplit(normalRange, protectedRanges)
+ && dataFiles.stream()
+ .anyMatch(
+ f ->
+
f.fileSize()
+
> largeFileThreshold
+ &&
!completedNormalFiles
+
.contains(
+
f
+
.fileName()))));
if (compactBlob) {
if (triggerNormalFile) {
@@ -464,9 +513,32 @@ public class DataEvolutionCompactCoordinator {
}
}
}
+ if (triggerNormalFile) {
+ // Dedicated compaction in the same batch may combine existing
file ranges.
+ // Its outputs must also fit inside a single normal output
file.
+ for (DataEvolutionCompactTask task : tasks) {
+
protectedRanges.add(checkContiguousRowRange(task.compactBefore()));
+ }
+ if (dataFiles.size() >= compactMinFileNum
+ || canSplit(normalRange, protectedRanges)) {
+ tasks.add(
+ 0,
+ new DataEvolutionNormalCompactTask(
+ partition, dataFiles, protectedRanges));
+ }
+ }
return tasks;
}
+ private boolean canSplit(Range normalRange, List<Range>
protectedRanges) {
+ return normalRange.from < normalRange.to
+ && Range.sortAndMergeOverlap(protectedRanges).stream()
+ .noneMatch(
+ range ->
+ range.from <= normalRange.from
+ && range.to >=
normalRange.to);
+ }
+
private CompactBin compactBin(List<DataFileMeta> files, long
groupWeight) {
CompactBin bin = new CompactBin(targetFileSize);
bin.add(files, groupWeight);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactDeletionVectorRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactDeletionVectorRewriter.java
index cb7e36b5fe..e6454246c7 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactDeletionVectorRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactDeletionVectorRewriter.java
@@ -42,6 +42,7 @@ import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.TreeMap;
import java.util.stream.Collectors;
import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
@@ -136,32 +137,31 @@ public class DataEvolutionCompactDeletionVectorRewriter {
continue;
}
- checkState(
- after.size() == 1,
- "Only one normal file should be generated by a single
compact task.");
Range beforeRange = checkContiguousRowRange(before);
- Range afterRange = after.get(0).nonNullRowIdRange();
+ Range afterRange = checkContiguousRowRange(after);
checkState(
beforeRange.equals(afterRange),
"Non-materialized data evolution compaction should keep
the same row-id "
+ "range, but compact before range is %s and
compact after range is %s.",
beforeRange,
afterRange);
- DeletionVector merged = newDeletionVector();
+ TreeMap<Long, DataFileMeta> outputFiles = new TreeMap<>();
+ for (DataFileMeta file : after) {
+ outputFiles.put(file.nonNullFirstRowId(), file);
+ }
+ Map<String, DeletionVector> rewritten = new LinkedHashMap<>();
- // Merge all old DeletionVectors, should consider row range offset
of each sub dv
+ // Redistribute old deletion positions across the output file
boundaries.
for (List<DataFileMeta> previousRowGroup :
rangeHelper.mergeOverlappingRanges(before)) {
DataFileMeta oldAnchor = retrieveAnchorFile(previousRowGroup,
file -> file);
moveDeletionVector(
maintainer,
- merged,
+ outputFiles,
+ rewritten,
oldAnchor.fileName(),
- oldAnchor.nonNullRowIdRange(),
- afterRange);
- }
- if (!merged.isEmpty()) {
- maintainer.notifyNewDeletionVector(after.get(0).fileName(),
merged);
+ oldAnchor.nonNullRowIdRange());
}
+ rewritten.forEach(maintainer::notifyNewDeletionVector);
}
return result;
}
@@ -185,27 +185,32 @@ public class DataEvolutionCompactDeletionVectorRewriter {
}
/**
- * 'Move' the deletion vectors of old anchor files to the new anchor file.
This may merge
- * several deletion vectors into one if row-level compaction is triggered,
+ * Moves old anchor-file deletion vectors to the new files, merging or
splitting them at the
+ * output row-id boundaries.
*/
private void moveDeletionVector(
AppendDeleteFileMaintainer maintainer,
- DeletionVector merged,
+ TreeMap<Long, DataFileMeta> outputFiles,
+ Map<String, DeletionVector> rewritten,
String oldFileName,
- Range oldRange,
- Range newRange) {
+ Range oldRange) {
DeletionVector old = maintainer.getDeletionVector(oldFileName);
if (old == null || old.isEmpty()) {
return;
}
- // Fast path for renaming deletion vectors only
- if (oldRange.equals(newRange)) {
- merged.merge(old);
+ // Fast path for renaming deletion vectors only.
+ DataFileMeta firstOutput =
outputFiles.floorEntry(oldRange.from).getValue();
+ if (oldRange.equals(firstOutput.nonNullRowIdRange())) {
+ rewritten
+ .computeIfAbsent(firstOutput.fileName(), ignored ->
newDeletionVector())
+ .merge(old);
} else {
old.forEachDeletedPosition(
position -> {
long absolutePosition = oldRange.from + position;
+ DataFileMeta output =
outputFiles.floorEntry(absolutePosition).getValue();
+ Range newRange = output.nonNullRowIdRange();
long newPosition = absolutePosition - newRange.from;
checkState(
newPosition >= 0 && newPosition <
newRange.count(),
@@ -213,7 +218,9 @@ public class DataEvolutionCompactDeletionVectorRewriter {
absolutePosition,
oldRange,
newRange);
- merged.delete(newPosition);
+ rewritten
+ .computeIfAbsent(output.fileName(), ignored ->
newDeletionVector())
+ .delete(newPosition);
});
}
maintainer.notifyRemovedDeletionVector(oldFileName);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactRangePlanner.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactRangePlanner.java
index d7abda0708..236f4fdf53 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactRangePlanner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactRangePlanner.java
@@ -157,7 +157,8 @@ final class DataEvolutionCompactRangePlanner {
candidateOptions.targetFileSize,
candidateOptions.blobTargetFileSize,
candidateOptions.openFileCost,
- candidateOptions.compactMinFileNum);
+ candidateOptions.compactMinFileNum,
+ candidateOptions.largeFileThreshold);
try {
collectDeletedIdentifiers(manifestGroup, deletedIdentifiers,
identifier);
collectCandidateRanges(manifestGroup, deletedIdentifiers,
identifier, candidateRanges);
@@ -489,6 +490,7 @@ final class DataEvolutionCompactRangePlanner {
private final boolean compactBlob;
private final boolean compactVector;
+ private final long largeFileThreshold;
private final long targetFileSize;
private final long blobTargetFileSize;
private final long openFileCost;
@@ -499,6 +501,7 @@ final class DataEvolutionCompactRangePlanner {
CandidateOptions(
boolean compactBlob,
boolean compactVector,
+ long largeFileThreshold,
long targetFileSize,
long blobTargetFileSize,
long openFileCost,
@@ -507,6 +510,7 @@ final class DataEvolutionCompactRangePlanner {
@Nullable Set<Integer> currentBlobFieldIds) {
this.compactBlob = compactBlob;
this.compactVector = compactVector;
+ this.largeFileThreshold = largeFileThreshold;
this.targetFileSize = targetFileSize;
this.blobTargetFileSize = blobTargetFileSize;
this.openFileCost = openFileCost;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
index 0019c4e17f..6f24a41bcc 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTask.java
@@ -61,7 +61,7 @@ public abstract class DataEvolutionCompactTask {
Map<String, String> options = new HashMap<>();
options.put(CoreOptions.TARGET_FILE_SIZE.key(), "99999 G");
options.put(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "99999 G");
- // Data evolution requires a single output file, so the row limit must
not roll it either.
+ // Compaction is size-based; the write-time row limit must not roll
its output.
options.put(CoreOptions.TARGET_FILE_ROW_NUM.key(),
String.valueOf(Long.MAX_VALUE));
return Collections.unmodifiableMap(options);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializer.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializer.java
index 40ae2073dd..e9cc9ed191 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactTaskSerializer.java
@@ -27,6 +27,7 @@ import org.apache.paimon.io.DataInputView;
import org.apache.paimon.io.DataOutputView;
import org.apache.paimon.io.DataOutputViewStreamWrapper;
import org.apache.paimon.table.source.DeletionFile;
+import org.apache.paimon.utils.Range;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
@@ -40,7 +41,7 @@ import static
org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
public class DataEvolutionCompactTaskSerializer
implements VersionedSerializer<DataEvolutionCompactTask> {
- private static final int CURRENT_VERSION = 3;
+ private static final int CURRENT_VERSION = 4;
private final DataFileMetaSerializer dataFileSerializer;
@@ -73,7 +74,14 @@ public class DataEvolutionCompactTaskSerializer
serializeBinaryRow(task.partition(), view);
dataFileSerializer.serializeList(task.compactBefore(), view);
view.writeInt(task.type().code());
- if (task.type() ==
DataEvolutionCompactTask.TaskType.MATERIALIZE_DELETION) {
+ if (task.type() == DataEvolutionCompactTask.TaskType.NORMAL) {
+ List<Range> ranges = ((DataEvolutionNormalCompactTask)
task).protectedRanges();
+ view.writeInt(ranges.size());
+ for (Range range : ranges) {
+ view.writeLong(range.from);
+ view.writeLong(range.to);
+ }
+ } else if (task.type() ==
DataEvolutionCompactTask.TaskType.MATERIALIZE_DELETION) {
DeletionFile.serializeList(
view, ((DataEvolutionMaterializeDeletionCompactTask)
task).deletionFiles());
}
@@ -117,7 +125,12 @@ public class DataEvolutionCompactTaskSerializer
DataEvolutionCompactTask.TaskType.fromCode(view.readInt());
switch (type) {
case NORMAL:
- return new DataEvolutionNormalCompactTask(partition, files);
+ int rangeCount = view.readInt();
+ List<Range> ranges = new ArrayList<>(rangeCount);
+ for (int i = 0; i < rangeCount; i++) {
+ ranges.add(new Range(view.readLong(), view.readLong()));
+ }
+ return new DataEvolutionNormalCompactTask(partition, files,
ranges);
case BLOB:
return new DataEvolutionBlobCompactTask(partition, files);
case MATERIALIZE_DELETION:
diff --git
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
index 7fe5739744..6a46a4425f 100644
---
a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
+++
b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java
@@ -20,12 +20,14 @@ package org.apache.paimon.append.dataevolution;
import org.apache.paimon.AppendOnlyFileStore;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.operation.AppendFileStoreWrite;
import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.reader.RecordReaderIterator;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
@@ -35,6 +37,7 @@ import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RecordWriter;
import org.apache.paimon.utils.SetUtils;
@@ -43,9 +46,12 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -62,9 +68,33 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
private static final Logger LOG =
LoggerFactory.getLogger(DataEvolutionNormalCompactTask.class);
+ private final List<Range> protectedRanges;
+
public DataEvolutionNormalCompactTask(BinaryRow partition,
List<DataFileMeta> files) {
+ this(partition, files, Collections.emptyList());
+ }
+
+ public DataEvolutionNormalCompactTask(
+ BinaryRow partition, List<DataFileMeta> files, List<Range>
protectedRanges) {
super(partition, files);
checkContiguousRowRange(files);
+ this.protectedRanges =
+
Collections.unmodifiableList(Range.sortAndMergeOverlap(protectedRanges));
+ }
+
+ public List<Range> protectedRanges() {
+ return protectedRanges;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ return super.equals(other)
+ && protectedRanges.equals(((DataEvolutionNormalCompactTask)
other).protectedRanges);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), protectedRanges);
}
@Override
@@ -86,7 +116,13 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
fieldNamesInBlobFile(table.rowType(),
options.blobInlineField()),
fieldNamesInVectorFile(table.rowType(),
options.withVectorFormat()));
- table = table.copy(DYNAMIC_WRITE_OPTIONS);
+ Map<String, String> writeOptions = new
HashMap<>(DYNAMIC_WRITE_OPTIONS);
+ if (options.dataEvolutionCompactionSplitLargeFiles()) {
+ // Buffer flushes may close files before reaching a safe
dedicated-file boundary.
+ writeOptions.put(CoreOptions.WRITE_BUFFER_FOR_APPEND.key(),
"false");
+ writeOptions.put(CoreOptions.TARGET_FILE_SIZE.key(),
Long.MAX_VALUE + " b");
+ }
+ table = table.copy(writeOptions);
long firstRowId = compactBefore.get(0).nonNullFirstRowId();
RowType readWriteType =
@@ -111,20 +147,26 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
storeWrite.withWriteType(readWriteType);
storeWrite.withFileSource(FileSource.COMPACT);
RecordWriter<InternalRow> writer = storeWrite.createWriter(partition,
0);
-
- reader.forEachRemaining(
- row -> {
- try {
- writer.write(row);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
-
- List<DataFileMeta> writeResult =
writer.prepareCommit(false).newFilesIncrement().newFiles();
- checkArgument(
- writeResult.size() == 1, "Data evolution compaction should
produce one file.");
-
+ List<Range> outputRanges =
+ options.dataEvolutionCompactionSplitLargeFiles()
+ ? planOutputRanges(options.targetFileSize(false))
+ :
Collections.singletonList(checkContiguousRowRange(compactBefore));
+ List<DataFileMeta> writeResult = new ArrayList<>();
+ try (RecordReaderIterator<InternalRow> iterator = new
RecordReaderIterator<>(reader)) {
+ for (Range range : outputRanges) {
+ for (long remaining = range.count(); remaining > 0;
remaining--) {
+ checkArgument(iterator.hasNext(), "Missing rows in normal
compaction input.");
+ writer.write(iterator.next());
+ }
+ List<DataFileMeta> output =
+
writer.prepareCommit(false).newFilesIncrement().newFiles();
+ checkArgument(
+ output.size() == 1,
+ "Each planned compaction range should produce one
normal file.");
+ writeResult.add(output.get(0));
+ }
+ checkArgument(!iterator.hasNext(), "Unexpected extra rows in
normal compaction input.");
+ }
try {
writer.close();
storeWrite.close();
@@ -132,22 +174,61 @@ public class DataEvolutionNormalCompactTask extends
DataEvolutionCompactTask {
LOG.warn("Failed to close reader and writer.", e);
}
- DataFileMeta dataFileMeta =
writeResult.get(0).assignFirstRowId(firstRowId);
- dataFileMeta =
- dataFileMeta.assignSequenceNumber(
- minSequenceId(compactBefore),
maxSequenceId(compactBefore));
- if (options.ignoreIndexColumnUpdate()) {
- long[] columnMaxSequenceNumbers =
- compactedColumnMaxSequenceNumbers(table, dataFileMeta);
+ long minSequenceNumber = minSequenceId(compactBefore);
+ long maxSequenceNumber = maxSequenceId(compactBefore);
+ long nextRowId = firstRowId;
+ long[] columnMaxSequenceNumbers =
+ options.ignoreIndexColumnUpdate() && !writeResult.isEmpty()
+ ? compactedColumnMaxSequenceNumbers(
+ table,
+ writeResult
+ .get(0)
+
.assignSequenceNumber(minSequenceNumber, maxSequenceNumber))
+ : null;
+ for (DataFileMeta file : writeResult) {
+ DataFileMeta dataFileMeta =
+ file.assignFirstRowId(nextRowId)
+ .assignSequenceNumber(minSequenceNumber,
maxSequenceNumber);
if (columnMaxSequenceNumbers != null) {
dataFileMeta =
dataFileMeta.withColumnMaxSequenceNumbers(columnMaxSequenceNumbers);
}
+ compactAfter.add(dataFileMeta);
+ nextRowId += dataFileMeta.rowCount();
}
- compactAfter.add(dataFileMeta);
+ checkSameRowRange("Normal file", compactBefore, compactAfter);
return commitMessage(compactBefore, compactAfter);
}
+ @VisibleForTesting
+ List<Range> planOutputRanges(long targetFileSize) {
+ Range inputRange = checkContiguousRowRange(compactBefore);
+ double inputSize =
compactBefore.stream().mapToDouble(DataFileMeta::fileSize).sum();
+ long targetRows = Math.max(1L, (long) (inputRange.count() *
(targetFileSize / inputSize)));
+ List<Range> result = new ArrayList<>();
+ int protectedIndex = 0;
+ long start = inputRange.from;
+ while (true) {
+ long end = start + Math.min(targetRows - 1, inputRange.to - start);
+ while (protectedIndex < protectedRanges.size()
+ && protectedRanges.get(protectedIndex).to <= end) {
+ protectedIndex++;
+ }
+ if (protectedIndex < protectedRanges.size()
+ && protectedRanges.get(protectedIndex).from <= end) {
+ end = protectedRanges.get(protectedIndex).to;
+ }
+ checkArgument(
+ end <= inputRange.to,
+ "Dedicated range must be contained in the normal
compaction range.");
+ result.add(new Range(start, end));
+ if (end == inputRange.to) {
+ return result;
+ }
+ start = end + 1;
+ }
+ }
+
@Nullable
private long[] compactedColumnMaxSequenceNumbers(
FileStoreTable table, DataFileMeta outputFile) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactCandidateRangeCollectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactCandidateRangeCollectorTest.java
index 29c368f9f7..a6a3db3911 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactCandidateRangeCollectorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/CompactCandidateRangeCollectorTest.java
@@ -19,6 +19,8 @@
package org.apache.paimon.append.dataevolution;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import java.util.ArrayList;
import java.util.List;
@@ -26,6 +28,7 @@ import java.util.List;
import static
org.apache.paimon.append.dataevolution.CompactCandidateRangeCollector.IGNORED_DEDICATED_FILE;
import static
org.apache.paimon.append.dataevolution.CompactCandidateRangeCollector.NORMAL_FILE;
import static
org.apache.paimon.append.dataevolution.CompactCandidateRangeCollector.VECTOR_FILE;
+import static
org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator.largeFileThreshold;
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link CompactCandidateRangeCollector}. */
@@ -45,6 +48,53 @@ class CompactCandidateRangeCollectorTest {
assertThat(collector.retainedWordCount()).isZero();
}
+ @Test
+ void testSplitLargeFilesUsesPhysicalSizeAndIncludesColumnUpdates() {
+ for (boolean enabled : new boolean[] {false, true}) {
+ CompactCandidateRangeCollector collector =
+ new CompactCandidateRangeCollector(
+ 16, 100L, 100L, 1000L, 10L, enabled ? 200L :
Long.MAX_VALUE);
+ collector.add(0, NORMAL_FILE, 0L, 10L, 199L);
+ collector.add(0, NORMAL_FILE, 10L, 10L, 200L);
+ collector.add(0, NORMAL_FILE, 20L, 10L, 201L);
+ collector.add(0, NORMAL_FILE, 20L, 10L, 10L);
+ // Dedicated files never trigger normal-file splitting.
+ collector.add(0, 3, 0L, 10L, 1000L);
+ if (enabled) {
+ assertThat(finish(collector)).containsExactly("20-29:2");
+ } else {
+ assertThat(finish(collector)).isEmpty();
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"1.0,100", "1.15,115", "1.5,150", "3.0,300"})
+ void testCustomLargeFileRatioUsesIndividualPhysicalFileSize(double ratio,
long threshold) {
+ CompactCandidateRangeCollector collector =
+ new CompactCandidateRangeCollector(
+ 16, 100L, 100L, 1000L, 10L, largeFileThreshold(100L,
ratio));
+ collector.add(0, NORMAL_FILE, 0L, 10L, threshold - 1);
+ collector.add(0, NORMAL_FILE, 10L, 10L, threshold);
+ collector.add(0, NORMAL_FILE, 20L, 10L, 10L);
+ collector.add(0, NORMAL_FILE, 20L, 10L, threshold + 1);
+ // Neither the sum of versions nor dedicated-file sizes bypass the
minimum file count.
+ collector.add(0, NORMAL_FILE, 30L, 10L, threshold * 3 / 4);
+ collector.add(0, NORMAL_FILE, 30L, 10L, threshold * 3 / 4);
+ collector.add(0, IGNORED_DEDICATED_FILE, 0L, 10L, 1000L);
+
+ assertThat(finish(collector)).containsExactly("20-29:2");
+ }
+
+ @Test
+ void testSplitThresholdDoesNotOverflow() {
+ CompactCandidateRangeCollector collector =
+ new CompactCandidateRangeCollector(
+ 16, Long.MAX_VALUE, 100L, 1L, 2L,
largeFileThreshold(Long.MAX_VALUE, 2.0d));
+ collector.add(0, NORMAL_FILE, 0L, 10L, Long.MAX_VALUE);
+ assertThat(finish(collector)).isEmpty();
+ }
+
@Test
void testSelectsUpdatedFilesEvenWhenOneLogicalRangeExceedsTarget() {
CompactCandidateRangeCollector collector = collector(100L, 100L, 1L,
2L);
@@ -156,7 +206,12 @@ class CompactCandidateRangeCollectorTest {
long openFileCost,
long compactMinFileNum) {
return new CompactCandidateRangeCollector(
- 16, targetFileSize, blobTargetFileSize, openFileCost,
compactMinFileNum);
+ 16,
+ targetFileSize,
+ blobTargetFileSize,
+ openFileCost,
+ compactMinFileNum,
+ Long.MAX_VALUE);
}
private List<String> finish(CompactCandidateRangeCollector collector) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinatorTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinatorTest.java
index cd784064bb..0ce6a252a5 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinatorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactCoordinatorTest.java
@@ -45,9 +45,13 @@ import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.SnapshotManager;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.util.ArrayList;
@@ -62,6 +66,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.LongFunction;
import java.util.stream.Collectors;
+import static
org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator.largeFileThreshold;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
@@ -89,6 +94,82 @@ public class DataEvolutionCompactCoordinatorTest {
.hasMessageContaining("materialize_deletion_vectors");
}
+ @Test
+ public void testLargeFileRatioOptions() {
+ assertThat(new CoreOptions(new
Options()).dataEvolutionCompactionLargeFileRatio())
+ .isEqualTo(2.0d);
+ for (double ratio : new double[] {1.0d, 1.15d, Double.MAX_VALUE}) {
+ Options options = new Options();
+
options.set(CoreOptions.DATA_EVOLUTION_COMPACTION_LARGE_FILE_RATIO, ratio);
+ assertThat(new
CoreOptions(options).dataEvolutionCompactionLargeFileRatio())
+ .isEqualTo(ratio);
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(
+ doubles = {
+ 0.0d,
+ 0.99d,
+ -1.0d,
+ Double.NaN,
+ Double.NEGATIVE_INFINITY,
+ Double.POSITIVE_INFINITY
+ })
+ public void testRejectsInvalidLargeFileRatio(double ratio) {
+ Options options = new Options();
+ options.set(CoreOptions.DATA_EVOLUTION_COMPACTION_LARGE_FILE_RATIO,
ratio);
+ assertThatThrownBy(() -> new
CoreOptions(options).dataEvolutionCompactionLargeFileRatio())
+ .isInstanceOf(IllegalArgumentException.class)
+
.hasMessageContaining(CoreOptions.DATA_EVOLUTION_COMPACTION_LARGE_FILE_RATIO.key());
+ }
+
+ @Test
+ public void testLargeFileThresholdPreservesByteBoundaries() {
+ // Direct double multiplication rounds 100 * 1.15 below 115.
+ assertThat(largeFileThreshold(100L, 1.15d)).isEqualTo(115L);
+ assertThat(largeFileThreshold(3L, 1.5d)).isEqualTo(4L);
+ assertThat(largeFileThreshold((1L << 53) + 1, 1.0d)).isEqualTo((1L <<
53) + 1);
+ assertThat(largeFileThreshold(Long.MAX_VALUE / 2,
2.0d)).isEqualTo(Long.MAX_VALUE - 1);
+ assertThat(largeFileThreshold(Long.MAX_VALUE / 2 + 1,
2.0d)).isEqualTo(Long.MAX_VALUE);
+ assertThat(largeFileThreshold(Long.MAX_VALUE,
1.0d)).isEqualTo(Long.MAX_VALUE);
+ assertThat(largeFileThreshold(100L,
Double.MAX_VALUE)).isEqualTo(Long.MAX_VALUE);
+ }
+
+ @ParameterizedTest
+ @CsvSource({"1.0,100", "1.15,115", "1.5,150", "3.0,300"})
+ public void testCustomLargeFileRatioUsesIndividualPhysicalFileSize(
+ double ratio, long threshold) {
+ long versionSize = threshold * 3 / 4;
+ List<ManifestEntry> entries =
+ Arrays.asList(
+ makeEntryWithSize("below.parquet", 0L, 10L, 0,
threshold - 1),
+ makeEntryWithSize("boundary.parquet", 10L, 10L, 0,
threshold),
+ makeEntryWithSize("base.parquet", 20L, 10L, 0, 10L),
+ makeEntryWithSize("large-update.parquet", 20L, 10L, 1,
threshold + 1),
+ makeEntryWithSize("version1.parquet", 30L, 10L, 0,
versionSize),
+ makeEntryWithSize("version2.parquet", 30L, 10L, 1,
versionSize),
+ makeBlobEntry("large.blob", 0L, 10L, 1000L),
+ makeVectorStoreEntry("large.vector.lance", 10L, 10L,
1000L));
+ DataEvolutionCompactCoordinator.CompactPlanner planner =
+ new DataEvolutionCompactCoordinator.CompactPlanner(
+ false,
+ false,
+ largeFileThreshold(100L, ratio),
+ 100L,
+ 100L,
+ 1000L,
+ 10L,
+ schemaId -> null,
+ null);
+
+ List<DataEvolutionCompactTask> tasks = planner.compactPlan(entries);
+
+ assertThat(tasks).hasSize(1);
+ assertThat(tasks.get(0).compactBefore())
+ .containsExactly(entries.get(2).file(), entries.get(3).file());
+ }
+
@Test
public void testCompactPlannerSingleFile() {
// Single file should not produce compaction tasks
@@ -104,6 +185,74 @@ public class DataEvolutionCompactCoordinatorTest {
assertThat(tasks).isEmpty();
}
+ @Test
+ public void testSplitLargeFilesUsesPhysicalSizeAndIncludesColumnUpdates() {
+ List<ManifestEntry> entries =
+ Arrays.asList(
+ makeEntryWithSize("below.parquet", 0L, 10L, 0, 199L),
+ makeEntryWithSize("boundary.parquet", 10L, 10L, 0,
200L),
+ makeEntryWithSize("large.parquet", 20L, 10L, 0, 201L),
+ makeEntryWithSize("update.parquet", 20L, 10L, 1, 10L));
+ for (boolean enabled : new boolean[] {false, true}) {
+ DataEvolutionCompactCoordinator.CompactPlanner planner =
+ new DataEvolutionCompactCoordinator.CompactPlanner(
+ false,
+ false,
+ enabled ? 200L : Long.MAX_VALUE,
+ 100L,
+ 100L,
+ 1000L,
+ 10L,
+ schemaId -> null,
+ null);
+ List<DataEvolutionCompactTask> tasks =
planner.compactPlan(entries);
+ if (enabled) {
+ assertThat(tasks).hasSize(1);
+ assertThat(tasks.get(0).compactBefore())
+ .containsExactly(entries.get(2).file(),
entries.get(3).file());
+ } else {
+ assertThat(tasks).isEmpty();
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testDoNotSplitInsideDedicatedFile(boolean vector) {
+ List<ManifestEntry> entries =
+ Arrays.asList(
+ makeEntryWithSize("large.parquet", 0L, 10L, 0, 1000L),
+ vector
+ ? makeVectorStoreEntry("whole.vector.lance",
0L, 10L, 10L)
+ : makeBlobEntry("whole.blob", 0L, 10L, 10L));
+ DataEvolutionCompactCoordinator.CompactPlanner planner =
+ new DataEvolutionCompactCoordinator.CompactPlanner(
+ false, false, 200L, 100L, 100L, 1L, 2L, schemaId ->
null, null);
+
+ assertThat(planner.compactPlan(entries)).isEmpty();
+ }
+
+ @Test
+ public void testSplitLargeFilesAndMergeSmallFilesKeepDedicatedFiles() {
+ List<ManifestEntry> entries =
+ Arrays.asList(
+ makeEntryWithSize("large.parquet", 0L, 10L, 0, 201L),
+ makeEntryWithSize("small1.parquet", 10L, 10L, 0, 20L),
+ makeEntryWithSize("small2.parquet", 20L, 10L, 0, 20L),
+ makeBlobEntry("original.blob", 0L, 5L, 1000L),
+ makeVectorStoreEntry("original.vector.lance", 5L, 5L,
1000L));
+ DataEvolutionCompactCoordinator.CompactPlanner planner =
+ new DataEvolutionCompactCoordinator.CompactPlanner(
+ false, false, 200L, 100L, 100L, 1L, 2L, schemaId ->
null, null);
+
+ List<DataEvolutionCompactTask> tasks = planner.compactPlan(entries);
+
+ assertThat(tasks).hasSize(2);
+
assertThat(tasks.get(0).compactBefore()).containsExactly(entries.get(0).file());
+ assertThat(tasks.get(1).compactBefore())
+ .containsExactly(entries.get(1).file(), entries.get(2).file());
+ }
+
@Test
public void testCompactPlannerContiguousFiles() {
// Multiple contiguous files should be grouped together
@@ -876,6 +1025,7 @@ public class DataEvolutionCompactCoordinatorTest {
return new DataEvolutionCompactCoordinator.CompactPlanner(
true,
false,
+ Long.MAX_VALUE,
targetFileSize,
targetFileSize,
openFileCost,
@@ -912,7 +1062,11 @@ public class DataEvolutionCompactCoordinatorTest {
createDataFileMeta("file2.parquet", 100L, 100L, 0,
1024));
DataEvolutionCompactTask task =
- new DataEvolutionNormalCompactTask(BinaryRow.EMPTY_ROW, files);
+ new DataEvolutionNormalCompactTask(
+ BinaryRow.EMPTY_ROW,
+ files,
+ Arrays.asList(
+ new Range(0L, 49L), new Range(50L, 149L), new
Range(150L, 199L)));
byte[] bytes = serializer.serialize(task);
DataEvolutionCompactTask deserialized =
@@ -973,6 +1127,76 @@ public class DataEvolutionCompactCoordinatorTest {
.hasMessageContaining("contiguous row range");
}
+ @Test
+ public void testPlanNormalOutputRangesAtDedicatedBoundaries() {
+ List<DataFileMeta> files =
+ Collections.singletonList(createDataFileMeta("file.parquet",
100, 10, 0, 1000));
+ DataEvolutionNormalCompactTask task =
+ new DataEvolutionNormalCompactTask(BinaryRow.EMPTY_ROW, files);
+ assertThat(task.planOutputRanges(400))
+ .containsExactly(new Range(100, 103), new Range(104, 107), new
Range(108, 109));
+
+ task =
+ new DataEvolutionNormalCompactTask(
+ BinaryRow.EMPTY_ROW,
+ files,
+ Arrays.asList(
+ new Range(105, 107), new Range(103, 106), new
Range(108, 109)));
+ // A cut after 103 lies inside overlapping dedicated files and moves
to 107.
+ // The adjacent dedicated file starting at 108 must not prevent this
boundary.
+ assertThat(task.planOutputRanges(400))
+ .containsExactly(new Range(100, 107), new Range(108, 109));
+ // A cut immediately before a dedicated file is also safe.
+ assertThat(task.planOutputRanges(300))
+ .containsExactly(new Range(100, 102), new Range(103, 107), new
Range(108, 109));
+ }
+
+ @Test
+ public void testPlanNormalOutputRangesUsesLogicalRowCountAcrossVersions() {
+ DataEvolutionNormalCompactTask task =
+ new DataEvolutionNormalCompactTask(
+ BinaryRow.EMPTY_ROW,
+ Arrays.asList(
+ createDataFileMeta("base.parquet", 100, 10, 0,
400),
+ createDataFileMeta("update.parquet", 100, 10,
1, 600)));
+ assertThat(task.planOutputRanges(500))
+ .containsExactly(new Range(100, 104), new Range(105, 109));
+ }
+
+ @Test
+ public void testPlanNormalOutputRangesAvoidsSizeAndRowIdOverflow() {
+ DataEvolutionNormalCompactTask task =
+ new DataEvolutionNormalCompactTask(
+ BinaryRow.EMPTY_ROW,
+ Arrays.asList(
+ createDataFileMeta(
+ "base.parquet", Long.MAX_VALUE - 10,
10, 0, Long.MAX_VALUE),
+ createDataFileMeta(
+ "update.parquet",
+ Long.MAX_VALUE - 10,
+ 10,
+ 1,
+ Long.MAX_VALUE)));
+ assertThat(task.planOutputRanges(Long.MAX_VALUE))
+ .containsExactly(
+ new Range(Long.MAX_VALUE - 10, Long.MAX_VALUE - 6),
+ new Range(Long.MAX_VALUE - 5, Long.MAX_VALUE - 1));
+ task =
+ new DataEvolutionNormalCompactTask(
+ BinaryRow.EMPTY_ROW,
+ Collections.singletonList(
+ createDataFileMeta(
+ "last.parquet",
+ Long.MAX_VALUE - 9,
+ 10,
+ 0,
+ Long.MAX_VALUE)));
+ assertThat(task.planOutputRanges(Long.MAX_VALUE / 2))
+ .containsExactly(
+ new Range(Long.MAX_VALUE - 9, Long.MAX_VALUE - 5),
+ new Range(Long.MAX_VALUE - 4, Long.MAX_VALUE));
+ }
+
@Test
public void testSerializerMaterializeDeletionTask() throws IOException {
DataEvolutionCompactTaskSerializer serializer = new
DataEvolutionCompactTaskSerializer();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactRangePlannerTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactRangePlannerTest.java
index 88278ec40f..93059aeea1 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactRangePlannerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionCompactRangePlannerTest.java
@@ -400,6 +400,7 @@ class DataEvolutionCompactRangePlannerTest extends
TableTestBase {
return new DataEvolutionCompactRangePlanner.CandidateOptions(
compactBlob,
compactVector,
+ Long.MAX_VALUE,
100L,
100L,
1L,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTaskTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTaskTest.java
index dd91e37f3a..6d15a8c345 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTaskTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTaskTest.java
@@ -21,10 +21,15 @@ package org.apache.paimon.append.dataevolution;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.BinaryVector;
+import org.apache.paimon.data.BlobData;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.TableTestBase;
@@ -33,11 +38,15 @@ import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Range;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.ArrayList;
import java.util.Arrays;
@@ -45,12 +54,17 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Random;
import java.util.stream.Collectors;
+import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
+import static org.apache.paimon.types.VectorType.isVectorStoreFile;
import static org.apache.paimon.utils.DataEvolutionUtils.fileFields;
import static org.assertj.core.api.Assertions.assertThat;
-/** Tests for column sequence propagation in {@link
DataEvolutionNormalCompactTask}. */
+/**
+ * Tests for splitting and column sequence propagation in {@link
DataEvolutionNormalCompactTask}.
+ */
public class DataEvolutionNormalCompactTaskTest extends TableTestBase {
private static final int ROW_COUNT = 100;
@@ -127,6 +141,581 @@ public class DataEvolutionNormalCompactTaskTest extends
TableTestBase {
assertThat(compacted.columnMaxSequenceNumbers()).isNull();
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testSplitHistoricalLargeFile(boolean updateColumn) throws
Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ int rowCount = 12000;
+ Random random = new Random(42);
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = 0; i < rowCount; i++) {
+ StringBuilder value = new StringBuilder();
+ for (int j = 0; j < 8; j++) {
+ value.append(Long.toHexString(random.nextLong()));
+ }
+ write.write(
+ GenericRow.of(
+ BinaryString.fromString("p0"),
+ i,
+ BinaryString.fromString(value.toString())));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ DataFileMeta original =
table.store().newScan().plan().files().get(0).file();
+ if (updateColumn) {
+ builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
+ builder.newWrite()
+ .withWriteType(
+
table.rowType().project(Arrays.asList("dt", "f0")));
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = 0; i < rowCount; i++) {
+ write.write(GenericRow.of(BinaryString.fromString("p0"), i
+ rowCount));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ assignFirstRowId(messages, original.nonNullFirstRowId());
+ commit.commit(messages);
+ }
+ }
+ List<InternalRow> expected = read(table);
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.TARGET_FILE_SIZE.key(), "128 kb");
+ options.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "10");
+ options.put(CoreOptions.GLOBAL_INDEX_COLUMN_UPDATE_ACTION.key(),
"ignore");
+ table = table.copy(options);
+ long targetSize = table.coreOptions().targetFileSize(false);
+ assertThat(original.fileSize()).isGreaterThan(3 * targetSize);
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ assertThat(new DataEvolutionCompactCoordinator(table, false, false,
snapshot).plan())
+ .isEmpty();
+
+
options.put(CoreOptions.DATA_EVOLUTION_COMPACTION_SPLIT_LARGE_FILES.key(),
"true");
+
options.put(CoreOptions.DATA_EVOLUTION_COMPACTION_LARGE_FILE_RATIO.key(),
"1000.0");
+ table = table.copy(options);
+ assertThat(new DataEvolutionCompactCoordinator(table, false, false,
snapshot).plan())
+ .isEmpty();
+
options.put(CoreOptions.DATA_EVOLUTION_COMPACTION_LARGE_FILE_RATIO.key(),
"3.0");
+ table = table.copy(options);
+ List<DataEvolutionCompactTask> tasks =
+ new DataEvolutionCompactCoordinator(table, false, false,
snapshot).plan();
+ assertThat(tasks).hasSize(1);
+ DataEvolutionCompactTaskSerializer serializer = new
DataEvolutionCompactTaskSerializer();
+ DataEvolutionCompactTask task =
+ serializer.deserialize(serializer.getVersion(),
serializer.serialize(tasks.get(0)));
+ List<CommitMessage> messages = new ArrayList<>();
+ messages.add(task.doCompact(table, "split-large-file"));
+ List<DataFileMeta> output = task.compactAfter();
+ assertThat(output.size()).isGreaterThan(1);
+ long nextRowId = original.nonNullFirstRowId();
+ long maxSequenceNumber =
+ task.compactBefore().stream()
+ .mapToLong(DataFileMeta::maxSequenceNumber)
+ .max()
+ .getAsLong();
+ for (DataFileMeta file : output) {
+ assertThat(file.nonNullFirstRowId()).isEqualTo(nextRowId);
+ assertThat(file.fileSize()).isLessThan(2 * targetSize);
+
assertThat(file.minSequenceNumber()).isEqualTo(original.minSequenceNumber());
+ assertThat(file.maxSequenceNumber()).isEqualTo(maxSequenceNumber);
+ if (updateColumn) {
+ assertThat(columnSequence(file,
table.rowType().getField("f1").id()))
+ .isEqualTo(original.maxSequenceNumber());
+ assertThat(columnSequence(file,
table.rowType().getField("f0").id()))
+ .isEqualTo(maxSequenceNumber);
+ }
+ nextRowId += file.rowCount();
+ }
+ assertThat(nextRowId).isEqualTo(original.nonNullFirstRowId() +
rowCount);
+ messages.addAll(
+ new DataEvolutionCompactionCommitPreparation(table,
snapshot).prepare(messages));
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(messages);
+ }
+ assertThat(read(table)).containsExactlyInAnyOrderElementsOf(expected);
+ assertThat(
+ new DataEvolutionCompactCoordinator(
+ table,
+ false,
+ false,
+
table.snapshotManager().latestSnapshot())
+ .plan())
+ .isEmpty();
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {1, 2000})
+ public void testSplitAtBlobBoundariesRetainsDedicatedFiles(int
estimatedRows) throws Exception {
+ FileStoreTable table = createBlobSegmentsTable();
+ List<DataFileMeta> dedicated = dedicatedFiles(table);
+ assertThat(dedicated).hasSize(3);
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
+ DataEvolutionCompactTask merged =
compactSingleTask(table.copy(options));
+ assertThat(merged.compactAfter()).hasSize(1);
+
assertThat(merged.compactAfter().get(0).nonNullRowIdRange()).isEqualTo(new
Range(0, 3749));
+
+ options.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "10");
+ long targetSize =
+ Math.max(1L, merged.compactAfter().get(0).fileSize() *
estimatedRows / 3750);
+ options.put(CoreOptions.TARGET_FILE_SIZE.key(), targetSize + " b");
+ options.put(CoreOptions.WRITE_BUFFER_FOR_APPEND.key(), "true");
+ options.put(CoreOptions.TARGET_FILE_ROW_NUM.key(), "10");
+
options.put(CoreOptions.DATA_EVOLUTION_COMPACTION_SPLIT_LARGE_FILES.key(),
"true");
+
options.put(CoreOptions.DATA_EVOLUTION_COMPACTION_LARGE_FILE_RATIO.key(),
"1.0");
+ table = table.copy(options);
+ DataEvolutionCompactTask split = compactSingleTask(table);
+ // Estimated cuts move to BLOB ends; adjacent ranges can share a
normal output.
+
assertThat(split.compactAfter().stream().map(DataFileMeta::nonNullRowIdRange))
+ .containsExactlyElementsOf(
+ estimatedRows == 1
+ ? Arrays.asList(
+ new Range(0, 1249),
+ new Range(1250, 2499),
+ new Range(2500, 3749))
+ : Arrays.asList(new Range(0, 2499), new
Range(2500, 3749)));
+ assertDedicatedFilesContained(table, dedicated);
+ assertBlobValues(table);
+ // Completed outputs must not be rewritten even if estimates leave
them oversized.
+ assertThat(
+ new DataEvolutionCompactCoordinator(
+ table,
+ false,
+ false,
+
table.snapshotManager().latestSnapshot())
+ .withCompletedNormalFiles(
+ split.compactAfter().stream()
+ .map(DataFileMeta::fileName)
+ .collect(Collectors.toSet()))
+ .plan())
+ .isEmpty();
+
+ options.put(CoreOptions.TARGET_FILE_SIZE.key(), "128 mb");
+ options.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
+ table = table.copy(options);
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ assertThat(new DataEvolutionCompactCoordinator(table, false, false,
snapshot).plan())
+ .hasSize(1);
+ assertThat(
+ new DataEvolutionCompactCoordinator(table, false,
false, snapshot)
+ .withCompletedNormalFiles(
+ split.compactAfter().stream()
+ .map(DataFileMeta::fileName)
+ .collect(Collectors.toSet()))
+ .plan())
+ .isEmpty();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testNormalSplitRespectsBlobCompactionOutputRange(boolean
mergeNormalVersions)
+ throws Exception {
+ FileStoreTable table = createBlobSegmentsTable();
+ List<DataFileMeta> originalBlobs = dedicatedFiles(table);
+ assertThat(originalBlobs).hasSize(3);
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
+ DataEvolutionCompactTask initialMerge =
compactSingleTask(table.copy(options));
+ assertThat(initialMerge.compactAfter()).hasSize(1);
+ DataFileMeta originalNormal = initialMerge.compactAfter().get(0);
+ if (mergeNormalVersions) {
+ writeProjectedRange(table, "id", 0, 3750, 0, true);
+ }
+
+ options.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
+ options.put(CoreOptions.TARGET_FILE_SIZE.key(), "1 b");
+
options.put(CoreOptions.DATA_EVOLUTION_COMPACTION_SPLIT_LARGE_FILES.key(),
"true");
+ table = table.copy(options);
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ List<DataEvolutionCompactTask> tasks =
+ new DataEvolutionCompactCoordinator(table, true, false,
snapshot).plan();
+ if (mergeNormalVersions) {
+ assertThat(tasks)
+ .extracting(DataEvolutionCompactTask::type)
+ .containsExactly(
+ DataEvolutionCompactTask.TaskType.NORMAL,
+ DataEvolutionCompactTask.TaskType.BLOB);
+ } else {
+ assertThat(tasks)
+ .extracting(DataEvolutionCompactTask::type)
+ .containsExactly(DataEvolutionCompactTask.TaskType.BLOB);
+ }
+ DataEvolutionCompactTaskSerializer serializer = new
DataEvolutionCompactTaskSerializer();
+ List<CommitMessage> messages = new ArrayList<>();
+ List<DataEvolutionCompactTask> restored = new ArrayList<>();
+ for (DataEvolutionCompactTask planned : tasks) {
+ DataEvolutionCompactTask task =
+ serializer.deserialize(serializer.getVersion(),
serializer.serialize(planned));
+ restored.add(task);
+ messages.add(task.doCompact(table, "compact-normal-and-blob"));
+ }
+ if (mergeNormalVersions) {
+ assertThat(restored.get(0).compactBefore()).hasSize(2);
+ }
+ assertThat(restored.get(restored.size() - 1).compactBefore())
+ .containsExactlyInAnyOrderElementsOf(originalBlobs);
+ // The old BLOB boundaries are safe individually, but the planned BLOB
merge removes them.
+ // Any normal rewrite must therefore retain the planned BLOB output's
complete range.
+ for (DataEvolutionCompactTask task : restored) {
+ assertThat(task.compactAfter()).hasSize(1);
+ assertThat(task.compactAfter().get(0).nonNullRowIdRange())
+ .isEqualTo(new Range(0, 3749));
+ }
+ messages.addAll(
+ new DataEvolutionCompactionCommitPreparation(table,
snapshot).prepare(messages));
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(messages);
+ }
+ if (!mergeNormalVersions) {
+ assertThat(
+ table.store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .filter(
+ file ->
+
!isBlobFile(file.fileName())
+ &&
!isVectorStoreFile(file.fileName())))
+ .containsExactly(originalNormal);
+ }
+ List<DataFileMeta> compactedBlobs = dedicatedFiles(table);
+
assertThat(compactedBlobs).hasSize(1).doesNotContainAnyElementsOf(originalBlobs);
+ assertDedicatedFilesContained(table, compactedBlobs);
+ assertBlobValues(table);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testOverlappingDedicatedRangesPreventSplit(boolean sameColumn)
throws Exception {
+ catalog.createTable(
+ identifier(),
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("b1", DataTypes.BLOB())
+ .column("b2", DataTypes.BLOB())
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .option(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "128
mb")
+ .build(),
+ false);
+ FileStoreTable table = getTableDefault();
+ writeProjectedRange(table, "id", 0, 3750, 0, false);
+ writeProjectedRange(table, "b1", 0, 1250, 0, true);
+ writeProjectedRange(table, "b1", 1250, 2500, 0, true);
+ String overlappingColumn = sameColumn ? "b1" : "b2";
+ writeProjectedRange(table, overlappingColumn, 0, 2500, 1, true);
+ writeProjectedRange(table, overlappingColumn, 2500, 1250, 1, true);
+ List<DataFileMeta> dedicated = dedicatedFiles(table);
+ assertThat(dedicated).hasSize(4);
+ assertThat(dedicated).allSatisfy(file ->
assertThat(file.rowCount()).isLessThan(3750));
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "10");
+ options.put(CoreOptions.TARGET_FILE_SIZE.key(), "1 b");
+
options.put(CoreOptions.DATA_EVOLUTION_COMPACTION_SPLIT_LARGE_FILES.key(),
"true");
+ FileStoreTable splitTable = table.copy(options);
+ assertThat(
+ new DataEvolutionCompactCoordinator(
+ splitTable,
+ false,
+ false,
+
table.snapshotManager().latestSnapshot())
+ .plan())
+ .isEmpty();
+
+ // Ordinary version merging remains useful even though no
dedicated-safe split is possible.
+ writeProjectedRange(table, "id", 0, 3750, 0, true);
+ options.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
+ DataEvolutionCompactTask merged =
compactSingleTask(table.copy(options));
+ assertThat(merged.compactBefore()).hasSize(2);
+ assertThat(merged.compactAfter()).hasSize(1);
+
assertThat(merged.compactAfter().get(0).nonNullRowIdRange()).isEqualTo(new
Range(0, 3749));
+ assertDedicatedFilesContained(table, dedicated);
+ ReadBuilder readBuilder = table.newReadBuilder();
+ List<Integer> ids = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(
+ row -> {
+ int id = row.getInt(0);
+ ids.add(id);
+ assertThat(row.getBlob(1).toData())
+ .containsExactly((byte) (id + (sameColumn ? 1
: 0)));
+ if (sameColumn) {
+ assertThat(row.isNullAt(2)).isTrue();
+ } else {
+
assertThat(row.getBlob(2).toData()).containsExactly((byte) (id + 1));
+ }
+ });
+ }
+ assertThat(ids)
+ .containsExactlyElementsOf(
+ java.util.stream.IntStream.range(0, 3750)
+ .boxed()
+ .collect(Collectors.toList()));
+ }
+
+ @Test
+ public void testFullRangeVectorPreventsSplit() throws Exception {
+ catalog.createTable(
+ identifier(),
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("vector", DataTypes.VECTOR(2,
DataTypes.FLOAT()))
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .option(CoreOptions.VECTOR_TARGET_FILE_SIZE.key(),
"128 mb")
+ .option(CoreOptions.VECTOR_FILE_FORMAT.key(), "json")
+ .option(CoreOptions.FILE_COMPRESSION.key(), "none")
+ .build(),
+ false);
+ FileStoreTable table = getTableDefault();
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = 0; i < 2500; i++) {
+ write.write(
+ GenericRow.of(i, BinaryVector.fromPrimitiveArray(new
float[] {i, i + 1})));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ catalog.alterTable(
+ identifier(),
+ Collections.singletonList(SchemaChange.renameColumn("vector",
"renamed_vector")),
+ false);
+ table = getTableDefault();
+ List<DataFileMeta> dedicated = dedicatedFiles(table);
+ assertThat(dedicated).hasSize(1);
+ assertThat(dedicated.get(0).nonNullRowIdRange()).isEqualTo(new
Range(0, 2499));
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.TARGET_FILE_SIZE.key(), "1 b");
+
options.put(CoreOptions.DATA_EVOLUTION_COMPACTION_SPLIT_LARGE_FILES.key(),
"true");
+ assertThat(
+ new DataEvolutionCompactCoordinator(
+ table.copy(options),
+ false,
+ false,
+
table.snapshotManager().latestSnapshot())
+ .plan())
+ .isEmpty();
+
+ writeProjectedRange(table, "id", 0, 2500, 0, true);
+ options.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
+ DataEvolutionCompactTask merged =
compactSingleTask(table.copy(options));
+ assertThat(merged.compactAfter()).hasSize(1);
+ assertDedicatedFilesContained(table, dedicated);
+ ReadBuilder readBuilder = table.newReadBuilder();
+ List<Integer> ids = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(
+ row -> {
+ int id = row.getInt(0);
+ ids.add(id);
+
assertThat(row.getVector(1).toFloatArray()).containsExactly(id, id + 1);
+ });
+ }
+ assertThat(ids)
+ .containsExactlyElementsOf(
+ java.util.stream.IntStream.range(0, 2500)
+ .boxed()
+ .collect(Collectors.toList()));
+ }
+
+ @Test
+ public void testEstimatedRangesIgnoreParquetBufferSize() throws Exception {
+ catalog.createTable(
+ identifier(),
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("payload", DataTypes.STRING())
+ .column("blob", DataTypes.BLOB())
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .option(CoreOptions.FILE_FORMAT.key(), "parquet")
+ .option(CoreOptions.FILE_COMPRESSION.key(), "snappy")
+ .option(CoreOptions.TARGET_FILE_SIZE.key(), "8 mb")
+ .option(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "128
mb")
+ .option(CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(),
"4 mb")
+ .option(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2")
+ .option(
+
CoreOptions.DATA_EVOLUTION_COMPACTION_SPLIT_LARGE_FILES.key(),
+ "true")
+ .option("parquet.enable.dictionary", "false")
+ .option("parquet.page.size", String.valueOf(16 * 1024
* 1024))
+ .build(),
+ false);
+ FileStoreTable table = getTableDefault();
+ char[] payloadChars = new char[10 * 1024];
+ Arrays.fill(payloadChars, 'a');
+ BinaryString payload = BinaryString.fromString(new
String(payloadChars));
+ for (int batch = 0; batch < 3; batch++) {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = batch * 1000; i < (batch + 1) * 1000; i++) {
+ write.write(GenericRow.of(i, payload, new BlobData(new
byte[] {(byte) i})));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+ List<DataFileMeta> dedicated = dedicatedFiles(table);
+ assertThat(dedicated).hasSize(3);
+
+ DataEvolutionCompactTask compacted = compactSingleTask(table);
+ assertThat(compacted.compactBefore()).hasSize(3);
+ List<DataFileMeta> outputs = compacted.compactAfter();
+ // The input files fit in the target based on their compressed sizes,
even though
+ // Parquet's uncompressed page buffer crosses that target every 1000
rows.
+
assertThat(outputs).extracting(DataFileMeta::rowCount).containsExactly(3000L);
+ assertThat(outputs)
+ .allSatisfy(
+ file ->
+ assertThat(file.fileSize())
+
.isLessThan(table.coreOptions().splitOpenFileCost()));
+ assertDedicatedFilesContained(table, dedicated);
+
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ assertThat(new DataEvolutionCompactCoordinator(table, false, false,
snapshot).plan())
+ .isEmpty();
+
+ List<Integer> ids = new ArrayList<>();
+ ReadBuilder readBuilder = table.newReadBuilder();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(
+ row -> {
+ int id = row.getInt(0);
+ ids.add(id);
+ assertThat(row.getString(1)).isEqualTo(payload);
+
assertThat(row.getBlob(2).toData()).containsExactly((byte) id);
+ });
+ }
+ assertThat(ids)
+ .containsExactlyElementsOf(
+ java.util.stream.IntStream.range(0, 3000)
+ .boxed()
+ .collect(Collectors.toList()));
+ }
+
+ private FileStoreTable createBlobSegmentsTable() throws Exception {
+ catalog.createTable(
+ identifier(),
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("blob", DataTypes.BLOB())
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .option(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "128
mb")
+ .build(),
+ false);
+ FileStoreTable table = getTableDefault();
+ for (int batch = 0; batch < 3; batch++) {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = batch * 1250; i < (batch + 1) * 1250; i++) {
+ write.write(GenericRow.of(i, new BlobData(new byte[]
{(byte) i})));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+ return table;
+ }
+
+ private void assertBlobValues(FileStoreTable table) throws Exception {
+ List<Integer> ids = new ArrayList<>();
+ ReadBuilder readBuilder = table.newReadBuilder();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(
+ row -> {
+ int id = row.getInt(0);
+ ids.add(id);
+
assertThat(row.getBlob(1).toData()).containsExactly((byte) id);
+ });
+ }
+ assertThat(ids)
+ .containsExactlyElementsOf(
+ java.util.stream.IntStream.range(0, 3750)
+ .boxed()
+ .collect(Collectors.toList()));
+ }
+
+ private void writeProjectedRange(
+ FileStoreTable table,
+ String column,
+ int from,
+ int count,
+ int valueOffset,
+ boolean existingRows)
+ throws Exception {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write =
+
builder.newWrite().withWriteType(table.rowType().project(column));
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int i = from; i < from + count; i++) {
+ write.write(
+ GenericRow.of(
+ "id".equals(column)
+ ? i
+ : new BlobData(new byte[] {(byte) (i +
valueOffset)})));
+ }
+ List<CommitMessage> messages = write.prepareCommit();
+ if (existingRows) {
+ assignFirstRowId(messages, from);
+ }
+ commit.commit(messages);
+ }
+ }
+
+ private DataEvolutionCompactTask compactSingleTask(FileStoreTable table)
throws Exception {
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ List<DataEvolutionCompactTask> tasks =
+ new DataEvolutionCompactCoordinator(table, false, false,
snapshot).plan();
+ assertThat(tasks).hasSize(1);
+ DataEvolutionCompactTaskSerializer serializer = new
DataEvolutionCompactTaskSerializer();
+ DataEvolutionCompactTask task =
+ serializer.deserialize(serializer.getVersion(),
serializer.serialize(tasks.get(0)));
+ List<CommitMessage> messages = new ArrayList<>();
+ messages.add(task.doCompact(table, "compact-dedicated-boundaries"));
+ messages.addAll(
+ new DataEvolutionCompactionCommitPreparation(table,
snapshot).prepare(messages));
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(messages);
+ }
+ return task;
+ }
+
+ private List<DataFileMeta> dedicatedFiles(FileStoreTable table) {
+ return table.store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .filter(file -> isBlobFile(file.fileName()) ||
isVectorStoreFile(file.fileName()))
+ .collect(Collectors.toList());
+ }
+
+ private void assertDedicatedFilesContained(FileStoreTable table,
List<DataFileMeta> expected) {
+
assertThat(dedicatedFiles(table)).containsExactlyInAnyOrderElementsOf(expected);
+ List<Range> normalRanges =
+ table.store().newScan().plan().files().stream()
+ .map(ManifestEntry::file)
+ .filter(
+ file ->
+ !isBlobFile(file.fileName())
+ &&
!isVectorStoreFile(file.fileName()))
+ .map(DataFileMeta::nonNullRowIdRange)
+ .collect(Collectors.toList());
+ for (DataFileMeta file : expected) {
+ Range dedicated = file.nonNullRowIdRange();
+ assertThat(
+ normalRanges.stream()
+ .anyMatch(
+ normal ->
+ normal.from <=
dedicated.from
+ && normal.to >=
dedicated.to))
+ .isTrue();
+ }
+ }
+
private void write() throws Exception {
createTableDefault();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
index 384f3fe2a3..8f189a7917 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionDeletionVectorTest.java
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator;
import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
+import
org.apache.paimon.append.dataevolution.DataEvolutionCompactTaskSerializer;
import
org.apache.paimon.append.dataevolution.DataEvolutionCompactionCommitPreparation;
import
org.apache.paimon.append.dataevolution.DataEvolutionDeletionVectorMaterializeCoordinator;
import org.apache.paimon.append.dataevolution.DataEvolutionRowIdReassigner;
@@ -67,6 +68,8 @@ import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RangeHelper;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.lang.reflect.Constructor;
@@ -82,6 +85,7 @@ import java.util.stream.Collectors;
import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
import static
org.apache.paimon.errors.ErrorMessages.DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE;
+import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET;
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile;
@@ -581,6 +585,115 @@ public class DataEvolutionDeletionVectorTest extends
DataEvolutionTestBase {
assertThat(liveDeletionVectorDataFileNames).doesNotContainAnyElementsOf(oldAnchorFiles);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testSplitLargeFilePreservesDeletionVectorsAndBlob(boolean
bitmap64)
+ throws Exception {
+ createTableDefault();
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.DELETION_VECTOR_BITMAP64.key(),
String.valueOf(bitmap64));
+ options.put(CoreOptions.BLOB_TARGET_FILE_SIZE.key(), "128 mb");
+ FileStoreTable table = getTableDefault().copy(options);
+ for (int batch = 0; batch < 3; batch++) {
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ for (int rowId = batch * 1250; rowId < (batch + 1) * 1250;
rowId++) {
+ write.write(
+ GenericRow.of(
+ rowId,
+ BinaryString.fromString("name-" + rowId),
+ BinaryString.fromString("base-" + rowId),
+ new BlobData(new byte[] {(byte) rowId})));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+ List<DataFileMeta> blobs =
+ currentDataFiles(table, BinaryRow.EMPTY_ROW).stream()
+ .filter(file -> isBlobFile(file.fileName()))
+ .collect(Collectors.toList());
+ assertThat(blobs).hasSize(3);
+ options.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "2");
+ compactDataEvolutionTable(table.copy(options), false);
+ Range range = new Range(0, 3749);
+ assertRegularFileRowRanges(
+ currentDataFiles(table, BinaryRow.EMPTY_ROW),
Collections.singletonList(range));
+ long[] deletedRowIds = {0, 1249, 1250, 2499, 2500, 3749};
+ commitDeletionVectors(table, Collections.singletonList(new
DvSpec(range, deletedRowIds)));
+ List<String> expected = readRows(table.newReadBuilder());
+ String oldAnchor = anchorFilesByRange(table).get(range);
+
+ options.put(CoreOptions.COMPACTION_MIN_FILE_NUM.key(), "10");
+ options.put(CoreOptions.TARGET_FILE_SIZE.key(), "1 b");
+
options.put(CoreOptions.DATA_EVOLUTION_COMPACTION_SPLIT_LARGE_FILES.key(),
"true");
+ table = table.copy(options);
+ Snapshot snapshot = table.snapshotManager().latestSnapshot();
+ List<DataEvolutionCompactTask> tasks =
+ new DataEvolutionCompactCoordinator(table, false, false,
snapshot).plan();
+ assertThat(tasks).hasSize(1);
+ DataEvolutionCompactTaskSerializer serializer = new
DataEvolutionCompactTaskSerializer();
+ DataEvolutionCompactTask task =
+ serializer.deserialize(serializer.getVersion(),
serializer.serialize(tasks.get(0)));
+ commit(table, prepareCompactionMessages(table, snapshot,
Collections.singletonList(task)));
+
+ List<DataFileMeta> files = currentDataFiles(table,
BinaryRow.EMPTY_ROW);
+ List<DataFileMeta> normalFiles =
+ files.stream()
+ .filter(file -> !isBlobFile(file.fileName()))
+ .collect(Collectors.toList());
+ assertRegularFileRowRanges(
+ files,
+ Arrays.asList(new Range(0, 1249), new Range(1250, 2499), new
Range(2500, 3749)));
+
assertThat(normalFiles.stream().mapToLong(DataFileMeta::rowCount).sum()).isEqualTo(3750);
+ assertThat(files.stream().filter(file -> isBlobFile(file.fileName())))
+ .containsExactlyInAnyOrderElementsOf(blobs);
+ for (DataFileMeta blob : blobs) {
+ Range blobRange = blob.nonNullRowIdRange();
+ assertThat(
+ normalFiles.stream()
+ .map(DataFileMeta::nonNullRowIdRange)
+ .anyMatch(
+ normal ->
+ normal.from <=
blobRange.from
+ && normal.to >=
blobRange.to))
+ .isTrue();
+ }
+
assertThat(readRows(table.newReadBuilder())).containsExactlyElementsOf(expected);
+ List<String> expectedAnchors = new ArrayList<>();
+ for (DataFileMeta file : normalFiles) {
+ Range fileRange = file.nonNullRowIdRange();
+ if (Arrays.stream(deletedRowIds)
+ .anyMatch(id -> id >= fileRange.from && id <=
fileRange.to)) {
+ expectedAnchors.add(file.fileName());
+ }
+ }
+ assertThat(liveDeletionVectorDataFileNames(table))
+ .containsExactlyInAnyOrderElementsOf(expectedAnchors)
+ .doesNotContain(oldAnchor);
+ ReadBuilder readBuilder =
+ table.newReadBuilder()
+
.withReadType(SpecialFields.rowTypeWithRowId(table.rowType()));
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(row ->
assertThat(row.getLong(4)).isEqualTo(row.getInt(0)));
+ }
+ ReadBuilder blobRead =
table.newReadBuilder().withReadType(table.rowType().project("f3"));
+ List<Byte> actualBlobs = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+ blobRead.newRead().createReader(blobRead.newScan().plan())) {
+ reader.forEachRemaining(row ->
actualBlobs.add(row.getBlob(0).toData()[0]));
+ }
+ List<Byte> expectedBlobs = new ArrayList<>();
+ for (int rowId = 0; rowId < 3750; rowId++) {
+ final int id = rowId;
+ if (Arrays.stream(deletedRowIds).noneMatch(deleted -> deleted ==
id)) {
+ expectedBlobs.add((byte) rowId);
+ }
+ }
+ assertThat(actualBlobs).containsExactlyElementsOf(expectedBlobs);
+ }
+
@Test
public void testCompactRenamesDeletionVectorForSameRowRange() throws
Exception {
createTableDefault();
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
index 832bd54780..218dc6d0a9 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
@@ -98,6 +98,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -106,7 +107,9 @@ import scala.collection.JavaConverters;
import scala.collection.Seq;
import static org.apache.paimon.CoreOptions.createCommitUser;
+import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
import static
org.apache.paimon.spark.utils.SparkProcedureUtils.readParallelism;
+import static org.apache.paimon.types.VectorType.isVectorStoreFile;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.spark.sql.types.DataTypes.StringType;
@@ -665,6 +668,7 @@ public class CompactProcedure extends BaseProcedure {
return;
}
AtomicReference<DataEvolutionCompactCoordinator> coordinatorRef = new
AtomicReference<>();
+ Set<String> completedNormalFiles = new HashSet<>();
Function<Snapshot, List<DataEvolutionCompactTask>> taskPlanner =
planningSnapshot -> {
DataEvolutionCompactCoordinator coordinator =
coordinatorRef.get();
@@ -685,11 +689,31 @@ public class CompactProcedure extends BaseProcedure {
false,
planningSnapshot,
candidateFilesPerBatch);
+
coordinator.withCompletedNormalFiles(completedNormalFiles);
coordinatorRef.set(coordinator);
}
return filterIdlePartitions(
coordinator.plan(), table, partitionPredicate,
partitionIdleTime);
};
+ Consumer<List<CommitMessage>> commitObserver = null;
+ if (table.coreOptions().dataEvolutionCompactionSplitLargeFiles()) {
+ commitObserver =
+ messages -> {
+ // Output sizes can still trigger splitting or
small-file merging.
+ // Skip bins containing only this invocation's
completed normal outputs.
+ for (CommitMessage message : messages) {
+ for (DataFileMeta file :
+ ((CommitMessageImpl) message)
+ .compactIncrement()
+ .compactAfter()) {
+ if (!isBlobFile(file.fileName())
+ &&
!isVectorStoreFile(file.fileName())) {
+ completedNormalFiles.add(file.fileName());
+ }
+ }
+ }
+ };
+ }
DataEvolutionRewriteExecutor.execute(
table,
snapshot,
@@ -699,7 +723,8 @@ public class CompactProcedure extends BaseProcedure {
commitConfigurer,
relation == null
? null
- : new DataEvolutionCompactMergeConflictRewriter(table,
relation)::rewrite);
+ : new DataEvolutionCompactMergeConflictRewriter(table,
relation)::rewrite,
+ commitObserver);
}
private static List<DataEvolutionCompactTask> filterIdlePartitions(
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DataEvolutionRewriteExecutor.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DataEvolutionRewriteExecutor.java
index b664b82f7e..e964f7b61c 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DataEvolutionRewriteExecutor.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DataEvolutionRewriteExecutor.java
@@ -46,6 +46,7 @@ import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
+import java.util.function.Consumer;
import java.util.function.Function;
import static org.apache.paimon.CoreOptions.createCommitUser;
@@ -84,6 +85,26 @@ final class DataEvolutionRewriteExecutor {
SparkSession sparkSession,
CommitConfigurer commitConfigurer,
@Nullable CommitMessageRewriter commitMessageRewriter) {
+ execute(
+ table,
+ initialSnapshot,
+ taskPlanner,
+ javaSparkContext,
+ sparkSession,
+ commitConfigurer,
+ commitMessageRewriter,
+ null);
+ }
+
+ static void execute(
+ FileStoreTable table,
+ Snapshot initialSnapshot,
+ Function<Snapshot, List<DataEvolutionCompactTask>> taskPlanner,
+ JavaSparkContext javaSparkContext,
+ SparkSession sparkSession,
+ CommitConfigurer commitConfigurer,
+ @Nullable CommitMessageRewriter commitMessageRewriter,
+ @Nullable Consumer<List<CommitMessage>> commitObserver) {
CommitMessageSerializer messageSerializer = new
CommitMessageSerializer();
String commitUser =
createCommitUser(table.coreOptions().toConfiguration());
Snapshot preparationSnapshot = initialSnapshot;
@@ -167,7 +188,8 @@ final class DataEvolutionRewriteExecutor {
commitUser,
sparkSession,
commitConfigurer,
- commitMessageRewriter);
+ commitMessageRewriter,
+ commitObserver);
checkArgument(
committedSnapshot.id() > preparationSnapshot.id(),
"Committed data evolution rewrite snapshot %s must
be newer than preparation snapshot %s.",
@@ -194,7 +216,8 @@ final class DataEvolutionRewriteExecutor {
String commitUser,
SparkSession sparkSession,
CommitConfigurer commitConfigurer,
- @Nullable CommitMessageRewriter commitMessageRewriter) {
+ @Nullable CommitMessageRewriter commitMessageRewriter,
+ @Nullable Consumer<List<CommitMessage>> commitObserver) {
int retryCount = 0;
long startMillis = System.currentTimeMillis();
RetryWaiter retryWaiter =
@@ -271,12 +294,18 @@ final class DataEvolutionRewriteExecutor {
}
throw conflict;
}
- return table.snapshotManager()
- .latestSnapshotOfUser(commitUser)
- .orElseThrow(
- () ->
- new IllegalStateException(
- "Cannot find the committed
data evolution rewrite snapshot."));
+ Snapshot committedSnapshot =
+ table.snapshotManager()
+ .latestSnapshotOfUser(commitUser)
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Cannot find the
committed data evolution rewrite snapshot."));
+ if (commitObserver != null) {
+ // Include successful retry artifacts, not just the
original staged outputs.
+ commitObserver.accept(preparedMessages);
+ }
+ return committedSnapshot;
} catch (RuntimeException conflict) {
if (commitMessageRewriter == null
|| !isMergeConflict(conflict)
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
index d9f9c3f865..5f57058fee 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
@@ -35,6 +35,7 @@ import
org.apache.paimon.spark.commands.{DataEvolutionCompactMergeConflictRewrit
import org.apache.paimon.spark.commands.CompactRowIdRangeIndex
import org.apache.paimon.spark.utils.SparkProcedureUtils
import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.sink.CommitMessageImpl
import org.apache.paimon.table.source.{DataSplit, EndOfScanException,
IncrementalSplit}
import org.apache.paimon.table.source.snapshot.SnapshotReader
import org.apache.paimon.utils.Range
@@ -1764,6 +1765,75 @@ abstract class CompactProcedureTestBase extends
PaimonSparkTestBase with StreamT
}
}
+ test("Paimon Procedure: split oversized files once per compact invocation
across batches") {
+ withTable("T") {
+ sql("""
+ |CREATE TABLE T (id INT, value STRING, picture BINARY, pt STRING)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'file.format' = 'avro',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'blob-field' = 'picture',
+ | 'compaction.min.file-num' = '2')
+ |PARTITIONED BY (pt)
+ |""".stripMargin)
+ for (pt <- Seq("p0", "p1")) {
+ sql(s"""
+ |INSERT INTO T SELECT /*+ REPARTITION(1) */
+ |id, concat('value-', id), CAST('blob' AS BINARY), '$pt' FROM
range(0, 1)
+ |""".stripMargin)
+ sql(s"""
+ |INSERT INTO T SELECT /*+ REPARTITION(1) */
+ |id, concat('value-', id), CAST('blob' AS BINARY), '$pt' FROM
range(1, 100)
+ |""".stripMargin)
+ }
+ val initial = loadTable("T")
+ CompactProcedure.executeDataEvolutionCompaction(
+ initial,
+ null,
+ null,
+ new JavaSparkContext(spark.sparkContext),
+ spark,
+ Int.box(2))
+ val targetSize = normalDataFiles(initial).map(_.fileSize()).min / 4
+ // Each normal file retains BLOB ranges for one row and 99 rows. The
estimated
+ // cut lands inside the latter, so the safe output remains the full 100
rows.
+ val table = initial.copy(
+ Map(
+ "target-file-size" -> s"$targetSize b",
+ "compaction.min.file-num" -> "10",
+ "data-evolution.compaction.split-large-files" -> "true").asJava)
+ assert(normalDataFiles(table).size == 2)
+ for (_ <- 0 until 2) {
+ val beforeFiles = normalDataFiles(table).map(_.fileName()).toSet
+ val beforeSnapshot = lastSnapshotId(table)
+ val attempts = new AtomicInteger()
+ CompactProcedure.executeDataEvolutionCompaction(
+ table,
+ null,
+ null,
+ null,
+ new JavaSparkContext(spark.sparkContext),
+ spark,
+ Int.box(1),
+ _ => assert(attempts.incrementAndGet() <= 2, "Recompacted this
invocation's output")
+ )
+
+ val afterFiles = normalDataFiles(table)
+ assert(attempts.get() == 2)
+ assert(lastSnapshotId(table) == beforeSnapshot + 2)
+ assert(afterFiles.size == 2)
+ assert(
+ afterFiles.forall(file => file.rowCount() == 100 && file.fileSize()
> 2 * targetSize))
+ assert(afterFiles.forall(file =>
!beforeFiles.contains(file.fileName())))
+ }
+ checkAnswer(
+ sql("SELECT id, value, pt FROM T ORDER BY id, pt"),
+ (0 until 100).flatMap(id => Seq("p0", "p1").map(pt => Row(id,
s"value-$id", pt))))
+ }
+ }
+
test("Paimon Procedure: materialize deletion vectors across planner
batches") {
withTable("T") {
sql("""
@@ -1969,6 +2039,8 @@ abstract class CompactProcedureTestBase extends
PaimonSparkTestBase with StreamT
PaimonRelation.getPaimonRelation(spark.table("T").queryExecution.analyzed)
val javaSparkContext = new JavaSparkContext(spark.sparkContext)
val attempts = new AtomicInteger()
+ val observedCommits = new AtomicInteger()
+ val observedFiles = new util.HashSet[String]()
val rewriteSnapshotId = new AtomicLong(-1L)
val mergeFileAfterRewrite = new AtomicReference[DataFileMeta]()
partialUpdate(table, "SELECT * FROM VALUES (1, 10), (2, 20) AS S(id,
value)")
@@ -2023,7 +2095,19 @@ abstract class CompactProcedureTestBase extends
PaimonSparkTestBase with StreamT
javaSparkContext,
spark,
configurer,
- messageRewriter
+ messageRewriter,
+ messages => {
+ observedCommits.incrementAndGet()
+ messages.asScala.foreach {
+ message =>
+ message
+ .asInstanceOf[CommitMessageImpl]
+ .compactIncrement()
+ .compactAfter()
+ .asScala
+ .foreach(file => observedFiles.add(file.fileName()))
+ }
+ }
)
Assertions.assertThat(attempts.get()).isEqualTo(2)
@@ -2036,6 +2120,10 @@ abstract class CompactProcedureTestBase extends
PaimonSparkTestBase with StreamT
file.fileName() != mergeFile.fileName())
assert(bridgeFiles.size == 1, bridgeFiles)
val bridgeFile = bridgeFiles.head
+ assert(observedCommits.get() == 1)
+ assert(observedFiles.size() == 2)
+ assert(observedFiles.contains(bridgeFile.fileName()))
+ assert(!observedFiles.contains(mergeFile.fileName()))
assert(bridgeFile.maxSequenceNumber() == rewriteSnapshotId.get())
assert(bridgeFile.maxSequenceNumber() < mergeFile.maxSequenceNumber())
assert(mergeFile.maxSequenceNumber() < table.latestSnapshot().get().id())
