diff --git a/docs/docs/multimodal-table/data-evolution.mdx b/docs/docs/multimodal-table/data-evolution.mdx index 02b1d271872b..c02437b42813 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 6a4955280dfb..bd231c88c047 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -506,12 +506,24 @@ Duration 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. + +
data-evolution.compaction.large-file-ratio
+ 2.0 + Double + 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. +
data-evolution.compaction.rewrite-row-ids
false Boolean 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. + +
data-evolution.compaction.split-large-files
+ false + Boolean + 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. +
data-evolution.enabled
false @@ -1817,7 +1829,7 @@
target-file-row-num
9223372036854775807 Long - 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-evolution append tables; its primary-key, blob and vector writers still fail fast when it is enabled. Disabled by default. + 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 data-evolution OOM. PyPaimon supports this for data-evolution append tables; its primary-key, blob and vector writers still fail fast when it is enabled. Disabled by default.
target-file-size
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 298126f3c66d..41b338cb9e41 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 InlineElement getDescription() { + "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 String toString() { .withDescription( "Whether to persist source when process merge into action on data evolution table."); + public static final ConfigOption 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 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 DATA_EVOLUTION_COMPACTION_REWRITE_ROW_IDS = key("data-evolution.compaction.rewrite-row-ids") .booleanType() @@ -4389,6 +4415,19 @@ public boolean deletionVectorBitmap64() { 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 947f944182f1..e991892fd4c2 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 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 @@ void finish(CandidateRangeConsumer consumer) { blobTargetFileSize, openFileCost, compactMinFileNum, + largeFileThreshold, consumer); if (chunks.size() == 1) { SortedEntryChunk chunk = chunks.get(0); @@ -402,6 +406,7 @@ private static final class CandidateAccumulator { 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 blobFields = new HashMap<>(); @@ -412,6 +417,7 @@ private static final class CandidateAccumulator { 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 @@ private CandidateAccumulator( 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 @@ private void startComponent(long start, long end, long fileSize) { normalEnd = end; normalFileCount = 1L; normalWeight = Math.max(fileSize, openFileCost); + largeFile = fileSize > largeFileThreshold; vectorFileCount = 0L; componentFileCount = 1; blobFields.clear(); @@ -472,6 +481,7 @@ private void addNormalFile(long start, long end, long fileSize) { 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 @@ private void finishComponent() { componentFileCount, normalFileCount, normalWeight, - dedicatedCandidate); + dedicatedCandidate, + largeFile); if (normalWeight > targetFileSize) { flushBin(); emitComponent(component); @@ -541,7 +552,9 @@ private void finishComponent() { } 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 @@ private static final class Component { private final long normalFileCount; private final long normalWeight; private final boolean dedicatedCandidate; + private final boolean largeFile; private Component( long start, @@ -578,13 +592,15 @@ private Component( 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 23f15772a034..5d80b8d97190 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 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 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 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 blobInlineFields = options.blobInlineField(); @@ -118,6 +126,7 @@ public DataEvolutionCompactCoordinator( new DataEvolutionCompactRangePlanner.CandidateOptions( compactBlob, compactVector, + largeFileThreshold, targetFileSize, options.blobTargetFileSize(), openFileCost, @@ -137,6 +146,7 @@ public DataEvolutionCompactCoordinator( new CompactPlanner( compactBlob, compactVector, + largeFileThreshold, targetFileSize, options.blobTargetFileSize(), openFileCost, @@ -145,6 +155,14 @@ public 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 static void validateOptions(CoreOptions options) { 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 fileNames) { + planner.completedNormalFiles = checkNotNull(fileNames); + return this; + } + public List plan() { // scan files in snapshot List entries = scanner.scan(); @@ -235,12 +259,14 @@ static class CompactPlanner { 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 schemaFetcher; @Nullable private final Set currentBlobFieldIds; + private Set completedNormalFiles = Collections.emptySet(); @VisibleForTesting CompactPlanner( @@ -252,6 +278,7 @@ static class CompactPlanner { this( compactBlob, compactVector, + Long.MAX_VALUE, targetFileSize, targetFileSize, openFileCost, @@ -266,6 +293,7 @@ static class CompactPlanner { CompactPlanner( boolean compactBlob, boolean compactVector, + long largeFileThreshold, long targetFileSize, long blobTargetFileSize, long openFileCost, @@ -274,6 +302,7 @@ static class CompactPlanner { @Nullable Set currentBlobFieldIds) { this.compactBlob = compactBlob; this.compactVector = compactVector; + this.largeFileThreshold = largeFileThreshold; this.targetFileSize = targetFileSize; this.blobTargetFileSize = blobTargetFileSize; this.openFileCost = openFileCost; @@ -314,12 +343,10 @@ List compactPlan(List input) { } } - 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 continuousDataRangeHelper = new RangeHelper<>( @@ -409,10 +436,32 @@ private List triggerTask( List dataFiles = compactBin.files(); List tasks = new ArrayList<>(); - boolean triggerNormalFile = dataFiles.size() >= compactMinFileNum; - if (triggerNormalFile) { - tasks.add(new DataEvolutionNormalCompactTask(partition, dataFiles)); + List 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 @@ private List triggerTask( } } } + 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 protectedRanges) { + return normalRange.from < normalRange.to + && Range.sortAndMergeOverlap(protectedRanges).stream() + .noneMatch( + range -> + range.from <= normalRange.from + && range.to >= normalRange.to); + } + private CompactBin compactBin(List 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 cb7e36b5fed1..e6454246c7c4 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.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 @@ private Map collectMaintainers( 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 outputFiles = new TreeMap<>(); + for (DataFileMeta file : after) { + outputFiles.put(file.nonNullFirstRowId(), file); + } + Map 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 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 @@ private boolean isMaterialized(CompactIncrement compactIncrement) { } /** - * '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 outputFiles, + Map 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 @@ private void moveDeletionVector( 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 d7abda070804..236f4fdf5367 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 @@ private Queue planManifestGroup(List manifestGroup 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 @@ static final class CandidateOptions { 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 @@ static final class CandidateOptions { CandidateOptions( boolean compactBlob, boolean compactVector, + long largeFileThreshold, long targetFileSize, long blobTargetFileSize, long openFileCost, @@ -507,6 +510,7 @@ static final class CandidateOptions { @Nullable Set 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 0019c4e17f92..6f24a41bcc16 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 @@ private static Map dynamicWriteOptions() { Map 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 40ae2073dd17..e9cc9ed19106 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.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 @@ public class DataEvolutionCompactTaskSerializer implements VersionedSerializer { - private static final int CURRENT_VERSION = 3; + private static final int CURRENT_VERSION = 4; private final DataFileMetaSerializer dataFileSerializer; @@ -73,7 +74,14 @@ private void serialize(DataEvolutionCompactTask task, DataOutputView view) throw 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 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 @@ private DataEvolutionCompactTask deserialize(int version, DataInputView view) DataEvolutionCompactTask.TaskType.fromCode(view.readInt()); switch (type) { case NORMAL: - return new DataEvolutionNormalCompactTask(partition, files); + int rangeCount = view.readInt(); + List 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 7fe573974410..6a46a4425fa7 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 @@ 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.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 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 protectedRanges; + public DataEvolutionNormalCompactTask(BinaryRow partition, List files) { + this(partition, files, Collections.emptyList()); + } + + public DataEvolutionNormalCompactTask( + BinaryRow partition, List files, List protectedRanges) { super(partition, files); checkContiguousRowRange(files); + this.protectedRanges = + Collections.unmodifiableList(Range.sortAndMergeOverlap(protectedRanges)); + } + + public List 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 CommitMessage doCompact(FileStoreTable table, String commitUser) throws E fieldNamesInBlobFile(table.rowType(), options.blobInlineField()), fieldNamesInVectorFile(table.rowType(), options.withVectorFormat())); - table = table.copy(DYNAMIC_WRITE_OPTIONS); + Map 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 CommitMessage doCompact(FileStoreTable table, String commitUser) throws E storeWrite.withWriteType(readWriteType); storeWrite.withFileSource(FileSource.COMPACT); RecordWriter writer = storeWrite.createWriter(partition, 0); - - reader.forEachRemaining( - row -> { - try { - writer.write(row); - } catch (Exception e) { - throw new RuntimeException(e); - } - }); - - List writeResult = writer.prepareCommit(false).newFilesIncrement().newFiles(); - checkArgument( - writeResult.size() == 1, "Data evolution compaction should produce one file."); - + List outputRanges = + options.dataEvolutionCompactionSplitLargeFiles() + ? planOutputRanges(options.targetFileSize(false)) + : Collections.singletonList(checkContiguousRowRange(compactBefore)); + List writeResult = new ArrayList<>(); + try (RecordReaderIterator 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 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 CommitMessage doCompact(FileStoreTable table, String commitUser) throws E 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 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 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 29c368f9f77e..a6a3db391157 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 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 @@ void testSelectsOnlyNormalFileBinsWhichCanCompact() { 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 @@ private CompactCandidateRangeCollector collector( long openFileCost, long compactMinFileNum) { return new CompactCandidateRangeCollector( - 16, targetFileSize, blobTargetFileSize, openFileCost, compactMinFileNum); + 16, + targetFileSize, + blobTargetFileSize, + openFileCost, + compactMinFileNum, + Long.MAX_VALUE); } private List 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 cd784064bbae..0ce6a252a523 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.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.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 @@ table, false, false, mock(Snapshot.class))) .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 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 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 void testCompactPlannerSingleFile() { assertThat(tasks).isEmpty(); } + @Test + public void testSplitLargeFilesUsesPhysicalSizeAndIncludesColumnUpdates() { + List 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 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 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 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 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 @@ private DataEvolutionCompactCoordinator.CompactPlanner blobPlanner( return new DataEvolutionCompactCoordinator.CompactPlanner( true, false, + Long.MAX_VALUE, targetFileSize, targetFileSize, openFileCost, @@ -912,7 +1062,11 @@ public void testSerializerBasic() throws IOException { 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 void testNormalCompactTaskRejectsDisjointRowRanges() { .hasMessageContaining("contiguous row range"); } + @Test + public void testPlanNormalOutputRangesAtDedicatedBoundaries() { + List 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 88278ec40f54..93059aeea1bb 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 @@ private DataEvolutionCompactRangePlanner.CandidateOptions candidateOptions( 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 dd91e37f3a48..6d15a8c345fa 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 @@ 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.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.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 void testOmitColumnSequencesUnlessUpdatesAreIgnored() throws Exception { 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 messages = write.prepareCommit(); + assignFirstRowId(messages, original.nonNullFirstRowId()); + commit.commit(messages); + } + } + List expected = read(table); + Map 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 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 messages = new ArrayList<>(); + messages.add(task.doCompact(table, "split-large-file")); + List 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 dedicated = dedicatedFiles(table); + assertThat(dedicated).hasSize(3); + Map 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 originalBlobs = dedicatedFiles(table); + assertThat(originalBlobs).hasSize(3); + Map 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 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 messages = new ArrayList<>(); + List 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 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 dedicated = dedicatedFiles(table); + assertThat(dedicated).hasSize(4); + assertThat(dedicated).allSatisfy(file -> assertThat(file.rowCount()).isLessThan(3750)); + Map 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 ids = new ArrayList<>(); + try (RecordReader 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 dedicated = dedicatedFiles(table); + assertThat(dedicated).hasSize(1); + assertThat(dedicated.get(0).nonNullRowIdRange()).isEqualTo(new Range(0, 2499)); + Map 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 ids = new ArrayList<>(); + try (RecordReader 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 dedicated = dedicatedFiles(table); + assertThat(dedicated).hasSize(3); + + DataEvolutionCompactTask compacted = compactSingleTask(table); + assertThat(compacted.compactBefore()).hasSize(3); + List 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 ids = new ArrayList<>(); + ReadBuilder readBuilder = table.newReadBuilder(); + try (RecordReader 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 ids = new ArrayList<>(); + ReadBuilder readBuilder = table.newReadBuilder(); + try (RecordReader 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 messages = write.prepareCommit(); + if (existingRows) { + assignFirstRowId(messages, from); + } + commit.commit(messages); + } + } + + private DataEvolutionCompactTask compactSingleTask(FileStoreTable table) throws Exception { + Snapshot snapshot = table.snapshotManager().latestSnapshot(); + List 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 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 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 expected) { + assertThat(dedicatedFiles(table)).containsExactlyInAnyOrderElementsOf(expected); + List 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 384f3fe2a306..8f189a791760 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.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.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 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 void testCompactRewritesDeletionVectors() throws Exception { assertThat(liveDeletionVectorDataFileNames).doesNotContainAnyElementsOf(oldAnchorFiles); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testSplitLargeFilePreservesDeletionVectorsAndBlob(boolean bitmap64) + throws Exception { + createTableDefault(); + Map 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 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 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 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 files = currentDataFiles(table, BinaryRow.EMPTY_ROW); + List 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 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 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 actualBlobs = new ArrayList<>(); + try (RecordReader reader = + blobRead.newRead().createReader(blobRead.newScan().plan())) { + reader.forEachRemaining(row -> actualBlobs.add(row.getBlob(0).toData()[0])); + } + List 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 832bd5478083..218dc6d0a958 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.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.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 @@ static void executeDataEvolutionCompaction( return; } AtomicReference coordinatorRef = new AtomicReference<>(); + Set completedNormalFiles = new HashSet<>(); Function> taskPlanner = planningSnapshot -> { DataEvolutionCompactCoordinator coordinator = coordinatorRef.get(); @@ -685,11 +689,31 @@ static void executeDataEvolutionCompaction( false, planningSnapshot, candidateFilesPerBatch); + coordinator.withCompletedNormalFiles(completedNormalFiles); coordinatorRef.set(coordinator); } return filterIdlePartitions( coordinator.plan(), table, partitionPredicate, partitionIdleTime); }; + Consumer> 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 @@ static void executeDataEvolutionCompaction( commitConfigurer, relation == null ? null - : new DataEvolutionCompactMergeConflictRewriter(table, relation)::rewrite); + : new DataEvolutionCompactMergeConflictRewriter(table, relation)::rewrite, + commitObserver); } private static List 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 b664b82f7eaf..e964f7b61c31 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.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 @@ static void execute( SparkSession sparkSession, CommitConfigurer commitConfigurer, @Nullable CommitMessageRewriter commitMessageRewriter) { + execute( + table, + initialSnapshot, + taskPlanner, + javaSparkContext, + sparkSession, + commitConfigurer, + commitMessageRewriter, + null); + } + + static void execute( + FileStoreTable table, + Snapshot initialSnapshot, + Function> taskPlanner, + JavaSparkContext javaSparkContext, + SparkSession sparkSession, + CommitConfigurer commitConfigurer, + @Nullable CommitMessageRewriter commitMessageRewriter, + @Nullable Consumer> commitObserver) { CommitMessageSerializer messageSerializer = new CommitMessageSerializer(); String commitUser = createCommitUser(table.coreOptions().toConfiguration()); Snapshot preparationSnapshot = initialSnapshot; @@ -167,7 +188,8 @@ static void execute( 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 @@ private static Snapshot commitWithMergeConflictRetry( String commitUser, SparkSession sparkSession, CommitConfigurer commitConfigurer, - @Nullable CommitMessageRewriter commitMessageRewriter) { + @Nullable CommitMessageRewriter commitMessageRewriter, + @Nullable Consumer> commitObserver) { int retryCount = 0; long startMillis = System.currentTimeMillis(); RetryWaiter retryWaiter = @@ -271,12 +294,18 @@ private static Snapshot commitWithMergeConflictRetry( } 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 d9f9c3f865fb..5f57058fee53 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())