[core] Split large normal files in data evolution compaction

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:

docs/generated/core_configuration.html

@@ -506,12 +506,24 @@
             <td>Duration</td>
             <td>The TTL in local index for cross partition upsert (primary keys not contain all partition fields), this can avoid maintaining too many indexes and lead to worse and worse performance, but please note that this may also cause data duplication.</td>
         </tr>
+        <tr>
+            <td><h5>data-evolution.compaction.large-file-ratio</h5></td>
+            <td style="word-wrap: break-word;">2.0</td>
+            <td>Double</td>
+            <td>Size multiplier relative to target-file-size for selecting large normal files when data-evolution.compaction.split-large-files is enabled. An individual file must strictly exceed this threshold. The value must be finite and at least 1.0. This does not change the target size of compacted output files.</td>
+        </tr>
         <tr>
             <td><h5>data-evolution.compaction.rewrite-row-ids</h5></td>
             <td style="word-wrap: break-word;">false</td>
             <td>Boolean</td>
             <td>Legacy compatibility option. Setting this option to true fails. Data-evolution compaction preserves row IDs and logical deletions. Use the 'materialize_deletion_vectors' procedure to apply deletion vectors to the latest table state and assign new row IDs. Reclaiming files retained by historical snapshots or tags requires snapshot expiration.</td>
         </tr>
+        <tr>
+            <td><h5>data-evolution.compaction.split-large-files</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether data-evolution compaction selects normal data files larger than data-evolution.compaction.large-file-ratio times target-file-size, even below compaction.min.file-num when dedicated-file ranges allow splitting. Normal output ranges are estimated from input file sizes and row counts toward target-file-size, then adjusted to avoid cutting through any BLOB or VECTOR file range. Actual output sizes may differ from the target. Row IDs and logical deletions are preserved, and associated BLOB and VECTOR files are not rewritten by this option.</td>
+        </tr>
         <tr>
             <td><h5>data-evolution.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>
@@ -1817,7 +1829,7 @@
             <td><h5>target-file-row-num</h5></td>
             <td style="word-wrap: break-word;">9223372036854775807</td>
             <td>Long</td>
-            <td>Target number of rows per newly written data file; a file rolls when this or target-file-size is reached, whichever comes first. Enforced at bundle granularity, so a bundled write may exceed it by up to one bundle. Only constrains files at write time: compaction is size-based and may merge into larger files, and data-evolution compaction still produces a single file. Bounds per-file rows for wide columns to avoid data-evolution OOM. PyPaimon supports this for data-evolution append tables; its primary-key, blob and vector writers still fail fast when it is enabled. Disabled by default.</td>
+            <td>Target number of rows per newly written data file; a file rolls when this or target-file-size is reached, whichever comes first. Enforced at bundle granularity, so a bundled write may exceed it by up to one bundle. Only constrains files at write time: compaction is size-based and may merge into larger files, and data-evolution compaction 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.</td>
         </tr>
         <tr>
             <td><h5>target-file-size</h5></td>

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<Boolean> DATA_EVOLUTION_COMPACTION_SPLIT_LARGE_FILES =
+            key("data-evolution.compaction.split-large-files")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether data-evolution compaction selects normal data files larger than "
+                                    + "data-evolution.compaction.large-file-ratio times target-file-size, "
+                                    + "even below compaction.min.file-num when dedicated-file ranges allow splitting. "
+                                    + "Normal output ranges are estimated from input file sizes and row counts "
+                                    + "toward target-file-size, then adjusted to avoid cutting through any "
+                                    + "BLOB or VECTOR file range. Actual output sizes may differ from the target. "
+                                    + "Row IDs and logical deletions are preserved, and associated "
+                                    + "BLOB and VECTOR files are not rewritten by this option.");
+
+    public static final ConfigOption<Double> DATA_EVOLUTION_COMPACTION_LARGE_FILE_RATIO =
+            key("data-evolution.compaction.large-file-ratio")
+                    .doubleType()
+                    .defaultValue(2.0d)
+                    .withDescription(
+                            "Size multiplier relative to target-file-size for selecting large normal "
+                                    + "files when data-evolution.compaction.split-large-files is enabled. "
+                                    + "An individual file must strictly exceed this threshold. The value "
+                                    + "must be finite and at least 1.0. This does not change the target "
+                                    + "size of compacted output files.");
+
     public static final ConfigOption<Boolean> DATA_EVOLUTION_COMPACTION_REWRITE_ROW_IDS =
             key("data-evolution.compaction.rewrite-row-ids")
                     .booleanType()
@@ -4389,6 +4415,19 @@ public 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);
     }

paimon-core/src/main/java/org/apache/paimon/append/dataevolution/CompactCandidateRangeCollector.java

@@ -53,6 +53,7 @@ final class CompactCandidateRangeCollector {
     private final long blobTargetFileSize;
     private final long openFileCost;
     private final long compactMinFileNum;
+    private final long largeFileThreshold;
     private final List<SortedEntryChunk> sortedChunks = new ArrayList<>();
     private long[] words;
     private int chunkSize;
@@ -64,7 +65,8 @@ final class CompactCandidateRangeCollector {
             long targetFileSize,
             long blobTargetFileSize,
             long openFileCost,
-            long compactMinFileNum) {
+            long compactMinFileNum,
+            long largeFileThreshold) {
         checkArgument(expectedFileCount >= 0, "Expected live file count cannot be negative.");
         checkArgument(targetFileSize > 0, "Target file size must be positive.");
         checkArgument(blobTargetFileSize > 0, "Blob target file size must be positive.");
@@ -75,6 +77,7 @@ final class CompactCandidateRangeCollector {
         this.blobTargetFileSize = blobTargetFileSize;
         this.openFileCost = openFileCost;
         this.compactMinFileNum = compactMinFileNum;
+        this.largeFileThreshold = largeFileThreshold;
         int initialEntries = Math.max(16, Math.min(expectedFileCount, ENTRY_CHUNK_SIZE));
         this.words = new long[Math.multiplyExact(initialEntries, ENTRY_WORDS)];
     }
@@ -137,6 +140,7 @@ 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<Integer, BlobFieldAccumulator> 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,20 +584,23 @@ 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,
                 long end,
                 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;
         }
     }