reader = ConcatRecordReader.create(sectionReaders);
+ return finishMergeReader(ConcatRecordReader.create(sectionReaders), keepDelete);
+ }
+
+ /**
+ * Merges one real-bucket split with routed postpone records.
+ *
+ * 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 createPostponeMergeReader(
+ @Nullable DataSplit realSplit, RecordReader postponeRecords)
+ throws IOException {
+ List realFiles =
+ realSplit == null ? Collections.emptyList() : realSplit.dataFiles();
+
+ RowType valueType = actualReadType();
+ UserDefinedSeqComparator udsComparator = null;
+ SortBufferWriteBuffer postponeBuffer = null;
+ long nextSequenceNumber = -1L;
+ try (RecordReader records = postponeRecords) {
+ RecordReader.RecordIterator 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 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 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 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 finishMergeReader(
+ RecordReader 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 43c51d7ee7d7..f75be9a2b5a6 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 static int getWriteId(String fileName) {
e);
}
}
+
+ /** Returns the unique prefix shared by files produced by the same postpone writer. */
+ public static String getWriterPrefix(String fileName) {
+ int separator = fileName.lastIndexOf("-w-");
+ if (separator < 0) {
+ throw new IllegalArgumentException(
+ "Data file name " + fileName + " does not match the postpone writer pattern.");
+ }
+ return fileName.substring(0, separator + 3);
+ }
}
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 8a600eee9ed8..0463bb3f2d73 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 @@
/** 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 groupPostponeFiles(List splits) {
+ Map, List> 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 dataFiles = split.dataFiles();
+ List deletionFiles = split.deletionFiles().orElse(null);
+ for (int fileOrder = 0; fileOrder < dataFiles.size(); fileOrder++) {
+ DataFileMeta file = dataFiles.get(fileOrder);
+ Pair 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 result = new ArrayList<>(filesByWriter.size());
+ for (List writerFiles : filesByWriter.values()) {
+ writerFiles.sort(Comparator.comparingLong(file -> file.file.creationTimeEpochMillis()));
+
+ PostponeFile first = writerFiles.get(0);
+ List dataFiles = new ArrayList<>(writerFiles.size());
+ List 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 knownNumBuckets =
+ getKnownNumBuckets(table, snapshotId, partitionFilter);
+ Long targetRowNumPerBucket =
+ table.coreOptions().postponeTargetRowNumPerBucket().orElse(null);
+ Map 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 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 static Map getKnownNumBuckets(FileStoreTable table) {
public static Map getKnownNumBuckets(
FileStoreTable table, long snapshotId) {
- return getKnownNumBuckets(
- table.store()
- .newScan()
- .withSnapshot(snapshotId)
- .onlyReadRealBuckets()
- .readSimpleEntries());
+ return getKnownNumBuckets(table, snapshotId, null);
+ }
+
+ static Map 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 getKnownNumBuckets(
@@ -157,11 +300,19 @@ public static Map getPostponeRowCounts(FileStoreTable table) {
/** Returns row counts of active postpone files in the specified snapshot. */
public static Map getPostponeRowCounts(FileStoreTable table, long snapshotId) {
- return getPostponeRowCounts(
+ return getPostponeRowCounts(table, snapshotId, null);
+ }
+
+ static Map 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 getPostponeRowCounts(Iterator iterator) {
@@ -196,6 +347,98 @@ public static FileStoreTable tableForCommit(FileStoreTable table) {
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 knownNumBuckets;
+ @Nullable private final Long targetRowNumPerBucket;
+ private final Map postponeRowCounts;
+ private final int defaultBucketNum;
+
+ private PostponeBucketAssigner(
+ Map knownNumBuckets,
+ @Nullable Long targetRowNumPerBucket,
+ Map 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 000000000000..b2e6bc460e6f
--- /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 realSplits;
+ private final List postponeSplits;
+ private final PostponeBucketRouter bucketRouter;
+ private final RowType keyType;
+ private final RowType resultReadType;
+ private final RowType mergeReadType;
+ private final long numPotentialBuckets;
+
+ PostponeMergePlan(
+ List realSplits,
+ List 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 realSplits() {
+ return realSplits;
+ }
+
+ public List postponeSplits() {
+ return postponeSplits;
+ }
+
+ @Override
+ public List splits() {
+ List 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 realSplits,
+ List postponeSplits,
+ PostponeBucketRouter bucketRouter) {
+ Map> realBuckets = new HashMap<>();
+ for (DataSplit split : realSplits) {
+ realBuckets
+ .computeIfAbsent(split.partition(), ignored -> new HashSet<>())
+ .add(split.bucket());
+ }
+
+ Set postponePartitions = new HashSet<>();
+ for (DataSplit split : postponeSplits) {
+ postponePartitions.add(split.partition());
+ }
+
+ long count = 0;
+ for (Map.Entry> 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 000000000000..83ed07150f10
--- /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 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 createBucketMergeReader(
+ @Nullable DataSplit realBucketSplit, RecordReader bucketPostponeRecords)
+ throws IOException {
+ RecordReader 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 000000000000..d634df8bacb3
--- /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 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, List> split =
+ splitPartitionPredicatesAndDataPredicates(
+ predicate, schema.logicalRowType(), schema.partitionKeys());
+ List primaryKeys = schema.trimmedPrimaryKeys();
+ Set nonPrimaryKeys =
+ schema.fieldNames().stream()
+ .filter(name -> !primaryKeys.contains(name))
+ .collect(Collectors.toSet());
+ List 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 ff8249c22153..06fd4abf8027 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.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 @@ protected void runTest(List input) throws IOException {
assertThat(expected).isEmpty();
}
+ @Test
+ public void testCreateReader() throws Exception {
+ List input = ReusingTestData.generateData(100, addOnly());
+ Queue expected = new LinkedList<>(getExpected(input));
+ prepareTable(input);
+
+ try (RecordReaderIterator 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 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 db65eb8d1660..d573b5bd905a 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.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.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.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 void testValueProjection() throws Exception {
}
}
+ @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 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 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 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 baseExpected =
+ realRecords.stream()
+ .collect(
+ Collectors.toMap(
+ record -> record.key().getInt(0),
+ record -> record.value().getString(1).toString()));
+
+ RowType projectedValueType = valueType.project("v");
+ try (RecordReaderIterator reader =
+ new RecordReaderIterator<>(
+ store.newRead()
+ .withReadType(projectedValueType)
+ .createPostponeMergeReader(
+ realSplit,
+ new IteratorRecordReader<>(
+ Collections.emptyList().iterator())))) {
+ Map 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 reader =
+ new RecordReaderIterator<>(
+ store.newRead()
+ .withReadType(projectedValueType)
+ .withIOManager(ioManager)
+ .createPostponeMergeReader(
+ realSplit,
+ new IteratorRecordReader<>(
+ postponeRecords.iterator())))) {
+ Map 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 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 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 sequences = new HashMap<>();
+ try (RecordReaderIterator reader =
+ new RecordReaderIterator<>(store.newRead().createPostponeReader(split))) {
+ while (reader.hasNext()) {
+ KeyValue record = reader.next();
+ sequences.put(record.key().getInt(0), record.sequenceNumber());
+ }
+ }
+ Map 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 writeThenRead(
List data,
RowType readKeyType,
@@ -274,6 +446,17 @@ private TestFileStore createStore(
KeyValueFieldsExtractor extractor,
MergeFunctionFactory 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 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 @@ private TestFileStore createStore(
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 251f06c5f691..40b36bfb9672 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.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 void testGetKnownNumBucketsFromSnapshot() {
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 void testGetPostponeRowCountsFromSnapshot() {
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 void testGetLevel0BucketsFromSnapshot() {
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 splits =
+ Arrays.asList(
+ dataSplit(partition1, writer1Newest, writer2, sameWriteIdFromAnotherCommit),
+ dataSplit(partition1, writer1First),
+ dataSplit(partition2, otherPartition));
+
+ List 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 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 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 @@ private static SimpleFileEntry fileEntry(
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 75863dd6427b..972713a5d0a5 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 @@
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.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.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 void testLimitPushdownBoundaryCases() throws Exception {
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 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 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 values = new ArrayList<>();
+ try (IOManager ioManager = IOManager.create(tempDir.resolve("postpone-merge").toString());
+ RecordReaderIterator 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 dffd53beb9c4..2fb0b2ffc9f1 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,8 +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 {
@@ -41,4 +44,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 ae0d96e8f8e1..0a14f4fbd7a6 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,8 +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 {
@@ -41,4 +44,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 e3046f564f02..f3ed62814470 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
@@ -39,15 +39,17 @@ 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.{Aggregate, Assignment, ColumnDefinition, CTERelationRef, InsertAction, LogicalPlan, MergeAction, MergeIntoTable, MergeRows, SubqueryAlias, TableSpec, UnresolvedWith, UpdateAction}
import org.apache.spark.sql.catalyst.plans.logical.MergeRows.Keep
+import org.apache.spark.sql.catalyst.plans.physical.{ClusteredDistribution, Distribution}
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}
@@ -292,6 +294,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 df600a6a75e8..c9d19a73ece1 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 844c4de52b1d..c17bc35ba812 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 000000000000..debb4e345008
--- /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 000000000000..5dcdb802ffda
--- /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/execution/PaimonStrategy.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
index 2686ef7d7666..eb8205beadce 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.plans.logical.{CopyIntoLocationCommand, CopyIntoLocationSource, CopyIntoTableCommand, CreateOrReplaceTagCommand, CreatePaimonView, DeleteTagCommand, DropPaimonView, LateralVectorSearch, PaimonCallCommand, PaimonDropPartitions, PaimonTableValuedFunctions, RenameTagCommand, ResolvedIdentifier, ShowPaimonViews, ShowTagsCommand, TruncatePaimonTableWithFilter}
@@ -42,12 +42,13 @@ 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.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.util.ArrayData
import org.apache.spark.sql.connector.catalog.{Identifier, PaimonLookupCatalog, TableCatalog}
-import org.apache.spark.sql.execution.{PaimonDescribeTableExec, SparkPlan, SparkStrategy}
-import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Implicits, DataSourceV2Relation}
+import org.apache.spark.sql.execution.{FilterExec, PaimonDescribeTableExec, ProjectExec, SparkPlan, SparkStrategy}
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Implicits, DataSourceV2Relation, DataSourceV2ScanRelation}
import org.apache.spark.sql.execution.shim.{PaimonCreateTableAsSelectStrategy, PaimonReplaceTableAsSelectStrategy, PaimonReplaceTableStrategy}
import org.apache.spark.sql.paimon.shims.SparkShimLoader
@@ -64,6 +65,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 000000000000..175709a94a99
--- /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 000000000000..12f654acad2a
--- /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 388889bbeaaa..f655b4b55b97 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
@@ -113,6 +113,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 0e136e0e97fc..a3141f97b79f 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 f27af93cf604..ed331ff61778 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 85c923325dd3..5b7f9aee941e 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,17 +29,19 @@ 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}
+import org.apache.spark.sql.catalyst.plans.physical.Distribution
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}
@@ -191,8 +193,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 72ca496a1245..187ddf82defc 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 ebafe4093335..80806858289c 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 8877446d7615..82d5af6c638b 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,8 +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 {
@@ -47,4 +50,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 9bde530b28c7..9561756ce128 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
@@ -38,6 +38,7 @@ import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference,
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.{Aggregate, Assignment, CTERelationRef, InsertAction, LogicalPlan, MergeAction, MergeIntoTable, SubqueryAlias, TableSpec, UnresolvedWith, UpdateAction}
+import org.apache.spark.sql.catalyst.plans.physical.Distribution
// NOTE: `MergeRows` / `MergeRows.Keep` were introduced in Spark 3.4. We access them only via
// reflection inside the `mergeRowsKeep*` method bodies so that loading `Spark3Shim` does not fail
// on Spark 3.2 / 3.3 runtimes that still ship `paimon-spark3-common` (the module targets 3.5.8 at
@@ -47,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
@@ -275,6 +277,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 7a7cdc70f531..ccbed7431320 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
@@ -39,15 +39,17 @@ 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.{Aggregate, Assignment, ColumnDefinition, CTERelationRef, InsertAction, LogicalPlan, MergeAction, MergeIntoTable, MergeRows, SubqueryAlias, TableSpec, UnresolvedWith, UpdateAction}
import org.apache.spark.sql.catalyst.plans.logical.MergeRows.{Copy, Insert, Keep, Update}
+import org.apache.spark.sql.catalyst.plans.physical.{ClusteredDistribution, Distribution}
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
@@ -276,6 +278,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 =