[core] Support Deletion Vectors for Chain Table

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,44 @@ 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.
 - 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.
 
-After creating a chain table, you can read and write data in the following ways.
+## Write Data
+
+After creating a chain table, you can write data to the snapshot or delta branch.
+
+### 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 +179,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 +196,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 +217,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"');

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeFunction.java

@@ -167,6 +167,10 @@ public void withIOManager(@Nullable IOManager ioManager) {
             this.ioManager = ioManager;
         }
 
+        public MergeFunctionFactory<KeyValue> wrapped() {
+            return wrapped;
+        }
+
         @Override
         public MergeFunction<KeyValue> create(@Nullable RowType readType) {
             return new LookupMergeFunction(

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

@@ -263,7 +263,11 @@ public RecordReader<KeyValue> createChainReader(ChainSplit chainSplit) throws IO
         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(
@@ -281,18 +285,21 @@ public RecordReader<KeyValue> createMergeReader(
         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(
                     () ->
@@ -366,4 +373,11 @@ private RecordReader<KeyValue> projectOuter(RecordReader<KeyValue> reader) {
     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;
+    }
 }

paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java

@@ -1162,9 +1162,6 @@ public static void validateChainTable(TableSchema schema, CoreOptions options) {
                     changelogProducer == ChangelogProducer.NONE
                             || 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");
             Preconditions.checkArgument(
                     options.partitionTimestampPattern() != null,
                     "Partition timestamp pattern is required for chain table.");

paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java

@@ -35,6 +35,7 @@
 import org.apache.paimon.table.source.DataFilePlan;
 import org.apache.paimon.table.source.DataSplit;
 import org.apache.paimon.table.source.DataTableScan;
+import org.apache.paimon.table.source.DeletionFile;
 import org.apache.paimon.table.source.InnerTableRead;
 import org.apache.paimon.table.source.Split;
 import org.apache.paimon.table.source.TableRead;
@@ -454,9 +455,20 @@ public Plan plan() {
                         for (Map.Entry<Integer, List<DataSplit>> entry : bucketSplits.entrySet()) {
                             HashMap<String, String> fileBucketPathMapping = new HashMap<>();
                             HashMap<String, String> fileBranchMapping = new HashMap<>();
+                            List<DeletionFile> deletionFiles = new ArrayList<>();
+                            List<DataFileMeta> dataFiles = new ArrayList<>();
                             List<DataSplit> splitList = entry.getValue();
                             for (DataSplit dataSplit : splitList) {
-                                for (DataFileMeta file : dataSplit.dataFiles()) {
+                                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;
+                                    dataFiles.add(file);
+                                    deletionFiles.add(deletionFile);
                                     fileBucketPathMapping.put(
                                             file.fileName(), dataSplit.bucketPath());
                                     String branch =
@@ -469,13 +481,10 @@ public Plan plan() {
                             ChainSplit split =
                                     new ChainSplit(
                                             partitionPairs.getKey(),
-                                            entry.getValue().stream()
-                                                    .flatMap(
-                                                            dataSplit ->
-                                                                    dataSplit.dataFiles().stream())
-                                                    .collect(Collectors.toList()),
+                                            dataFiles,
                                             fileBranchMapping,
-                                            fileBucketPathMapping);
+                                            fileBucketPathMapping,
+                                            deletionFiles);
                             splits.add(split);
                         }
                     }
@@ -523,18 +532,7 @@ private Set<BinaryRow> preloadTargetSnapshotSplits(List<Split> splits) {
 
             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(

paimon-core/src/main/java/org/apache/paimon/table/source/ChainSplit.java

@@ -27,6 +27,8 @@
 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.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 static ChainSplit from(DataSplit dataSplit, String branch) {
                 dataSplit.partition(),
                 dataSplit.dataFiles(),
                 fileBranchMapping,
-                fileBucketPathMapping);
+                fileBucketPathMapping,
+                dataSplit.deletionFiles().orElse(null));
+    }
+
+    @Override
+    public Optional<List<DeletionFile>> deletionFiles() {
+        return Optional.ofNullable(dataDeletionFiles);
     }
 
     @Override
@@ -108,7 +123,21 @@ public long rowCount() {
 
     @Override
     public OptionalLong mergedRowCount() {
-        return OptionalLong.empty();
+        if (dataDeletionFiles == null || dataDeletionFiles.isEmpty()) {
+            return OptionalLong.empty();
+        }
+        long deletedRows = 0L;
+        for (DeletionFile deletionFile : dataDeletionFiles) {
+            if (deletionFile == null) {
+                continue;
+            }
+            if (deletionFile.cardinality() == null) {
+                return OptionalLong.empty();
+            } else {
+                deletedRows += deletionFile.cardinality();
+            }
+        }
+        return OptionalLong.of(rowCount() - deletedRows);
     }
 
     @Override
@@ -158,6 +187,7 @@ protected void assign(ChainSplit other) {
         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 +214,14 @@ public void serialize(DataOutputView out) throws IOException {
             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 +247,17 @@ public static ChainSplit deserialize(DataInputView in) throws IOException {
             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);
     }
 }