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

docs/generated/core_configuration.html

@@ -1266,6 +1266,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

@@ -2637,6 +2637,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()
@@ -4272,6 +4280,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,51 @@ public void forEach(
         }
     }
 
+    /**
+     * Creates a sorted reader which merges records with the same key.
+     *
+     * <p>The returned reader owns this buffer. Closing it clears the buffer and releases spill
+     * files. No records may be added after this method is called.
+     */
+    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();