You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/07/26 10:47:46 UTC
(paimon) branch master updated: [spark] Support merge-on-read for postpone bucket tables (#8802)
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new cb51c68131 [spark] Support merge-on-read for postpone bucket tables (#8802)
cb51c68131 is described below
commit cb51c681316e3a6eb46cc8a2db6433b9972bb4c1
Author: Zouxxyy <zo...@alibaba-inc.com>
AuthorDate: Sun Jul 26 18:47:41 2026 +0800
[spark] Support merge-on-read for postpone bucket tables (#8802)
---
docs/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 12 +
.../paimon/mergetree/SortBufferWriteBuffer.java | 41 +++
.../paimon/operation/MergeFileSplitRead.java | 185 +++++++++++-
.../postpone/PostponeBucketFileStoreWrite.java | 10 +
.../org/apache/paimon/table/PostponeUtils.java | 261 ++++++++++++++++-
.../paimon/table/source/PostponeMergePlan.java | 124 ++++++++
.../paimon/table/source/PostponeMergeRead.java | 87 ++++++
.../table/source/PostponeMergeReadBuilder.java | 256 ++++++++++++++++
.../mergetree/SortBufferWriteBufferTestBase.java | 20 ++
.../paimon/operation/MergeFileSplitReadTest.java | 185 +++++++++++-
.../org/apache/paimon/table/PostponeUtilsTest.java | 116 +++++++-
.../apache/paimon/table/source/TableScanTest.java | 195 ++++++++++++
.../spark/sql/paimon/shims/MinorVersionShim.scala | 17 +-
.../spark/sql/paimon/shims/MinorVersionShim.scala | 17 +-
.../apache/spark/sql/paimon/shims/Spark4Shim.scala | 16 +-
.../org/apache/paimon/spark/PaimonBaseScan.scala | 57 +++-
.../apache/paimon/spark/PaimonScanBuilder.scala | 4 +
.../paimon/spark/PostponeMergeInputScan.scala | 235 +++++++++++++++
.../apache/paimon/spark/PostponeMergeOnRead.scala | 113 +++++++
...OptimizeMetadataOnlyDeleteFromPaimonTable.scala | 13 +-
.../paimon/spark/execution/PaimonStrategy.scala | 40 ++-
.../spark/execution/PostponeMergeOnReadExec.scala | 247 ++++++++++++++++
.../DisablePostponeCarrierShuffleCoalescing.scala | 63 ++++
.../extensions/PaimonSparkSessionExtensions.scala | 5 +-
.../procedure/SparkPostponeCompactProcedure.scala | 61 +---
.../paimon/spark/read/BinPackingSplits.scala | 62 ++--
.../apache/spark/sql/paimon/shims/SparkShim.scala | 19 +-
.../paimon/spark/sql/PostponeBucketTableTest.scala | 326 ++++++++++++++++++++-
.../spark/sql/PrimaryKeyVectorSearchTest.scala | 14 +
.../spark/sql/paimon/shims/MinorVersionShim.scala | 17 +-
.../apache/spark/sql/paimon/shims/Spark3Shim.scala | 16 +-
.../apache/spark/sql/paimon/shims/Spark4Shim.scala | 16 +-
33 files changed, 2727 insertions(+), 129 deletions(-)
diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index b1da97ffea..435fabede8 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1278,6 +1278,12 @@ For an internal format table in a REST catalog, it also makes the catalog own th
<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>
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 2def6e9ffc..b395635acd 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2701,6 +2701,14 @@ public class CoreOptions implements Serializable {
.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 class CoreOptions implements Serializable {
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);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
index 6f6d84c935..594252ca3b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
@@ -35,6 +35,7 @@ import org.apache.paimon.memory.MemorySegmentPool;
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 class SortBufferWriteBuffer implements WriteBuffer {
}
}
+ /** 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();
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
index 689117f1d2..50bb08cfc6 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
@@ -33,6 +33,7 @@ import org.apache.paimon.io.KeyValueFileReaderFactory;
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.MergeFunctionFactory;
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 javax.annotation.Nullable;
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 @@ import static org.apache.paimon.predicate.PredicateBuilder.splitAnd;
*/
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 class MergeFileSplitRead implements SplitRead<KeyValue> {
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 class MergeFileSplitRead implements SplitRead<KeyValue> {
@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 class MergeFileSplitRead implements SplitRead<KeyValue> {
}
}
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 class MergeFileSplitRead implements SplitRead<KeyValue> {
}
}
+ /** 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 class MergeFileSplitRead implements SplitRead<KeyValue> {
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);
diff --git a/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java
index 43c51d7ee7..f75be9a2b5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java
@@ -241,4 +241,14 @@ public class PostponeBucketFileStoreWrite extends MemoryFileStoreWrite<KeyValue>
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);
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java b/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java
index 8a600eee9e..0463bb3f2d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java
@@ -18,17 +18,33 @@
package org.apache.paimon.table;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.bucket.BucketFunction;
+import org.apache.paimon.codegen.CodeGenUtils;
+import org.apache.paimon.codegen.Projection;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.SimpleFileEntry;
+import org.apache.paimon.operation.FileStoreScan;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.postpone.PostponeBucketFileStoreWrite;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.DeletionFile;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
import javax.annotation.Nullable;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -43,6 +59,129 @@ import static org.apache.paimon.CoreOptions.WRITE_ONLY;
/** Utils for postpone table. */
public class PostponeUtils {
+ /**
+ * Groups each partition's postpone files by writer and creation time. Equal creation times keep
+ * scan order; different writers have no relative order.
+ */
+ public static List<DataSplit> groupPostponeFiles(List<DataSplit> splits) {
+ Map<Pair<BinaryRow, String>, List<PostponeFile>> filesByWriter = new LinkedHashMap<>();
+ for (DataSplit split : splits) {
+ if (split.isStreaming() || split.bucket() != BucketMode.POSTPONE_BUCKET) {
+ throw new IllegalArgumentException(
+ "Postpone file grouping requires batch splits from the postpone bucket.");
+ }
+
+ List<DataFileMeta> dataFiles = split.dataFiles();
+ List<DeletionFile> deletionFiles = split.deletionFiles().orElse(null);
+ for (int fileOrder = 0; fileOrder < dataFiles.size(); fileOrder++) {
+ DataFileMeta file = dataFiles.get(fileOrder);
+ Pair<BinaryRow, String> writer =
+ Pair.of(
+ split.partition(),
+ PostponeBucketFileStoreWrite.getWriterPrefix(file.fileName()));
+ filesByWriter
+ .computeIfAbsent(writer, ignored -> new ArrayList<>())
+ .add(
+ new PostponeFile(
+ split,
+ file,
+ deletionFiles == null ? null : deletionFiles.get(fileOrder),
+ deletionFiles != null));
+ }
+ }
+
+ List<DataSplit> result = new ArrayList<>(filesByWriter.size());
+ for (List<PostponeFile> writerFiles : filesByWriter.values()) {
+ writerFiles.sort(Comparator.comparingLong(file -> file.file.creationTimeEpochMillis()));
+
+ PostponeFile first = writerFiles.get(0);
+ List<DataFileMeta> dataFiles = new ArrayList<>(writerFiles.size());
+ List<DeletionFile> deletionFiles = new ArrayList<>(writerFiles.size());
+ boolean hasDeletionFiles = false;
+ boolean rawConvertible = true;
+ for (PostponeFile file : writerFiles) {
+ dataFiles.add(file.file);
+ deletionFiles.add(file.deletionFile);
+ hasDeletionFiles |= file.deletionFilesPresent;
+ rawConvertible &= file.split.rawConvertible();
+ }
+
+ DataSplit split = first.split;
+ DataSplit.Builder builder =
+ DataSplit.builder()
+ .withSnapshot(split.snapshotId())
+ .withPartition(split.partition())
+ .withBucket(split.bucket())
+ .withBucketPath(split.bucketPath())
+ .withTotalBuckets(split.totalBuckets())
+ .withDataFiles(dataFiles)
+ .isStreaming(false)
+ .rawConvertible(rawConvertible);
+ if (hasDeletionFiles) {
+ builder.withDataDeletionFiles(deletionFiles);
+ }
+ result.add(builder.build());
+ }
+ return result;
+ }
+
+ public static PostponeBucketAssigner createPostponeBucketAssigner(
+ FileStoreTable table, long snapshotId, int defaultParallelism) {
+ return createPostponeBucketAssigner(table, snapshotId, defaultParallelism, null);
+ }
+
+ private static PostponeBucketAssigner createPostponeBucketAssigner(
+ FileStoreTable table,
+ long snapshotId,
+ int defaultParallelism,
+ @Nullable PartitionPredicate partitionFilter) {
+ Map<BinaryRow, Integer> knownNumBuckets =
+ getKnownNumBuckets(table, snapshotId, partitionFilter);
+ Long targetRowNumPerBucket =
+ table.coreOptions().postponeTargetRowNumPerBucket().orElse(null);
+ Map<BinaryRow, Long> postponeRowCounts =
+ targetRowNumPerBucket == null
+ ? Collections.emptyMap()
+ : getPostponeRowCounts(table, snapshotId, partitionFilter);
+ int defaultBucketNum =
+ table.coreOptions()
+ .toConfiguration()
+ .contains(CoreOptions.POSTPONE_DEFAULT_BUCKET_NUM)
+ ? table.coreOptions().postponeDefaultBucketNum()
+ : defaultParallelism;
+ return new PostponeBucketAssigner(
+ knownNumBuckets, targetRowNumPerBucket, postponeRowCounts, defaultBucketNum);
+ }
+
+ public static PostponeBucketRouter createPostponeBucketRouter(
+ FileStoreTable table,
+ long snapshotId,
+ int defaultParallelism,
+ @Nullable PartitionPredicate partitionFilter) {
+ List<String> trimmedPrimaryKeys = table.schema().trimmedPrimaryKeys();
+ int[] bucketKeyMapping =
+ table.schema().bucketKeys().stream()
+ .mapToInt(trimmedPrimaryKeys::indexOf)
+ .toArray();
+ for (int index : bucketKeyMapping) {
+ if (index < 0) {
+ throw new IllegalArgumentException(
+ "Postpone bucket key must be part of the trimmed primary key.");
+ }
+ }
+ RowType keyType =
+ new RowType(
+ PrimaryKeyTableUtils.PrimaryKeyFieldsExtractor.EXTRACTOR.keyFields(
+ table.schema()));
+ return new PostponeBucketRouter(
+ createPostponeBucketAssigner(
+ table, snapshotId, defaultParallelism, partitionFilter),
+ keyType,
+ table.schema().logicalBucketKeyType(),
+ bucketKeyMapping,
+ table.coreOptions().bucketFunctionType());
+ }
+
public static int computeBucketNumByRowCount(long rowCount, long targetRowNumPerBucket) {
if (targetRowNumPerBucket <= 0) {
throw new IllegalArgumentException(
@@ -100,12 +239,16 @@ public class PostponeUtils {
public static Map<BinaryRow, Integer> getKnownNumBuckets(
FileStoreTable table, long snapshotId) {
- return getKnownNumBuckets(
- table.store()
- .newScan()
- .withSnapshot(snapshotId)
- .onlyReadRealBuckets()
- .readSimpleEntries());
+ return getKnownNumBuckets(table, snapshotId, null);
+ }
+
+ static Map<BinaryRow, Integer> getKnownNumBuckets(
+ FileStoreTable table, long snapshotId, @Nullable PartitionPredicate partitionFilter) {
+ FileStoreScan scan = table.store().newScan().withSnapshot(snapshotId).onlyReadRealBuckets();
+ if (partitionFilter != null) {
+ scan.withPartitionFilter(partitionFilter);
+ }
+ return getKnownNumBuckets(scan.readSimpleEntries());
}
private static Map<BinaryRow, Integer> getKnownNumBuckets(
@@ -157,11 +300,19 @@ public class PostponeUtils {
/** Returns row counts of active postpone files in the specified snapshot. */
public static Map<BinaryRow, Long> getPostponeRowCounts(FileStoreTable table, long snapshotId) {
- return getPostponeRowCounts(
+ return getPostponeRowCounts(table, snapshotId, null);
+ }
+
+ static Map<BinaryRow, Long> getPostponeRowCounts(
+ FileStoreTable table, long snapshotId, @Nullable PartitionPredicate partitionFilter) {
+ SnapshotReader reader =
table.newSnapshotReader()
.withSnapshot(snapshotId)
- .withBucket(BucketMode.POSTPONE_BUCKET)
- .readFileIterator());
+ .withBucket(BucketMode.POSTPONE_BUCKET);
+ if (partitionFilter != null) {
+ reader.withPartitionFilter(partitionFilter);
+ }
+ return getPostponeRowCounts(reader.readFileIterator());
}
private static Map<BinaryRow, Long> getPostponeRowCounts(Iterator<ManifestEntry> iterator) {
@@ -196,6 +347,98 @@ public class PostponeUtils {
Collections.singletonMap(BUCKET.key(), String.valueOf(BucketMode.POSTPONE_BUCKET)));
}
+ /** Snapshot-bound bucket-count assignment. */
+ public static final class PostponeBucketAssigner implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Map<BinaryRow, Integer> knownNumBuckets;
+ @Nullable private final Long targetRowNumPerBucket;
+ private final Map<BinaryRow, Long> postponeRowCounts;
+ private final int defaultBucketNum;
+
+ private PostponeBucketAssigner(
+ Map<BinaryRow, Integer> knownNumBuckets,
+ @Nullable Long targetRowNumPerBucket,
+ Map<BinaryRow, Long> postponeRowCounts,
+ int defaultBucketNum) {
+ this.knownNumBuckets = knownNumBuckets;
+ this.targetRowNumPerBucket = targetRowNumPerBucket;
+ this.postponeRowCounts = postponeRowCounts;
+ this.defaultBucketNum = defaultBucketNum;
+ }
+
+ public int assign(BinaryRow partition) {
+ return determineBucketNum(
+ partition,
+ knownNumBuckets,
+ targetRowNumPerBucket,
+ postponeRowCounts,
+ defaultBucketNum);
+ }
+ }
+
+ /** Snapshot-bound routing metadata for postpone records. */
+ public static final class PostponeBucketRouter implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final PostponeBucketAssigner bucketAssigner;
+ private final RowType keyType;
+ private final RowType bucketKeyType;
+ private final int[] bucketKeyMapping;
+ private final CoreOptions.BucketFunctionType bucketFunctionType;
+ @Nullable private transient Projection bucketKeyProjection;
+ @Nullable private transient BucketFunction bucketFunction;
+
+ private PostponeBucketRouter(
+ PostponeBucketAssigner bucketAssigner,
+ RowType keyType,
+ RowType bucketKeyType,
+ int[] bucketKeyMapping,
+ CoreOptions.BucketFunctionType bucketFunctionType) {
+ this.bucketAssigner = bucketAssigner;
+ this.keyType = keyType;
+ this.bucketKeyType = bucketKeyType;
+ this.bucketKeyMapping = bucketKeyMapping;
+ this.bucketFunctionType = bucketFunctionType;
+ }
+
+ public int bucket(BinaryRow partition, InternalRow trimmedPrimaryKey) {
+ if (bucketKeyProjection == null) {
+ bucketKeyProjection = CodeGenUtils.newProjection(keyType, bucketKeyMapping);
+ }
+ if (bucketFunction == null) {
+ bucketFunction = BucketFunction.create(bucketFunctionType, bucketKeyType);
+ }
+ return bucketFunction.bucket(
+ bucketKeyProjection.apply(trimmedPrimaryKey), numBuckets(partition));
+ }
+
+ public int numBuckets(BinaryRow partition) {
+ return bucketAssigner.assign(partition);
+ }
+ }
+
+ private static final class PostponeFile {
+
+ private final DataSplit split;
+ private final DataFileMeta file;
+ @Nullable private final DeletionFile deletionFile;
+ private final boolean deletionFilesPresent;
+
+ private PostponeFile(
+ DataSplit split,
+ DataFileMeta file,
+ @Nullable DeletionFile deletionFile,
+ boolean deletionFilesPresent) {
+ this.split = split;
+ this.file = file;
+ this.deletionFile = deletionFile;
+ this.deletionFilesPresent = deletionFilesPresent;
+ }
+ }
+
/** A real bucket which requires background compaction. */
public static final class CompactBucket implements Serializable {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergePlan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergePlan.java
new file mode 100644
index 0000000000..b2e6bc460e
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergePlan.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.source;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.table.PostponeUtils.PostponeBucketRouter;
+import org.apache.paimon.types.RowType;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** Plan for an execution-engine postpone merge read. */
+public final class PostponeMergePlan implements TableScan.Plan {
+
+ private final List<DataSplit> realSplits;
+ private final List<DataSplit> postponeSplits;
+ private final PostponeBucketRouter bucketRouter;
+ private final RowType keyType;
+ private final RowType resultReadType;
+ private final RowType mergeReadType;
+ private final long numPotentialBuckets;
+
+ PostponeMergePlan(
+ List<DataSplit> realSplits,
+ List<DataSplit> postponeSplits,
+ PostponeBucketRouter bucketRouter,
+ RowType keyType,
+ RowType resultReadType,
+ RowType mergeReadType) {
+ this.realSplits = realSplits;
+ this.postponeSplits = postponeSplits;
+ this.bucketRouter = bucketRouter;
+ this.keyType = keyType;
+ this.resultReadType = resultReadType;
+ this.mergeReadType = mergeReadType;
+ this.numPotentialBuckets = numPotentialBuckets(realSplits, postponeSplits, bucketRouter);
+ }
+
+ public List<DataSplit> realSplits() {
+ return realSplits;
+ }
+
+ public List<DataSplit> postponeSplits() {
+ return postponeSplits;
+ }
+
+ @Override
+ public List<Split> splits() {
+ List<Split> splits = new ArrayList<>(realSplits);
+ splits.addAll(postponeSplits);
+ return splits;
+ }
+
+ public PostponeBucketRouter bucketRouter() {
+ return bucketRouter;
+ }
+
+ public RowType keyType() {
+ return keyType;
+ }
+
+ public RowType resultReadType() {
+ return resultReadType;
+ }
+
+ public RowType mergeReadType() {
+ return mergeReadType;
+ }
+
+ /** Number of (partition, bucket) groups which this plan may merge. */
+ public long numPotentialBuckets() {
+ return numPotentialBuckets;
+ }
+
+ private static long numPotentialBuckets(
+ List<DataSplit> realSplits,
+ List<DataSplit> postponeSplits,
+ PostponeBucketRouter bucketRouter) {
+ Map<BinaryRow, Set<Integer>> realBuckets = new HashMap<>();
+ for (DataSplit split : realSplits) {
+ realBuckets
+ .computeIfAbsent(split.partition(), ignored -> new HashSet<>())
+ .add(split.bucket());
+ }
+
+ Set<BinaryRow> postponePartitions = new HashSet<>();
+ for (DataSplit split : postponeSplits) {
+ postponePartitions.add(split.partition());
+ }
+
+ long count = 0;
+ for (Map.Entry<BinaryRow, Set<Integer>> entry : realBuckets.entrySet()) {
+ BinaryRow partition = entry.getKey();
+ count +=
+ postponePartitions.remove(partition)
+ ? bucketRouter.numBuckets(partition)
+ : entry.getValue().size();
+ }
+ for (BinaryRow partition : postponePartitions) {
+ count += bucketRouter.numBuckets(partition);
+ }
+ return count;
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergeRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergeRead.java
new file mode 100644
index 0000000000..83ed07150f
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergeRead.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.source;
+
+import org.apache.paimon.KeyValue;
+import org.apache.paimon.KeyValueFileStore;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.operation.MergeFileSplitRead;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/** Reads execution-engine-routed inputs with Paimon's merge semantics. */
+public final class PostponeMergeRead {
+
+ private final FileStoreTable table;
+ @Nullable private final Predicate filter;
+ private final RowType resultReadType;
+ private final RowType mergeReadType;
+
+ @Nullable private IOManager ioManager;
+
+ PostponeMergeRead(
+ FileStoreTable table,
+ @Nullable Predicate filter,
+ RowType resultReadType,
+ RowType mergeReadType) {
+ this.table = table;
+ this.filter = filter;
+ this.resultReadType = resultReadType;
+ this.mergeReadType = mergeReadType;
+ }
+
+ public PostponeMergeRead withIOManager(IOManager ioManager) {
+ this.ioManager = ioManager;
+ return this;
+ }
+
+ /** Reads one writer's postpone files serially with key-only predicate pushdown. */
+ public RecordReader<KeyValue> createPostponeReader(DataSplit split) throws IOException {
+ return newMergeRead(mergeReadType).createPostponeReader(split);
+ }
+
+ /** Merges one real-bucket split and routed postpone records for the same target bucket. */
+ public RecordReader<InternalRow> createBucketMergeReader(
+ @Nullable DataSplit realBucketSplit, RecordReader<KeyValue> bucketPostponeRecords)
+ throws IOException {
+ RecordReader<KeyValue> reader =
+ newMergeRead(resultReadType)
+ .createPostponeMergeReader(realBucketSplit, bucketPostponeRecords);
+ return KeyValueTableRead.unwrap(reader, table.schema().options());
+ }
+
+ private MergeFileSplitRead newMergeRead(RowType readType) {
+ MergeFileSplitRead read =
+ ((KeyValueFileStore) table.store()).newRead().withReadType(readType);
+ if (filter != null) {
+ read.withFilter(filter);
+ }
+ if (ioManager != null) {
+ read.withIOManager(ioManager);
+ }
+ return read;
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergeReadBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergeReadBuilder.java
new file mode 100644
index 0000000000..d634df8bac
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PostponeMergeReadBuilder.java
@@ -0,0 +1,256 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.source;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.KeyValueFileStore;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.metrics.MetricRegistry;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.PostponeUtils;
+import org.apache.paimon.table.PrimaryKeyTableUtils;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
+import org.apache.paimon.tag.BatchReadTagCreator;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.partition.PartitionPredicate.splitPartitionPredicatesAndDataPredicates;
+import static org.apache.paimon.predicate.PredicateBuilder.and;
+import static org.apache.paimon.predicate.PredicateBuilder.excludePredicateWithFields;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Builds a postpone merge read for an execution engine. */
+public final class PostponeMergeReadBuilder implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final FileStoreTable table;
+ private final Snapshot snapshot;
+
+ @Nullable private Predicate filter;
+ @Nullable private PartitionPredicate partitionFilter;
+ @Nullable private RowType readType;
+ @Nullable private transient MetricRegistry metricRegistry;
+ @Nullable private transient String readProtectionTagName;
+ private int defaultBucketNum = 1;
+
+ private PostponeMergeReadBuilder(FileStoreTable table, Snapshot snapshot) {
+ this.table = table;
+ this.snapshot = snapshot;
+ }
+
+ /** Creates a snapshot-bound builder when the selected partitions contain postpone files. */
+ public static Optional<PostponeMergeReadBuilder> create(
+ FileStoreTable table, @Nullable PartitionPredicate partitionFilter) {
+ checkArgument(
+ table.bucketMode() == BucketMode.POSTPONE_MODE && !table.primaryKeys().isEmpty(),
+ "Postpone merge read requires a primary-key postpone bucket table.");
+
+ // Let the ordinary compacted-full scanner select its snapshot.
+ if (table.coreOptions().startupMode() == CoreOptions.StartupMode.COMPACTED_FULL) {
+ return Optional.empty();
+ }
+
+ Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest(table);
+ if (snapshot == null) {
+ return Optional.empty();
+ }
+
+ SnapshotReader postponeReader =
+ table.newSnapshotReader()
+ .withSnapshot(snapshot)
+ .withBucket(BucketMode.POSTPONE_BUCKET);
+ if (partitionFilter != null) {
+ postponeReader.withPartitionFilter(partitionFilter);
+ }
+ if (!postponeReader.readFileIterator().hasNext()) {
+ return Optional.empty();
+ }
+
+ validateReadMode(table);
+ PostponeMergeReadBuilder builder = new PostponeMergeReadBuilder(table, snapshot);
+ builder.withPartitionFilter(partitionFilter);
+ return Optional.of(builder);
+ }
+
+ private PostponeMergeReadBuilder withPartitionFilter(
+ @Nullable PartitionPredicate partitionPredicate) {
+ if (partitionPredicate == null) {
+ return this;
+ }
+ partitionFilter =
+ partitionFilter == null
+ ? partitionPredicate
+ : PartitionPredicate.and(
+ Arrays.asList(partitionFilter, partitionPredicate));
+ return this;
+ }
+
+ /** Applies safe pushdowns; the execution engine must evaluate the predicate after merging. */
+ public PostponeMergeReadBuilder withFilter(@Nullable Predicate predicate) {
+ if (predicate == null) {
+ return this;
+ }
+ filter = filter == null ? predicate : PredicateBuilder.and(filter, predicate);
+ splitPartitionPredicatesAndDataPredicates(predicate, table.rowType(), table.partitionKeys())
+ .getLeft()
+ .ifPresent(this::withPartitionFilter);
+ return this;
+ }
+
+ public PostponeMergeReadBuilder withReadType(RowType readType) {
+ this.readType = readType;
+ return this;
+ }
+
+ public PostponeMergeReadBuilder withMetricRegistry(MetricRegistry metricRegistry) {
+ this.metricRegistry = metricRegistry;
+ return this;
+ }
+
+ public PostponeMergeReadBuilder withDefaultBucketNum(int defaultBucketNum) {
+ checkArgument(defaultBucketNum > 0, "Default postpone bucket number must be positive.");
+ this.defaultBucketNum = defaultBucketNum;
+ return this;
+ }
+
+ public PostponeMergePlan plan() {
+ RowType resultReadType = resultReadType();
+ RowType mergeReadType = mergeReadType(resultReadType);
+
+ SnapshotReader realReader =
+ table.newSnapshotReader()
+ .withSnapshot(snapshot)
+ .onlyReadRealBuckets()
+ .withReadType(resultReadType);
+ if (metricRegistry != null) {
+ realReader.withMetricRegistry(metricRegistry);
+ }
+ if (filter != null) {
+ realReader.withFilter(filter, safeKeyPredicate(table.schema(), filter));
+ }
+ if (partitionFilter != null) {
+ realReader.withPartitionFilter(partitionFilter);
+ }
+
+ SnapshotReader postponeReader =
+ table.newSnapshotReader()
+ .withSnapshot(snapshot)
+ .withBucket(BucketMode.POSTPONE_BUCKET);
+ if (filter != null) {
+ // A normal bucket selector cannot prune bucket -2.
+ postponeReader.withFilter(filter, null);
+ }
+ if (partitionFilter != null) {
+ postponeReader.withPartitionFilter(partitionFilter);
+ }
+
+ PostponeMergePlan plan =
+ new PostponeMergePlan(
+ realReader.read().dataSplits(),
+ PostponeUtils.groupPostponeFiles(postponeReader.read().dataSplits()),
+ PostponeUtils.createPostponeBucketRouter(
+ table, snapshot.id(), defaultBucketNum, partitionFilter),
+ keyType(),
+ resultReadType,
+ mergeReadType);
+ maybeCreateReadProtectionTag(snapshot.id());
+ return plan;
+ }
+
+ @Nullable
+ public String readProtectionTagName() {
+ return readProtectionTagName;
+ }
+
+ public PostponeMergeRead newRead() {
+ RowType resultReadType = resultReadType();
+ return new PostponeMergeRead(table, filter, resultReadType, mergeReadType(resultReadType));
+ }
+
+ private RowType resultReadType() {
+ return readType == null ? table.rowType() : readType;
+ }
+
+ private RowType mergeReadType(RowType resultReadType) {
+ return ((KeyValueFileStore) table.store()).newRead().adjustReadType(resultReadType);
+ }
+
+ private RowType keyType() {
+ return new RowType(
+ PrimaryKeyTableUtils.PrimaryKeyFieldsExtractor.EXTRACTOR.keyFields(table.schema()));
+ }
+
+ private void maybeCreateReadProtectionTag(long snapshotId) {
+ if (table.coreOptions().scanPlanAutoTagTimeRetained() == null) {
+ return;
+ }
+ BatchReadTagCreator creator =
+ new BatchReadTagCreator(
+ table.tagManager(),
+ table.snapshotManager(),
+ table.coreOptions().scanPlanAutoTagTimeRetained());
+ readProtectionTagName = creator.createReadTag(snapshotId);
+ }
+
+ @Nullable
+ private static Predicate safeKeyPredicate(TableSchema schema, Predicate predicate) {
+ Pair<Optional<PartitionPredicate>, List<Predicate>> split =
+ splitPartitionPredicatesAndDataPredicates(
+ predicate, schema.logicalRowType(), schema.partitionKeys());
+ List<String> primaryKeys = schema.trimmedPrimaryKeys();
+ Set<String> nonPrimaryKeys =
+ schema.fieldNames().stream()
+ .filter(name -> !primaryKeys.contains(name))
+ .collect(Collectors.toSet());
+ List<Predicate> keyFilters = excludePredicateWithFields(split.getRight(), nonPrimaryKeys);
+ return keyFilters.isEmpty() ? null : and(keyFilters);
+ }
+
+ private static void validateReadMode(FileStoreTable table) {
+ if (table.coreOptions().queryAuthEnabled()) {
+ throw new UnsupportedOperationException(
+ "Postpone merge-on-read does not support query authorization.");
+ }
+ CoreOptions.StartupMode startupMode = table.coreOptions().startupMode();
+ if (startupMode == CoreOptions.StartupMode.INCREMENTAL
+ || startupMode == CoreOptions.StartupMode.FROM_FILE_CREATION_TIME
+ || startupMode == CoreOptions.StartupMode.FROM_CREATION_TIMESTAMP) {
+ throw new UnsupportedOperationException(
+ "Postpone merge-on-read requires a full snapshot scan, but found scan mode '"
+ + startupMode
+ + "'.");
+ }
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/SortBufferWriteBufferTestBase.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/SortBufferWriteBufferTestBase.java
index ff8249c221..06fd4abf80 100644
--- a/paimon-core/src/test/java/org/apache/paimon/mergetree/SortBufferWriteBufferTestBase.java
+++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/SortBufferWriteBufferTestBase.java
@@ -33,6 +33,7 @@ import org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction;
import org.apache.paimon.mergetree.compact.aggregate.AggregateMergeFunction;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.RecordReaderIterator;
import org.apache.paimon.sort.BinaryInMemorySortBuffer;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
@@ -128,6 +129,25 @@ public abstract class SortBufferWriteBufferTestBase {
assertThat(expected).isEmpty();
}
+ @Test
+ public void testCreateReader() throws Exception {
+ List<ReusingTestData> input = ReusingTestData.generateData(100, addOnly());
+ Queue<ReusingTestData> expected = new LinkedList<>(getExpected(input));
+ prepareTable(input);
+
+ try (RecordReaderIterator<KeyValue> reader =
+ new RecordReaderIterator<>(
+ table.createReader(KEY_COMPARATOR, createMergeFunction()))) {
+ while (reader.hasNext()) {
+ assertThat(expected.isEmpty()).isFalse();
+ expected.poll().assertEquals(reader.next());
+ }
+ }
+
+ assertThat(expected).isEmpty();
+ assertThat(table.isEmpty()).isTrue();
+ }
+
private void prepareTable(List<ReusingTestData> input) throws IOException {
ReusingKeyValue reuse = new ReusingKeyValue();
for (ReusingTestData data : input) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/MergeFileSplitReadTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/MergeFileSplitReadTest.java
index db65eb8d16..d573b5bd90 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/MergeFileSplitReadTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/MergeFileSplitReadTest.java
@@ -22,11 +22,14 @@ import org.apache.paimon.KeyValue;
import org.apache.paimon.TestFileStore;
import org.apache.paimon.TestKeyValueGenerator;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIOFinder;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.mergetree.compact.DeduplicateMergeFunction;
import org.apache.paimon.mergetree.compact.MergeFunction;
@@ -37,6 +40,8 @@ import org.apache.paimon.schema.KeyValueFieldsExtractor;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.table.PrimaryKeyTableUtils;
import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.types.BigIntType;
@@ -46,6 +51,7 @@ import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.IteratorRecordReader;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -219,6 +225,172 @@ public class MergeFileSplitReadTest {
}
}
+ @Test
+ public void testPostponeMergeReaderWithMultipleRealFilesAndValueProjection() throws Exception {
+ RowType keyType =
+ RowType.of(
+ new DataField(
+ SpecialFields.KEY_FIELD_ID_START,
+ SpecialFields.KEY_FIELD_PREFIX + "k",
+ new IntType(false)));
+ RowType valueType =
+ RowType.of(
+ new DataField(0, "k", new IntType(false)),
+ new DataField(1, "v", new VarCharType()));
+ TestFileStore store =
+ createStore(
+ RowType.of(),
+ keyType,
+ valueType,
+ PrimaryKeyTableUtils.PrimaryKeyFieldsExtractor.EXTRACTOR,
+ DeduplicateMergeFunction.factory());
+
+ List<KeyValue> realRecords = new ArrayList<>();
+ realRecords.add(keyValue(1, 0, RowKind.INSERT, "base-1", false));
+ realRecords.add(keyValue(2, 1, RowKind.INSERT, "base-2", false));
+ String largeValue = String.join("", Collections.nCopies(256, "x"));
+ for (int key = 10; key < 50; key++) {
+ realRecords.add(keyValue(key, key, RowKind.INSERT, largeValue + "-" + key, false));
+ }
+ store.commitData(realRecords, ignored -> BinaryRow.EMPTY_ROW, ignored -> 0);
+
+ long snapshotId = store.snapshotManager().latestSnapshotId();
+ List<DataFileMeta> realFiles =
+ store.newScan().withSnapshot(snapshotId).plan().files().stream()
+ .map(ManifestEntry::file)
+ .collect(Collectors.toList());
+ assertThat(realFiles).hasSizeGreaterThan(1);
+ DataSplit realSplit =
+ DataSplit.builder()
+ .withSnapshot(snapshotId)
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withDataFiles(realFiles)
+ .withBucketPath("not used")
+ .build();
+ List<KeyValue> postponeRecords =
+ Arrays.asList(
+ // Incoming sequence numbers are intentionally ignored.
+ keyValue(1, 200, RowKind.INSERT, "new-1", true),
+ keyValue(1, KeyValue.UNKNOWN_SEQUENCE, RowKind.INSERT, "newest-1", true),
+ keyValue(2, 300, RowKind.DELETE, null, true),
+ keyValue(3, KeyValue.UNKNOWN_SEQUENCE, RowKind.INSERT, "new-3", true));
+ Map<Integer, String> baseExpected =
+ realRecords.stream()
+ .collect(
+ Collectors.toMap(
+ record -> record.key().getInt(0),
+ record -> record.value().getString(1).toString()));
+
+ RowType projectedValueType = valueType.project("v");
+ try (RecordReaderIterator<KeyValue> reader =
+ new RecordReaderIterator<>(
+ store.newRead()
+ .withReadType(projectedValueType)
+ .createPostponeMergeReader(
+ realSplit,
+ new IteratorRecordReader<>(
+ Collections.<KeyValue>emptyList().iterator())))) {
+ Map<Integer, String> actual = new HashMap<>();
+ while (reader.hasNext()) {
+ KeyValue keyValue = reader.next();
+ actual.put(keyValue.key().getInt(0), keyValue.value().getString(0).toString());
+ }
+ assertThat(actual).containsExactlyInAnyOrderEntriesOf(baseExpected);
+ }
+
+ try (IOManager ioManager = IOManager.create(tempDir.resolve("io").toString());
+ RecordReaderIterator<KeyValue> reader =
+ new RecordReaderIterator<>(
+ store.newRead()
+ .withReadType(projectedValueType)
+ .withIOManager(ioManager)
+ .createPostponeMergeReader(
+ realSplit,
+ new IteratorRecordReader<>(
+ postponeRecords.iterator())))) {
+ Map<Integer, String> actual = new HashMap<>();
+ while (reader.hasNext()) {
+ KeyValue keyValue = reader.next();
+ assertThat(keyValue.value().getFieldCount()).isEqualTo(1);
+ actual.put(keyValue.key().getInt(0), keyValue.value().getString(0).toString());
+ }
+ Map<Integer, String> expected = new HashMap<>(baseExpected);
+ expected.put(1, "newest-1");
+ expected.remove(2);
+ expected.put(3, "new-3");
+ assertThat(actual).containsExactlyInAnyOrderEntriesOf(expected);
+ }
+ }
+
+ @Test
+ public void testPostponeReader() throws Exception {
+ RowType keyType =
+ RowType.of(
+ new DataField(
+ SpecialFields.KEY_FIELD_ID_START,
+ SpecialFields.KEY_FIELD_PREFIX + "k",
+ new IntType(false)));
+ RowType valueType =
+ RowType.of(
+ new DataField(0, "k", new IntType(false)),
+ new DataField(1, "v", new VarCharType()));
+ TestFileStore store =
+ createStore(
+ BucketMode.POSTPONE_BUCKET,
+ RowType.of(),
+ keyType,
+ valueType,
+ PrimaryKeyTableUtils.PrimaryKeyFieldsExtractor.EXTRACTOR,
+ DeduplicateMergeFunction.factory());
+
+ store.commitData(
+ Arrays.asList(
+ keyValue(1, 100, RowKind.INSERT, "one", false),
+ keyValue(2, 200, RowKind.INSERT, "two", false)),
+ ignored -> BinaryRow.EMPTY_ROW,
+ ignored -> BucketMode.POSTPONE_BUCKET);
+
+ long snapshotId = store.snapshotManager().latestSnapshotId();
+ List<ManifestEntry> files =
+ store.newScan()
+ .withSnapshot(snapshotId)
+ .withBucket(BucketMode.POSTPONE_BUCKET)
+ .plan()
+ .files();
+ assertThat(files).hasSize(1);
+ DataSplit split =
+ DataSplit.builder()
+ .withSnapshot(snapshotId)
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(BucketMode.POSTPONE_BUCKET)
+ .withDataFiles(Collections.singletonList(files.get(0).file()))
+ .withBucketPath("not used")
+ .build();
+
+ Map<Integer, Long> sequences = new HashMap<>();
+ try (RecordReaderIterator<KeyValue> reader =
+ new RecordReaderIterator<>(store.newRead().createPostponeReader(split))) {
+ while (reader.hasNext()) {
+ KeyValue record = reader.next();
+ sequences.put(record.key().getInt(0), record.sequenceNumber());
+ }
+ }
+ Map<Integer, Long> expected = new HashMap<>();
+ expected.put(1, 100L);
+ expected.put(2, 200L);
+ assertThat(sequences).containsExactlyInAnyOrderEntriesOf(expected);
+ }
+
+ private static KeyValue keyValue(
+ int key, long sequenceNumber, RowKind kind, String value, boolean projected) {
+ InternalRow row =
+ projected
+ ? GenericRow.of(value == null ? null : BinaryString.fromString(value))
+ : GenericRow.of(key, value == null ? null : BinaryString.fromString(value));
+ return new KeyValue().replace(GenericRow.of(key), sequenceNumber, kind, row);
+ }
+
private List<KeyValue> writeThenRead(
List<KeyValue> data,
RowType readKeyType,
@@ -274,6 +446,17 @@ public class MergeFileSplitReadTest {
KeyValueFieldsExtractor extractor,
MergeFunctionFactory<KeyValue> mfFactory)
throws Exception {
+ return createStore(1, partitionType, keyType, valueType, extractor, mfFactory);
+ }
+
+ private TestFileStore createStore(
+ int numBuckets,
+ RowType partitionType,
+ RowType keyType,
+ RowType valueType,
+ KeyValueFieldsExtractor extractor,
+ MergeFunctionFactory<KeyValue> mfFactory)
+ throws Exception {
Path path = new Path(tempDir.toUri());
SchemaManager schemaManager = new SchemaManager(FileIOFinder.find(path), path);
boolean valueCountMode = mfFactory.create() instanceof TestValueCountMergeFunction;
@@ -299,7 +482,7 @@ public class MergeFileSplitReadTest {
return new TestFileStore.Builder(
"avro",
tempDir.toString(),
- 1,
+ numBuckets,
partitionType,
keyType,
valueType,
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java
index 251f06c5f6..40b36bfb96 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java
@@ -25,6 +25,9 @@ import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.SimpleFileEntry;
import org.apache.paimon.operation.FileStoreScan;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
import org.junit.jupiter.api.Test;
@@ -50,6 +53,7 @@ public class PostponeUtilsTest {
@Test
public void testGetKnownNumBucketsFromSnapshot() {
BinaryRow partition = partition(1);
+ PartitionPredicate partitionFilter = mock(PartitionPredicate.class);
SimpleFileEntry entry = mock(SimpleFileEntry.class);
when(entry.partition()).thenReturn(partition);
when(entry.totalBuckets()).thenReturn(4);
@@ -61,14 +65,17 @@ public class PostponeUtilsTest {
FileStoreTable table = mock(FileStoreTable.class);
when(table.store()).thenReturn(store);
- assertThat(PostponeUtils.getKnownNumBuckets(table, 5L)).containsEntry(partition, 4);
+ assertThat(PostponeUtils.getKnownNumBuckets(table, 5L, partitionFilter))
+ .containsEntry(partition, 4);
verify(scan).withSnapshot(5L);
verify(scan).onlyReadRealBuckets();
+ verify(scan).withPartitionFilter(partitionFilter);
}
@Test
public void testGetPostponeRowCountsFromSnapshot() {
BinaryRow partition = partition(1);
+ PartitionPredicate partitionFilter = mock(PartitionPredicate.class);
DataFileMeta file = mock(DataFileMeta.class);
when(file.rowCount()).thenReturn(10L);
ManifestEntry entry = mock(ManifestEntry.class);
@@ -80,9 +87,11 @@ public class PostponeUtilsTest {
FileStoreTable table = mock(FileStoreTable.class);
when(table.newSnapshotReader()).thenReturn(reader);
- assertThat(PostponeUtils.getPostponeRowCounts(table, 5L)).containsEntry(partition, 10L);
+ assertThat(PostponeUtils.getPostponeRowCounts(table, 5L, partitionFilter))
+ .containsEntry(partition, 10L);
verify(reader).withSnapshot(5L);
verify(reader).withBucket(BucketMode.POSTPONE_BUCKET);
+ verify(reader).withPartitionFilter(partitionFilter);
}
@Test
@@ -111,6 +120,80 @@ public class PostponeUtilsTest {
verify(scan).onlyReadRealBuckets();
}
+ @Test
+ public void testGroupPostponeFilesByPartitionAndWriter() {
+ BinaryRow partition1 = partition(1);
+ BinaryRow partition2 = partition(2);
+ DataFileMeta writer1Newest = dataFile("data-u-c-s-1-w-newest", 20L);
+ DataFileMeta writer2 = dataFile("data-u-c-s-2-w-only", 5L);
+ DataFileMeta sameWriteIdFromAnotherCommit = dataFile("data-u-other-s-1-w-only", 15L);
+ DataFileMeta writer1First = dataFile("data-u-c-s-1-w-first", 10L);
+ DataFileMeta otherPartition = dataFile("data-u-c-s-1-w-other-partition", 1L);
+ List<DataSplit> splits =
+ Arrays.asList(
+ dataSplit(partition1, writer1Newest, writer2, sameWriteIdFromAnotherCommit),
+ dataSplit(partition1, writer1First),
+ dataSplit(partition2, otherPartition));
+
+ List<DataSplit> grouped = PostponeUtils.groupPostponeFiles(splits);
+
+ assertThat(grouped).hasSize(4);
+ assertThat(grouped.get(0).partition()).isEqualTo(partition1);
+ assertThat(grouped.get(0).dataFiles())
+ .extracting(DataFileMeta::fileName)
+ .containsExactly("data-u-c-s-1-w-first", "data-u-c-s-1-w-newest");
+ assertThat(grouped.get(1).partition()).isEqualTo(partition1);
+ assertThat(grouped.get(1).dataFiles())
+ .extracting(DataFileMeta::fileName)
+ .containsExactly("data-u-c-s-2-w-only");
+ assertThat(grouped.get(2).partition()).isEqualTo(partition1);
+ assertThat(grouped.get(2).dataFiles())
+ .extracting(DataFileMeta::fileName)
+ .containsExactly("data-u-other-s-1-w-only");
+ assertThat(grouped.get(3).partition()).isEqualTo(partition2);
+ assertThat(grouped.get(3).dataFiles())
+ .extracting(DataFileMeta::fileName)
+ .containsExactly("data-u-c-s-1-w-other-partition");
+ }
+
+ @Test
+ public void testGroupPostponeFilesKeepsScanOrderForEqualCreationTime() {
+ BinaryRow partition = partition(1);
+ DataFileMeta first = dataFile("data-u-c-s-1-w-z", 10L);
+ DataFileMeta second = dataFile("data-u-c-s-1-w-a", 10L);
+
+ List<DataSplit> grouped =
+ PostponeUtils.groupPostponeFiles(
+ Arrays.asList(dataSplit(partition, first), dataSplit(partition, second)));
+
+ assertThat(grouped).hasSize(1);
+ assertThat(grouped.get(0).dataFiles())
+ .extracting(DataFileMeta::fileName)
+ .containsExactly("data-u-c-s-1-w-z", "data-u-c-s-1-w-a");
+ }
+
+ @Test
+ public void testGroupPostponeFilesKeepsDeletionFilesAligned() {
+ BinaryRow partition = partition(1);
+ DataFileMeta newest = dataFile("data-u-c-s-1-w-newest", 20L);
+ DataFileMeta first = dataFile("data-u-c-s-1-w-first", 10L);
+ DeletionFile newestDeletion = mock(DeletionFile.class);
+ DeletionFile firstDeletion = mock(DeletionFile.class);
+
+ List<DataSplit> grouped =
+ PostponeUtils.groupPostponeFiles(
+ Arrays.asList(
+ dataSplit(partition, newest, newestDeletion),
+ dataSplit(partition, first, firstDeletion)));
+
+ assertThat(grouped).hasSize(1);
+ assertThat(grouped.get(0).dataFiles())
+ .extracting(DataFileMeta::fileName)
+ .containsExactly("data-u-c-s-1-w-first", "data-u-c-s-1-w-newest");
+ assertThat(grouped.get(0).deletionFiles().orElseThrow(AssertionError::new))
+ .containsExactly(firstDeletion, newestDeletion);
+ }
+
@Test
public void testTableForPostponeCompact() {
FileStoreTable table = mock(FileStoreTable.class);
@@ -202,4 +285,33 @@ public class PostponeUtilsTest {
when(entry.level()).thenReturn(level);
return entry;
}
+
+ private static DataFileMeta dataFile(String name, long creationTime) {
+ DataFileMeta file = mock(DataFileMeta.class);
+ when(file.fileName()).thenReturn(name);
+ when(file.creationTimeEpochMillis()).thenReturn(creationTime);
+ return file;
+ }
+
+ private static DataSplit dataSplit(BinaryRow partition, DataFileMeta... files) {
+ return DataSplit.builder()
+ .withPartition(partition)
+ .withBucket(BucketMode.POSTPONE_BUCKET)
+ .withBucketPath("postpone")
+ .withTotalBuckets(BucketMode.POSTPONE_BUCKET)
+ .withDataFiles(Arrays.asList(files))
+ .build();
+ }
+
+ private static DataSplit dataSplit(
+ BinaryRow partition, DataFileMeta file, DeletionFile deletionFile) {
+ return DataSplit.builder()
+ .withPartition(partition)
+ .withBucket(BucketMode.POSTPONE_BUCKET)
+ .withBucketPath("postpone")
+ .withTotalBuckets(BucketMode.POSTPONE_BUCKET)
+ .withDataFiles(Collections.singletonList(file))
+ .withDataDeletionFiles(Collections.singletonList(deletionFile))
+ .build();
+ }
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java
index 75863dd642..972713a5d0 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java
@@ -21,6 +21,7 @@ package org.apache.paimon.table.source;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.disk.IOManager;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.options.Options;
@@ -28,12 +29,14 @@ import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.predicate.TopN;
+import org.apache.paimon.reader.RecordReaderIterator;
import org.apache.paimon.stats.SimpleStats;
import org.apache.paimon.stats.SimpleStatsEvolutions;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.StreamTableCommit;
import org.apache.paimon.table.sink.StreamTableWrite;
import org.apache.paimon.table.source.snapshot.ScannerTestBase;
+import org.apache.paimon.tag.BatchReadTagCreator;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
@@ -41,9 +44,12 @@ import org.apache.paimon.utils.SnapshotManager;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import static org.apache.paimon.data.BinaryArray.fromLongArray;
import static org.apache.paimon.predicate.SortValue.NullOrdering.NULLS_FIRST;
@@ -378,6 +384,195 @@ public class TableScanTest extends ScannerTestBase {
commit.close();
}
+ @Test
+ public void testPostponeMergeReadBuilderAndPushDown() throws Exception {
+ StreamTableWrite write = table.newWrite(commitUser);
+ StreamTableCommit commit = table.newCommit(commitUser);
+ write.write(rowData(1, 10, 100L));
+ write.write(rowData(2, 20, 200L));
+ write.write(rowData(3, 30, 300L));
+ commit.commit(0, write.prepareCommit(true, 0));
+ write.close();
+ commit.close();
+
+ Map<String, String> dynamicOptions = new HashMap<>();
+ dynamicOptions.put(CoreOptions.BUCKET.key(), "-2");
+ dynamicOptions.put(CoreOptions.POSTPONE_MERGE_ON_READ.key(), "true");
+ FileStoreTable postponeTable = table.copy(dynamicOptions);
+ PredicateBuilder builder = new PredicateBuilder(postponeTable.rowType());
+
+ // The option must not change an ordinary Core scan.
+ assertThat(postponeTable.newScan().plan().splits()).hasSize(3);
+ assertThat(postponeTable.newScan().withLimit(1).plan().splits()).hasSize(1);
+ assertThat(PostponeMergeReadBuilder.create(postponeTable, null)).isEmpty();
+
+ StreamTableWrite postponeWrite = postponeTable.newWrite(commitUser);
+ StreamTableCommit postponeCommit = postponeTable.newCommit(commitUser);
+ postponeWrite.write(rowData(1, 10, 101L));
+ postponeCommit.commit(1, postponeWrite.prepareCommit(true, 1));
+ postponeWrite.close();
+ postponeCommit.close();
+
+ FileStoreTable compactedFullTable =
+ postponeTable.copy(
+ Collections.singletonMap(CoreOptions.SCAN_MODE.key(), "compacted-full"));
+ assertThat(PostponeMergeReadBuilder.create(compactedFullTable, null)).isEmpty();
+
+ // A postpone record may update any value, so real-file value statistics are unsafe.
+ Predicate valueFilter = builder.equal(2, 100L);
+ PostponeMergePlan mergePlan =
+ PostponeMergeReadBuilder.create(postponeTable, null)
+ .get()
+ .withFilter(valueFilter)
+ .withReadType(postponeTable.rowType().project("b"))
+ .withDefaultBucketNum(1)
+ .plan();
+ assertThat(mergePlan.realSplits()).hasSize(3);
+ assertThat(mergePlan.postponeSplits()).hasSize(1);
+ assertThat(mergePlan.resultReadType().getFieldNames()).containsExactly("b");
+
+ // Partition and trimmed-primary-key filters remain safe.
+ Predicate partitionFilter = builder.equal(0, 1);
+ assertThat(
+ PostponeMergeReadBuilder.create(postponeTable, null)
+ .get()
+ .withFilter(partitionFilter)
+ .plan()
+ .realSplits())
+ .hasSize(1);
+ Predicate primaryKeyFilter = builder.equal(1, 10);
+ assertThat(
+ PostponeMergeReadBuilder.create(postponeTable, null)
+ .get()
+ .withFilter(primaryKeyFilter)
+ .plan()
+ .realSplits())
+ .hasSize(1);
+
+ // A builder remains bound to the snapshot selected during its probe.
+ PostponeMergeReadBuilder pinnedBuilder =
+ PostponeMergeReadBuilder.create(postponeTable, null).get();
+ long pinnedSnapshotId = postponeTable.snapshotManager().latestSnapshotId();
+ postponeWrite = postponeTable.newWrite(commitUser);
+ postponeCommit = postponeTable.newCommit(commitUser);
+ postponeWrite.write(rowData(2, 20, 201L));
+ postponeCommit.commit(2, postponeWrite.prepareCommit(true, 2));
+ postponeWrite.close();
+ postponeCommit.close();
+ PostponeMergePlan pinnedPlan = pinnedBuilder.plan();
+ assertThat(pinnedPlan.postponeSplits()).hasSize(1);
+ assertThat(pinnedPlan.realSplits())
+ .allSatisfy(split -> assertThat(split.snapshotId()).isEqualTo(pinnedSnapshotId));
+ assertThat(pinnedPlan.postponeSplits())
+ .allSatisfy(split -> assertThat(split.snapshotId()).isEqualTo(pinnedSnapshotId));
+
+ FileStoreTable protectedTable =
+ postponeTable.copy(
+ Collections.singletonMap(
+ CoreOptions.SCAN_PLAN_AUTO_TAG_FOR_READ_TIME_RETAINED.key(),
+ "1 h"));
+ PostponeMergeReadBuilder protectedBuilder =
+ PostponeMergeReadBuilder.create(protectedTable, null).get();
+ protectedBuilder.plan();
+ String readProtectionTag = protectedBuilder.readProtectionTagName();
+ assertThat(readProtectionTag).isNotNull();
+ assertThat(protectedTable.tagManager().tagExists(readProtectionTag)).isTrue();
+ new BatchReadTagCreator(
+ protectedTable.tagManager(),
+ protectedTable.snapshotManager(),
+ protectedTable.coreOptions().scanPlanAutoTagTimeRetained())
+ .deleteReadTag(readProtectionTag);
+ }
+
+ @Test
+ public void testPostponeMergePlanAndRead() throws Exception {
+ StreamTableWrite realWrite = table.newWrite(commitUser);
+ StreamTableCommit realCommit = table.newCommit(commitUser);
+ realWrite.write(rowData(1, 10, 100L));
+ realWrite.write(rowData(1, 20, 200L));
+ realCommit.commit(0, realWrite.prepareCommit(true, 0));
+ realWrite.close();
+ realCommit.close();
+
+ Map<String, String> dynamicOptions = new HashMap<>();
+ dynamicOptions.put(CoreOptions.BUCKET.key(), "-2");
+ FileStoreTable postponeTable = table.copy(dynamicOptions);
+ StreamTableWrite postponeWrite = postponeTable.newWrite(commitUser);
+ StreamTableCommit postponeCommit = postponeTable.newCommit(commitUser);
+ postponeWrite.write(rowData(1, 10, 101L));
+ postponeWrite.write(rowData(1, 30, 300L));
+ postponeCommit.commit(1, postponeWrite.prepareCommit(true, 1));
+ postponeWrite.write(rowData(1, 10, 102L));
+ postponeCommit.commit(2, postponeWrite.prepareCommit(true, 2));
+ postponeWrite.close();
+ postponeCommit.close();
+
+ Predicate partitionFilter = new PredicateBuilder(postponeTable.rowType()).equal(0, 1);
+ PostponeMergeReadBuilder readBuilder =
+ PostponeMergeReadBuilder.create(postponeTable, null)
+ .get()
+ .withFilter(partitionFilter)
+ .withReadType(postponeTable.rowType().project("b"))
+ .withDefaultBucketNum(1);
+ PostponeMergePlan plan = readBuilder.plan();
+
+ assertThat(plan.realSplits()).hasSize(1);
+ assertThat(plan.postponeSplits()).hasSize(1);
+ assertThat(plan.postponeSplits().get(0).dataFiles()).hasSize(2);
+ assertThat(plan.mergeReadType().getFieldNames()).containsExactly("b");
+
+ List<Long> values = new ArrayList<>();
+ try (IOManager ioManager = IOManager.create(tempDir.resolve("postpone-merge").toString());
+ RecordReaderIterator<org.apache.paimon.data.InternalRow> rows =
+ new RecordReaderIterator<>(
+ readBuilder
+ .newRead()
+ .withIOManager(ioManager)
+ .createBucketMergeReader(
+ plan.realSplits().get(0),
+ readBuilder
+ .newRead()
+ .withIOManager(ioManager)
+ .createPostponeReader(
+ plan.postponeSplits().get(0))))) {
+ while (rows.hasNext()) {
+ values.add(rows.next().getLong(0));
+ }
+ }
+ assertThat(values).containsExactlyInAnyOrder(102L, 200L, 300L);
+ }
+
+ @Test
+ public void testPostponeMergePlanPotentialBuckets() throws Exception {
+ StreamTableWrite realWrite = table.newWrite(commitUser);
+ StreamTableCommit realCommit = table.newCommit(commitUser);
+ realWrite.write(rowData(1, 10, 100L));
+ realWrite.write(rowData(2, 20, 200L));
+ realCommit.commit(0, realWrite.prepareCommit(true, 0));
+ realWrite.close();
+ realCommit.close();
+
+ FileStoreTable postponeTable =
+ table.copy(Collections.singletonMap(CoreOptions.BUCKET.key(), "-2"));
+ StreamTableWrite postponeWrite = postponeTable.newWrite(commitUser);
+ StreamTableCommit postponeCommit = postponeTable.newCommit(commitUser);
+ postponeWrite.write(rowData(1, 10, 101L));
+ postponeWrite.write(rowData(3, 30, 300L));
+ postponeCommit.commit(1, postponeWrite.prepareCommit(true, 1));
+ postponeWrite.close();
+ postponeCommit.close();
+
+ PostponeMergePlan plan =
+ PostponeMergeReadBuilder.create(postponeTable, null)
+ .get()
+ .withDefaultBucketNum(4)
+ .plan();
+
+ // Partition 1 uses its known bucket, partition 2 is real-only, and the new partition 3
+ // may route to any of the four default buckets.
+ assertThat(plan.numPotentialBuckets()).isEqualTo(6);
+ }
+
@Test
public void testPushDownTopN() throws Exception {
createAppendOnlyTable();
diff --git a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/paimon/shims/MinorVersionShim.scala b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/paimon/shims/MinorVersionShim.scala
index aed66477bc..ffcdee388c 100644
--- a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/paimon/shims/MinorVersionShim.scala
+++ b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/paimon/shims/MinorVersionShim.scala
@@ -18,9 +18,11 @@
package org.apache.spark.sql.paimon.shims
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression}
import org.apache.spark.sql.catalyst.plans.logical.{CTERelationRef, LogicalPlan, MergeAction, MergeIntoTable}
import org.apache.spark.sql.catalyst.plans.physical.{ClusteredDistribution, Distribution}
+import org.apache.spark.sql.connector.read.Scan
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
object MinorVersionShim {
@@ -45,4 +47,17 @@ object MinorVersionShim {
// Spark 3.2 has no `notMatchedBySourceActions` field on `MergeIntoTable` (added in 3.4).
def notMatchedBySourceActions(merge: MergeIntoTable): Seq[MergeAction] = Seq.empty
+
+ def createDataSourceV2ScanRelation(
+ relation: DataSourceV2ScanRelation,
+ scan: Scan,
+ output: Seq[AttributeReference]): DataSourceV2ScanRelation = {
+ DataSourceV2ScanRelation(relation.relation, scan, output)
+ }
+
+ def createClusteredDistribution(
+ expressions: Seq[Expression],
+ requiredNumPartitions: Option[Int]): Distribution = {
+ ClusteredDistribution(expressions, requiredNumPartitions)
+ }
}
diff --git a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/spark/sql/paimon/shims/MinorVersionShim.scala b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/spark/sql/paimon/shims/MinorVersionShim.scala
index 41e5ed4c12..b66081d065 100644
--- a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/spark/sql/paimon/shims/MinorVersionShim.scala
+++ b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/spark/sql/paimon/shims/MinorVersionShim.scala
@@ -18,9 +18,11 @@
package org.apache.spark.sql.paimon.shims
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression}
import org.apache.spark.sql.catalyst.plans.logical.{CTERelationRef, LogicalPlan, MergeAction, MergeIntoTable}
import org.apache.spark.sql.catalyst.plans.physical.{ClusteredDistribution, Distribution}
+import org.apache.spark.sql.connector.read.Scan
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
object MinorVersionShim {
@@ -48,4 +50,17 @@ object MinorVersionShim {
// Spark 3.3 has no `notMatchedBySourceActions` field on `MergeIntoTable` (added in 3.4).
def notMatchedBySourceActions(merge: MergeIntoTable): Seq[MergeAction] = Seq.empty
+
+ def createDataSourceV2ScanRelation(
+ relation: DataSourceV2ScanRelation,
+ scan: Scan,
+ output: Seq[AttributeReference]): DataSourceV2ScanRelation = {
+ DataSourceV2ScanRelation(relation.relation, scan, output, None)
+ }
+
+ def createClusteredDistribution(
+ expressions: Seq[Expression],
+ requiredNumPartitions: Option[Int]): Distribution = {
+ ClusteredDistribution(expressions, requiredNumPartitions = requiredNumPartitions)
+ }
}
diff --git a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
index 854dda9fa2..f9e7ba1c0b 100644
--- a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
+++ b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
@@ -44,11 +44,12 @@ import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.util.{ArrayData, GeneratedColumn, IdentityColumn, ResolveDefaultColumns}
import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column, Identifier, StagingTableCatalog, Table, TableCatalog}
import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.connector.read.Scan
import org.apache.spark.sql.connector.write.BatchWrite
import org.apache.spark.sql.execution.{SparkFormatTable, SparkPlan}
import org.apache.spark.sql.execution.datasources.{PartitioningAwareFileIndex, PartitionSpec}
import org.apache.spark.sql.execution.datasources.v2.{AtomicReplaceTableAsSelectExec, AtomicReplaceTableExec, ReplaceTableAsSelectExec, ReplaceTableExec}
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
import org.apache.spark.sql.execution.streaming.{FileStreamSink, MetadataLogFileIndex}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DataTypes, StructType, VariantType}
@@ -301,6 +302,19 @@ class Spark4Shim extends SparkShim {
relation.copy(table = table, output = output)
}
+ override def createDataSourceV2ScanRelation(
+ relation: DataSourceV2ScanRelation,
+ scan: Scan,
+ output: Seq[AttributeReference]): DataSourceV2ScanRelation = {
+ DataSourceV2ScanRelation(relation.relation, scan, output, None, None)
+ }
+
+ override def createClusteredDistribution(
+ expressions: Seq[Expression],
+ requiredNumPartitions: Option[Int]): Distribution = {
+ ClusteredDistribution(expressions, requiredNumPartitions = requiredNumPartitions)
+ }
+
// Spark 4.0 still has `SubstituteUnresolvedOrdinals` (Spark 4.1 removed it because the new
// resolver framework handles ordinals inline). `PaimonViewResolver` applies the shim's early
// rules to the parsed view text before storing, so we must substitute `ORDER BY 1` →
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
index df600a6a75..c9d19a73ec 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
@@ -22,8 +22,9 @@ import org.apache.paimon.CoreOptions
import org.apache.paimon.globalindex.GlobalIndexResult
import org.apache.paimon.partition.PartitionPredicate
import org.apache.paimon.predicate.{Predicate, PredicateBuilder, VectorSearch}
+import org.apache.paimon.spark.PostponeMergeOnRead.MergePlan
import org.apache.paimon.spark.metric.SparkMetricRegistry
-import org.apache.paimon.spark.read.{BaseScan, BatchReadTagCleanupListener, PaimonSupportsRuntimeFiltering, SparkHybridSearchBuilderImpl, SparkVectorSearchBuilderImpl}
+import org.apache.paimon.spark.read.{BaseScan, BatchReadTagCleanupListener, PaimonStatistics, PaimonSupportsRuntimeFiltering, SparkHybridSearchBuilderImpl, SparkVectorSearchBuilderImpl}
import org.apache.paimon.spark.sources.PaimonMicroBatchStream
import org.apache.paimon.spark.util.OptionUtils
import org.apache.paimon.table.{DataTable, FileStoreTable, InnerTable}
@@ -31,8 +32,9 @@ import org.apache.paimon.table.source.{InnerTableScan, Split}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.SQLConfHelper
+import org.apache.spark.sql.connector.expressions.NamedReference
import org.apache.spark.sql.connector.metric.{CustomMetric, CustomTaskMetric}
-import org.apache.spark.sql.connector.read.Batch
+import org.apache.spark.sql.connector.read.{Batch, Statistics}
import org.apache.spark.sql.connector.read.streaming.MicroBatchStream
import scala.collection.JavaConverters._
@@ -42,7 +44,9 @@ abstract class PaimonBaseScan(table: InnerTable)
with PaimonSupportsRuntimeFiltering
with SQLConfHelper {
- private lazy val paimonMetricsRegistry: SparkMetricRegistry = SparkMetricRegistry()
+ private[spark] lazy val paimonMetricsRegistry: SparkMetricRegistry = SparkMetricRegistry()
+
+ @transient private lazy val postponeMergeOnRead = new PostponeMergeOnRead(this)
protected def getInputSplits: Array[Split] = {
val scan = readBuilder
@@ -52,15 +56,17 @@ abstract class PaimonBaseScan(table: InnerTable)
.withMetricRegistry(paimonMetricsRegistry)
val plan = scan.plan()
+ registerReadProtectionTagCleanup(scan.readProtectionTagName)
+ plan.splits().asScala.toArray
+ }
- Option(scan.readProtectionTagName).foreach {
+ final private[spark] def registerReadProtectionTagCleanup(tagName: String): Unit = {
+ Option(tagName).foreach {
name =>
BatchReadTagCleanupListener
.getOrCreate(SparkSession.active)
.registerCleanup(name, table)
}
-
- plan.splits().asScala.toArray
}
private def evalGlobalIndexSearch(): GlobalIndexResult = {
@@ -82,6 +88,10 @@ abstract class PaimonBaseScan(table: InnerTable)
null
}
+ override def filterAttributes(): Array[NamedReference] = {
+ if (postponeMergeOnRead.enabled) Array.empty else super.filterAttributes()
+ }
+
private def evalVectorSearch(): GlobalIndexResult = {
PaimonBaseScan.evalVectorSearch(
table,
@@ -129,14 +139,39 @@ abstract class PaimonBaseScan(table: InnerTable)
}
override def toBatch: Batch = {
- ensureNoFullScan()
- super.toBatch
+ if (postponeMergeOnRead.enabled) {
+ throw new UnsupportedOperationException(
+ "Postpone merge-on-read must be executed by PostponeMergeOnReadExec.")
+ } else {
+ ensureNoFullScan()
+ super.toBatch
+ }
}
override def toMicroBatchStream(checkpointLocation: String): MicroBatchStream = {
+ if (PostponeMergeOnRead.enabled(table)) {
+ throw new UnsupportedOperationException(
+ "Option 'postpone.merge-on-read' is only supported for batch reads.")
+ }
new PaimonMicroBatchStream(table.asInstanceOf[DataTable], readBuilder, checkpointLocation)
}
+ override def estimateStatistics: Statistics = {
+ planPostponeMerge(SparkSession.active.sparkContext.defaultParallelism) match {
+ case Some(plan) =>
+ PaimonStatistics(
+ plan.corePlan.splits().asScala.toArray,
+ readTableRowType,
+ table.rowType(),
+ table.statistics())
+ case None => super.estimateStatistics
+ }
+ }
+
+ final private[spark] def planPostponeMerge(defaultBucketNum: Int): Option[MergePlan] = {
+ postponeMergeOnRead.plan(defaultBucketNum)
+ }
+
override def supportedCustomMetrics: Array[CustomMetric] = {
super.supportedCustomMetrics ++
Array(
@@ -151,7 +186,9 @@ abstract class PaimonBaseScan(table: InnerTable)
paimonMetricsRegistry.buildSparkScanMetrics()
}
- private def ensureNoFullScan(): Unit = {
+ final protected def ensureNoFullScan(): Unit = ensureNoFullScan(0L)
+
+ final private[spark] def ensureNoFullScan(externallyReadFiles: Long): Unit = {
if (OptionUtils.readAllowFullScan()) {
return
}
@@ -161,7 +198,7 @@ abstract class PaimonBaseScan(table: InnerTable)
val skippedFiles = paimonMetricsRegistry.buildSparkScanMetrics().collectFirst {
case m: PaimonSkippedTableFilesTaskMetric => m.value
}
- if (skippedFiles.contains(0)) {
+ if (skippedFiles.exists(_ <= externallyReadFiles)) {
throw new RuntimeException("Full scan is not supported.")
}
case _ =>
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
index 844c4de52b..c17bc35ba8 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
@@ -97,6 +97,10 @@ class PaimonScanBuilder(val table: InnerTable)
// Spark does not support push down aggregation for streaming scan.
override def pushAggregation(aggregation: Aggregation): Boolean = {
+ if (PostponeMergeOnRead.createReadBuilder(table, pushedPartitionFilters).isDefined) {
+ return false
+ }
+
if (localScan.isDefined) {
return true
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeInputScan.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeInputScan.scala
new file mode 100644
index 0000000000..debb4e3450
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeInputScan.scala
@@ -0,0 +1,235 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark
+
+import org.apache.paimon.KeyValue
+import org.apache.paimon.data.serializer.InternalRowSerializer
+import org.apache.paimon.reader.RecordReaderIterator
+import org.apache.paimon.spark.PostponeMergeInputScan._
+import org.apache.paimon.spark.PostponeMergeOnRead.MergePlan
+import org.apache.paimon.table.BucketMode
+import org.apache.paimon.table.PostponeUtils.PostponeBucketRouter
+import org.apache.paimon.table.source.{DataSplit, DeletionFile, PostponeMergePlan, PostponeMergeReadBuilder, SplitSerializer}
+import org.apache.paimon.types.RowType
+import org.apache.paimon.utils.SerializationUtils
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
+import org.apache.spark.sql.connector.read.{Batch, InputPartition, PartitionReader, PartitionReaderFactory, Scan}
+import org.apache.spark.sql.types.{BinaryType, ByteType, IntegerType, LongType, StructField, StructType}
+
+import scala.collection.JavaConverters._
+
+/** Internal DSv2 scan which materializes real-split markers and routed postpone records. */
+private[spark] case class PostponeMergeInputScan(mergePlan: MergePlan) extends Scan {
+
+ override def readSchema(): StructType = CARRIER_SCHEMA
+
+ override def toBatch: Batch =
+ PostponeMergeInputBatch(mergePlan.readBuilder, mergePlan.corePlan)
+
+ override def description(): String = "PaimonPostponeMergeInput"
+}
+
+private[spark] object PostponeMergeInputScan {
+
+ val PARTITION_COLUMN = "__paimon_postpone_partition"
+ val BUCKET_COLUMN = "__paimon_postpone_bucket"
+ val INPUT_KIND_COLUMN = "__paimon_postpone_input_kind"
+ val REAL_SPLIT_COLUMN = "__paimon_postpone_real_split"
+ val KEY_COLUMN = "__paimon_postpone_key"
+ val WRITER_LOCAL_ORDER_COLUMN = "__paimon_postpone_writer_local_order"
+ val ROW_KIND_COLUMN = "__paimon_postpone_row_kind"
+ val VALUE_COLUMN = "__paimon_postpone_value"
+
+ val REAL_SPLIT: Byte = 0
+ val POSTPONE_RECORD: Byte = 1
+
+ val CARRIER_SCHEMA: StructType = StructType(
+ Seq(
+ StructField(PARTITION_COLUMN, BinaryType, nullable = false),
+ StructField(BUCKET_COLUMN, IntegerType, nullable = false),
+ StructField(INPUT_KIND_COLUMN, ByteType, nullable = false),
+ StructField(REAL_SPLIT_COLUMN, BinaryType, nullable = true),
+ StructField(KEY_COLUMN, BinaryType, nullable = true),
+ StructField(WRITER_LOCAL_ORDER_COLUMN, LongType, nullable = false),
+ StructField(ROW_KIND_COLUMN, ByteType, nullable = false),
+ StructField(VALUE_COLUMN, BinaryType, nullable = true)
+ ))
+
+ val PARTITION_ORDINAL: Int = CARRIER_SCHEMA.fieldIndex(PARTITION_COLUMN)
+ val BUCKET_ORDINAL: Int = CARRIER_SCHEMA.fieldIndex(BUCKET_COLUMN)
+ val INPUT_KIND_ORDINAL: Int = CARRIER_SCHEMA.fieldIndex(INPUT_KIND_COLUMN)
+ val REAL_SPLIT_ORDINAL: Int = CARRIER_SCHEMA.fieldIndex(REAL_SPLIT_COLUMN)
+ val KEY_ORDINAL: Int = CARRIER_SCHEMA.fieldIndex(KEY_COLUMN)
+ val WRITER_LOCAL_ORDER_ORDINAL: Int = CARRIER_SCHEMA.fieldIndex(WRITER_LOCAL_ORDER_COLUMN)
+ val ROW_KIND_ORDINAL: Int = CARRIER_SCHEMA.fieldIndex(ROW_KIND_COLUMN)
+ val VALUE_ORDINAL: Int = CARRIER_SCHEMA.fieldIndex(VALUE_COLUMN)
+
+ private case class PostponeMergeInputBatch(
+ readBuilder: PostponeMergeReadBuilder,
+ corePlan: PostponeMergePlan)
+ extends Batch {
+
+ override def planInputPartitions(): Array[InputPartition] = {
+ val realPartitions = corePlan
+ .realSplits()
+ .asScala
+ .groupBy(bucketKey)
+ .map {
+ case (_, splits) =>
+ PaimonInputPartition(mergeRealSplits(splits.toSeq))
+ }
+ val postponePartitions = corePlan.postponeSplits().asScala.map(PaimonInputPartition(_))
+ (realPartitions ++ postponePartitions).toArray[InputPartition]
+ }
+
+ override def createReaderFactory(): PartitionReaderFactory = {
+ PostponeMergeInputReaderFactory(
+ readBuilder,
+ corePlan.keyType(),
+ corePlan.mergeReadType(),
+ corePlan.bucketRouter())
+ }
+ }
+
+ private case class PostponeMergeInputReaderFactory(
+ readBuilder: PostponeMergeReadBuilder,
+ keyType: RowType,
+ mergeReadType: RowType,
+ bucketRouter: PostponeBucketRouter)
+ extends PartitionReaderFactory {
+
+ override def createReader(partition: InputPartition): PartitionReader[InternalRow] = {
+ partition match {
+ case input: PaimonInputPartition =>
+ val split = input.splits.head.asInstanceOf[DataSplit]
+ if (split.bucket() == BucketMode.POSTPONE_BUCKET) {
+ new PostponePartitionReader(split, readBuilder, keyType, mergeReadType, bucketRouter)
+ } else {
+ new RealBucketPartitionReader(split)
+ }
+ case other =>
+ throw new IllegalArgumentException(
+ "Unsupported postpone merge input partition: " + other.getClass.getName + ".")
+ }
+ }
+ }
+
+ private class RealBucketPartitionReader(split: DataSplit) extends PartitionReader[InternalRow] {
+
+ private val row = new GenericInternalRow(
+ Array[Any](
+ SerializationUtils.serializeBinaryRow(split.partition()),
+ split.bucket(),
+ REAL_SPLIT,
+ SplitSerializer.serialize(split),
+ null,
+ 0L,
+ 0.toByte,
+ null))
+ private var emitted = false
+
+ override def next(): Boolean = {
+ if (!emitted) {
+ emitted = true
+ true
+ } else {
+ false
+ }
+ }
+
+ override def get(): InternalRow = row
+
+ override def close(): Unit = {}
+ }
+
+ /** Reads one writer and attaches an ordinal which preserves its order after the shuffle. */
+ private class PostponePartitionReader(
+ split: DataSplit,
+ readBuilder: PostponeMergeReadBuilder,
+ keyType: RowType,
+ mergeReadType: RowType,
+ bucketRouter: PostponeBucketRouter)
+ extends PartitionReader[InternalRow] {
+
+ private val keySerializer = new InternalRowSerializer(keyType)
+ private val valueSerializer = new InternalRowSerializer(mergeReadType)
+ private val partitionBytes = SerializationUtils.serializeBinaryRow(split.partition())
+ private val records =
+ new RecordReaderIterator[KeyValue](readBuilder.newRead().createPostponeReader(split))
+ private val current = new GenericInternalRow(
+ Array[Any](partitionBytes, 0, POSTPONE_RECORD, null, null, 0L, 0.toByte, null))
+ private var nextWriterLocalOrder = 0L
+
+ override def next(): Boolean = {
+ if (!records.hasNext) {
+ false
+ } else {
+ val keyValue = records.next()
+ val key = keySerializer.toBinaryRow(keyValue.key())
+ val value = valueSerializer.toBinaryRow(keyValue.value())
+ current.setInt(BUCKET_ORDINAL, bucketRouter.bucket(split.partition(), key))
+ current.update(KEY_ORDINAL, SerializationUtils.serializeBinaryRow(key))
+ current.setLong(WRITER_LOCAL_ORDER_ORDINAL, nextWriterLocalOrder)
+ current.setByte(ROW_KIND_ORDINAL, keyValue.valueKind().toByteValue)
+ current.update(VALUE_ORDINAL, SerializationUtils.serializeBinaryRow(value))
+ nextWriterLocalOrder = Math.addExact(nextWriterLocalOrder, 1L)
+ true
+ }
+ }
+
+ override def get(): InternalRow = current
+
+ override def close(): Unit = records.close()
+ }
+
+ private def bucketKey(split: DataSplit) = (split.partition(), split.bucket())
+
+ private def mergeRealSplits(splits: Seq[DataSplit]): DataSplit = {
+ if (splits.size == 1) {
+ splits.head
+ } else {
+ val first = splits.head
+ val builder = DataSplit
+ .builder()
+ .withSnapshot(first.snapshotId())
+ .withPartition(first.partition())
+ .withBucket(first.bucket())
+ .withBucketPath(first.bucketPath())
+ .withTotalBuckets(first.totalBuckets())
+ .withDataFiles(splits.flatMap(_.dataFiles().asScala).asJava)
+ .isStreaming(first.isStreaming())
+ .rawConvertible(splits.forall(_.rawConvertible()))
+
+ if (splits.exists(_.deletionFiles().isPresent)) {
+ val deletionFiles = splits.flatMap {
+ split =>
+ if (split.deletionFiles().isPresent) {
+ split.deletionFiles().get().asScala
+ } else {
+ Seq.fill(split.dataFiles().size())(null: DeletionFile)
+ }
+ }
+ builder.withDataDeletionFiles(deletionFiles.asJava)
+ }
+ builder.build()
+ }
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeOnRead.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeOnRead.scala
new file mode 100644
index 0000000000..5dcdb802ff
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PostponeMergeOnRead.scala
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark
+
+import org.apache.paimon.partition.PartitionPredicate
+import org.apache.paimon.predicate.PredicateBuilder
+import org.apache.paimon.spark.PostponeMergeOnRead.MergePlan
+import org.apache.paimon.table.{BucketMode, FileStoreTable, Table}
+import org.apache.paimon.table.source.{PostponeMergePlan, PostponeMergeReadBuilder}
+
+import scala.collection.JavaConverters._
+
+final private[spark] class PostponeMergeOnRead(scan: PaimonBaseScan) {
+
+ @transient private lazy val mergeReadBuilder = {
+ val builder =
+ PostponeMergeOnRead.createReadBuilder(scan.table, scan.pushedPartitionFilters.toArray)
+ if (
+ builder.isDefined &&
+ (scan.pushedVectorSearch.isDefined ||
+ scan.pushedHybridSearch.isDefined ||
+ scan.pushedFullTextSearch.isDefined)
+ ) {
+ throw new UnsupportedOperationException(
+ "Option 'postpone.merge-on-read' does not support vector, hybrid or full-text search.")
+ }
+ builder
+ }
+
+ @transient private var mergePlan: MergePlan = _
+
+ def enabled: Boolean = mergeReadBuilder.isDefined
+
+ def plan(defaultBucketNum: Int): Option[MergePlan] = synchronized {
+ mergeReadBuilder.map {
+ builder =>
+ if (mergePlan == null) {
+ if (scan.metadataFields.nonEmpty) {
+ throw new UnsupportedOperationException(
+ "Option 'postpone.merge-on-read' does not support metadata columns: " +
+ scan.metadataFields.map(_.name).mkString(", "))
+ }
+
+ builder
+ .withReadType(scan.readTableRowType)
+ .withDefaultBucketNum(defaultBucketNum)
+ .withMetricRegistry(scan.paimonMetricsRegistry)
+ if (scan.pushedDataFilters.nonEmpty) {
+ builder.withFilter(PredicateBuilder.and(scan.pushedDataFilters.toList.asJava))
+ }
+
+ val corePlan = builder.plan()
+ scan.registerReadProtectionTagCleanup(builder.readProtectionTagName())
+ val postponeFiles =
+ corePlan.postponeSplits().asScala.iterator.map(_.dataFiles().size().toLong).sum
+ scan.ensureNoFullScan(postponeFiles)
+ mergePlan = MergePlan(builder, corePlan, scan.coreOptions.blobAsDescriptor())
+ }
+ mergePlan
+ }
+ }
+}
+
+private[spark] object PostponeMergeOnRead {
+
+ private[spark] def enabled(table: Table): Boolean = {
+ table match {
+ case fileStoreTable: FileStoreTable => configured(fileStoreTable)
+ case _ => false
+ }
+ }
+
+ private[spark] def createReadBuilder(
+ table: Table,
+ partitionFilters: Array[PartitionPredicate]): Option[PostponeMergeReadBuilder] = {
+ table match {
+ case fileStoreTable: FileStoreTable if configured(fileStoreTable) =>
+ val partitionFilter =
+ if (partitionFilters.isEmpty) null
+ else PartitionPredicate.and(partitionFilters.toList.asJava)
+ val result = PostponeMergeReadBuilder.create(fileStoreTable, partitionFilter)
+ if (result.isPresent) Some(result.get) else None
+ case _ => None
+ }
+ }
+
+ private def configured(table: FileStoreTable): Boolean = {
+ table.coreOptions().postponeMergeOnRead() &&
+ table.bucketMode() == BucketMode.POSTPONE_MODE &&
+ !table.primaryKeys().isEmpty
+ }
+
+ private[spark] case class MergePlan(
+ readBuilder: PostponeMergeReadBuilder,
+ corePlan: PostponeMergePlan,
+ blobAsDescriptor: Boolean)
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/OptimizeMetadataOnlyDeleteFromPaimonTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/OptimizeMetadataOnlyDeleteFromPaimonTable.scala
index 7f28182599..aa56627a73 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/OptimizeMetadataOnlyDeleteFromPaimonTable.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/OptimizeMetadataOnlyDeleteFromPaimonTable.scala
@@ -49,13 +49,12 @@ import scala.collection.JavaConverters._
* <p>Note: this rule must be placed before [[MergePaimonScalarSubqueries]], because
* [[MergePaimonScalarSubqueries]] will merge subqueries.
*/
-object OptimizeMetadataOnlyDeleteFromPaimonTable
+case class OptimizeMetadataOnlyDeleteFromPaimonTable(spark: SparkSession)
extends Rule[LogicalPlan]
with ExpressionHelper
with Logging {
- lazy val spark: SparkSession = PaimonSparkSession.active
- lazy val resolver: Resolver = spark.sessionState.conf.resolver
+ private def resolver: Resolver = spark.sessionState.conf.resolver
override def apply(plan: LogicalPlan): LogicalPlan = {
plan.transform {
@@ -166,3 +165,11 @@ object OptimizeMetadataOnlyDeleteFromPaimonTable
subquery.updateResult()
}
}
+
+object OptimizeMetadataOnlyDeleteFromPaimonTable {
+
+ def isMetadataOnlyDelete(table: FileStoreTable, condition: Expression): Boolean = {
+ OptimizeMetadataOnlyDeleteFromPaimonTable(PaimonSparkSession.active)
+ .isMetadataOnlyDelete(table, condition)
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
index 77c50acf70..03fbbb42c4 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
@@ -24,7 +24,7 @@ import org.apache.paimon.globalindex.{GlobalIndexResult, IndexedSplit, ScoredGlo
import org.apache.paimon.partition.PartitionPredicate
import org.apache.paimon.partition.PartitionPredicate.splitPartitionPredicatesAndDataPredicates
import org.apache.paimon.predicate.{Predicate, PredicateBuilder}
-import org.apache.paimon.spark.{PaimonRecordReaderIterator, SparkCatalog, SparkGenericCatalog, SparkTable, SparkUtils}
+import org.apache.paimon.spark.{PaimonRecordReaderIterator, PaimonScan, PostponeMergeInputScan, PostponeMergeOnRead, SparkCatalog, SparkGenericCatalog, SparkTable, SparkUtils}
import org.apache.paimon.spark.catalog.{SparkBaseCatalog, SupportView}
import org.apache.paimon.spark.catalyst.analysis.ResolvedPaimonView
import org.apache.paimon.spark.catalyst.optimizer.RepartitionLateralVectorSearchInput
@@ -43,14 +43,15 @@ import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.{ResolvedNamespace, ResolvedTable}
-import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet, Expression, GenericInternalRow, JoinedRow, PredicateHelper, UnsafeProjection}
+import org.apache.spark.sql.catalyst.expressions.{And, Attribute, AttributeSet, Expression, GenericInternalRow, JoinedRow, PredicateHelper, UnsafeProjection}
import org.apache.spark.sql.catalyst.optimizer.BuildRight
+import org.apache.spark.sql.catalyst.planning.PhysicalOperation
import org.apache.spark.sql.catalyst.plans.logical.{AddPartitions, CreateTableAsSelect, DescribeRelation, DropPartitions, LogicalPlan, RepairTable, ReplaceTable, ReplaceTableAsSelect, ShowCreateTable}
import org.apache.spark.sql.catalyst.plans.physical.{Distribution, UnspecifiedDistribution}
import org.apache.spark.sql.catalyst.util.ArrayData
import org.apache.spark.sql.connector.catalog.{Identifier, PaimonLookupCatalog, TableCatalog}
-import org.apache.spark.sql.execution.{GlobalLimitExec, PaimonDescribeTableExec, SparkPlan, SparkStrategy, UnaryExecNode}
-import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Implicits, DataSourceV2Relation}
+import org.apache.spark.sql.execution.{FilterExec, GlobalLimitExec, PaimonDescribeTableExec, ProjectExec, SparkPlan, SparkStrategy, UnaryExecNode}
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Implicits, DataSourceV2Relation, DataSourceV2ScanRelation}
import org.apache.spark.sql.execution.exchange.ShuffleExchangeLike
import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BroadcastNestedLoopJoinExec}
import org.apache.spark.sql.execution.shim.{PaimonCreateTableAsSelectStrategy, PaimonReplaceTableAsSelectStrategy, PaimonReplaceTableStrategy}
@@ -69,6 +70,37 @@ case class PaimonStrategy(spark: SparkSession)
override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case PhysicalOperation(projects, filters, relation: DataSourceV2ScanRelation) =>
+ relation.scan match {
+ case scan: PaimonScan if PostponeMergeOnRead.enabled(scan.table) =>
+ scan.planPostponeMerge(spark.sparkContext.defaultParallelism) match {
+ case Some(mergePlan) =>
+ val inputScan = PostponeMergeInputScan(mergePlan)
+ val inputOutput =
+ org.apache.spark.sql.PaimonUtils.toAttributes(inputScan.readSchema())
+ val inputRelation =
+ SparkShimLoader.shim.createDataSourceV2ScanRelation(
+ relation,
+ inputScan,
+ inputOutput)
+ val merge =
+ PostponeMergeOnReadExec(
+ relation.output,
+ mergePlan,
+ PostponeMergeOnReadExec.computeShufflePartitions(
+ mergePlan.corePlan,
+ scan.coreOptions,
+ spark.sessionState.conf),
+ planLater(inputRelation)
+ )
+ val filtered =
+ filters.reduceLeftOption(And).map(FilterExec(_, merge)).getOrElse(merge)
+ ProjectExec(projects, filtered) :: Nil
+ case None => Nil
+ }
+ case _ => Nil
+ }
+
case ctas: CreateTableAsSelect =>
PaimonCreateTableAsSelectStrategy(spark)(ctas)
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PostponeMergeOnReadExec.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PostponeMergeOnReadExec.scala
new file mode 100644
index 0000000000..175709a94a
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PostponeMergeOnReadExec.scala
@@ -0,0 +1,247 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.execution
+
+import org.apache.paimon.{CoreOptions, KeyValue}
+import org.apache.paimon.data.{InternalRow => PaimonInternalRow}
+import org.apache.paimon.reader.RecordReaderIterator
+import org.apache.paimon.spark.PostponeMergeInputScan._
+import org.apache.paimon.spark.PostponeMergeOnRead.MergePlan
+import org.apache.paimon.spark.SparkUtils
+import org.apache.paimon.spark.data.SparkInternalRow
+import org.apache.paimon.spark.read.BinPackingSplits
+import org.apache.paimon.table.source.{DataSplit, PostponeMergePlan, PostponeMergeReadBuilder, SplitSerializer}
+import org.apache.paimon.types.{RowKind, RowType}
+import org.apache.paimon.utils.{IteratorRecordReader, SerializationUtils}
+
+import org.apache.spark.TaskContext
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute, SortOrder, UnsafeProjection}
+import org.apache.spark.sql.catalyst.plans.physical.{Distribution, Partitioning, UnknownPartitioning}
+import org.apache.spark.sql.execution.{SparkPlan, UnaryExecNode}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.paimon.shims.SparkShimLoader
+
+import java.util.Arrays
+
+import scala.collection.JavaConverters._
+
+/** Merges a Spark-clustered stream of real splits and postpone records through Paimon Core. */
+private[spark] case class PostponeMergeOnReadExec(
+ override val output: Seq[Attribute],
+ @transient mergePlan: MergePlan,
+ numShufflePartitions: Int,
+ child: SparkPlan)
+ extends UnaryExecNode {
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ Seq(
+ SparkShimLoader.shim.createClusteredDistribution(
+ Seq(child.output(PARTITION_ORDINAL), child.output(BUCKET_ORDINAL)),
+ Some(numShufflePartitions)))
+ }
+
+ override def requiredChildOrdering: Seq[Seq[SortOrder]] = {
+ Seq(
+ Seq(
+ SortOrder(child.output(PARTITION_ORDINAL), Ascending),
+ SortOrder(child.output(BUCKET_ORDINAL), Ascending),
+ SortOrder(child.output(INPUT_KIND_ORDINAL), Ascending),
+ SortOrder(child.output(WRITER_LOCAL_ORDER_ORDINAL), Ascending)
+ ))
+ }
+
+ override def outputPartitioning: Partitioning = {
+ UnknownPartitioning(child.outputPartitioning.numPartitions)
+ }
+
+ override def outputOrdering: Seq[SortOrder] = Nil
+
+ override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan = {
+ copy(child = newChild)
+ }
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val readBuilder = mergePlan.readBuilder
+ val resultRowType = mergePlan.corePlan.resultReadType()
+ val blobAsDescriptor = mergePlan.blobAsDescriptor
+ val outputAttributes = output
+
+ child.execute().mapPartitions {
+ rows =>
+ val unsafeProjection = UnsafeProjection.create(outputAttributes, outputAttributes)
+ new PostponeMergeOnReadExec.SortedBucketMergeIterator(
+ rows,
+ readBuilder,
+ resultRowType,
+ blobAsDescriptor)
+ .map(row => unsafeProjection(row): InternalRow)
+ }
+ }
+}
+
+private[spark] object PostponeMergeOnReadExec {
+
+ private case class BucketKey(partition: Array[Byte], bucket: Int)
+
+ private[spark] def computeShufflePartitions(
+ plan: PostponeMergePlan,
+ coreOptions: CoreOptions,
+ conf: SQLConf): Int = {
+ val maxShufflePartitions = conf.numShufflePartitions
+ val openCostInBytes = BinPackingSplits.openCostInBytes(coreOptions, conf)
+ val estimatedSize =
+ BinPackingSplits.estimatedSize(plan.splits().asScala, openCostInBytes)
+ val targetSize = BinPackingSplits.filesMaxPartitionBytes(coreOptions, conf)
+ val sizeParallelism =
+ if (estimatedSize <= 0) 1L else (estimatedSize - 1L) / targetSize + 1L
+ val usefulParallelism = Math.max(
+ 1L,
+ Math.min(maxShufflePartitions.toLong, Math.min(plan.numPotentialBuckets(), sizeParallelism)))
+ // Leave room for different bucket groups which hash to the same reducer.
+ val withHashHeadroom =
+ if (usefulParallelism == 1L) 1L
+ else usefulParallelism + (usefulParallelism + 1L) / 2L
+ Math.min(maxShufflePartitions.toLong, withHashHeadroom).toInt
+ }
+
+ private def deserializeSplit(serialized: Array[Byte]): DataSplit = {
+ SplitSerializer.deserialize(serialized).asInstanceOf[DataSplit]
+ }
+
+ private def sameBucket(row: InternalRow, bucketKey: BucketKey): Boolean = {
+ row.getInt(BUCKET_ORDINAL) == bucketKey.bucket &&
+ Arrays.equals(row.getBinary(PARTITION_ORDINAL), bucketKey.partition)
+ }
+
+ private class SortedBucketMergeIterator(
+ rows: Iterator[InternalRow],
+ readBuilder: PostponeMergeReadBuilder,
+ resultRowType: RowType,
+ blobAsDescriptor: Boolean)
+ extends Iterator[InternalRow]
+ with AutoCloseable {
+
+ private val bufferedRows = rows.buffered
+ private val ioManager = SparkUtils.createIOManager()
+ private val read = readBuilder.newRead().withIOManager(ioManager)
+ private val sparkRow = SparkInternalRow.create(resultRowType, blobAsDescriptor)
+ private var currentReader: RecordReaderIterator[PaimonInternalRow] = _
+ private var nextRow: InternalRow = _
+ private var closed = false
+
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ => close()))
+
+ override def hasNext: Boolean = {
+ advanceIfNeeded()
+ nextRow != null
+ }
+
+ override def next(): InternalRow = {
+ if (!hasNext) {
+ throw new NoSuchElementException
+ }
+ val result = nextRow
+ nextRow = null
+ result
+ }
+
+ private def advanceIfNeeded(): Unit = {
+ while (nextRow == null && !closed) {
+ if (currentReader == null && !openNextReader()) {
+ close()
+ return
+ }
+ if (currentReader.hasNext) {
+ nextRow = sparkRow.replace(currentReader.next())
+ } else {
+ closeCurrentReader()
+ }
+ }
+ }
+
+ private def openNextReader(): Boolean = {
+ if (!bufferedRows.hasNext) {
+ false
+ } else {
+ val first = bufferedRows.head
+ val bucketKey = BucketKey(first.getBinary(PARTITION_ORDINAL), first.getInt(BUCKET_ORDINAL))
+ val realSplit =
+ if (
+ sameBucket(bufferedRows.head, bucketKey) &&
+ bufferedRows.head.getByte(INPUT_KIND_ORDINAL) == REAL_SPLIT
+ ) {
+ deserializeSplit(bufferedRows.next().getBinary(REAL_SPLIT_ORDINAL))
+ } else {
+ null
+ }
+
+ val postponeRecords = new Iterator[KeyValue] {
+ override def hasNext: Boolean = {
+ bufferedRows.hasNext &&
+ sameBucket(bufferedRows.head, bucketKey) &&
+ bufferedRows.head.getByte(INPUT_KIND_ORDINAL) == POSTPONE_RECORD
+ }
+
+ override def next(): KeyValue = {
+ if (!hasNext) {
+ throw new NoSuchElementException
+ }
+ val row = bufferedRows.next()
+ new KeyValue().replace(
+ SerializationUtils.deserializeBinaryRow(row.getBinary(KEY_ORDINAL)),
+ RowKind.fromByteValue(row.getByte(ROW_KIND_ORDINAL)),
+ SerializationUtils.deserializeBinaryRow(row.getBinary(VALUE_ORDINAL))
+ )
+ }
+ }
+
+ currentReader = new RecordReaderIterator[PaimonInternalRow](
+ read.createBucketMergeReader(
+ realSplit,
+ new IteratorRecordReader[KeyValue](postponeRecords.asJava)))
+ if (bufferedRows.hasNext && sameBucket(bufferedRows.head, bucketKey)) {
+ throw new IllegalStateException(
+ "Unexpected postpone merge carrier kind " +
+ bufferedRows.head.getByte(INPUT_KIND_ORDINAL) + " for one bucket.")
+ }
+ true
+ }
+ }
+
+ private def closeCurrentReader(): Unit = {
+ if (currentReader != null) {
+ currentReader.close()
+ currentReader = null
+ }
+ }
+
+ override def close(): Unit = {
+ if (!closed) {
+ closed = true
+ try {
+ closeCurrentReader()
+ } finally {
+ ioManager.close()
+ }
+ }
+ }
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/adaptive/DisablePostponeCarrierShuffleCoalescing.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/adaptive/DisablePostponeCarrierShuffleCoalescing.scala
new file mode 100644
index 0000000000..12f654acad
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/adaptive/DisablePostponeCarrierShuffleCoalescing.scala
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.execution.adaptive
+
+import org.apache.paimon.spark.PostponeMergeInputScan
+import org.apache.paimon.spark.execution.PostponeMergeOnReadExec
+
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.exchange.{ENSURE_REQUIREMENTS, REPARTITION_BY_NUM, ShuffleExchangeExec}
+
+/** Prevents AQE from coalescing reducers which read real buckets behind carrier markers. */
+object DisablePostponeCarrierShuffleCoalescing extends Rule[SparkPlan] {
+
+ override def apply(plan: SparkPlan): SparkPlan = {
+ plan.transformUp {
+ case merge: PostponeMergeOnReadExec =>
+ merge.mapChildren(disableCoalescing)
+ }
+ }
+
+ private def disableCoalescing(plan: SparkPlan): SparkPlan = {
+ plan.transformUp {
+ case exchange: ShuffleExchangeExec
+ if exchange.shuffleOrigin == ENSURE_REQUIREMENTS &&
+ containsPostponeCarrierScan(exchange.child) =>
+ withFixedPartitionCount(exchange)
+ }
+ }
+
+ private def containsPostponeCarrierScan(plan: SparkPlan): Boolean = {
+ plan.find {
+ case scan: BatchScanExec => scan.scan.isInstanceOf[PostponeMergeInputScan]
+ case _ => false
+ }.isDefined
+ }
+
+ private def withFixedPartitionCount(exchange: ShuffleExchangeExec): ShuffleExchangeExec = {
+ // Replace only shuffleOrigin, preserving version-specific constructor fields.
+ val arguments = exchange.productIterator.map(_.asInstanceOf[AnyRef]).toArray
+ arguments(2) = REPARTITION_BY_NUM
+ val copied = exchange.makeCopy(arguments).asInstanceOf[ShuffleExchangeExec]
+ copied.copyTagsFrom(exchange)
+ copied
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
index 69b3b5f28b..95c2657036 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala
@@ -23,7 +23,7 @@ import org.apache.paimon.spark.catalyst.optimizer.{MergePaimonScalarSubqueries,
import org.apache.paimon.spark.catalyst.plans.logical.PaimonTableValuedFunctions
import org.apache.paimon.spark.commands.BucketExpression
import org.apache.paimon.spark.execution.{OldCompatibleStrategy, PaimonStrategy}
-import org.apache.paimon.spark.execution.adaptive.DisableUnnecessaryPaimonBucketedScan
+import org.apache.paimon.spark.execution.adaptive.{DisablePostponeCarrierShuffleCoalescing, DisableUnnecessaryPaimonBucketedScan}
import org.apache.spark.sql.SparkSessionExtensions
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
@@ -100,7 +100,7 @@ class PaimonSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
// optimization rules
extensions.injectOptimizerRule(spark => ReplacePaimonFunctions(spark))
- extensions.injectOptimizerRule(_ => OptimizeMetadataOnlyDeleteFromPaimonTable)
+ extensions.injectOptimizerRule(spark => OptimizeMetadataOnlyDeleteFromPaimonTable(spark))
// TODO: Enable MAP selected-key pushdown after core reader supports
// __PAIMON_MAP_SELECTED_KEYS read type.
extensions.injectOptimizerRule(_ => MergePaimonScalarSubqueries)
@@ -114,6 +114,7 @@ class PaimonSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
// query stage preparation
extensions.injectQueryStagePrepRule(_ => DisableUnnecessaryPaimonBucketedScan)
+ extensions.injectQueryStagePrepRule(_ => DisablePostponeCarrierShuffleCoalescing)
}
/**
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkPostponeCompactProcedure.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkPostponeCompactProcedure.scala
index 0e136e0e97..a3141f97b7 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkPostponeCompactProcedure.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkPostponeCompactProcedure.scala
@@ -18,7 +18,6 @@
package org.apache.paimon.spark.procedure
-import org.apache.paimon.CoreOptions
import org.apache.paimon.data.BinaryRow
import org.apache.paimon.io.{CompactIncrement, DataFileMeta, DataIncrement}
import org.apache.paimon.operation.FileSystemWriteRestore
@@ -30,6 +29,7 @@ import org.apache.paimon.spark.schema.SparkSystemColumns.{BUCKET_COL, ROW_KIND_C
import org.apache.paimon.spark.util.{ScanPlanHelper, SparkRowUtils}
import org.apache.paimon.spark.write.{PaimonDataWrite, WriteTaskResult}
import org.apache.paimon.table.{BlobDescriptorReaderFactory, BucketMode, FileStoreTable, PostponeUtils}
+import org.apache.paimon.table.PostponeUtils.PostponeBucketAssigner
import org.apache.paimon.table.sink.{CommitMessage, CommitMessageImpl}
import org.apache.paimon.utils.{SerializationUtils, UriReaderFactory}
@@ -58,34 +58,17 @@ case class SparkPostponeCompactProcedure(
@transient relation: DataSourceV2Relation) {
private val LOG = LoggerFactory.getLogger(getClass)
- private def createPostponePartitionBucketComputer(snapshotId: Long) = {
- val knownNumBuckets = PostponeUtils.getKnownNumBuckets(table, snapshotId)
- val targetRowNumPerBucket: Option[java.lang.Long] =
- table.coreOptions.postponeTargetRowNumPerBucket
- val postponeRowCounts =
- if (targetRowNumPerBucket.isDefined) {
- PostponeUtils.getPostponeRowCounts(table, snapshotId)
- } else {
- Collections.emptyMap[BinaryRow, java.lang.Long]()
- }
- val defaultBucketNum =
- if (table.coreOptions.toConfiguration.contains(CoreOptions.POSTPONE_DEFAULT_BUCKET_NUM)) {
- table.coreOptions.postponeDefaultBucketNum
- } else {
- spark.sparkContext.defaultParallelism
- }
-
- SparkPostponeCompactProcedure.PostponePartitionBucketComputer(
- knownNumBuckets,
- targetRowNumPerBucket,
- postponeRowCounts,
- defaultBucketNum)
+ private def createPostponeBucketAssigner(snapshotId: Long) = {
+ PostponeUtils.createPostponeBucketAssigner(
+ table,
+ snapshotId,
+ spark.sparkContext.defaultParallelism)
}
private def newDataWrite(
realTable: FileStoreTable,
rowKindColIdx: Int,
- postponePartitionBucketComputer: SparkPostponeCompactProcedure.PostponePartitionBucketComputer,
+ bucketAssigner: PostponeBucketAssigner,
uriReaderFactoryForBlobDescriptor: UriReaderFactory): PaimonDataWrite = {
val rowType = table.rowType()
val coreOptions = table.coreOptions()
@@ -98,7 +81,7 @@ case class SparkPostponeCompactProcedure(
Option.apply(coreOptions.fullCompactionDeltaCommits()),
None,
uriReaderFactoryForBlobDescriptor,
- Some(postponePartitionBucketComputer)
+ Some(partition => bucketAssigner.assign(partition))
)
dataWrite
}
@@ -124,8 +107,7 @@ case class SparkPostponeCompactProcedure(
return
}
val snapshotId = snapshot.id()
- val postponePartitionBucketComputer =
- createPostponePartitionBucketComputer(snapshotId)
+ val bucketAssigner = createPostponeBucketAssigner(snapshotId)
val realTable = PostponeUtils.tableForPostponeCompact(table, 1, snapshotId)
// Read data splits from the POSTPONE_BUCKET (-2)
@@ -165,7 +147,7 @@ case class SparkPostponeCompactProcedure(
table,
bucketColIdx,
encoderGroupWithBucketCol,
- postponePartitionBucketComputer
+ partition => bucketAssigner.assign(partition)
)
val dataFrame = withInitBucketCol
.mapPartitions(processor.processPartition)(encoderGroupWithBucketCol.encoder)
@@ -223,11 +205,7 @@ case class SparkPostponeCompactProcedure(
Iterator.empty
} else {
val dataWrite =
- newDataWrite(
- realTable,
- rowWorkAndKind._2,
- postponePartitionBucketComputer,
- uriReaderFactory)
+ newDataWrite(realTable, rowWorkAndKind._2, bucketAssigner, uriReaderFactory)
dataWrite.write.withWriteRestore(
new FileSystemWriteRestore(
realTable.coreOptions(),
@@ -329,21 +307,4 @@ object SparkPostponeCompactProcedure {
private[procedure] case class PostponeCompactKey(partition: IndexedSeq[Byte], bucket: Int)
extends Serializable
- private[procedure] case class PostponePartitionBucketComputer(
- knownNumBuckets: java.util.Map[BinaryRow, Integer],
- targetRowNumPerBucket: Option[java.lang.Long],
- postponeRowCounts: java.util.Map[BinaryRow, java.lang.Long],
- defaultBucketNum: Int)
- extends (BinaryRow => Integer)
- with Serializable {
-
- override def apply(p: BinaryRow): Integer = {
- PostponeUtils.determineBucketNum(
- p,
- knownNumBuckets,
- targetRowNumPerBucket.orNull,
- postponeRowCounts,
- defaultBucketNum)
- }
- }
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BinPackingSplits.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BinPackingSplits.scala
index f27af93cf6..ed331ff617 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BinPackingSplits.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BinPackingSplits.scala
@@ -43,31 +43,11 @@ case class BinPackingSplits(coreOptions: CoreOptions, readRowSizeRatio: Double =
private lazy val deletionVectors: Boolean = coreOptions.deletionVectorsEnabled()
- private lazy val filesMaxPartitionBytes: Long = {
- val options = coreOptions.toConfiguration
- var _filesMaxPartitionBytes = SOURCE_SPLIT_TARGET_SIZE.defaultValue().getBytes
-
- if (conf.contains(SQLConf.FILES_MAX_PARTITION_BYTES.key)) {
- _filesMaxPartitionBytes = conf.getConf(SQLConf.FILES_MAX_PARTITION_BYTES)
- }
- if (options.containsKey(SOURCE_SPLIT_TARGET_SIZE.key())) {
- _filesMaxPartitionBytes = options.get(SOURCE_SPLIT_TARGET_SIZE).getBytes
- }
- _filesMaxPartitionBytes
- }
-
- private lazy val openCostInBytes: Long = {
- val options = coreOptions.toConfiguration
- var _openCostBytes = SOURCE_SPLIT_OPEN_FILE_COST.defaultValue().getBytes
+ private lazy val filesMaxPartitionBytes: Long =
+ BinPackingSplits.filesMaxPartitionBytes(coreOptions, conf)
- if (conf.contains(SQLConf.FILES_OPEN_COST_IN_BYTES.key)) {
- _openCostBytes = conf.getConf(SQLConf.FILES_OPEN_COST_IN_BYTES)
- }
- if (options.containsKey(SOURCE_SPLIT_OPEN_FILE_COST.key())) {
- _openCostBytes = options.get(SOURCE_SPLIT_OPEN_FILE_COST).getBytes
- }
- _openCostBytes
- }
+ private lazy val openCostInBytes: Long =
+ BinPackingSplits.openCostInBytes(coreOptions, conf)
private lazy val leafNodeDefaultParallelism: Int = {
conf
@@ -232,8 +212,7 @@ case class BinPackingSplits(coreOptions: CoreOptions, readRowSizeRatio: Double =
val defaultMaxSplitBytes = filesMaxPartitionBytes
val minPartitionNum = conf.filesMinPartitionNum.getOrElse(leafNodeDefaultParallelism)
- val totalRawBytes =
- dataSplits.map(s => SplitUtils.splitSize(s) + SplitUtils.fileCount(s) * openCostInBytes).sum
+ val totalRawBytes = BinPackingSplits.estimatedSize(dataSplits, openCostInBytes)
val bytesPerCore = totalRawBytes / minPartitionNum
val maxSplitBytes = Math.min(defaultMaxSplitBytes, Math.max(openCostInBytes, bytesPerCore))
@@ -246,3 +225,34 @@ case class BinPackingSplits(coreOptions: CoreOptions, readRowSizeRatio: Double =
maxSplitBytes
}
}
+
+object BinPackingSplits {
+
+ private[spark] def filesMaxPartitionBytes(coreOptions: CoreOptions, conf: SQLConf): Long = {
+ val options = coreOptions.toConfiguration
+ if (options.containsKey(SOURCE_SPLIT_TARGET_SIZE.key())) {
+ options.get(SOURCE_SPLIT_TARGET_SIZE).getBytes
+ } else if (conf.contains(SQLConf.FILES_MAX_PARTITION_BYTES.key)) {
+ conf.getConf(SQLConf.FILES_MAX_PARTITION_BYTES)
+ } else {
+ SOURCE_SPLIT_TARGET_SIZE.defaultValue().getBytes
+ }
+ }
+
+ private[spark] def openCostInBytes(coreOptions: CoreOptions, conf: SQLConf): Long = {
+ val options = coreOptions.toConfiguration
+ if (options.containsKey(SOURCE_SPLIT_OPEN_FILE_COST.key())) {
+ options.get(SOURCE_SPLIT_OPEN_FILE_COST).getBytes
+ } else if (conf.contains(SQLConf.FILES_OPEN_COST_IN_BYTES.key)) {
+ conf.getConf(SQLConf.FILES_OPEN_COST_IN_BYTES)
+ } else {
+ SOURCE_SPLIT_OPEN_FILE_COST.defaultValue().getBytes
+ }
+ }
+
+ private[spark] def estimatedSize(splits: Iterable[Split], openCostInBytes: Long): Long = {
+ splits
+ .map(split => SplitUtils.splitSize(split) + SplitUtils.fileCount(split) * openCostInBytes)
+ .sum
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
index b65079d067..df21168bd3 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
@@ -29,7 +29,7 @@ import org.apache.paimon.types.{DataType, RowType}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.FunctionIdentifier
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression}
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
import org.apache.spark.sql.catalyst.parser.ParserInterface
import org.apache.spark.sql.catalyst.plans.logical.{Assignment, CTERelationRef, InsertAction, LogicalPlan, MergeAction, MergeIntoTable, SubqueryAlias, TableSpec, UnresolvedWith, UpdateAction}
@@ -38,9 +38,10 @@ import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.util.ArrayData
import org.apache.spark.sql.connector.catalog.{Column, Identifier, StagingTableCatalog, Table, TableCatalog}
import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.connector.read.Scan
import org.apache.spark.sql.connector.write.BatchWrite
import org.apache.spark.sql.execution.SparkPlan
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
import org.apache.spark.sql.types.StructType
import java.util.{Map => JMap}
@@ -194,8 +195,18 @@ trait SparkShim {
def copyDataSourceV2Relation(
relation: DataSourceV2Relation,
table: Table,
- output: Seq[org.apache.spark.sql.catalyst.expressions.AttributeReference])
- : DataSourceV2Relation
+ output: Seq[AttributeReference]): DataSourceV2Relation
+
+ /** Creates an internal scan relation whose constructor changed across Spark minor versions. */
+ def createDataSourceV2ScanRelation(
+ relation: DataSourceV2ScanRelation,
+ scan: Scan,
+ output: Seq[AttributeReference]): DataSourceV2ScanRelation
+
+ /** Creates a clustered distribution whose constructor changed in Spark 3.3. */
+ def createClusteredDistribution(
+ expressions: Seq[Expression],
+ requiredNumPartitions: Option[Int]): Distribution
/**
* Returns the list of "early" substitution rules Paimon needs to apply on a parsed view plan.
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PostponeBucketTableTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PostponeBucketTableTest.scala
index 72ca496a12..187ddf82de 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PostponeBucketTableTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PostponeBucketTableTest.scala
@@ -20,12 +20,14 @@ package org.apache.paimon.spark.sql
import org.apache.paimon.catalog.{Catalog, CatalogLoader, DelegateCatalog, Identifier}
import org.apache.paimon.fs.Path
-import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.spark.{PaimonScan, PaimonSparkTestBase}
import org.apache.paimon.spark.procedure.SparkPostponeCompactProcedure
import org.apache.paimon.table.{CatalogEnvironment, FileStoreTableFactory}
import org.apache.spark.TaskContext
import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
class PostponeBucketTableTest extends PaimonSparkTestBase {
@@ -133,6 +135,328 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
}
}
+ test("Postpone bucket table: Spark merge on read") {
+ withTable("t", "normal_t", "postpone_only_t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING
+ |) TBLPROPERTIES (
+ | 'primary-key' = 'k',
+ | 'bucket' = '-2',
+ | 'postpone.batch-write-fixed-bucket' = 'true',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'deletion-vectors.merge-on-read' = 'true',
+ | 'postpone.default-bucket-num' = '1',
+ | 'source.split.target-size' = '1 B'
+ |)
+ |""".stripMargin)
+
+ // Force multiple real splits in bucket 0.
+ sql("INSERT INTO t VALUES (1, 'base-1')")
+ sql("INSERT INTO t VALUES (2, 'base-2')")
+ val baseSnapshotId = loadTable("t").latestSnapshot().get().id()
+
+ withSparkSQLConf("spark.paimon.postpone.merge-on-read" -> "true") {
+ checkAnswer(sql("SELECT * FROM t ORDER BY k"), Seq(Row(1, "base-1"), Row(2, "base-2")))
+ val plan = sql("SELECT * FROM t").queryExecution.executedPlan.toString()
+ assert(!plan.contains("PostponeMergeOnRead"), plan)
+
+ val aggregate = sql("SELECT count(*) FROM t")
+ checkAnswer(aggregate, Seq(Row(2L)))
+ assert(
+ aggregate.queryExecution.executedPlan.collect {
+ case _: BaseAggregateExec => true
+ }.isEmpty,
+ aggregate.queryExecution.executedPlan)
+ }
+
+ sql("CREATE TABLE normal_t (k INT, v STRING) USING paimon")
+ sql("INSERT INTO normal_t VALUES (10, 'normal')")
+
+ sql("""
+ |CREATE TABLE postpone_only_t (
+ | k INT,
+ | v STRING
+ |) TBLPROPERTIES (
+ | 'primary-key' = 'k',
+ | 'bucket' = '-2',
+ | 'postpone.batch-write-fixed-bucket' = 'false',
+ | 'postpone.default-bucket-num' = '4'
+ |)
+ |""".stripMargin)
+ sql("INSERT INTO postpone_only_t VALUES (4, 'only-postpone')")
+
+ withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" -> "false") {
+ // Conflicting records share one writer order.
+ sql("INSERT INTO t VALUES (1, 'new-1'), (1, 'newest-1'), (3, 'new-3')")
+ sql("DELETE FROM t WHERE k = 2")
+ }
+
+ // MOR disabled: postpone records remain hidden.
+ checkAnswer(sql("SELECT * FROM t ORDER BY k"), Seq(Row(1, "base-1"), Row(2, "base-2")))
+ checkAnswer(sql("SELECT * FROM postpone_only_t"), Seq.empty)
+
+ withSparkSQLConf("spark.paimon.postpone.merge-on-read" -> "true") {
+ // The option must not affect ordinary tables.
+ checkAnswer(sql("SELECT * FROM normal_t"), Seq(Row(10, "normal")))
+ val postponeOnly = sql("SELECT * FROM postpone_only_t")
+ assert(postponeOnly.queryExecution.optimizedPlan.stats.rowCount.contains(BigInt(1)))
+ val postponeScan = postponeOnly.queryExecution.optimizedPlan.collectFirst {
+ case relation: DataSourceV2ScanRelation if relation.scan.isInstanceOf[PaimonScan] =>
+ relation.scan.asInstanceOf[PaimonScan]
+ }.get
+ assert(postponeScan.planPostponeMerge(spark.sparkContext.defaultParallelism).isDefined)
+ assert(postponeScan.filterAttributes().isEmpty)
+ assert(
+ intercept[UnsupportedOperationException](postponeScan.toBatch).getMessage
+ .contains("PostponeMergeOnReadExec"))
+ checkAnswer(postponeOnly, Seq(Row(4, "only-postpone")))
+ checkAnswer(sql("SELECT * FROM t ORDER BY k"), Seq(Row(1, "newest-1"), Row(3, "new-3")))
+ checkAnswer(sql("SELECT v FROM t ORDER BY v"), Seq(Row("new-3"), Row("newest-1")))
+ checkAnswer(sql("SELECT * FROM t WHERE v = 'base-1'"), Seq.empty)
+ checkAnswer(sql("SELECT * FROM t WHERE v = 'new-3'"), Seq(Row(3, "new-3")))
+ checkAnswer(sql("SELECT k FROM t WHERE k = 3"), Seq(Row(3)))
+ checkAnswer(sql("SELECT count(*) FROM t"), Seq(Row(2L)))
+ checkAnswer(sql("SELECT count(*), max(k) FROM t"), Seq(Row(2, 3)))
+ checkAnswer(sql("SELECT v FROM t ORDER BY k LIMIT 1"), Seq(Row("newest-1")))
+ if (gteqSpark3_3) {
+ checkAnswer(
+ sql(s"SELECT * FROM t VERSION AS OF $baseSnapshotId ORDER BY k"),
+ Seq(Row(1, "base-1"), Row(2, "base-2")))
+ }
+
+ val query = sql("SELECT * FROM t")
+ val mergeScan = query.queryExecution.optimizedPlan.collectFirst {
+ case relation: DataSourceV2ScanRelation if relation.scan.isInstanceOf[PaimonScan] =>
+ relation.scan.asInstanceOf[PaimonScan]
+ }.get
+ val realSplits = mergeScan
+ .planPostponeMerge(spark.sparkContext.defaultParallelism)
+ .get
+ .corePlan
+ .realSplits()
+ assert(realSplits.size() > 1, realSplits)
+
+ val plan = query.queryExecution.executedPlan.toString()
+ assert(plan.contains("PostponeMergeOnRead"), plan)
+ assert(plan.contains("PaimonPostponeMergeInput"), plan)
+ assert(plan.contains("Exchange hashpartitioning"), plan)
+ assert(!plan.contains("PostponeArrivalOrder"), plan)
+ assert(plan.contains("Sort [__paimon_postpone_partition"), plan)
+ assert(plan.contains("__paimon_postpone_writer_local_order"), plan)
+ }
+
+ withSparkSQLConf(
+ "spark.paimon.postpone.merge-on-read" -> "true",
+ "spark.sql.adaptive.enabled" -> "true",
+ "spark.sql.adaptive.coalescePartitions.enabled" -> "true",
+ "spark.sql.adaptive.coalescePartitions.parallelismFirst" -> "false",
+ "spark.sql.adaptive.advisoryPartitionSizeInBytes" -> (1L << 30).toString,
+ "spark.sql.files.maxPartitionBytes" -> (1L << 30).toString,
+ "spark.sql.shuffle.partitions" -> "8"
+ ) {
+ val query = sql("SELECT * FROM postpone_only_t")
+ checkAnswer(query, Seq(Row(4, "only-postpone")))
+ assert(query.rdd.getNumPartitions == 1, query.queryExecution.executedPlan)
+ }
+
+ withSparkSQLConf(
+ "spark.paimon.postpone.merge-on-read" -> "true",
+ "spark.sql.adaptive.enabled" -> "true",
+ "spark.sql.adaptive.coalescePartitions.enabled" -> "true",
+ "spark.sql.adaptive.coalescePartitions.parallelismFirst" -> "false",
+ "spark.sql.adaptive.advisoryPartitionSizeInBytes" -> (1L << 30).toString,
+ "spark.sql.files.maxPartitionBytes" -> "1",
+ "spark.sql.shuffle.partitions" -> "8"
+ ) {
+ val query = sql("SELECT * FROM postpone_only_t")
+ checkAnswer(query, Seq(Row(4, "only-postpone")))
+ assert(query.rdd.getNumPartitions == 6, query.queryExecution.executedPlan)
+ }
+
+ withSparkSQLConf(
+ "spark.paimon.postpone.merge-on-read" -> "true",
+ "spark.paimon.scan.mode" -> "compacted-full") {
+ val compactedFull = sql("SELECT * FROM t ORDER BY k")
+ checkAnswer(compactedFull, Seq(Row(1, "base-1"), Row(2, "base-2")))
+ assert(!compactedFull.queryExecution.executedPlan.toString.contains("PostponeMergeOnRead"))
+ }
+ }
+ }
+
+ test("Postpone bucket table: Spark merge on read respects full scan protection") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING,
+ | pt INT
+ |) TBLPROPERTIES (
+ | 'primary-key' = 'k, pt',
+ | 'bucket' = '-2',
+ | 'postpone.batch-write-fixed-bucket' = 'false'
+ |)
+ |PARTITIONED BY (pt)
+ |""".stripMargin)
+ sql("INSERT INTO t VALUES (1, 'a', 1), (2, 'b', 2)")
+
+ withSparkSQLConf(
+ "spark.paimon.postpone.merge-on-read" -> "true",
+ "spark.paimon.read.allow.fullScan" -> "false") {
+ assert(
+ intercept[Exception](sql("SELECT * FROM t").collect()).getMessage
+ .contains("Full scan is not supported."))
+ assert(
+ intercept[Exception](sql("SELECT * FROM t WHERE v IS NOT NULL").collect()).getMessage
+ .contains("Full scan is not supported."))
+ checkAnswer(sql("SELECT * FROM t WHERE pt = 1"), Seq(Row(1, "a", 1)))
+ }
+ }
+ }
+
+ test("Postpone bucket table: Spark merge on read custom bucket key") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k1 INT,
+ | k2 INT,
+ | v STRING
+ |) TBLPROPERTIES (
+ | 'primary-key' = 'k1, k2',
+ | 'bucket-key' = 'k1',
+ | 'bucket' = '-2',
+ | 'postpone.default-bucket-num' = '3',
+ | 'postpone.batch-write-fixed-bucket' = 'true'
+ |)
+ |""".stripMargin)
+
+ sql("INSERT INTO t VALUES (1, 1, 'a'), (1, 2, 'b'), (2, 1, 'c')")
+ withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" -> "false") {
+ sql("INSERT INTO t VALUES (1, 1, 'new-a'), (1, 2, 'new-b'), (3, 1, 'd')")
+ }
+
+ withSparkSQLConf("spark.paimon.postpone.merge-on-read" -> "true") {
+ checkAnswer(
+ sql("SELECT * FROM t ORDER BY k1, k2"),
+ Seq(Row(1, 1, "new-a"), Row(1, 2, "new-b"), Row(2, 1, "c"), Row(3, 1, "d")))
+ checkAnswer(sql("SELECT v FROM t WHERE k1 = 1 AND k2 = 2"), Seq(Row("new-b")))
+ }
+ }
+ }
+
+ test("Postpone bucket table: Spark merge on read sequence field") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING,
+ | seq INT
+ |) TBLPROPERTIES (
+ | 'primary-key' = 'k',
+ | 'bucket' = '-2',
+ | 'sequence.field' = 'seq',
+ | 'postpone.batch-write-fixed-bucket' = 'true'
+ |)
+ |""".stripMargin)
+
+ sql("INSERT INTO t VALUES (1, 'base-1', 10), (2, 'base-2', 10)")
+ withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" -> "false") {
+ sql("INSERT INTO t VALUES (1, 'older', 5), (2, 'newer', 20)")
+ }
+
+ withSparkSQLConf("spark.paimon.postpone.merge-on-read" -> "true") {
+ checkAnswer(
+ sql("SELECT * FROM t ORDER BY k"),
+ Seq(Row(1, "base-1", 10), Row(2, "newer", 20)))
+ // Core must retain seq after Spark projects v.
+ checkAnswer(sql("SELECT v FROM t ORDER BY v"), Seq(Row("base-1"), Row("newer")))
+ }
+ }
+ }
+
+ test("Postpone bucket table: Spark merge on read keeps partitions isolated") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING,
+ | pt STRING
+ |) TBLPROPERTIES (
+ | 'primary-key' = 'k, pt',
+ | 'bucket' = '-2',
+ | 'postpone.batch-write-fixed-bucket' = 'true'
+ |)
+ |PARTITIONED BY (pt)
+ |""".stripMargin)
+
+ sql("INSERT INTO t VALUES (1, 'base-a', 'a'), (1, 'base-b', 'b')")
+ withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" -> "false") {
+ sql("INSERT INTO t VALUES (1, 'new-a', 'a')")
+ }
+
+ withSparkSQLConf("spark.paimon.postpone.merge-on-read" -> "true") {
+ checkAnswer(
+ sql("SELECT * FROM t ORDER BY pt"),
+ Seq(Row(1, "new-a", "a"), Row(1, "base-b", "b")))
+ checkAnswer(sql("SELECT * FROM t WHERE pt = 'a'"), Seq(Row(1, "new-a", "a")))
+ checkAnswer(sql("SELECT * FROM t WHERE pt = 'b'"), Seq(Row(1, "base-b", "b")))
+
+ val mergePlan = sql("SELECT * FROM t WHERE pt = 'a'").queryExecution.executedPlan.toString
+ val ordinaryPlan =
+ sql("SELECT * FROM t WHERE pt = 'b'").queryExecution.executedPlan.toString
+ assert(mergePlan.contains("PostponeMergeOnRead"), mergePlan)
+ assert(!ordinaryPlan.contains("PostponeMergeOnRead"), ordinaryPlan)
+ }
+ }
+ }
+
+ test("Postpone bucket table: Spark merge on read merge engines") {
+ withTable("partial_t", "aggregation_t") {
+ sql("""
+ |CREATE TABLE partial_t (
+ | k INT,
+ | v1 STRING,
+ | v2 STRING
+ |) TBLPROPERTIES (
+ | 'primary-key' = 'k',
+ | 'bucket' = '-2',
+ | 'merge-engine' = 'partial-update',
+ | 'postpone.batch-write-fixed-bucket' = 'true'
+ |)
+ |""".stripMargin)
+ sql("INSERT INTO partial_t VALUES (1, 'a', CAST(NULL AS STRING))")
+
+ sql("""
+ |CREATE TABLE aggregation_t (
+ | k INT,
+ | total BIGINT
+ |) TBLPROPERTIES (
+ | 'primary-key' = 'k',
+ | 'bucket' = '-2',
+ | 'merge-engine' = 'aggregation',
+ | 'fields.total.aggregate-function' = 'sum',
+ | 'postpone.batch-write-fixed-bucket' = 'true'
+ |)
+ |""".stripMargin)
+ sql("INSERT INTO aggregation_t VALUES (1, 10L)")
+
+ withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" -> "false") {
+ sql("INSERT INTO partial_t VALUES (1, CAST(NULL AS STRING), 'b')")
+ sql("INSERT INTO aggregation_t VALUES (1, 3L)")
+ }
+
+ withSparkSQLConf("spark.paimon.postpone.merge-on-read" -> "true") {
+ checkAnswer(sql("SELECT * FROM partial_t"), Seq(Row(1, "a", "b")))
+ checkAnswer(sql("SELECT * FROM aggregation_t"), Seq(Row(1, 13L)))
+ checkAnswer(sql("SELECT v1 FROM partial_t"), Seq(Row("a")))
+ checkAnswer(sql("SELECT total FROM aggregation_t"), Seq(Row(13L)))
+ checkAnswer(sql("SELECT * FROM partial_t WHERE v2 = 'b'"), Seq(Row(1, "a", "b")))
+ checkAnswer(sql("SELECT * FROM aggregation_t WHERE total = 13"), Seq(Row(1, 13L)))
+ }
+ }
+ }
+
test("Postpone bucket table: write postpone bucket then write fix bucket") {
withTable("t") {
sql("""
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeyVectorSearchTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeyVectorSearchTest.scala
index ebafe40933..8080685828 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeyVectorSearchTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeyVectorSearchTest.scala
@@ -155,6 +155,20 @@ class PrimaryKeyVectorSearchTest extends PaimonSparkTestBase {
assert(spark.sql("SELECT bucket FROM `T$buckets`").collect().exists(_.getInt(0) == -2))
assert(spark.sql("SELECT file_path FROM `T$files` WHERE level = 0").collect().nonEmpty)
+ withSparkSQLConf("spark.paimon.postpone.merge-on-read" -> "true") {
+ val error = intercept[Exception] {
+ spark
+ .sql("""
+ |SELECT __paimon_search_score
+ |FROM vector_search('T', 'embedding', array(0.0f, 0.0f), 2)
+ |""".stripMargin)
+ .collect()
+ }
+ assert(
+ error.getMessage.contains("does not support vector, hybrid or full-text search"),
+ error.getMessage)
+ }
+
spark.sql("CALL sys.compact(table => 'T')")
assert(
diff --git a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/MinorVersionShim.scala b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/MinorVersionShim.scala
index 0784530909..6c9e991d99 100644
--- a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/MinorVersionShim.scala
+++ b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/MinorVersionShim.scala
@@ -18,9 +18,11 @@
package org.apache.spark.sql.paimon.shims
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, Expression}
import org.apache.spark.sql.catalyst.plans.logical.{CTERelationRef, LogicalPlan, MergeAction, MergeIntoTable}
import org.apache.spark.sql.catalyst.plans.physical.{ClusteredDistribution, Distribution}
+import org.apache.spark.sql.connector.read.Scan
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
object MinorVersionShim {
@@ -54,4 +56,17 @@ object MinorVersionShim {
def notMatchedBySourceActions(merge: MergeIntoTable): Seq[MergeAction] =
merge.notMatchedBySourceActions
+
+ def createDataSourceV2ScanRelation(
+ relation: DataSourceV2ScanRelation,
+ scan: Scan,
+ output: Seq[AttributeReference]): DataSourceV2ScanRelation = {
+ DataSourceV2ScanRelation(relation.relation, scan, output, None, None)
+ }
+
+ def createClusteredDistribution(
+ expressions: Seq[Expression],
+ requiredNumPartitions: Option[Int]): Distribution = {
+ ClusteredDistribution(expressions, requiredNumPartitions = requiredNumPartitions)
+ }
}
diff --git a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
index f14c212590..5ffbf6a145 100644
--- a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
+++ b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
@@ -48,11 +48,12 @@ import org.apache.spark.sql.catalyst.util.{ArrayData, GeneratedColumn, ResolveDe
import org.apache.spark.sql.connector.catalog.{Column, Identifier, StagingTableCatalog, Table, TableCatalog}
import org.apache.spark.sql.connector.catalog.CatalogV2Util.structTypeToV2Columns
import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.connector.read.Scan
import org.apache.spark.sql.connector.write.BatchWrite
import org.apache.spark.sql.execution.{SparkFormatTable, SparkPlan}
import org.apache.spark.sql.execution.datasources.{PartitioningAwareFileIndex, PartitionSpec}
import org.apache.spark.sql.execution.datasources.v2.{AtomicReplaceTableAsSelectExec, AtomicReplaceTableExec, ReplaceTableAsSelectExec, ReplaceTableExec}
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
import org.apache.spark.sql.execution.streaming.{FileStreamSink, MetadataLogFileIndex}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.StructType
@@ -281,6 +282,19 @@ class Spark3Shim extends SparkShim {
relation.copy(table = table, output = output)
}
+ override def createDataSourceV2ScanRelation(
+ relation: DataSourceV2ScanRelation,
+ scan: Scan,
+ output: Seq[AttributeReference]): DataSourceV2ScanRelation = {
+ MinorVersionShim.createDataSourceV2ScanRelation(relation, scan, output)
+ }
+
+ override def createClusteredDistribution(
+ expressions: Seq[Expression],
+ requiredNumPartitions: Option[Int]): Distribution = {
+ MinorVersionShim.createClusteredDistribution(expressions, requiredNumPartitions)
+ }
+
override def earlyBatchRules(): Seq[Rule[LogicalPlan]] =
Seq(CTESubstitution, SubstituteUnresolvedOrdinals)
diff --git a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
index 91664858ac..00e0b1ae4f 100644
--- a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
+++ b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
@@ -44,11 +44,12 @@ import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.util.{ArrayData, GeneratedColumn, IdentityColumn, ResolveDefaultColumns}
import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column, Identifier, StagingTableCatalog, Table, TableCatalog}
import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.connector.read.Scan
import org.apache.spark.sql.connector.write.BatchWrite
import org.apache.spark.sql.execution.{SparkFormatTable, SparkPlan}
import org.apache.spark.sql.execution.datasources.{PartitioningAwareFileIndex, PartitionSpec}
import org.apache.spark.sql.execution.datasources.v2.{AtomicReplaceTableAsSelectExec, AtomicReplaceTableExec, ReplaceTableAsSelectExec, ReplaceTableExec}
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
import org.apache.spark.sql.execution.streaming.runtime.MetadataLogFileIndex
import org.apache.spark.sql.execution.streaming.sinks.FileStreamSink
import org.apache.spark.sql.internal.SQLConf
@@ -285,6 +286,19 @@ class Spark4Shim extends SparkShim {
relation.copy(table = table, output = output)
}
+ override def createDataSourceV2ScanRelation(
+ relation: DataSourceV2ScanRelation,
+ scan: Scan,
+ output: Seq[AttributeReference]): DataSourceV2ScanRelation = {
+ DataSourceV2ScanRelation(relation.relation, scan, output, None, None)
+ }
+
+ override def createClusteredDistribution(
+ expressions: Seq[Expression],
+ requiredNumPartitions: Option[Int]): Distribution = {
+ ClusteredDistribution(expressions, requiredNumPartitions = requiredNumPartitions)
+ }
+
override def earlyBatchRules(): Seq[Rule[LogicalPlan]] = Seq(CTESubstitution)
override def mergeRowsKeepCopy(condition: Expression, output: Seq[Expression]): AnyRef =