feat(read): support deletion vectors for data-evolution tables

Port the read side of Apache Paimon d2be7eace (#8380). A data-evolution
table may now enable deletion-vectors.enabled: the deletion vector of a
row range group is maintained against the group's anchor file, so every
reader of the group applies it shifted by the file's offset inside the
anchor row id range, and the blob fallback path drops the deleted row
ids from its placeholder gap segments. That keeps the readers of a group
positionally aligned for the column merge.

- Add DataEvolutionUtils::RetrieveAnchorFile, matching Java's rule of
  the oldest normal file compared by (max_sequence_number, file_name)
  and skipping blob and vector-store files.
- Drop the SchemaValidation rule forbidding the combination.
- Support committing a deletion vector index file for a table without
  buckets, which a data-evolution table always is: combine such index
  files by file name, as Java's GlobalCombiner does, and drop the
  conflict detection rule that refused deletion vectors in
  BUCKET_UNAWARE mode. Without this the deletes another engine issues
  could not reach the table through this commit path at all.
- Subtract deletion vector cardinality from DataSplit::MergedRowCount
  and report it unavailable when a deletion file has no cardinality.
- Skip the limit push down when a non-partition filter or a row range
  index is present, since a split's metadata row count is then only an
  upper bound of what the read returns.
- Apply the row ranges selection and the group deletion vectors to the
  blob view pre-read, so a reference held only by a dropped row is not
  resolved.

Paimon C++ does not produce these deletion vectors; bitmap64 vectors and
compaction of such a table remain unsupported. See
docs/source/user_guide/compaction.rst.

Behavior change worth calling out, since it reaches append and primary
key tables too: ApplyPushDownLimit used to consider only raw convertible
splits, silently leaving a merged split out of the pruned plan. It now
asks every split for its row count and abandons the push down as soon as
one cannot be counted from metadata, which is the same conservative
answer the raw convertible path already gave. That is what lets a
data-evolution split, which is not raw convertible but does have an
exact count, take part in the pruning.

Author: SteNicholas

docs/source/user_guide/compaction.rst

@@ -47,6 +47,20 @@ not block writes.
    (``bucket > 0``). Dynamic bucketing (``bucket = -1``) does not support
    compaction. Tables with blob columns also skip compaction.
 
+.. note::
+   A data-evolution table (``data-evolution.enabled = true``) may enable
+   ``deletion-vectors.enabled``. Paimon C++ reads such a table: the deletion
+   vector of a row range group is applied to every file of the group, so a
+   deleted row disappears from all merged columns. Only the default 32-bit
+   deletion vectors can be read; ``deletion-vectors.bitmap64`` is not supported
+   yet, and a read fails when it actually encounters a 64-bit deletion vector.
+   Paimon C++ currently reads but does not write deletion vectors for
+   data-evolution tables, so the deletes themselves have to be issued by
+   another engine. Compacting such a table is supported by Paimon Java but is
+   not ported to Paimon C++ yet: auto compaction never runs on it, since data
+   evolution requires ``bucket = -1``, and the dedicated compaction entry point
+   rejects it outright rather than dropping the deletes.
+
 Auto Compaction
 ~~~~~~~~~~~~~~~
 During each flush, the writer triggers a best-effort auto compaction. The

src/paimon/CMakeLists.txt

@@ -411,6 +411,7 @@ set(PAIMON_CORE_SRCS
     core/utils/branch_manager.cpp
     core/utils/blob_view_lookup.cpp
     core/utils/consumer_manager.cpp
+    core/utils/data_evolution_utils.cpp
     core/utils/field_mapping.cpp
     core/utils/nested_projection_utils.cpp
     core/utils/file_store_path_factory.cpp
@@ -862,6 +863,7 @@ if(PAIMON_BUILD_TESTS)
                     core/utils/blob_view_lookup_test.cpp
                     core/utils/branch_manager_test.cpp
                     core/utils/consumer_manager_test.cpp
+                    core/utils/data_evolution_utils_test.cpp
                     core/utils/file_store_path_factory_cache_test.cpp
                     core/utils/field_mapping_test.cpp
                     core/utils/nested_projection_utils_test.cpp

src/paimon/core/manifest/index_manifest_file_handler.cpp

@@ -19,15 +19,17 @@
 #include "paimon/core/manifest/index_manifest_file_handler.h"
 
 #include <set>
+#include <string>
 #include <unordered_map>
 #include <utility>
 
+#include "fmt/format.h"
 #include "paimon/core/deletionvectors/deletion_vectors_index_file.h"
 namespace paimon {
 
 using BucketIdentifier = std::tuple<BinaryRow, int32_t, std::string>;
 
-std::vector<IndexManifestEntry> IndexManifestFileHandler::BucketedCombiner::Combine(
+Result<std::vector<IndexManifestEntry>> IndexManifestFileHandler::BucketedCombiner::Combine(
     const std::vector<IndexManifestEntry>& prev_index_files,
     const std::vector<IndexManifestEntry>& new_index_files) const {
     std::unordered_map<BucketIdentifier, IndexManifestEntry> index_entries;
@@ -67,7 +69,7 @@ std::vector<IndexManifestEntry> IndexManifestFileHandler::BucketedCombiner::Comb
     return result_entries;
 }
 
-std::vector<IndexManifestEntry> IndexManifestFileHandler::GlobalFileNameCombiner::Combine(
+Result<std::vector<IndexManifestEntry>> IndexManifestFileHandler::GlobalFileNameCombiner::Combine(
     const std::vector<IndexManifestEntry>& prev_index_files,
     const std::vector<IndexManifestEntry>& new_index_files) const {
     std::map<std::string, IndexManifestEntry> index_entries;
@@ -104,6 +106,81 @@ std::vector<IndexManifestEntry> IndexManifestFileHandler::GlobalFileNameCombiner
     return result_entries;
 }
 
+namespace {
+std::vector<std::string> DeletionVectorDataFiles(const IndexManifestEntry& entry) {
+    if (entry.index_file->DvRanges() == std::nullopt) {
+        return {};
+    }
+    return entry.index_file->DvRanges().value().key_vec();
+}
+}  // namespace
+
+Result<std::vector<IndexManifestEntry>>
+IndexManifestFileHandler::GlobalDeletionVectorCombiner::Combine(
+    const std::vector<IndexManifestEntry>& prev_index_files,
+    const std::vector<IndexManifestEntry>& new_index_files) const {
+    std::map<std::string, IndexManifestEntry> index_entries;
+    std::set<std::string> covered_data_files;
+    for (const auto& entry : prev_index_files) {
+        index_entries.insert_or_assign(entry.index_file->FileName(), entry);
+        for (const auto& data_file : DeletionVectorDataFiles(entry)) {
+            covered_data_files.insert(data_file);
+        }
+    }
+
+    std::vector<IndexManifestEntry> removed;
+    removed.reserve(new_index_files.size());
+    std::vector<IndexManifestEntry> added;
+    added.reserve(new_index_files.size());
+
+    for (const auto& entry : new_index_files) {
+        if (entry.kind == FileKind::Delete()) {
+            removed.push_back(entry);
+        } else if (entry.kind == FileKind::Add()) {
+            added.push_back(entry);
+        }
+    }
+
+    // The deleted entry is processed first, so that an index file taking over the data files of
+    // the one it replaces is not rejected as a second vector for them.
+    for (const auto& entry : removed) {
+        const std::string& file_name = entry.index_file->FileName();
+        if (index_entries.erase(file_name) == 0) {
+            return Status::Invalid(fmt::format(
+                "Trying to delete deletion vector index file {} which does not exist.", file_name));
+        }
+        for (const auto& data_file : DeletionVectorDataFiles(entry)) {
+            if (covered_data_files.erase(data_file) == 0) {
+                return Status::Invalid(
+                    fmt::format("Trying to delete the deletion vector of data file {}, which does "
+                                "not exist.",
+                                data_file));
+            }
+        }
+    }
+    for (const auto& entry : added) {
+        const std::string& file_name = entry.index_file->FileName();
+        if (index_entries.find(file_name) != index_entries.end()) {
+            return Status::Invalid(fmt::format(
+                "Trying to add deletion vector index file {} which is already added.", file_name));
+        }
+        for (const auto& data_file : DeletionVectorDataFiles(entry)) {
+            if (!covered_data_files.insert(data_file).second) {
+                return Status::Invalid(fmt::format(
+                    "Trying to add a second deletion vector for data file {}.", data_file));
+            }
+        }
+        index_entries.insert_or_assign(file_name, entry);
+    }
+
+    std::vector<IndexManifestEntry> result_entries;
+    result_entries.reserve(index_entries.size());
+    for (const auto& [_, entry] : index_entries) {
+        result_entries.push_back(entry);
+    }
+    return result_entries;
+}
+
 Result<std::string> IndexManifestFileHandler::Write(
     const std::optional<std::string>& previous_index_manifest,
     const std::vector<IndexManifestEntry>& new_index_entries, int32_t bucket_mode,
@@ -138,8 +215,8 @@ Result<std::string> IndexManifestFileHandler::Write(
             GetIndexManifestFileCombine(index_type, bucket_mode));
         std::vector<IndexManifestEntry> typed_previous_entries = previous[index_type];
         std::vector<IndexManifestEntry> typed_current_entries = current[index_type];
-        std::vector<IndexManifestEntry> combined_entries =
-            combiner->Combine(typed_previous_entries, typed_current_entries);
+        PAIMON_ASSIGN_OR_RAISE(std::vector<IndexManifestEntry> combined_entries,
+                               combiner->Combine(typed_previous_entries, typed_current_entries));
 
         index_entries.insert(index_entries.end(), combined_entries.begin(), combined_entries.end());
     }
@@ -168,7 +245,7 @@ IndexManifestFileHandler::GetIndexManifestFileCombine(const std::string& index_t
         return std::make_unique<GlobalFileNameCombiner>();
     }
     if (index_type == DeletionVectorsIndexFile::DELETION_VECTORS_INDEX && bucket_mode == -1) {
-        return Status::NotImplemented("not yet support dv with BUCKET_UNAWARE mode");
+        return std::make_unique<GlobalDeletionVectorCombiner>();
     }
     return std::make_unique<BucketedCombiner>();
 }

src/paimon/core/manifest/index_manifest_file_handler.h

@@ -43,23 +43,36 @@ class IndexManifestFileHandler {
     class IndexManifestFileCombiner {
      public:
         virtual ~IndexManifestFileCombiner() = default;
-        virtual std::vector<IndexManifestEntry> Combine(
+        virtual Result<std::vector<IndexManifestEntry>> Combine(
             const std::vector<IndexManifestEntry>& prev_index_files,
             const std::vector<IndexManifestEntry>& new_index_files) const = 0;
     };
 
     /// Combine previous and new index files by partition, bucket and index type.
     class BucketedCombiner : public IndexManifestFileCombiner {
      public:
-        std::vector<IndexManifestEntry> Combine(
+        Result<std::vector<IndexManifestEntry>> Combine(
             const std::vector<IndexManifestEntry>& prev_index_files,
             const std::vector<IndexManifestEntry>& new_index_files) const override;
     };
 
     /// Combine previous and new index files by file name.
     class GlobalFileNameCombiner : public IndexManifestFileCombiner {
      public:
-        std::vector<IndexManifestEntry> Combine(
+        Result<std::vector<IndexManifestEntry>> Combine(
+            const std::vector<IndexManifestEntry>& prev_index_files,
+            const std::vector<IndexManifestEntry>& new_index_files) const override;
+    };
+
+    /// Combine previous and new deletion vector index files by file name, for a table without
+    /// buckets, where the bucket cannot tell two index files apart.
+    ///
+    /// A data file is covered by at most one deletion vector, so a delta that adds a second one
+    /// for a data file, or drops one that is not there, was built against a different base and
+    /// is rejected rather than silently changing which rows the read returns.
+    class GlobalDeletionVectorCombiner : public IndexManifestFileCombiner {
+     public:
+        Result<std::vector<IndexManifestEntry>> Combine(
             const std::vector<IndexManifestEntry>& prev_index_files,
             const std::vector<IndexManifestEntry>& new_index_files) const override;
     };

src/paimon/core/manifest/index_manifest_file_handler_test.cpp

@@ -20,6 +20,7 @@
 
 #include <memory>
 #include <optional>
+#include <set>
 #include <string>
 #include <vector>
 
@@ -218,17 +219,107 @@ TEST_F(IndexManifestFileHandlerTest, GlobalCombinerOverwritesDuplicateAddedEntri
     ASSERT_EQ(written_entries[0].index_file->RowCount(), 20);
 }
 
-TEST_F(IndexManifestFileHandlerTest, DvWithBucketUnawareModeReturnsNotImplemented) {
+TEST_F(IndexManifestFileHandlerTest, GlobalDvCombinerReplacesIndexFileInBucketUnawareMode) {
     ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(/*bucket_mode=*/-1));
 
     auto partition = BinaryRow::EmptyRow();
+    // an unaware bucket table writes every index file under bucket 0, so only the index file
+    // name tells the two entries apart
+    std::vector<IndexManifestEntry> previous_entries = {
+        MakeDvEntry(FileKind::Add(), partition, /*bucket=*/0, "dv-0", {"data-0.orc"}, 1),
+        MakeDvEntry(FileKind::Add(), partition, /*bucket=*/0, "dv-1", {"data-1.orc"}, 1)};
+
+    ASSERT_OK_AND_ASSIGN(std::string previous_manifest,
+                         IndexManifestFileHandler::Write(
+                             /*previous_index_manifest=*/std::nullopt, previous_entries,
+                             /*bucket_mode=*/-1, index_manifest_file.get()));
+
+    // updating the vector of data-0.orc replaces its index file, and the data file it covers
+    // moves to the new one
     std::vector<IndexManifestEntry> new_entries = {
+        MakeDvEntry(FileKind::Delete(), partition, /*bucket=*/0, "dv-0", {"data-0.orc"}, 1),
+        MakeDvEntry(FileKind::Add(), partition, /*bucket=*/0, "dv-0-new", {"data-0.orc"}, 2)};
+
+    ASSERT_OK_AND_ASSIGN(
+        std::string current_manifest,
+        IndexManifestFileHandler::Write(previous_manifest, new_entries,
+                                        /*bucket_mode=*/-1, index_manifest_file.get()));
+
+    std::vector<IndexManifestEntry> written_entries;
+    ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr, &written_entries));
+    ASSERT_EQ(written_entries.size(), 2);
+    std::set<std::string> written_file_names;
+    for (const auto& entry : written_entries) {
+        written_file_names.insert(entry.index_file->FileName());
+    }
+    ASSERT_EQ(written_file_names, (std::set<std::string>{"dv-0-new", "dv-1"}));
+
+    // the same replacement with the added entry listed first: the delta carries no order the
+    // commit path is bound to, so the added entry must not be read as a second vector for
+    // data-0.orc just because its delete trails it
+    std::vector<IndexManifestEntry> reordered_entries = {
+        MakeDvEntry(FileKind::Add(), partition, /*bucket=*/0, "dv-0-newer", {"data-0.orc"}, 3),
+        MakeDvEntry(FileKind::Delete(), partition, /*bucket=*/0, "dv-0-new", {"data-0.orc"}, 2)};
+
+    ASSERT_OK_AND_ASSIGN(
+        std::string reordered_manifest,
+        IndexManifestFileHandler::Write(current_manifest, reordered_entries,
+                                        /*bucket_mode=*/-1, index_manifest_file.get()));
+
+    written_entries.clear();
+    ASSERT_OK(index_manifest_file->Read(reordered_manifest, /*filter=*/nullptr, &written_entries));
+    ASSERT_EQ(written_entries.size(), 2);
+    written_file_names.clear();
+    for (const auto& entry : written_entries) {
+        written_file_names.insert(entry.index_file->FileName());
+    }
+    ASSERT_EQ(written_file_names, (std::set<std::string>{"dv-0-newer", "dv-1"}));
+}
+
+TEST_F(IndexManifestFileHandlerTest, GlobalDvCombinerRejectsInconsistentDelta) {
+    ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(/*bucket_mode=*/-1));
+
+    auto partition = BinaryRow::EmptyRow();
+    std::vector<IndexManifestEntry> previous_entries = {
         MakeDvEntry(FileKind::Add(), partition, /*bucket=*/0, "dv-0", {"data-0.orc"}, 1)};
 
-    ASSERT_NOK_WITH_MSG(IndexManifestFileHandler::Write(
-                            /*previous_index_manifest=*/std::nullopt, new_entries,
-                            /*bucket_mode=*/-1, index_manifest_file.get()),
-                        "not yet support dv with BUCKET_UNAWARE mode");
+    ASSERT_OK_AND_ASSIGN(std::string previous_manifest,
+                         IndexManifestFileHandler::Write(
+                             /*previous_index_manifest=*/std::nullopt, previous_entries,
+                             /*bucket_mode=*/-1, index_manifest_file.get()));
+
+    // every delta below was built against a base the previous manifest is not, so applying it
+    // would leave the read either with two vectors for one data file or with none
+    {
+        // adding a vector for data-0.orc without dropping the one it already has
+        std::vector<IndexManifestEntry> new_entries = {
+            MakeDvEntry(FileKind::Add(), partition, /*bucket=*/0, "dv-1", {"data-0.orc"}, 1)};
+
+        ASSERT_NOK_WITH_MSG(
+            IndexManifestFileHandler::Write(previous_manifest, new_entries,
+                                            /*bucket_mode=*/-1, index_manifest_file.get()),
+            "Trying to add a second deletion vector for data file data-0.orc");
+    }
+    {
+        // adding an index file the manifest already holds
+        std::vector<IndexManifestEntry> new_entries = {
+            MakeDvEntry(FileKind::Add(), partition, /*bucket=*/0, "dv-0", {"data-1.orc"}, 1)};
+
+        ASSERT_NOK_WITH_MSG(
+            IndexManifestFileHandler::Write(previous_manifest, new_entries,
+                                            /*bucket_mode=*/-1, index_manifest_file.get()),
+            "Trying to add deletion vector index file dv-0 which is already added.");
+    }
+    {
+        // deleting an index file the manifest does not hold
+        std::vector<IndexManifestEntry> new_entries = {
+            MakeDvEntry(FileKind::Delete(), partition, /*bucket=*/0, "dv-1", {"data-1.orc"}, 1)};
+
+        ASSERT_NOK_WITH_MSG(
+            IndexManifestFileHandler::Write(previous_manifest, new_entries,
+                                            /*bucket_mode=*/-1, index_manifest_file.get()),
+            "Trying to delete deletion vector index file dv-1 which does not exist.");
+    }
 }
 
 }  // namespace paimon::test

src/paimon/core/operation/commit/conflict_detection.cpp

@@ -48,7 +48,6 @@
 #include "paimon/core/operation/commit/manifest_entry_changes.h"
 #include "paimon/core/operation/commit/row_id_column_conflict_checker.h"
 #include "paimon/core/schema/table_schema.h"
-#include "paimon/core/table/bucket_mode.h"
 #include "paimon/core/utils/field_mapping.h"
 #include "paimon/core/utils/file_store_path_factory.h"
 #include "paimon/core/utils/snapshot_manager.h"
@@ -115,12 +114,6 @@ Status ConflictDetection::CheckConflicts(
         row_id_column_conflict_checker,
     const Snapshot::CommitKind& commit_kind) const {
     std::string base_commit_user = latest_snapshot.CommitUser();
-    if (options_.DeletionVectorsEnabled() &&
-        ResolveBucketMode(options_.GetBucket(), table_schema_) == BucketMode::BUCKET_UNAWARE) {
-        return Status::NotImplemented(
-            "check conflicts failed. not yet support dv with BUCKET_UNAWARE mode");
-    }
-
     std::vector<ManifestEntry> all_entries = base_entries;
     all_entries.insert(all_entries.end(), delta_entries.begin(), delta_entries.end());
     PAIMON_RETURN_NOT_OK(CheckBucketKeepSame(all_entries, commit_kind, base_commit_user,