[spark] Support merge-on-read for postpone bucket tables

docs/generated/core_configuration.html

@@ -1278,6 +1278,12 @@
             <td>Integer</td>
             <td>Bucket number for the partitions compacted for the first time in postpone bucket tables.</td>
         </tr>
+        <tr>
+            <td><h5>postpone.merge-on-read</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to merge records in the postpone bucket with records in real buckets during batch reads. This requires an execution engine capable of routing and shuffling postpone records to their target real buckets.</td>
+        </tr>
         <tr>
             <td><h5>postpone.target-row-num-per-bucket</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -2701,6 +2701,14 @@ public String toString() {
                     .withDescription(
                             "Whether to write the data into fixed bucket for batch writing a postpone bucket table.");
 
+    public static final ConfigOption<Boolean> POSTPONE_MERGE_ON_READ =
+            key("postpone.merge-on-read")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to merge records in the postpone bucket with records in real buckets during batch reads. "
+                                    + "This requires an execution engine capable of routing and shuffling postpone records to their target real buckets.");
+
     public static final ConfigOption<Integer> POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM =
             key("postpone.batch-write-fixed-bucket.max-parallelism")
                     .intType()
@@ -4364,6 +4372,10 @@ public boolean postponeBatchWriteFixedBucket() {
         return options.get(POSTPONE_BATCH_WRITE_FIXED_BUCKET);
     }
 
+    public boolean postponeMergeOnRead() {
+        return options.get(POSTPONE_MERGE_ON_READ);
+    }
+
     public int postponeBatchWriteFixedBucketMaxParallelism() {
         return options.get(POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM);
     }

paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java

@@ -35,6 +35,7 @@
 import org.apache.paimon.mergetree.compact.MergeFunction;
 import org.apache.paimon.mergetree.compact.ReducerMergeFunctionWrapper;
 import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.sort.BinaryExternalSortBuffer;
 import org.apache.paimon.sort.BinaryInMemorySortBuffer;
 import org.apache.paimon.sort.SortBuffer;
@@ -181,6 +182,46 @@ public void forEach(
         }
     }
 
+    /** Returns a merged reader which releases this buffer when closed. */
+    public RecordReader<KeyValue> createReader(
+            Comparator<InternalRow> keyComparator, MergeFunction<KeyValue> mergeFunction)
+            throws IOException {
+        MergeIterator mergeIterator =
+                new MergeIterator(null, buffer.sortedIterator(), keyComparator, mergeFunction);
+        return new RecordReader<KeyValue>() {
+
+            private boolean read;
+            private boolean closed;
+
+            @Nullable
+            @Override
+            public RecordIterator<KeyValue> readBatch() {
+                if (read) {
+                    return null;
+                }
+                read = true;
+                return new RecordIterator<KeyValue>() {
+                    @Nullable
+                    @Override
+                    public KeyValue next() throws IOException {
+                        return mergeIterator.hasNext() ? mergeIterator.next() : null;
+                    }
+
+                    @Override
+                    public void releaseBatch() {}
+                };
+            }
+
+            @Override
+            public void close() {
+                if (!closed) {
+                    closed = true;
+                    clear();
+                }
+            }
+        };
+    }
+
     @Override
     public void clear() {
         buffer.clear();

paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java

@@ -33,6 +33,7 @@
 import org.apache.paimon.mergetree.DropDeleteReader;
 import org.apache.paimon.mergetree.MergeSorter;
 import org.apache.paimon.mergetree.MergeTreeReaders;
+import org.apache.paimon.mergetree.SortBufferWriteBuffer;
 import org.apache.paimon.mergetree.SortedRun;
 import org.apache.paimon.mergetree.compact.ConcatRecordReader;
 import org.apache.paimon.mergetree.compact.IntervalPartition;
@@ -41,6 +42,7 @@
 import org.apache.paimon.mergetree.compact.MergeFunctionWrapper;
 import org.apache.paimon.mergetree.compact.ReducerMergeFunctionWrapper;
 import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.reader.EmptyRecordReader;
 import org.apache.paimon.reader.ReaderSupplier;
 import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.schema.TableSchema;
@@ -58,6 +60,8 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Set;
@@ -74,7 +78,9 @@
  */
 public class MergeFileSplitRead implements SplitRead<KeyValue> {
 
+    private final CoreOptions options;
     private final TableSchema tableSchema;
+    private final RowType keyType;
     private final FileIO fileIO;
     private final KeyValueFileReaderFactory.Builder readerFactoryBuilder;
     private final Comparator<InternalRow> keyComparator;
@@ -85,6 +91,7 @@ public class MergeFileSplitRead implements SplitRead<KeyValue> {
 
     @Nullable private RowType readKeyType;
     @Nullable private RowType outerReadType;
+    @Nullable private IOManager ioManager;
 
     @Nullable private List<Predicate> filtersForKeys;
     @Nullable private List<Predicate> filtersForAll;
@@ -99,7 +106,9 @@ public MergeFileSplitRead(
             Comparator<InternalRow> keyComparator,
             MergeFunctionFactory<KeyValue> mfFactory,
             KeyValueFileReaderFactory.Builder readerFactoryBuilder) {
+        this.options = options;
         this.tableSchema = schema;
+        this.keyType = keyType;
         this.readerFactoryBuilder = readerFactoryBuilder;
         this.fileIO = readerFactoryBuilder.fileIO();
         this.keyComparator = keyComparator;
@@ -131,6 +140,21 @@ public MergeFileSplitRead withReadKeyType(RowType readKeyType) {
 
     @Override
     public MergeFileSplitRead withReadType(RowType readType) {
+        RowType adjustedReadType = adjustReadType(readType);
+
+        readerFactoryBuilder.withReadValueType(adjustedReadType);
+        mergeSorter.setProjectedValueType(adjustedReadType);
+
+        // Project away fields added for merging.
+        if (adjustedReadType != readType) {
+            outerReadType = readType;
+        }
+
+        return this;
+    }
+
+    /** Returns the value type required internally by the configured merge function. */
+    public RowType adjustReadType(RowType readType) {
         RowType tableRowType = tableSchema.logicalRowType();
         RowType adjustedReadType = readType;
 
@@ -150,20 +174,12 @@ public MergeFileSplitRead withReadType(RowType readType) {
             }
         }
         adjustedReadType = mfFactory.adjustReadType(adjustedReadType);
-
-        readerFactoryBuilder.withReadValueType(adjustedReadType);
-        mergeSorter.setProjectedValueType(adjustedReadType);
-
-        // When finalReadType != readType, need to project the outer read type
-        if (adjustedReadType != readType) {
-            outerReadType = readType;
-        }
-
-        return this;
+        return adjustedReadType;
     }
 
     @Override
     public MergeFileSplitRead withIOManager(IOManager ioManager) {
+        this.ioManager = ioManager;
         this.mergeSorter.setIOManager(ioManager);
         if (mfFactory instanceof LookupMergeFunction.Factory) {
             ((LookupMergeFunction.Factory) mfFactory).withIOManager(ioManager);
@@ -246,6 +262,20 @@ public RecordReader<KeyValue> createReader(DataSplit split) throws IOException {
         }
     }
 
+    /** Reads a writer-grouped postpone split with key predicates only. */
+    public RecordReader<KeyValue> createPostponeReader(DataSplit split) throws IOException {
+        if (split.isStreaming() || split.bucket() != BucketMode.POSTPONE_BUCKET) {
+            throw new IllegalArgumentException(
+                    "Postpone reader requires a batch split from the postpone bucket.");
+        }
+        return createNoMergeReader(
+                split.partition(),
+                split.bucket(),
+                split.dataFiles(),
+                split.deletionFiles().orElse(null),
+                true);
+    }
+
     public RecordReader<KeyValue> createChainReader(ChainSplit chainSplit) throws IOException {
         List<DataFileMeta> files = chainSplit.dataFiles();
         ChainReadContext chainReadContext =
@@ -306,7 +336,140 @@ public RecordReader<KeyValue> createMergeReader(
                                     mergeFuncWrapper,
                                     mergeSorter));
         }
-        RecordReader<KeyValue> reader = ConcatRecordReader.create(sectionReaders);
+        return finishMergeReader(ConcatRecordReader.create(sectionReaders), keepDelete);
+    }
+
+    /**
+     * Merges one real-bucket split with routed postpone records.
+     *
+     * <p>The postpone reader is consumed and closed eagerly. Its input order defines sequence
+     * numbers after the real files, while user-defined sequence fields take precedence. A null
+     * split represents a postpone-only bucket.
+     */
+    public RecordReader<KeyValue> createPostponeMergeReader(
+            @Nullable DataSplit realSplit, RecordReader<KeyValue> postponeRecords)
+            throws IOException {
+        List<DataFileMeta> realFiles =
+                realSplit == null ? Collections.emptyList() : realSplit.dataFiles();
+
+        RowType valueType = actualReadType();
+        UserDefinedSeqComparator udsComparator = null;
+        SortBufferWriteBuffer postponeBuffer = null;
+        long nextSequenceNumber = -1L;
+        try (RecordReader<KeyValue> records = postponeRecords) {
+            RecordReader.RecordIterator<KeyValue> batch;
+            while ((batch = records.readBatch()) != null) {
+                try {
+                    KeyValue record;
+                    while ((record = batch.next()) != null) {
+                        if (postponeBuffer == null) {
+                            for (DataFileMeta file : realFiles) {
+                                nextSequenceNumber =
+                                        Math.max(nextSequenceNumber, file.maxSequenceNumber());
+                            }
+                            udsComparator = createUdsComparator();
+                            postponeBuffer =
+                                    new SortBufferWriteBuffer(
+                                            keyType,
+                                            valueType,
+                                            udsComparator,
+                                            mergeSorter.memoryPool(),
+                                            options.writeBufferSpillable(),
+                                            options.writeBufferSpillDiskSize(),
+                                            options.localSortMaxNumFileHandles(),
+                                            options.spillCompressOptions(),
+                                            ioManager);
+                        }
+                        nextSequenceNumber = Math.addExact(nextSequenceNumber, 1L);
+                        if (!postponeBuffer.put(
+                                nextSequenceNumber,
+                                record.valueKind(),
+                                record.key(),
+                                record.value())) {
+                            throw new IOException(
+                                    "Postpone merge read buffer is full. Enable write-buffer-spillable or increase sort-spill-buffer-size.");
+                        }
+                    }
+                } finally {
+                    batch.releaseBatch();
+                }
+            }
+        } catch (IOException | RuntimeException e) {
+            if (postponeBuffer != null) {
+                postponeBuffer.clear();
+            }
+            throw e;
+        }
+
+        if (postponeBuffer == null) {
+            return realSplit == null ? new EmptyRecordReader<>() : createReader(realSplit);
+        }
+
+        final RecordReader<KeyValue> postponeReader;
+        try {
+            postponeReader =
+                    postponeBuffer.createReader(keyComparator, mfFactory.create(valueType));
+        } catch (IOException | RuntimeException e) {
+            postponeBuffer.clear();
+            throw e;
+        }
+
+        if (realFiles.isEmpty()) {
+            return finishMergeReader(postponeReader, forceKeepDelete);
+        }
+
+        // Postpone records make real-file value filters unsafe before the final merge.
+        final RecordReader<KeyValue> realBucketReader;
+        try {
+            DeletionVector.Factory dvFactory =
+                    DeletionVector.factory(
+                            fileIO, realFiles, realSplit.deletionFiles().orElse(null));
+            KeyValueFileReaderFactory realFileReaderFactory =
+                    readerFactoryBuilder.build(
+                            realSplit.partition(),
+                            realSplit.bucket(),
+                            dvFactory,
+                            false,
+                            filtersForKeys);
+            realBucketReader =
+                    MergeTreeReaders.readerForMergeTree(
+                            new IntervalPartition(realFiles, keyComparator).partition(),
+                            realFileReaderFactory,
+                            keyComparator,
+                            udsComparator,
+                            new ReducerMergeFunctionWrapper(mfFactory.create(valueType)),
+                            mergeSorter);
+        } catch (IOException | RuntimeException e) {
+            closeAndSuppress(postponeReader, e);
+            throw e;
+        }
+
+        RecordReader<KeyValue> reader;
+        try {
+            reader =
+                    mergeSorter.mergeSortNoSpill(
+                            Arrays.asList(() -> realBucketReader, () -> postponeReader),
+                            keyComparator,
+                            udsComparator,
+                            new ReducerMergeFunctionWrapper(mfFactory.create(valueType)));
+        } catch (IOException | RuntimeException e) {
+            closeAndSuppress(realBucketReader, e);
+            closeAndSuppress(postponeReader, e);
+            throw e;
+        }
+        return finishMergeReader(reader, forceKeepDelete);
+    }
+
+    private static void closeAndSuppress(RecordReader<?> reader, Exception exception) {
+        try {
+            reader.close();
+        } catch (IOException closeException) {
+            exception.addSuppressed(closeException);
+        }
+    }
+
+    private RecordReader<KeyValue> finishMergeReader(
+            RecordReader<KeyValue> reader, boolean keepDelete) {
 
         if (!keepDelete) {
             reader = new DropDeleteReader(reader);

paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java

@@ -241,4 +241,14 @@ public static int getWriteId(String fileName) {
                     e);
         }
     }
+
+    /** Returns the unique prefix shared by files produced by the same postpone writer. */
+    public static String getWriterPrefix(String fileName) {
+        int separator = fileName.lastIndexOf("-w-");
+        if (separator < 0) {
+            throw new IllegalArgumentException(
+                    "Data file name " + fileName + " does not match the postpone writer pattern.");
+        }
+        return fileName.substring(0, separator + 3);
+    }
 }