This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 2665f44c04 [core] Support Deletion Vectors for Chain Table (#8574)
2665f44c04 is described below
commit 2665f44c049c73bf9c188cf643b9fa4ae66ce77c
Author: Juntao Zhang <[email protected]>
AuthorDate: Thu Jul 30 23:00:32 2026 +0800
[core] Support Deletion Vectors for Chain Table (#8574)
---
docs/docs/primary-key-table/chain-table.mdx | 49 +-
.../ExposeDeletionKeyValueReader.java | 90 ++++
.../paimon/io/ChainKeyValueFileReaderFactory.java | 28 ++
.../paimon/io/KeyValueDataFileRecordReader.java | 5 -
.../paimon/io/KeyValueFileReaderFactory.java | 41 +-
.../mergetree/compact/LookupMergeFunction.java | 4 +
.../paimon/operation/MergeFileSplitRead.java | 24 +-
.../org/apache/paimon/schema/SchemaValidation.java | 6 +-
.../apache/paimon/table/ChainGroupReadTable.java | 14 +-
.../org/apache/paimon/table/source/ChainSplit.java | 45 +-
.../paimon/table/source/SplitSerializer.java | 25 +-
.../org/apache/paimon/utils/ChainTableUtils.java | 31 +-
.../apache/paimon/utils/SerializationUtils.java | 19 +
.../paimon/table/ChainTableFileStoreTableTest.java | 511 +++++++++++++++++++++
.../apache/paimon/table/source/ChainSplitTest.java | 46 +-
.../paimon/table/source/SplitSerializerTest.java | 21 +-
.../test/resources/compatibility/split-v1-chain | Bin 1359 -> 1419 bytes
.../test/resources/compatibility/split-v1-fallback | Bin 1376 -> 1436 bytes
.../flink/FlinkChainTableDeletionVectorITCase.java | 339 ++++++++++++++
.../apache/paimon/flink/FlinkChainTableITCase.java | 192 +-------
.../paimon/flink/FlinkChainTableITCaseBase.java | 218 +++++++++
.../apache/paimon/spark/SparkChainTableITCase.java | 99 ++++
22 files changed, 1527 insertions(+), 280 deletions(-)
diff --git a/docs/docs/primary-key-table/chain-table.mdx
b/docs/docs/primary-key-table/chain-table.mdx
index 031c0a03e4..38e5be64de 100644
--- a/docs/docs/primary-key-table/chain-table.mdx
+++ b/docs/docs/primary-key-table/chain-table.mdx
@@ -46,6 +46,8 @@ Based on the regular table, chain table introduces snapshot
and delta branches t
data respectively. When writing, you specify the branch to write full or
incremental data. When reading, paimon
automatically chooses the appropriate strategy based on the read mode, such as
full, incremental, or hybrid.
+## Enable Chain Table
+
To enable chain table, you must config `chain-table.enabled` to true in the
table options when creating the
table, and the snapshot and delta branch need to be created as well.
@@ -129,29 +131,46 @@ ALTER TABLE `default`.`t$branch_delta` SET (
Notice that:
- Chain table is only supported for primary key table, which means you should
define `bucket` and `bucket-key` for the table.
- Chain table should ensure that the schema of each branch is consistent.
-- Deletion vector is not supported for chain table.
+- Deletion vector is only supported for chain table with the `DEDUPLICATE`
merge engine.
- The delta branch must use the `DEDUPLICATE` merge engine (default) if you
plan
to use streaming read or lookup join. Other merge engine types are not
supported
on the delta branch for these incremental read paths. Batch read is not
affected.
+- Chain table requires `sequence.field`, so the `FIRST_ROW` and `AGGREGATE`
merge engines are not supported.
+
+## Write Data
+
+After creating a chain table, you can write data to the snapshot or delta
branch.
-After creating a chain table, you can read and write data in the following
ways.
+### Full Write
+
+Write data to `t$branch_snapshot`.
-- Full Write: Write data to t$branch_snapshot.
```sql
insert overwrite `default`.`t$branch_snapshot` partition (date = '20250810')
values ('1', '1', '1');
```
-- Incremental Write: Write data to t$branch_delta.
+### Incremental Write
+
+Write data to `t$branch_delta`.
+
```sql
insert overwrite `default`.`t$branch_delta` partition (date = '20250811')
values ('2', '1', '1');
```
-- Full Query: If the snapshot branch has full partition, read it directly;
otherwise, read on chain merge mode.
+## Read Data
+
+You can query chain table in full, incremental, or hybrid mode.
+
+### Full Query
+
+If the snapshot branch has full partition, read it directly; otherwise, read
on chain merge mode.
+
```sql
select t1, t2, t3 from default.t where date = '20250811'
```
+
you will get the following result:
```text
+---+----+-----+
@@ -162,10 +181,14 @@ you will get the following result:
+---+----+-----+
```
-- Incremental Query: Read the incremental partition from t$branch_delta
+### Incremental Query
+
+Read the incremental partition from `t$branch_delta`.
+
```sql
select t1, t2, t3 from `default`.`t$branch_delta` where date = '20250811'
```
+
you will get the following result:
```text
+---+----+-----+
@@ -175,12 +198,16 @@ you will get the following result:
+---+----+-----+
```
-- Hybrid Query: Read both full and incremental data simultaneously.
+### Hybrid Query
+
+Read both full and incremental data simultaneously.
+
```sql
select t1, t2, t3 from default.t where date = '20250811'
union all
select t1, t2, t3 from `default`.`t$branch_delta` where date = '20250811'
```
+
you will get the following result:
```text
+---+----+-----+
@@ -192,9 +219,11 @@ you will get the following result:
+---+----+-----+
```
-- Chain Table Compaction: Merge data from snapshot and delta branches into the
snapshot branch.
- This is useful for periodically compacting incremental data into full
snapshots.
- You can use the `compact_chain_table` procedure to merge a specific
partition:
+## Chain Table Compaction
+
+Merge data from snapshot and delta branches into the snapshot branch.
+This is useful for periodically compacting incremental data into full
snapshots.
+You can use the `compact_chain_table` procedure to merge a specific partition:
```sql
CALL sys.compact_chain_table(table => 'default.t', partition =>
'date="20250811"');
diff --git
a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ExposeDeletionKeyValueReader.java
b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ExposeDeletionKeyValueReader.java
new file mode 100644
index 0000000000..d67ce96c00
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/ExposeDeletionKeyValueReader.java
@@ -0,0 +1,90 @@
+/*
+ * 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.deletionvectors;
+
+import org.apache.paimon.KeyValue;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.reader.FileRecordIterator;
+import org.apache.paimon.reader.FileRecordReader;
+import org.apache.paimon.types.RowKind;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * A {@link FileRecordReader} that is aware of a {@link DeletionVector}. For
each record returned by
+ * the wrapped reader, if the corresponding position is marked as deleted in
the deletion vector,
+ * the record's value kind is replaced with {@link RowKind#DELETE} so that
downstream merge
+ * functions see the deletion as a real tombstone record.
+ */
+public class ExposeDeletionKeyValueReader implements
FileRecordReader<KeyValue> {
+
+ private final FileRecordReader<KeyValue> reader;
+ private final DeletionVectorJudger deletionVector;
+
+ public ExposeDeletionKeyValueReader(
+ FileRecordReader<KeyValue> reader, DeletionVector deletionVector) {
+ this.reader = reader;
+ this.deletionVector = deletionVector;
+ }
+
+ @Nullable
+ @Override
+ public FileRecordIterator<KeyValue> readBatch() throws IOException {
+ FileRecordIterator<KeyValue> iterator = reader.readBatch();
+ if (iterator == null) {
+ return null;
+ }
+ return new FileRecordIterator<KeyValue>() {
+ @Override
+ public long returnedPosition() {
+ return iterator.returnedPosition();
+ }
+
+ @Override
+ public Path filePath() {
+ return iterator.filePath();
+ }
+
+ @Nullable
+ @Override
+ public KeyValue next() throws IOException {
+ KeyValue kv = iterator.next();
+ if (kv == null) {
+ return null;
+ }
+ if (deletionVector.isDeleted(returnedPosition())) {
+ kv.replaceValueKind(RowKind.DELETE);
+ }
+ return kv;
+ }
+
+ @Override
+ public void releaseBatch() {
+ iterator.releaseBatch();
+ }
+ };
+ }
+
+ @Override
+ public void close() throws IOException {
+ reader.close();
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/ChainKeyValueFileReaderFactory.java
b/paimon-core/src/main/java/org/apache/paimon/io/ChainKeyValueFileReaderFactory.java
index a06028b2e0..0b957bf6f2 100644
---
a/paimon-core/src/main/java/org/apache/paimon/io/ChainKeyValueFileReaderFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/io/ChainKeyValueFileReaderFactory.java
@@ -19,10 +19,14 @@
package org.apache.paimon.io;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.KeyValue;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.deletionvectors.DeletionVector;
+import org.apache.paimon.deletionvectors.ExposeDeletionKeyValueReader;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.reader.FileRecordReader;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.types.RowType;
@@ -30,9 +34,11 @@ import org.apache.paimon.utils.FormatReaderMapping;
import javax.annotation.Nullable;
+import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
/** A specific implementation about {@link KeyValueFileReaderFactory} for
chain read. */
public class ChainKeyValueFileReaderFactory extends KeyValueFileReaderFactory {
@@ -101,6 +107,28 @@ public class ChainKeyValueFileReaderFactory extends
KeyValueFileReaderFactory {
return chainReadContext.logicalPartition();
}
+ protected FileRecordReader<KeyValue> createRecordReader(
+ DataFileMeta file,
+ FileRecordReader<InternalRow> fileRecordReader,
+ boolean overrideSequenceWithSnapshotId)
+ throws IOException {
+ Optional<DeletionVector> deletionVector =
dvFactory.create(file.fileName());
+ KeyValueDataFileRecordReader reader =
+ new KeyValueDataFileRecordReader(
+ fileRecordReader,
+ keyType,
+ valueType,
+ file.level(),
+ overrideSequenceWithSnapshotId,
+ file.minSequenceNumber());
+
+ if (deletionVector.isPresent() && !deletionVector.get().isEmpty()) {
+ return new ExposeDeletionKeyValueReader(reader,
deletionVector.get());
+ }
+
+ return reader;
+ }
+
public static Builder newBuilder(KeyValueFileReaderFactory.Builder
wrapped) {
return new Builder(wrapped);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileRecordReader.java
b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileRecordReader.java
index 2538c08a21..1e4db70464 100644
---
a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileRecordReader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileRecordReader.java
@@ -39,11 +39,6 @@ public class KeyValueDataFileRecordReader implements
FileRecordReader<KeyValue>
private final boolean overrideSequenceWithSnapshotId;
private final long snapshotId;
- public KeyValueDataFileRecordReader(
- FileRecordReader<InternalRow> reader, RowType keyType, RowType
valueType, int level) {
- this(reader, keyType, valueType, level, false,
KeyValue.UNKNOWN_SEQUENCE);
- }
-
public KeyValueDataFileRecordReader(
FileRecordReader<InternalRow> reader,
RowType keyType,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java
b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java
index c594e4e082..20a6fe395c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileReaderFactory.java
@@ -62,8 +62,8 @@ public class KeyValueFileReaderFactory implements
FileReaderFactory<KeyValue> {
private final FileIO fileIO;
private final SchemaManager schemaManager;
private final TableSchema schema;
- private final RowType keyType;
- private final RowType valueType;
+ protected final RowType keyType;
+ protected final RowType valueType;
private final FormatReaderMapping.Builder formatReaderMappingBuilder;
private final DataFilePathFactory pathFactory;
@@ -73,7 +73,7 @@ public class KeyValueFileReaderFactory implements
FileReaderFactory<KeyValue> {
private final boolean snapshotSequenceOrdering;
private final Map<FormatKey, FormatReaderMapping> formatReaderMappings;
private final BinaryRow partition;
- private final DeletionVector.Factory dvFactory;
+ protected final DeletionVector.Factory dvFactory;
protected KeyValueFileReaderFactory(
FileIO fileIO,
@@ -127,6 +127,26 @@ public class KeyValueFileReaderFactory implements
FileReaderFactory<KeyValue> {
return createRecordReader(file, true, null);
}
+ protected FileRecordReader<KeyValue> createRecordReader(
+ DataFileMeta file,
+ FileRecordReader<InternalRow> fileRecordReader,
+ boolean overrideSequenceWithSnapshotId)
+ throws IOException {
+ Optional<DeletionVector> deletionVector =
dvFactory.create(file.fileName());
+ if (deletionVector.isPresent() && !deletionVector.get().isEmpty()) {
+ fileRecordReader =
+ new ApplyDeletionVectorReader(fileRecordReader,
deletionVector.get());
+ }
+
+ return new KeyValueDataFileRecordReader(
+ fileRecordReader,
+ keyType,
+ valueType,
+ file.level(),
+ overrideSequenceWithSnapshotId,
+ file.minSequenceNumber());
+ }
+
private FileRecordReader<KeyValue> createRecordReader(
DataFileMeta file, boolean reuseFormat, @Nullable Integer
orcPoolSize)
throws IOException {
@@ -166,12 +186,6 @@ public class KeyValueFileReaderFactory implements
FileReaderFactory<KeyValue> {
-1,
Collections.emptyMap());
- Optional<DeletionVector> deletionVector =
dvFactory.create(file.fileName());
- if (deletionVector.isPresent() && !deletionVector.get().isEmpty()) {
- fileRecordReader =
- new ApplyDeletionVectorReader(fileRecordReader,
deletionVector.get());
- }
-
// In snapshot-ordering mode, APPEND files carry the commit snapshot
id in
// minSequenceNumber (stamped at commit time); override per-record
sequence with it so
// later snapshots win during merge. COMPACT files already carry the
snapshot id in their
@@ -185,13 +199,8 @@ public class KeyValueFileReaderFactory implements
FileReaderFactory<KeyValue> {
+ "Legacy files without fileSource cannot be
ordered by commit snapshot id.");
overrideSequenceWithSnapshotId = file.fileSource().get() ==
FileSource.APPEND;
}
- return new KeyValueDataFileRecordReader(
- fileRecordReader,
- keyType,
- valueType,
- file.level(),
- overrideSequenceWithSnapshotId,
- file.minSequenceNumber());
+
+ return createRecordReader(file, fileRecordReader,
overrideSequenceWithSnapshotId);
}
public static Builder builder(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeFunction.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeFunction.java
index cb494b86d0..309d97a903 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeFunction.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeFunction.java
@@ -167,6 +167,10 @@ public class LookupMergeFunction implements
MergeFunction<KeyValue> {
this.ioManager = ioManager;
}
+ public MergeFunctionFactory<KeyValue> wrapped() {
+ return wrapped;
+ }
+
@Override
public MergeFunction<KeyValue> create(@Nullable RowType readType) {
return new LookupMergeFunction(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
index 50bb08cfc6..2714177959 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
@@ -293,7 +293,11 @@ public class MergeFileSplitRead implements
SplitRead<KeyValue> {
ChainKeyValueFileReaderFactory nonOverlappedSectionFactory =
builder.build(null, dvFactory, false, filtersForAll,
chainReadContext);
return createMergeReader(
- files, overlappedSectionFactory, nonOverlappedSectionFactory,
forceKeepDelete);
+ files,
+ overlappedSectionFactory,
+ nonOverlappedSectionFactory,
+ forceKeepDelete,
+ new
ReducerMergeFunctionWrapper(unwrapLookup(mfFactory).create(actualReadType())));
}
public RecordReader<KeyValue> createMergeReader(
@@ -311,18 +315,21 @@ public class MergeFileSplitRead implements
SplitRead<KeyValue> {
KeyValueFileReaderFactory nonOverlappedSectionFactory =
readerFactoryBuilder.build(partition, bucket, dvFactory,
false, filtersForAll);
return createMergeReader(
- files, overlappedSectionFactory, nonOverlappedSectionFactory,
keepDelete);
+ files,
+ overlappedSectionFactory,
+ nonOverlappedSectionFactory,
+ keepDelete,
+ new
ReducerMergeFunctionWrapper(mfFactory.create(actualReadType())));
}
public RecordReader<KeyValue> createMergeReader(
List<DataFileMeta> files,
KeyValueFileReaderFactory overlappedSectionFactory,
KeyValueFileReaderFactory nonOverlappedSectionFactory,
- boolean keepDelete)
+ boolean keepDelete,
+ MergeFunctionWrapper<KeyValue> mergeFuncWrapper)
throws IOException {
List<ReaderSupplier<KeyValue>> sectionReaders = new ArrayList<>();
- MergeFunctionWrapper<KeyValue> mergeFuncWrapper =
- new
ReducerMergeFunctionWrapper(mfFactory.create(actualReadType()));
for (List<SortedRun> section : new IntervalPartition(files,
keyComparator).partition()) {
sectionReaders.add(
() ->
@@ -529,4 +536,11 @@ public class MergeFileSplitRead implements
SplitRead<KeyValue> {
public UserDefinedSeqComparator createUdsComparator() {
return UserDefinedSeqComparator.create(actualReadType(),
sequenceFields, sequenceOrder);
}
+
+ private static MergeFunctionFactory<KeyValue> unwrapLookup(
+ MergeFunctionFactory<KeyValue> factory) {
+ return factory instanceof LookupMergeFunction.Factory
+ ? ((LookupMergeFunction.Factory) factory).wrapped()
+ : factory;
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index fc1c539455..8c058def52 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -1513,8 +1513,10 @@ public class SchemaValidation {
|| changelogProducer == ChangelogProducer.INPUT,
"Changelog producer must be none or input for chain
table.");
Preconditions.checkArgument(
- !options.deletionVectorsEnabled(),
- "Chain table do not support enable deletion vector");
+ !options.deletionVectorsEnabled()
+ || options.deletionVectorsEnabled()
+ && options.mergeEngine() ==
MergeEngine.DEDUPLICATE,
+ "Chain tables only support deletion vectors with the
deduplicate merge engine.");
Preconditions.checkArgument(
options.partitionTimestampPattern() != null,
"Partition timestamp pattern is required for chain
table.");
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
index cf3c32be27..cb6541662a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
@@ -24,7 +24,6 @@ import org.apache.paimon.codegen.RecordComparator;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
-import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
@@ -520,18 +519,7 @@ public class ChainGroupReadTable extends
FallbackReadFileStoreTable {
for (Split split : mainScan.plan().splits()) {
DataSplit dataSplit = (DataSplit) split;
- HashMap<String, String> fileBucketPathMapping = new
HashMap<>();
- HashMap<String, String> fileBranchMapping = new HashMap<>();
- for (DataFileMeta file : dataSplit.dataFiles()) {
- fileBucketPathMapping.put(file.fileName(), ((DataSplit)
split).bucketPath());
- fileBranchMapping.put(file.fileName(),
options.scanFallbackSnapshotBranch());
- }
- splits.add(
- new ChainSplit(
- dataSplit.partition(),
- dataSplit.dataFiles(),
- fileBranchMapping,
- fileBucketPathMapping));
+ splits.add(ChainSplit.from(dataSplit,
options.scanFallbackSnapshotBranch()));
}
snapshotPartitions.addAll(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/ChainSplit.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/ChainSplit.java
index d296b099da..bad2dc1c7b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/ChainSplit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/ChainSplit.java
@@ -27,6 +27,8 @@ import org.apache.paimon.io.DataOutputView;
import org.apache.paimon.io.DataOutputViewStreamWrapper;
import org.apache.paimon.utils.SerializationUtils;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
@@ -35,6 +37,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.OptionalLong;
/**
@@ -45,22 +48,28 @@ public class ChainSplit implements Split {
private static final long serialVersionUID = 1L;
- private static final int VERSION = 1;
+ private static final int VERSION_1 = 1;
+ private static final int VERSION = 2;
private BinaryRow logicalPartition;
private List<DataFileMeta> dataFiles;
private Map<String, String> fileBranchMapping;
private Map<String, String> fileBucketPathMapping;
+ /** Deletion files corresponding to {@link #dataFiles}, in the same order.
*/
+ @Nullable private List<DeletionFile> dataDeletionFiles;
+
public ChainSplit(
BinaryRow logicalPartition,
List<DataFileMeta> dataFiles,
Map<String, String> fileBranchMapping,
- Map<String, String> fileBucketPathMapping) {
+ Map<String, String> fileBucketPathMapping,
+ @Nullable List<DeletionFile> dataDeletionFiles) {
this.logicalPartition = logicalPartition;
this.dataFiles = dataFiles;
this.fileBranchMapping = fileBranchMapping;
this.fileBucketPathMapping = fileBucketPathMapping;
+ this.dataDeletionFiles = dataDeletionFiles;
}
public BinaryRow logicalPartition() {
@@ -94,7 +103,13 @@ public class ChainSplit implements Split {
dataSplit.partition(),
dataSplit.dataFiles(),
fileBranchMapping,
- fileBucketPathMapping);
+ fileBucketPathMapping,
+ dataSplit.deletionFiles().orElse(null));
+ }
+
+ @Override
+ public Optional<List<DeletionFile>> deletionFiles() {
+ return Optional.ofNullable(dataDeletionFiles);
}
@Override
@@ -108,6 +123,10 @@ public class ChainSplit implements Split {
@Override
public OptionalLong mergedRowCount() {
+ // Chain tables merge and deduplicate rows across snapshot and delta
files by primary key,
+ // so the post-merge row count cannot be determined from per-file
metadata alone.
+ // Returning empty prevents LIMIT pushdown from discarding splits
based on an inaccurate
+ // estimate.
return OptionalLong.empty();
}
@@ -158,6 +177,7 @@ public class ChainSplit implements Split {
this.dataFiles = other.dataFiles;
this.fileBranchMapping = other.fileBranchMapping;
this.fileBucketPathMapping = other.fileBucketPathMapping;
+ this.dataDeletionFiles = other.dataDeletionFiles;
}
public void serialize(DataOutputView out) throws IOException {
@@ -184,13 +204,14 @@ public class ChainSplit implements Split {
out.writeUTF(entry.getKey());
out.writeUTF(entry.getValue());
}
+
+ // Serialize deletionFiles
+ DeletionFile.serializeList(out, dataDeletionFiles);
}
public static ChainSplit deserialize(DataInputView in) throws IOException {
int version = in.readInt();
- if (version != VERSION) {
- throw new UnsupportedOperationException("Unsupported version: " +
version);
- }
+ SerializationUtils.checkVersion(version, VERSION_1, VERSION,
"ChainSplit");
BinaryRow logicalPartition =
SerializationUtils.deserializeBinaryRow(in);
@@ -216,7 +237,17 @@ public class ChainSplit implements Split {
fileBranchMapping.put(key, value);
}
+ // Deserialize deletionFiles (only for version > 1)
+ List<DeletionFile> deletionFiles = null;
+ if (version > VERSION_1) {
+ deletionFiles = DeletionFile.deserializeList(in,
DeletionFile::deserialize);
+ }
+
return new ChainSplit(
- logicalPartition, dataFiles, fileBranchMapping,
fileBucketPathMapping);
+ logicalPartition,
+ dataFiles,
+ fileBranchMapping,
+ fileBucketPathMapping,
+ deletionFiles);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/SplitSerializer.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/SplitSerializer.java
index d1f8ad6253..070956b811 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/SplitSerializer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/SplitSerializer.java
@@ -90,7 +90,7 @@ public class SplitSerializer {
((IndexedSplit) split).serialize(out);
} else if (split instanceof ChainSplit) {
out.writeInt(CHAIN_SPLIT);
- writeChainSplit((ChainSplit) split, out);
+ ((ChainSplit) split).serialize(out);
} else if (split instanceof IncrementalSplit) {
out.writeInt(INCREMENTAL_SPLIT);
writeIncrementalSplit((IncrementalSplit) split, out);
@@ -126,7 +126,7 @@ public class SplitSerializer {
case INDEXED_SPLIT:
return IndexedSplit.deserialize(in);
case CHAIN_SPLIT:
- return readChainSplit(in);
+ return ChainSplit.deserialize(in);
case QUERY_AUTH_SPLIT:
return readQueryAuthSplit(in);
case FALLBACK_DATA_SPLIT:
@@ -177,22 +177,6 @@ public class SplitSerializer {
isStreaming);
}
- private static void writeChainSplit(ChainSplit split, DataOutputView out)
throws IOException {
- serializeBinaryRow(split.logicalPartition(), out);
- writeDataFiles(split.dataFiles(), out);
- writeStringMap(out, split.fileBucketPathMapping());
- writeStringMap(out, split.fileBranchMapping());
- }
-
- private static ChainSplit readChainSplit(DataInputView in) throws
IOException {
- BinaryRow logicalPartition = deserializeBinaryRow(in);
- List<DataFileMeta> dataFiles = readDataFiles(in);
- Map<String, String> fileBucketPathMapping = readStringMap(in);
- Map<String, String> fileBranchMapping = readStringMap(in);
- return new ChainSplit(
- logicalPartition, dataFiles, fileBranchMapping,
fileBucketPathMapping);
- }
-
private static void writeQueryAuthSplit(QueryAuthSplit split,
DataOutputView out)
throws IOException {
serialize(split.split(), out);
@@ -301,11 +285,6 @@ public class SplitSerializer {
}
}
- private static Map<String, String> readStringMap(DataInputView in) throws
IOException {
- Map<String, String> map = readNullableStringMap(in);
- return map == null ? new HashMap<>() : map;
- }
-
@Nullable
private static Map<String, String> readNullableStringMap(DataInputView in)
throws IOException {
if (!in.readBoolean()) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
b/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
index 35659f06d8..4a7bb1b073 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
@@ -34,6 +34,7 @@ import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.ChainSplit;
import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.types.RowType;
import javax.annotation.Nullable;
@@ -47,6 +48,7 @@ import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.function.BiFunction;
import java.util.function.Function;
@@ -462,15 +464,21 @@ public class ChainTableUtils {
for (Map.Entry<Integer, List<DataSplit>> entry :
bucketSplits.entrySet()) {
Map<String, String> fileBranchMapping = new HashMap<>();
Map<String, String> fileBucketPathMapping = new HashMap<>();
+ Map<String, DeletionFile> fileDeletionMapping = new HashMap<>();
List<DataFileMeta> bucketFiles = new ArrayList<>();
- for (DataSplit ds : entry.getValue()) {
- for (DataFileMeta file : ds.dataFiles()) {
- fileBucketPathMapping.put(file.fileName(),
ds.bucketPath());
+ for (DataSplit dataSplit : entry.getValue()) {
+ Optional<List<DeletionFile>> deletionFilesOpt =
dataSplit.deletionFiles();
+ for (int i = 0; i < dataSplit.dataFiles().size(); i++) {
+ DataFileMeta file = dataSplit.dataFiles().get(i);
+ DeletionFile deletionFile =
+ deletionFilesOpt.isPresent() ?
deletionFilesOpt.get().get(i) : null;
+ fileBucketPathMapping.put(file.fileName(),
dataSplit.bucketPath());
String branch =
snapshotFileNames.contains(file.fileName())
? snapshotBranch
: deltaBranch;
fileBranchMapping.put(file.fileName(), branch);
+ fileDeletionMapping.put(file.fileName(), deletionFile);
bucketFiles.add(file);
}
}
@@ -490,7 +498,8 @@ public class ChainTableUtils {
logicalPartition,
groupFiles,
subMapping(fileBranchMapping, groupFiles),
- subMapping(fileBucketPathMapping,
groupFiles)));
+ subMapping(fileBucketPathMapping, groupFiles),
+ subDeletionFiles(fileDeletionMapping,
groupFiles)));
}
}
return result;
@@ -547,6 +556,20 @@ public class ChainTableUtils {
return sub;
}
+ private static List<DeletionFile> subDeletionFiles(
+ Map<String, DeletionFile> mapping, List<DataFileMeta> files) {
+ boolean hasDeletionFile = false;
+ List<DeletionFile> result = new ArrayList<>(files.size());
+ for (DataFileMeta file : files) {
+ DeletionFile deletionFile = mapping.get(file.fileName());
+ if (deletionFile != null) {
+ hasDeletionFile = true;
+ }
+ result.add(deletionFile);
+ }
+ return hasDeletionFile ? result : null;
+ }
+
private static Integer addToBucketMap(
DataSplit ds, Map<Integer, List<DataSplit>> bucketSplits, Integer
bucketInAll) {
Integer totalBuckets = ds.totalBuckets();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/SerializationUtils.java
b/paimon-core/src/main/java/org/apache/paimon/utils/SerializationUtils.java
index f4b43c5195..9ec9b6486f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/SerializationUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/SerializationUtils.java
@@ -103,4 +103,23 @@ public class SerializationUtils {
public static BinaryRow deserializeBinaryRow(DataInputView input) throws
IOException {
return deserializeBinaryRow(deserializedBytes(input));
}
+
+ /**
+ * Checks that the given version is within the supported range.
+ *
+ * @param version the version read from serialized bytes
+ * @param minVersion the minimum supported version (inclusive)
+ * @param maxVersion the maximum supported version (inclusive)
+ * @param context the context name included in the error message
+ * @throws IOException if the version is out of range
+ */
+ public static void checkVersion(int version, int minVersion, int
maxVersion, String context)
+ throws IOException {
+ if (version < minVersion || version > maxVersion) {
+ throw new IOException(
+ String.format(
+ "Unsupported %s version: %d, expected between %d
and %d",
+ context, version, minVersion, maxVersion));
+ }
+ }
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
new file mode 100644
index 0000000000..f0027ab440
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java
@@ -0,0 +1,511 @@
+/*
+ * 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;
+
+import org.apache.paimon.CoreOptions;
+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.IOManagerImpl;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.InnerTableWrite;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.source.ChainSplit;
+import org.apache.paimon.table.source.DeletionFile;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
+
+import static org.apache.paimon.CoreOptions.BUCKET;
+import static org.apache.paimon.CoreOptions.BUCKET_KEY;
+import static org.apache.paimon.CoreOptions.CHAIN_TABLE_CHAIN_PARTITION_KEYS;
+import static org.apache.paimon.CoreOptions.CHAIN_TABLE_ENABLED;
+import static org.apache.paimon.CoreOptions.CHANGELOG_PRODUCER;
+import static org.apache.paimon.CoreOptions.DELETION_VECTORS_ENABLED;
+import static org.apache.paimon.CoreOptions.MERGE_ENGINE;
+import static
org.apache.paimon.CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE;
+import static org.apache.paimon.CoreOptions.PARTITION_TIMESTAMP_FORMATTER;
+import static org.apache.paimon.CoreOptions.PARTITION_TIMESTAMP_PATTERN;
+import static org.apache.paimon.CoreOptions.PATH;
+import static org.apache.paimon.CoreOptions.SEQUENCE_FIELD;
+import static org.apache.paimon.catalog.Identifier.DEFAULT_MAIN_BRANCH;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link ChainTableFileStoreTable}. */
+public class ChainTableFileStoreTableTest {
+ @TempDir java.nio.file.Path tempDir;
+
+ private static final String SNAPSHOT_BRANCH = "snapshot";
+ private static final String DELTA_BRANCH = "delta";
+
+ private String commitUser;
+ private String tableName;
+
+ @BeforeEach
+ public void beforeEach() {
+ String uuid = UUID.randomUUID().toString();
+ commitUser = uuid;
+ tableName = "chain_t_" + uuid.replace("-", "");
+ }
+
+ @Test
+ public void testChainTableRejectsAggregate() throws Exception {
+ assertThatThrownBy(
+ () ->
+ createChainTable(
+ options -> {
+ options.set(
+ MERGE_ENGINE,
+
CoreOptions.MergeEngine.AGGREGATE);
+
options.set("fields.seq.aggregate-function", "max");
+
options.set("fields.v.aggregate-function", "min");
+ }))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Should not define aggregation on sequence field:
'seq'.");
+ }
+
+ @Test
+ public void testChainTableRejectsFirstRow() throws Exception {
+ assertThatThrownBy(
+ () ->
+ createChainTable(
+ options -> {
+ options.set(
+ MERGE_ENGINE,
+
CoreOptions.MergeEngine.FIRST_ROW);
+ }))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Do not support use sequence field on FIRST_ROW
merge engine.");
+ }
+
+ @Test
+ public void testChainTableWithPartialUpdateDelete() throws Exception {
+ createPartialUpdateChainTable(
+ options -> {
+ options.set(PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE, true);
+ });
+ FileStoreTable chainTable = loadTable();
+ FileStoreTable snapshotTable =
chainTable.switchToBranch(SNAPSHOT_BRANCH);
+ FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH);
+ writeWithCommit(
+ snapshotTable,
+ row(1L, 1L, "a", null, "20250810"),
+ row(2L, 1L, null, "B", "20250810"));
+
+ writeWithCommit(
+ deltaTable,
+ row(1L, 2L, null, "A1", "20250811"),
+ row(2L, 2L, "b1", null, "20250811"),
+ row(5L, 1L, "e", "E", "20250811"));
+
+ writeWithCommit(
+ deltaTable,
+ row(RowKind.DELETE, 1L, 3L, null, null, "20250811"),
+ row(RowKind.UPDATE_BEFORE, 2L, 3L, "b1", "B", "20250811"),
+ row(RowKind.UPDATE_AFTER, 2L, 4L, "b2", "B", "20250812"),
+ row(RowKind.DELETE, 5L, 2L, null, null, "20250812"));
+
+ assertThat(getResult(loadTable(), ImmutableMap.of("dt", "20250811")))
+ .containsExactlyInAnyOrder(
+ row(2L, 2L, "b1", "B", "20250811"), row(5L, 1L, "e",
"E", "20250811"));
+
+ assertThat(getResult(loadTable(), ImmutableMap.of("dt", "20250812")))
+ .containsExactlyInAnyOrder(row(2L, 4L, "b2", "B", "20250812"));
+ }
+
+ @Test
+ public void testChainTableRejectsPartialUpdateWithDeletionVectors() throws
Exception {
+ assertThatThrownBy(
+ () ->
+ createPartialUpdateChainTable(
+ options -> {
+
options.set(DELETION_VECTORS_ENABLED, true);
+ options.set(
+
PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE, true);
+ }))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Chain tables only support deletion vectors with the
deduplicate merge engine.");
+ }
+
+ private static Stream<Arguments> dvChangelogParams() {
+ return Stream.of(
+ Arguments.of(false, CoreOptions.ChangelogProducer.NONE),
+ Arguments.of(true, CoreOptions.ChangelogProducer.NONE),
+ Arguments.of(false, CoreOptions.ChangelogProducer.INPUT),
+ Arguments.of(true, CoreOptions.ChangelogProducer.INPUT));
+ }
+
+ @ParameterizedTest
+ @MethodSource("dvChangelogParams")
+ public void testChainTableWithDelete(
+ boolean deletionVectors, CoreOptions.ChangelogProducer changelog)
throws Exception {
+ createChainTable(
+ options -> {
+ options.set(DELETION_VECTORS_ENABLED, deletionVectors);
+ options.set(CHANGELOG_PRODUCER, changelog);
+ });
+ FileStoreTable chainTable = loadTable();
+ FileStoreTable snapshotTable =
chainTable.switchToBranch(SNAPSHOT_BRANCH);
+ writeWithCommit(
+ snapshotTable,
+ row(1L, 1L, "1", "CN", "20250810", "20"),
+ row(2L, 1L, "2", "CN", "20250810", "20"),
+ row(3L, 1L, "3", "US", "20250810", "20"));
+ writeWithCommit(snapshotTable, row(RowKind.DELETE, 2L, 1L, "2", "CN",
"20250810", "20"));
+
+ assertThat(getResult(snapshotTable, ImmutableMap.of("dt", "20250810",
"hour", "20")))
+ .containsExactlyInAnyOrder(
+ row(1L, 1L, "1", "CN", "20250810", "20"),
+ row(3L, 1L, "3", "US", "20250810", "20"));
+
+ FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH);
+ writeWithCommit(
+ deltaTable,
+ row(1L, 2L, "1-1", "CN", "20250810", "21"),
+ row(4L, 1L, "4", "CN", "20250810", "21"),
+ row(5L, 1L, "5", "US", "20250810", "21"),
+ row(6L, 1L, "6", "UK", "20250810", "21"));
+ writeWithCommit(deltaTable, row(RowKind.DELETE, 4L, 2L, "4", "CN",
"20250810", "21"));
+
+ assertThat(getResult(deltaTable, ImmutableMap.of("dt", "20250810",
"hour", "21")))
+ .containsExactlyInAnyOrder(
+ row(1L, 2L, "1-1", "CN", "20250810", "21"),
+ row(5L, 1L, "5", "US", "20250810", "21"),
+ row(6L, 1L, "6", "UK", "20250810", "21"));
+
+ assertThat(getResult(loadTable(), ImmutableMap.of("dt", "20250810",
"hour", "21")))
+ .containsExactlyInAnyOrder(
+ row(1L, 2L, "1-1", "CN", "20250810", "21"),
+ row(3L, 1L, "3", "US", "20250810", "21"),
+ row(5L, 1L, "5", "US", "20250810", "21"),
+ row(6L, 1L, "6", "UK", "20250810", "21"));
+
+ assertChainSplitCarriesDeletionFiles(
+ deletionVectors, ImmutableMap.of("region", "CN", "dt",
"20250810", "hour", "21"));
+
+ // test cross partition delete
+ writeWithCommit(deltaTable, row(RowKind.DELETE, 5L, 2L, "5", "US",
"20250810", "22"));
+ assertThat(getResult(loadTable(), ImmutableMap.of("dt", "20250810",
"hour", "22")))
+ .containsExactlyInAnyOrder(row(3L, 1L, "3", "US", "20250810",
"22"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("dvChangelogParams")
+ public void testChainTableWithUpdate(
+ boolean deletionVectors, CoreOptions.ChangelogProducer changelog)
throws Exception {
+ createChainTable(
+ options -> {
+ options.set(DELETION_VECTORS_ENABLED, deletionVectors);
+ options.set(CHANGELOG_PRODUCER, changelog);
+ });
+ FileStoreTable chainTable = loadTable();
+ FileStoreTable snapshotTable =
chainTable.switchToBranch(SNAPSHOT_BRANCH);
+ writeWithCommit(
+ snapshotTable,
+ row(1L, 1L, "1", "CN", "20250810", "20"),
+ row(2L, 1L, "2", "CN", "20250810", "20"));
+ writeWithCommit(
+ snapshotTable,
+ row(RowKind.UPDATE_BEFORE, 2L, 1L, "2", "CN", "20250810",
"20"),
+ row(RowKind.UPDATE_AFTER, 2L, 2L, "2-1", "CN", "20250810",
"20"));
+
+ FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH);
+ writeWithCommit(
+ deltaTable,
+ row(1L, 2L, "1-1", "CN", "20250811", "20"),
+ row(3L, 1L, "3", "CN", "20250811", "20"));
+ writeWithCommit(
+ deltaTable,
+ row(RowKind.UPDATE_BEFORE, 3L, 1L, "3", "CN", "20250811",
"20"),
+ row(RowKind.UPDATE_AFTER, 3L, 2L, "3-1", "CN", "20250811",
"20"));
+
+ assertThat(getResult(loadTable(), ImmutableMap.of("dt", "20250811",
"hour", "20")))
+ .containsExactlyInAnyOrder(
+ row(RowKind.INSERT, 1L, 2L, "1-1", "CN", "20250811",
"20"),
+ row(RowKind.UPDATE_AFTER, 2L, 2L, "2-1", "CN",
"20250811", "20"),
+ row(RowKind.UPDATE_AFTER, 3L, 2L, "3-1", "CN",
"20250811", "20"));
+
+ assertChainSplitCarriesDeletionFiles(
+ deletionVectors, ImmutableMap.of("region", "CN", "dt",
"20250811", "hour", "20"));
+
+ // test cross partition update
+ writeWithCommit(
+ deltaTable, row(RowKind.UPDATE_BEFORE, 1L, 3L, "1-1", "CN",
"20250811", "22"));
+ writeWithCommit(
+ deltaTable, row(RowKind.UPDATE_AFTER, 1L, 4L, "1-2", "CN",
"20250811", "21"));
+ assertThat(getResult(loadTable(), ImmutableMap.of("dt", "20250811",
"hour", "22")))
+ .containsExactlyInAnyOrder(
+ row(RowKind.UPDATE_AFTER, 1L, 4L, "1-2", "CN",
"20250811", "22"),
+ row(RowKind.UPDATE_AFTER, 2L, 2L, "2-1", "CN",
"20250811", "22"),
+ row(RowKind.UPDATE_AFTER, 3L, 2L, "3-1", "CN",
"20250811", "22"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("dvChangelogParams")
+ public void testChainTableCrossPartitionDeletionVectors(
+ boolean deletionVectors, CoreOptions.ChangelogProducer changelog)
throws Exception {
+ createChainTable(
+ options -> {
+ options.set(DELETION_VECTORS_ENABLED, deletionVectors);
+ options.set(CHANGELOG_PRODUCER, changelog);
+ });
+ FileStoreTable chainTable = loadTable();
+ FileStoreTable snapshotTable =
chainTable.switchToBranch(SNAPSHOT_BRANCH);
+ writeWithCommit(snapshotTable, row(1L, 1L, "1-1", "CN", "20250811",
"20"));
+
+ FileStoreTable deltaTable = chainTable.switchToBranch(DELTA_BRANCH);
+ writeWithCommit(
+ deltaTable,
+ row(1L, 2L, "1-2", "CN", "20250811", "21"),
+ row(2L, 1L, "2-1", "CN", "20250811", "21"));
+ writeWithCommit(deltaTable, row(RowKind.DELETE, 1L, 3L, null, "CN",
"20250811", "21"));
+
+ assertThat(getResult(loadTable(), ImmutableMap.of("dt", "20250811",
"hour", "21")))
+ .containsExactlyInAnyOrder(row(2L, 1L, "2-1", "CN",
"20250811", "21"));
+ assertChainSplitCarriesDeletionFiles(
+ deletionVectors, ImmutableMap.of("region", "CN", "dt",
"20250811", "hour", "21"));
+
+ writeWithCommit(
+ deltaTable, row(RowKind.UPDATE_BEFORE, 2L, 2L, "2-1", "CN",
"20250811", "22"));
+ assertThat(getResult(loadTable(), ImmutableMap.of("dt", "20250811",
"hour", "22")))
+ .isEmpty();
+ assertChainSplitCarriesDeletionFiles(
+ deletionVectors, ImmutableMap.of("region", "CN", "dt",
"20250811", "hour", "22"));
+ }
+
+ private void assertChainSplitCarriesDeletionFiles(
+ boolean deletionVectors, Map<String, String> partitionFilter) {
+ if (!deletionVectors) {
+ return;
+ }
+ TableScan.Plan plan =
+
loadTable().newReadBuilder().withPartitionFilter(partitionFilter).newScan().plan();
+
+ boolean foundChainSplit = false;
+ for (Split split : plan.splits()) {
+ if (split instanceof FallbackReadFileStoreTable.FallbackSplit) {
+ foundChainSplit = true;
+ ChainSplit chainSplit =
+ (ChainSplit)
((FallbackReadFileStoreTable.FallbackSplit) split).wrapped();
+ Optional<List<DeletionFile>> deletionFilesOpt =
chainSplit.deletionFiles();
+ assertThat(deletionFilesOpt).isPresent();
+ long nonNullDeletionFiles =
+
deletionFilesOpt.get().stream().filter(Objects::nonNull).count();
+ assertThat(nonNullDeletionFiles)
+ .as("ChainSplit should carry deletion files from both
branches")
+ .isGreaterThan(0);
+ }
+ }
+ assertThat(foundChainSplit).as("Should have found at least one
ChainSplit").isTrue();
+ }
+
+ private FileStoreTable loadTable() {
+ Path tablePath = new Path(tempDir.toUri().toString(), tableName);
+ LocalFileIO fileIO = LocalFileIO.create();
+ Options options = new Options();
+ options.set(CoreOptions.PATH, tablePath.toString());
+ String branchName = CoreOptions.branch(options.toMap());
+ Optional<TableSchema> schemaOpt = new SchemaManager(fileIO, tablePath,
branchName).latest();
+ assertThat(schemaOpt.isPresent()).isTrue();
+ return FileStoreTableFactory.create(
+ fileIO, tablePath, schemaOpt.get(),
CatalogEnvironment.empty());
+ }
+
+ private void createChainTable(Consumer<Options> optionCustomizer) throws
Exception {
+ Path tablePath = new Path(tempDir.toUri().toString(), tableName);
+ LocalFileIO fileIO = LocalFileIO.create();
+ SchemaManager mainSchemaManager = new SchemaManager(fileIO, tablePath);
+
+ Options options = new Options();
+ options.set(BUCKET, 1);
+ options.set(BUCKET_KEY, "k");
+ options.set(SEQUENCE_FIELD, "seq");
+ options.set(MERGE_ENGINE, CoreOptions.MergeEngine.DEDUPLICATE);
+ options.set(CHAIN_TABLE_ENABLED, true);
+ options.set(PARTITION_TIMESTAMP_PATTERN, "$dt $hour:00:00");
+ options.set(PARTITION_TIMESTAMP_FORMATTER, "yyyyMMdd HH:mm:ss");
+ options.set(CHAIN_TABLE_CHAIN_PARTITION_KEYS, "dt,hour");
+ options.set(PATH, tablePath.toString());
+ optionCustomizer.accept(options);
+
+ Schema schema =
+ new Schema(
+ RowType.of(
+ new org.apache.paimon.types.DataType[]
{
+ DataTypes.BIGINT(),
+ DataTypes.BIGINT(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"k", "seq", "v",
"region", "dt", "hour"})
+ .getFields(),
+ Arrays.asList("region", "dt", "hour"),
+ Arrays.asList("region", "dt", "hour", "k"),
+ options.toMap(),
+ "");
+
+ mainSchemaManager.createTable(schema);
+ FileStoreTable table = loadTable();
+
+ // chain table setup procedure.
+ table.createBranch(SNAPSHOT_BRANCH);
+ table.createBranch(DELTA_BRANCH);
+
+ configureBranchOptions(fileIO, tablePath, DEFAULT_MAIN_BRANCH);
+ configureBranchOptions(fileIO, tablePath, SNAPSHOT_BRANCH);
+ configureBranchOptions(fileIO, tablePath, DELTA_BRANCH);
+
+ Optional<TableSchema> schemaOpt = mainSchemaManager.latest();
+ assertThat(schemaOpt.isPresent()).isTrue();
+ }
+
+ private void createPartialUpdateChainTable(Consumer<Options>
optionCustomizer)
+ throws Exception {
+ Path tablePath = new Path(tempDir.toUri().toString(), tableName);
+ LocalFileIO fileIO = LocalFileIO.create();
+ SchemaManager mainSchemaManager = new SchemaManager(fileIO, tablePath);
+
+ Options options = new Options();
+ options.set(BUCKET, 1);
+ options.set(BUCKET_KEY, "k");
+ options.set(SEQUENCE_FIELD, "seq");
+ options.set(MERGE_ENGINE, CoreOptions.MergeEngine.PARTIAL_UPDATE);
+ options.set(CHAIN_TABLE_ENABLED, true);
+ options.set(PARTITION_TIMESTAMP_PATTERN, "$dt");
+ options.set(PARTITION_TIMESTAMP_FORMATTER, "yyyyMMdd");
+ options.set(CHAIN_TABLE_CHAIN_PARTITION_KEYS, "dt");
+ options.set(PATH, tablePath.toString());
+ optionCustomizer.accept(options);
+
+ Schema schema =
+ new Schema(
+ RowType.of(
+ new org.apache.paimon.types.DataType[]
{
+ DataTypes.BIGINT(),
+ DataTypes.BIGINT(),
+ DataTypes.STRING(),
+ DataTypes.STRING(),
+ DataTypes.STRING()
+ },
+ new String[] {"k", "seq", "v1", "v2",
"dt"})
+ .getFields(),
+ Collections.singletonList("dt"),
+ Arrays.asList("dt", "k"),
+ options.toMap(),
+ "");
+
+ mainSchemaManager.createTable(schema);
+ FileStoreTable table = loadTable();
+
+ table.createBranch(SNAPSHOT_BRANCH);
+ table.createBranch(DELTA_BRANCH);
+
+ configureBranchOptions(fileIO, tablePath, DEFAULT_MAIN_BRANCH);
+ configureBranchOptions(fileIO, tablePath, SNAPSHOT_BRANCH);
+ configureBranchOptions(fileIO, tablePath, DELTA_BRANCH);
+
+ Optional<TableSchema> schemaOpt = mainSchemaManager.latest();
+ assertThat(schemaOpt.isPresent()).isTrue();
+ }
+
+ private void configureBranchOptions(LocalFileIO fileIO, Path tablePath,
String branchName)
+ throws Exception {
+ SchemaManager branchSchemaManager = new SchemaManager(fileIO,
tablePath, branchName);
+ branchSchemaManager.commitChanges(
+ SchemaChange.setOption(
+ CoreOptions.SCAN_FALLBACK_SNAPSHOT_BRANCH.key(),
SNAPSHOT_BRANCH),
+
SchemaChange.setOption(CoreOptions.SCAN_FALLBACK_DELTA_BRANCH.key(),
DELTA_BRANCH));
+ }
+
+ private GenericRow row(RowKind kind, Object... values) {
+ for (int i = 0; i < values.length; i++) {
+ if (values[i] != null && values[i] instanceof String) {
+ values[i] = BinaryString.fromString((String) values[i]);
+ }
+ }
+ return GenericRow.ofKind(kind, values);
+ }
+
+ private GenericRow row(Object... values) {
+ return row(RowKind.INSERT, values);
+ }
+
+ private List<GenericRow> getResult(FileStoreTable table, Map<String,
String> partitionFilter)
+ throws Exception {
+ ReadBuilder readBuilder = table.newReadBuilder();
+ TableScan.Plan plan =
readBuilder.withPartitionFilter(partitionFilter).newScan().plan();
+ List<GenericRow> result = new ArrayList<>();
+ RowType rowType = table.schema().logicalRowType();
+ InternalRowSerializer serializer = new InternalRowSerializer(rowType);
+
+ try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(plan)) {
+ reader.forEachRemaining(row -> result.add((GenericRow)
serializer.copy(row)));
+ }
+ return result;
+ }
+
+ private void writeWithCommit(FileStoreTable table, GenericRow... rows)
throws Exception {
+ try (InnerTableWrite write =
+ table.newWrite(commitUser).withIOManager(new
IOManagerImpl(tempDir.toString()))) {
+ for (GenericRow r : rows) {
+ write.write(r);
+ }
+ try (StreamTableCommit commit = table.newCommit(commitUser)) {
+ List<CommitMessage> messages = write.prepareCommit(true, 0);
+ commit.commit(0, messages);
+ }
+ }
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/ChainSplitTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/ChainSplitTest.java
index e281d7557f..402ac581bb 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/ChainSplitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/ChainSplitTest.java
@@ -29,6 +29,7 @@ import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -56,12 +57,55 @@ public class ChainSplitTest {
}
ChainSplit split =
new ChainSplit(
- logicalPartition, dataFiles, fileBranchMapping,
fileBucketPathMapping);
+ logicalPartition,
+ dataFiles,
+ fileBranchMapping,
+ fileBucketPathMapping,
+ null);
byte[] bytes = InstantiationUtil.serializeObject(split);
ChainSplit newSplit =
InstantiationUtil.deserializeObject(bytes,
ChainSplit.class.getClassLoader());
assertThat(fileBucketPathMapping).isEqualTo(newSplit.fileBucketPathMapping());
assertThat(fileBranchMapping).isEqualTo(newSplit.fileBranchMapping());
+ assertThat(newSplit.deletionFiles()).isEmpty();
+ assertThat(newSplit).isEqualTo(split);
+ }
+
+ @Test
+ public void testChainSplitSerdeWithDeletionFiles() throws IOException,
ClassNotFoundException {
+ BinaryRow logicalPartition = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(logicalPartition);
+ writer.writeString(0, BinaryString.fromString("20251202"));
+ DataFileTestDataGenerator gen =
DataFileTestDataGenerator.builder().build();
+ List<DataFileMeta> dataFiles = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ dataFiles.add(gen.next().meta);
+ }
+ Map<String, String> fileBucketPathMapping = new HashMap<>();
+ Map<String, String> fileBranchMapping = new HashMap<>();
+ List<DeletionFile> deletionFiles =
+ Arrays.asList(
+ new DeletionFile("dv1", 0, 100, 1L),
+ null,
+ new DeletionFile("dv3", 0, 50, 1L));
+ for (DataFileMeta dataFile : dataFiles) {
+ fileBucketPathMapping.put(dataFile.fileName(),
"dt=20251202/bucket_0");
+ fileBranchMapping.put(dataFile.fileName(), "branch_0");
+ }
+ ChainSplit split =
+ new ChainSplit(
+ logicalPartition,
+ dataFiles,
+ fileBranchMapping,
+ fileBucketPathMapping,
+ deletionFiles);
+ byte[] bytes = InstantiationUtil.serializeObject(split);
+ ChainSplit newSplit =
+ InstantiationUtil.deserializeObject(bytes,
ChainSplit.class.getClassLoader());
+
assertThat(fileBucketPathMapping).isEqualTo(newSplit.fileBucketPathMapping());
+ assertThat(fileBranchMapping).isEqualTo(newSplit.fileBranchMapping());
+ assertThat(newSplit.deletionFiles().isPresent());
+ assertThat(newSplit.deletionFiles().get()).isEqualTo(deletionFiles);
assertThat(newSplit).isEqualTo(split);
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/SplitSerializerTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/SplitSerializerTest.java
index 7cc16771f9..0fd6f982b5 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/SplitSerializerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/SplitSerializerTest.java
@@ -44,6 +44,7 @@ import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
@@ -184,20 +185,33 @@ public class SplitSerializerTest {
.withBucketPath("dt=20260707/bucket-5")
.withDataFiles(
Collections.singletonList(dataFile("chain-file", 0, 21, 30, 300L)))
+ .withDataDeletionFiles(
+ Collections.singletonList(
+ new DeletionFile("deletion_file", 100,
22, null)))
.rawConvertible(false)
.build();
Map<String, String> fileBucketPathMapping = new LinkedHashMap<>();
Map<String, String> fileBranchMapping = new LinkedHashMap<>();
+ List<DeletionFile> deletionFiles = new ArrayList<>();
List<DataFileMeta> files = new ArrayList<>();
for (DataSplit split : Arrays.asList(left, right)) {
- files.addAll(split.dataFiles());
- for (DataFileMeta file : split.dataFiles()) {
+ Optional<List<DeletionFile>> deletionFilesOpt =
split.deletionFiles();
+ for (int i = 0; i < split.dataFiles().size(); i++) {
+ DataFileMeta file = split.dataFiles().get(i);
+ DeletionFile deletionFile =
+ deletionFilesOpt.isPresent() ?
deletionFilesOpt.get().get(i) : null;
+ files.add(file);
+ deletionFiles.add(deletionFile);
fileBucketPathMapping.put(file.fileName(), split.bucketPath());
fileBranchMapping.put(file.fileName(), split == left ?
"snapshot" : "delta");
}
}
return new ChainSplit(
- DataFileTestUtils.row(2026), files, fileBranchMapping,
fileBucketPathMapping);
+ DataFileTestUtils.row(2026),
+ files,
+ fileBranchMapping,
+ fileBucketPathMapping,
+ deletionFiles);
}
private static QueryAuthSplit queryAuthSplit(Split split) {
@@ -266,6 +280,7 @@ public class SplitSerializerTest {
assertThat(actual).isEqualTo(expected);
assertThat(actual.fileBucketPathMapping()).isEqualTo(expected.fileBucketPathMapping());
assertThat(actual.fileBranchMapping()).isEqualTo(expected.fileBranchMapping());
+ assertThat(actual.deletionFiles()).isEqualTo(expected.deletionFiles());
}
private static void assertFallbackSplitEquals(
diff --git a/paimon-core/src/test/resources/compatibility/split-v1-chain
b/paimon-core/src/test/resources/compatibility/split-v1-chain
index d9f12c9765..08f61c52fa 100644
Binary files a/paimon-core/src/test/resources/compatibility/split-v1-chain and
b/paimon-core/src/test/resources/compatibility/split-v1-chain differ
diff --git a/paimon-core/src/test/resources/compatibility/split-v1-fallback
b/paimon-core/src/test/resources/compatibility/split-v1-fallback
index 976a41d75f..f83872aea1 100644
Binary files a/paimon-core/src/test/resources/compatibility/split-v1-fallback
and b/paimon-core/src/test/resources/compatibility/split-v1-fallback differ
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkChainTableDeletionVectorITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkChainTableDeletionVectorITCase.java
new file mode 100644
index 0000000000..31ab4ed04d
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkChainTableDeletionVectorITCase.java
@@ -0,0 +1,339 @@
+/*
+ * 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.flink;
+
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.CloseableIterator;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for chain table deletion-vector support using Flink SQL. */
+public class FlinkChainTableDeletionVectorITCase extends
FlinkChainTableITCaseBase {
+
+ @Test
+ public void testChainTableWithDeletionVectors() throws Exception {
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ sql(
+ "CREATE TABLE chain_dv_t1 ("
+ + " t1 BIGINT,"
+ + " t2 BIGINT,"
+ + " t3 STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,t1',"
+ + " 'bucket-key' = 't1',"
+ + " 'bucket' = '1',"
+ + " 'sequence.field' = 't2',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'deletion-vectors.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd',"
+ + " 'compaction.min.file-num' = '100',"
+ + " 'num-sorted-run.compaction-trigger' = '20'"
+ + ")");
+
+ setupChainTableBranches("chain_dv_t1");
+
+ sql(
+ "INSERT INTO `chain_dv_t1$branch_snapshot` PARTITION (dt =
'20260222')"
+ + " VALUES (1, 1, '1'), (6, 1, '1')");
+ sql(
+ "INSERT INTO `chain_dv_t1$branch_snapshot` PARTITION (dt =
'20260223')"
+ + " VALUES (1, 2, '2'), (2, 2, '2'), (3, 1, '1')");
+
+ sql(
+ "INSERT INTO `chain_dv_t1$branch_delta` PARTITION (dt =
'20260224')"
+ + " VALUES (4, 1, '1'), (5, 1, '1')");
+
+ // Delete rows from both branches to produce deletion vectors
+ sql("DELETE FROM `chain_dv_t1$branch_snapshot` WHERE dt = '20260223'
AND t1 = 3");
+ sql("DELETE FROM `chain_dv_t1$branch_delta` WHERE dt = '20260224' AND
t1 = 4");
+ sql("UPDATE `chain_dv_t1$branch_delta` SET t2=5, t3='5' WHERE dt =
'20260224' AND t1 = 5");
+ assertThat(collectResult("SELECT * FROM chain_dv_t1 WHERE dt =
'20260224'"))
+ .containsExactlyInAnyOrder(
+ "+I[1, 2, 2, 20260224]", "+I[2, 2, 2, 20260224]",
"+I[5, 5, 5, 20260224]");
+ assertThat(
+ collectResult(
+ "SELECT * FROM
`chain_dv_t1$branch_delta$table_indexes` WHERE `partition`='{20260224}'"))
+ .isNotEmpty();
+ assertThat(
+ collectResult(
+ "SELECT * FROM
`chain_dv_t1$branch_snapshot$table_indexes` WHERE `partition`='{20260223}'"))
+ .isNotEmpty();
+
+ // Compact both snapshot partitions that have deletion vectors, then
verify no pending DVs
+ // remain in the snapshot branch via the $table_indexes system table.
+ sql("CALL sys.compact_chain_table('default.chain_dv_t1',
'dt=20260224')");
+
+ assertThat(
+ collectResult(
+ "SELECT * FROM `chain_dv_t1$branch_snapshot`
WHERE dt = '20260223'"))
+ .containsExactlyInAnyOrder("+I[1, 2, 2, 20260223]", "+I[2, 2,
2, 20260223]");
+
+ assertThat(collectResult("SELECT * FROM `chain_dv_t1$branch_delta`
WHERE dt = '20260224'"))
+ .containsExactlyInAnyOrder("+I[5, 5, 5, 20260224]");
+
+ assertThat(
+ collectResult(
+ "SELECT * FROM `chain_dv_t1$branch_snapshot`
WHERE dt = '20260224'"))
+ .containsExactlyInAnyOrder(
+ "+I[1, 2, 2, 20260224]", "+I[2, 2, 2, 20260224]",
"+I[5, 5, 5, 20260224]");
+ assertThat(
+ collectResult(
+ "SELECT * FROM
`chain_dv_t1$branch_snapshot$table_indexes` WHERE `partition`='{20260224}'"))
+ .as("Compacted snapshot files should not have deletion
vectors")
+ .isEmpty();
+ assertThat(
+ collectResult(
+ "SELECT * FROM
`chain_dv_t1$branch_snapshot$files` WHERE `partition`='{20260224}'"))
+ .as("Compacted snapshot files should not have deletion
vectors")
+ .isNotEmpty();
+ }
+
+ @Test
+ @Timeout(120)
+ public void testStreamingReadWithDeletionVectors() throws Exception {
+ String tableName = "chain_dv_stream";
+ sql(
+ "CREATE TABLE "
+ + tableName
+ + " ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " region STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (region, dt) WITH ("
+ + " 'primary-key' = 'region,dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '1',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'changelog-producer' = 'input',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'deletion-vectors.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd',"
+ + " 'chain-table.chain-partition-keys' = 'dt',"
+ + " 'continuous.discovery-interval' = '1ms'"
+ + ")");
+
+ String db = tEnv.getCurrentDatabase();
+ setupChainTableBranches(tableName);
+
+ // Write snapshot data and delete one row to produce a deletion vector
in snapshot branch.
+ sql(
+ "INSERT INTO `"
+ + tableName
+ + "$branch_snapshot` PARTITION (region = 'CN', dt =
'20260223')"
+ + " VALUES (1, 1, '1'), (2, 1, '2'), (3, 1, '3')");
+ sql("DELETE FROM `" + tableName + "$branch_snapshot` WHERE region =
'CN' AND k = 3");
+
+ CloseableIterator<Row> it = sEnv.executeSql("SELECT * FROM " +
tableName).collect();
+
+ List<String> startingRows = collectRows(it, 2);
+ assertThat(startingRows)
+ .as("Starting phase should apply the snapshot deletion vector")
+ .containsExactlyInAnyOrder(
+ "+I[1, 1, 1, CN, 20260223]", "+I[2, 1, 2, CN,
20260223]");
+
+ // Write delta with cross-partition delete, update and insert.
+ writeChangelogToBranchWithRegion(
+ db,
+ tableName,
+ "delta",
+ Row.ofKind(RowKind.DELETE, 1L, 2L, "1", "CN", "20260224"),
+ Row.ofKind(RowKind.UPDATE_BEFORE, 2L, 2L, "2", "CN",
"20260224"),
+ Row.ofKind(RowKind.UPDATE_AFTER, 2L, 3L, "2-1", "CN",
"20260224"),
+ Row.ofKind(RowKind.INSERT, 4L, 1L, "4", "CN", "20260224"));
+
+ // Incremental phase: explicit changelog should stream through.
+ List<String> deltaRows = collectRows(it, 4);
+ assertThat(deltaRows)
+ .as("Incremental delta changelog should stream through with
deletion vectors")
+ .containsExactlyInAnyOrder(
+ "-D[1, 2, 1, CN, 20260224]",
+ "-U[2, 2, 2, CN, 20260224]",
+ "+U[2, 3, 2-1, CN, 20260224]",
+ "+I[4, 1, 4, CN, 20260224]");
+
+ it.close();
+ }
+
+ /**
+ * Tests streaming read with {@code
chain-table.streaming.merge-snapshot=true} and deletion
+ * vectors enabled. Verifies that the full-load phase correctly applies
cross-branch deletes
+ * that are stored as deletion-vector bitmaps in the delta branch.
+ */
+ @Test
+ @Timeout(120)
+ public void testStreamingReadWithMergeSnapshotAndDeletionVectors() throws
Exception {
+ String tableName = "chain_merge_dv_stream";
+ sql(
+ "CREATE TABLE "
+ + tableName
+ + " ("
+ + " k BIGINT, seq BIGINT, v STRING, dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '1',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'changelog-producer' = 'input',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'deletion-vectors.enabled' = 'true',"
+ + " 'chain-table.streaming.merge-snapshot' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd',"
+ + " 'continuous.discovery-interval' = '1ms'"
+ + ")");
+
+ String db = tEnv.getCurrentDatabase();
+ setupChainTableBranches(tableName);
+
+ // Write snapshot data at dt=20250808
+ sql(
+ "INSERT INTO `"
+ + tableName
+ + "$branch_snapshot` PARTITION (dt = '20250808')"
+ + " VALUES (1, 1, 'snap_1'), (2, 1, 'snap_2')");
+
+ // Write delta data spanning dt=20250809 and dt=20250810:
+ // - delete k=1 at dt=20250809 (stored as DV bitmap in delta)
+ // - update k=2: -U old snapshot value at dt=20250809, +U new delta
value at dt=20250810
+ // - insert k=3 at dt=20250810
+ writeChangelogToBranch(
+ db,
+ tableName,
+ "delta",
+ Row.ofKind(RowKind.DELETE, 1L, 2L, "snap_1", "20250809"),
+ Row.ofKind(RowKind.UPDATE_BEFORE, 2L, 2L, "snap_2",
"20250809"),
+ Row.ofKind(RowKind.UPDATE_AFTER, 2L, 3L, "delta_2",
"20250810"),
+ Row.ofKind(RowKind.INSERT, 3L, 1L, "delta_3", "20250810"));
+
+ CloseableIterator<Row> it = sEnv.executeSql("SELECT * FROM " +
tableName).collect();
+
+ // Starting (merge mode): snapshot@20250808 is anchored to the latest
delta partition
+ // dt=20250810. k=1 is deleted by the delta DV; k=2 is updated; k=3 is
newly inserted.
+ List<String> startingRows = collectRows(it, 2);
+ assertThat(startingRows)
+ .as(
+ "Starting with merge-snapshot and DV: cross-branch
delete/update should be "
+ + "applied")
+ .containsExactlyInAnyOrder(
+ "+U[2, 3, delta_2, 20250810]", "+I[3, 1, delta_3,
20250810]");
+
+ // Incremental: write new delta and verify it streams through
+ writeChangelogToBranch(
+ db, tableName, "delta", Row.ofKind(RowKind.INSERT, 4L, 1L,
"delta_4", "20250811"));
+
+ List<String> incr = collectRows(it, 1);
+ assertThat(incr)
+ .as("Incremental: new delta data should stream through")
+ .containsExactlyInAnyOrder("+I[4, 1, delta_4, 20250811]");
+
+ it.close();
+ }
+
+ /**
+ * Tests lookup join on a chain table with deletion vectors enabled.
Writes changelog records
+ * (including a cross-partition delete) to the delta branch and verifies
that lookup join can
+ * read matching rows from both snapshot and delta branches.
+ */
+ @Test
+ public void testLookupJoinWithDeletionVectors() throws Exception {
+ sql(
+ "CREATE TABLE chain_dim_dv ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '1',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'chain-table.streaming.merge-snapshot' = 'true',"
+ + " 'deletion-vectors.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_dv");
+
+ String db = tEnv.getCurrentDatabase();
+
+ // Snapshot branch has k=1,2,3 in partition 20250808
+ sql(
+ "INSERT OVERWRITE `chain_dim_dv$branch_snapshot` PARTITION (dt
= '20250808')"
+ + " VALUES (1, 1, 'snap_1'), (2, 1, 'snap_2'), (3, 1,
'snap_3')");
+
+ // Delta branch updates k=2 and inserts k=4 in partition 20250809,
+ // then deletes k=1 via a -D changelog record in a different partition
(20250809).
+ writeChangelogToBranch(
+ db,
+ "chain_dim_dv",
+ "delta",
+ Row.ofKind(RowKind.DELETE, 1L, 2L, "snap_1", "20250809"),
+ Row.ofKind(RowKind.UPDATE_BEFORE, 2L, 2L, "snap_2",
"20250809"),
+ Row.ofKind(RowKind.UPDATE_AFTER, 2L, 3L, "delta_2_updated",
"20250809"),
+ Row.ofKind(RowKind.INSERT, 4L, 1L, "delta_4", "20250809"));
+
+ // Create source table
+ sql(
+ "CREATE TABLE source_dv_delete ("
+ + " id BIGINT,"
+ + " proc_time AS PROCTIME()"
+ + ") WITH ("
+ + " 'connector' = 'paimon'"
+ + ")");
+ sql("INSERT INTO source_dv_delete VALUES (1), (2), (3), (4)");
+
+ // Lookup join on k
+ List<String> result =
+ collectResult(
+ "SELECT S.id, D.k, D.v "
+ + "FROM source_dv_delete AS S "
+ + "LEFT JOIN chain_dim_dv /*+
OPTIONS('lookup.cache' = 'full') */ "
+ + "FOR SYSTEM_TIME AS OF S.proc_time AS D "
+ + "ON S.id = D.k");
+
+ // Verify lookup results:
+ // Merge Snapshot in Phase 1 dim data:
+ // Snapshot@20250808 (anchor): k=1(snap_1), k=2(snap_2), k=3(snap_3)
+ // Delta@20250809 (with anchor merge) changelog: delete k=1, update
k=2 to
+ // delta_2_updated, insert k=4
+ // Expected lookup matches: k=1(null), k=2(delta_2_updated),
k=3(snap_3), k=4(delta_4)
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ "+I[1, null, null]",
+ "+I[2, 2, delta_2_updated]",
+ "+I[3, 3, snap_3]",
+ "+I[4, 4, delta_4]");
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkChainTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkChainTableITCase.java
index a89a103ab1..203f617e4d 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkChainTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkChainTableITCase.java
@@ -23,7 +23,6 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.flink.lookup.FullCacheLookupTable;
import org.apache.paimon.flink.lookup.LookupFileStoreTable;
-import org.apache.paimon.flink.sink.FlinkSinkBuilder;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
@@ -35,15 +34,11 @@ import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.DataTableScan;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableScan;
-import org.apache.paimon.utils.BlockingIterator;
-import org.apache.flink.api.common.JobStatus;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.ExternalizedCheckpointRetention;
import org.apache.flink.core.execution.JobClient;
-import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
@@ -58,14 +53,11 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
-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 java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static java.lang.String.format;
@@ -73,67 +65,13 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** IT cases for chain table using Flink SQL. */
-public class FlinkChainTableITCase extends CatalogITCaseBase {
+public class FlinkChainTableITCase extends FlinkChainTableITCaseBase {
@SuppressWarnings("unused")
static boolean isFlink2OrLater() {
return isFlinkVersionGreaterThanOrEqualTo("2.0");
}
- private List<String> collectResult(String query) throws Exception {
- List<String> result = new ArrayList<>();
- try (CloseableIterator<Row> it = tEnv.executeSql(query).collect()) {
- while (it.hasNext()) {
- result.add(it.next().toString());
- }
- }
- return result;
- }
-
- private void createChainTable(String tableName) {
- sql(
- "CREATE TABLE %s ("
- + " t1 BIGINT,"
- + " t2 BIGINT,"
- + " t3 STRING,"
- + " dt STRING"
- + ") PARTITIONED BY (dt) WITH ("
- + " 'primary-key' = 'dt,t1',"
- + " 'bucket-key' = 't1',"
- + " 'bucket' = '2',"
- + " 'sequence.field' = 't2',"
- + " 'merge-engine' = 'deduplicate',"
- + " 'chain-table.enabled' = 'true',"
- + " 'partition.timestamp-pattern' = '$dt',"
- + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
- + ")",
- tableName);
- }
-
- private void setupChainTableBranches(String tableName) {
- sql("CALL sys.create_branch('%s.%s', 'snapshot')",
tEnv.getCurrentDatabase(), tableName);
- sql("CALL sys.create_branch('%s.%s', 'delta')",
tEnv.getCurrentDatabase(), tableName);
-
- sql(
- "ALTER TABLE %s SET ("
- + " 'scan.fallback-snapshot-branch' = 'snapshot',"
- + " 'scan.fallback-delta-branch' = 'delta'"
- + ")",
- tableName);
- sql(
- "ALTER TABLE `%s$branch_snapshot` SET ("
- + " 'scan.fallback-snapshot-branch' = 'snapshot',"
- + " 'scan.fallback-delta-branch' = 'delta'"
- + ")",
- tableName);
- sql(
- "ALTER TABLE `%s$branch_delta` SET ("
- + " 'scan.fallback-snapshot-branch' = 'snapshot',"
- + " 'scan.fallback-delta-branch' = 'delta'"
- + ")",
- tableName);
- }
-
@Test
public void testChainTable() throws Exception {
createChainTable("chain_test");
@@ -697,97 +635,6 @@ public class FlinkChainTableITCase extends
CatalogITCaseBase {
"+I[2, 2, 1-1, CN, 20250811]", "+I[4, 1, 1, CN,
20250811]");
}
- /** Write Row data (with RowKind) to a specific branch using DataStream
API. */
- private void writeChangelogToBranch(String db, String tableName, String
branch, Row... rows)
- throws Exception {
- FileStoreTable table = paimonTable(tableName + "$branch_" + branch);
-
- StreamExecutionEnvironment env =
- streamExecutionEnvironmentBuilder()
- .streamingMode()
- .checkpointIntervalMs(100)
- .parallelism(1)
- .build();
-
- DataStream<Row> stream = env.fromCollection(Arrays.asList(rows));
-
- new FlinkSinkBuilder(table)
- .forRow(
- stream,
- DataTypes.ROW(
- DataTypes.FIELD("k", DataTypes.BIGINT()),
- DataTypes.FIELD("seq", DataTypes.BIGINT()),
- DataTypes.FIELD("v", DataTypes.STRING()),
- DataTypes.FIELD("dt", DataTypes.STRING())))
- .build();
- env.execute();
- }
-
- /**
- * Write Row data (with RowKind and a group partition column) to a
specific branch using
- * DataStream API.
- */
- private void writeChangelogToBranchWithRegion(
- String db, String tableName, String branch, Row... rows) throws
Exception {
- FileStoreTable table = paimonTable(tableName + "$branch_" + branch);
-
- StreamExecutionEnvironment env =
- streamExecutionEnvironmentBuilder()
- .streamingMode()
- .checkpointIntervalMs(100)
- .parallelism(1)
- .build();
-
- DataStream<Row> stream = env.fromCollection(Arrays.asList(rows));
-
- new FlinkSinkBuilder(table)
- .forRow(
- stream,
- DataTypes.ROW(
- DataTypes.FIELD("k", DataTypes.BIGINT()),
- DataTypes.FIELD("seq", DataTypes.BIGINT()),
- DataTypes.FIELD("v", DataTypes.STRING()),
- DataTypes.FIELD("region", DataTypes.STRING()),
- DataTypes.FIELD("dt", DataTypes.STRING())))
- .build();
- env.execute();
- }
-
- /**
- * Collect n rows from a streaming iterator with a timeout. If no data
arrives within
- * timeoutSeconds, the iterator is closed and an AssertionError is thrown.
This is necessary
- * because it.next() blocks indefinitely when no data is available, and
JUnit @Timeout cannot
- * interrupt it.
- */
- /**
- * Collects {@code n} rows from a streaming iterator using the
project-standard {@link
- * BlockingIterator}.
- */
- private List<String> collectRows(CloseableIterator<Row> it, int n) throws
Exception {
- return BlockingIterator.of(it).collect(n, 30,
TimeUnit.SECONDS).stream()
- .map(Row::toString)
- .collect(Collectors.toList());
- }
-
- /**
- * Polls the given table until it contains at least {@code minRows} rows.
Used instead of
- * fixed-duration Thread.sleep to avoid flaky tests on slow CI.
- */
- private void waitForRowCount(String tableName, int minRows) throws
Exception {
- long deadline = System.currentTimeMillis() + 60_000;
- int count = 0;
- while (System.currentTimeMillis() < deadline) {
- List<Row> rows = sql("SELECT * FROM " + tableName);
- count = rows.size();
- if (count >= minRows) {
- return;
- }
- Thread.sleep(1000);
- }
- throw new AssertionError(
- "Timed out waiting for " + minRows + " rows in " + tableName +
", got " + count);
- }
-
/**
* Tests the streaming read lifecycle for a chain table with
changelog-producer=input.
*
@@ -3685,41 +3532,4 @@ public class FlinkChainTableITCase extends
CatalogITCaseBase {
lookupTable.close();
}
}
-
- /** Helper method: poll a query until the result matches the expected
condition or timeout. */
- private void waitForQueryResult(
- String query, java.util.function.Predicate<List<String>>
condition) throws Exception {
- long startTime = System.currentTimeMillis();
- while (System.currentTimeMillis() - startTime < 30000L) {
- List<String> results = collectResult(query);
- if (condition.test(results)) {
- return;
- }
- Thread.sleep(500);
- }
- throw new RuntimeException("Timed out waiting for query result: " +
query);
- }
-
- /** Helper method: wait for job to reach target status. */
- private void waitForJobRunning(JobClient jobClient) throws Exception {
- long startTime = System.currentTimeMillis();
- JobStatus currentStatus = null;
- while (System.currentTimeMillis() - startTime < (long) 30000) {
- CompletableFuture<JobStatus> statusFuture =
jobClient.getJobStatus();
- currentStatus = statusFuture.get();
-
- if (currentStatus == JobStatus.RUNNING) {
- return;
- }
-
- if (currentStatus.isGloballyTerminalState()) {
- throw new RuntimeException(
- "Job terminated unexpectedly with status: " +
currentStatus);
- }
-
- Thread.sleep(500);
- }
- throw new RuntimeException(
- "Timed out waiting for job status running. Current status: " +
currentStatus);
- }
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkChainTableITCaseBase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkChainTableITCaseBase.java
new file mode 100644
index 0000000000..7409e03579
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkChainTableITCaseBase.java
@@ -0,0 +1,218 @@
+/*
+ * 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.flink;
+
+import org.apache.paimon.flink.sink.FlinkSinkBuilder;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.BlockingIterator;
+
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** Base class for chain table Flink IT cases providing shared helper methods.
*/
+public abstract class FlinkChainTableITCaseBase extends CatalogITCaseBase {
+
+ protected List<String> collectResult(String query) throws Exception {
+ List<String> result = new ArrayList<>();
+ try (CloseableIterator<Row> it = tEnv.executeSql(query).collect()) {
+ while (it.hasNext()) {
+ result.add(it.next().toString());
+ }
+ }
+ return result;
+ }
+
+ protected void createChainTable(String tableName) {
+ sql(
+ "CREATE TABLE %s ("
+ + " t1 BIGINT,"
+ + " t2 BIGINT,"
+ + " t3 STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,t1',"
+ + " 'bucket-key' = 't1',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 't2',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")",
+ tableName);
+ }
+
+ protected void setupChainTableBranches(String tableName) {
+ sql("CALL sys.create_branch('%s.%s', 'snapshot')",
tEnv.getCurrentDatabase(), tableName);
+ sql("CALL sys.create_branch('%s.%s', 'delta')",
tEnv.getCurrentDatabase(), tableName);
+
+ sql(
+ "ALTER TABLE %s SET ("
+ + " 'scan.fallback-snapshot-branch' = 'snapshot',"
+ + " 'scan.fallback-delta-branch' = 'delta'"
+ + ")",
+ tableName);
+ sql(
+ "ALTER TABLE `%s$branch_snapshot` SET ("
+ + " 'scan.fallback-snapshot-branch' = 'snapshot',"
+ + " 'scan.fallback-delta-branch' = 'delta'"
+ + ")",
+ tableName);
+ sql(
+ "ALTER TABLE `%s$branch_delta` SET ("
+ + " 'scan.fallback-snapshot-branch' = 'snapshot',"
+ + " 'scan.fallback-delta-branch' = 'delta'"
+ + ")",
+ tableName);
+ }
+
+ /** Write Row data (with RowKind) to a specific branch using DataStream
API. */
+ protected void writeChangelogToBranch(String db, String tableName, String
branch, Row... rows)
+ throws Exception {
+ FileStoreTable table = paimonTable(tableName + "$branch_" + branch);
+
+ StreamExecutionEnvironment env =
+ streamExecutionEnvironmentBuilder()
+ .streamingMode()
+ .checkpointIntervalMs(100)
+ .parallelism(1)
+ .build();
+
+ DataStream<Row> stream = env.fromCollection(Arrays.asList(rows));
+
+ new FlinkSinkBuilder(table)
+ .forRow(
+ stream,
+ DataTypes.ROW(
+ DataTypes.FIELD("k", DataTypes.BIGINT()),
+ DataTypes.FIELD("seq", DataTypes.BIGINT()),
+ DataTypes.FIELD("v", DataTypes.STRING()),
+ DataTypes.FIELD("dt", DataTypes.STRING())))
+ .build();
+ env.execute();
+ }
+
+ /**
+ * Write Row data (with RowKind and a group partition column) to a
specific branch using
+ * DataStream API.
+ */
+ protected void writeChangelogToBranchWithRegion(
+ String db, String tableName, String branch, Row... rows) throws
Exception {
+ FileStoreTable table = paimonTable(tableName + "$branch_" + branch);
+
+ StreamExecutionEnvironment env =
+ streamExecutionEnvironmentBuilder()
+ .streamingMode()
+ .checkpointIntervalMs(100)
+ .parallelism(1)
+ .build();
+
+ DataStream<Row> stream = env.fromCollection(Arrays.asList(rows));
+
+ new FlinkSinkBuilder(table)
+ .forRow(
+ stream,
+ DataTypes.ROW(
+ DataTypes.FIELD("k", DataTypes.BIGINT()),
+ DataTypes.FIELD("seq", DataTypes.BIGINT()),
+ DataTypes.FIELD("v", DataTypes.STRING()),
+ DataTypes.FIELD("region", DataTypes.STRING()),
+ DataTypes.FIELD("dt", DataTypes.STRING())))
+ .build();
+ env.execute();
+ }
+
+ /**
+ * Collects {@code n} rows from a streaming iterator using the
project-standard {@link
+ * BlockingIterator}.
+ */
+ protected List<String> collectRows(CloseableIterator<Row> it, int n)
throws Exception {
+ return BlockingIterator.of(it).collect(n, 30,
TimeUnit.SECONDS).stream()
+ .map(Row::toString)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Polls the given table until it contains at least {@code minRows} rows.
Used instead of
+ * fixed-duration Thread.sleep to avoid flaky tests on slow CI.
+ */
+ protected void waitForRowCount(String tableName, int minRows) throws
Exception {
+ long deadline = System.currentTimeMillis() + 60_000;
+ int count = 0;
+ while (System.currentTimeMillis() < deadline) {
+ List<Row> rows = sql("SELECT * FROM " + tableName);
+ count = rows.size();
+ if (count >= minRows) {
+ return;
+ }
+ Thread.sleep(1000);
+ }
+ throw new AssertionError(
+ "Timed out waiting for " + minRows + " rows in " + tableName +
", got " + count);
+ }
+
+ /** Helper method: poll a query until the result matches the expected
condition or timeout. */
+ protected void waitForQueryResult(
+ String query, java.util.function.Predicate<List<String>>
condition) throws Exception {
+ long startTime = System.currentTimeMillis();
+ while (System.currentTimeMillis() - startTime < 30000L) {
+ List<String> results = collectResult(query);
+ if (condition.test(results)) {
+ return;
+ }
+ Thread.sleep(500);
+ }
+ throw new RuntimeException("Timed out waiting for query result: " +
query);
+ }
+
+ /** Helper method: wait for job to reach target status. */
+ protected void waitForJobRunning(JobClient jobClient) throws Exception {
+ long startTime = System.currentTimeMillis();
+ JobStatus currentStatus = null;
+ while (System.currentTimeMillis() - startTime < (long) 30000) {
+ CompletableFuture<JobStatus> statusFuture =
jobClient.getJobStatus();
+ currentStatus = statusFuture.get();
+
+ if (currentStatus == JobStatus.RUNNING) {
+ return;
+ }
+
+ if (currentStatus.isGloballyTerminalState()) {
+ throw new RuntimeException(
+ "Job terminated unexpectedly with status: " +
currentStatus);
+ }
+
+ Thread.sleep(500);
+ }
+ throw new RuntimeException(
+ "Timed out waiting for job status running. Current status: " +
currentStatus);
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkChainTableITCase.java
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkChainTableITCase.java
index 6b183cdc62..ed3814ab78 100644
---
a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkChainTableITCase.java
+++
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkChainTableITCase.java
@@ -2421,4 +2421,103 @@ public class SparkChainTableITCase {
spark.sql("DROP TABLE IF EXISTS `my_db1`.`chain_test`;");
spark.close();
}
+
+ @Test
+ public void testChainTableWithDeletionVectors(@TempDir java.nio.file.Path
tempDir)
+ throws IOException {
+ Path warehousePath = new Path("file:" + tempDir.toString());
+ SparkSession.Builder builder =
createSparkSessionBuilder(warehousePath);
+ SparkSession spark = builder.getOrCreate();
+ spark.sql("CREATE DATABASE IF NOT EXISTS my_db1");
+ spark.sql("USE spark_catalog.my_db1");
+
+ spark.sql(
+ "CREATE TABLE IF NOT EXISTS `chain_dv_t1` (\n"
+ + " `t1` BIGINT COMMENT 't1',\n"
+ + " `t2` BIGINT COMMENT 't2',\n"
+ + " `t3` STRING COMMENT 't3'\n"
+ + ") PARTITIONED BY (`region` STRING COMMENT 'region',
`date` STRING COMMENT 'date')\n"
+ + "TBLPROPERTIES (\n"
+ + " 'chain-table.enabled' = 'true',\n"
+ + " 'deletion-vectors.enabled' = 'true',\n"
+ + " 'primary-key' = 'region,date,t1',\n"
+ + " 'sequence.field' = 't2',\n"
+ + " 'bucket-key' = 't1',\n"
+ + " 'bucket' = '1',\n"
+ + " 'partition.timestamp-pattern' = '$date',\n"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd',\n"
+ + " 'chain-table.chain-partition-keys' = 'date',\n"
+ + " 'compaction.min.file-num' = '100',\n"
+ + " 'num-sorted-run.compaction-trigger' = '20'\n"
+ + ")");
+
+ setupChainTableBranches(spark, "chain_dv_t1");
+
+ spark.sql(
+ "INSERT INTO TABLE `my_db1`.`chain_dv_t1$branch_snapshot`
PARTITION (region = 'CN', date = '20260222') VALUES (1, 1, '1'), (6, 1, '1')");
+ spark.sql(
+ "INSERT INTO TABLE `my_db1`.`chain_dv_t1$branch_snapshot`
PARTITION (region = 'CN', date = '20260223') VALUES (1, 2, '2'), (2, 2, '2'),
(3, 1, '1')");
+ spark.sql(
+ "INSERT INTO TABLE `my_db1`.`chain_dv_t1$branch_snapshot`
PARTITION (region = 'US', date = '20260223') VALUES (11, 1, '1')");
+
+ spark.sql(
+ "INSERT INTO TABLE `my_db1`.`chain_dv_t1$branch_delta`
PARTITION (region = 'CN', date = '20260224') VALUES (1, 3, '3'), (4, 1, '1'),
(5, 1, '1')");
+ spark.sql(
+ "INSERT INTO TABLE `my_db1`.`chain_dv_t1$branch_delta`
PARTITION (region = 'US', date = '20260224') VALUES (12, 1, '1')");
+ // Delete rows from both branches to produce deletion vectors
+ spark.sql("DELETE FROM `my_db1`.`chain_dv_t1$branch_snapshot` WHERE t1
= 3");
+ spark.sql("DELETE FROM `my_db1`.`chain_dv_t1$branch_delta` WHERE t1 =
4");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM
`my_db1`.`chain_dv_t1$branch_snapshot` WHERE date = '20260223'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,2,CN,20260223]", "[2,2,2,CN,20260223]",
"[11,1,1,US,20260223]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM
`my_db1`.`chain_dv_t1$branch_delta` WHERE date = '20260224'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,3,3,CN,20260224]", "[5,1,1,CN,20260224]",
"[12,1,1,US,20260224]");
+
+ assertThat(
+ spark.sql("SELECT * FROM `my_db1`.`chain_dv_t1` WHERE
date = '20260224'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,3,3,CN,20260224]",
+ "[2,2,2,CN,20260224]",
+ "[5,1,1,CN,20260224]",
+ "[11,1,1,US,20260224]",
+ "[12,1,1,US,20260224]");
+
+ spark.sql(
+ "CALL sys.compact_chain_table(table => 'my_db1.chain_dv_t1',
partition => 'date=\"20260224\"')");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM
`my_db1`.`chain_dv_t1$branch_snapshot` WHERE date = '20260224'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,3,3,CN,20260224]",
+ "[2,2,2,CN,20260224]",
+ "[5,1,1,CN,20260224]",
+ "[11,1,1,US,20260224]",
+ "[12,1,1,US,20260224]");
+
+ spark.sql("DROP TABLE IF EXISTS `my_db1`.`chain_dv_t1`;");
+ spark.close();
+ }
}
