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.

- Pick the anchor by Java's rule, the oldest normal file compared by
  (max_sequence_number, file_name), since the engine writing the vectors
  keys them by the same file.
- 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.
- Narrow the conflict detection rule that refused deletion vectors in
  BUCKET_UNAWARE mode outright, which kept the deletes another engine
  issues from reaching the table through this commit path at all, to
  refusing a delta that deletes data files. That is the only shape whose
  conflicts need the data file to deletion vector pairing Java builds in
  buildBaseEntriesWithDV, which is not ported; adding files cannot
  orphan a vector, and Paimon C++ produces no other shape here, since
  it never writes deletion vectors and never compacts such a table.
- Subtract deletion vector cardinality from DataSplit::MergedRowCount
  and report it unavailable when a deletion file has no cardinality.
- 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.
- Read a row range group made only of blob files, which a split can pack
  next to a group carrying a deletion vector, without an anchor: no
  writer can key a vector by such a group, so it reads undeleted rather
  than failing the whole split on the missing anchor. Paimon Java throws
  here.

Paimon C++ does not produce these deletion vectors; bitmap64 vectors and
compaction of such a table remain unsupported. The read behavior and its
limitations are documented in docs/source/user_guide/read.rst, the
compaction restriction in docs/source/user_guide/compaction.rst.

Two behavior changes worth calling out, since they reach 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.

The new CanPushDownLimit guard also fixes a wrong result that predates
deletion vectors: the push down never consulted the non-partition
predicate, so a limited query carrying a value filter pruned splits on
row counts the filter had not been applied to yet, and could return
fewer rows than the limit asked for while more still existed. It is now
skipped whenever a non-partition filter or a row range index is present,
since the metadata count is only an upper bound of what such a read
returns.

Author: SteNicholas

docs/source/user_guide/compaction.rst

@@ -47,6 +47,21 @@ not block writes.
    (``bucket > 0``). Dynamic bucketing (``bucket = -1``) does not support
    compaction. Tables with blob columns also skip compaction.
 
+.. _data-evolution-deletion-vectors-compaction:
+
+.. note::
+   A data-evolution table (``data-evolution.enabled = true``) may enable
+   ``deletion-vectors.enabled``. Paimon C++ never compacts such a table: auto
+   compaction never runs on it, since data evolution requires ``bucket = -1``,
+   and ``AppendCompactCoordinator::Run``, the dedicated compaction entry point,
+   rejects it outright rather than dropping the deletes. This matches the
+   Paimon Java revision this support was ported from, which ends the compaction
+   scan as soon as deletion vectors are enabled; later Java releases do compact
+   such a table, and Paimon C++ has not caught up.
+
+   Reading such a table is supported; see
+   :ref:`data-evolution-deletion-vectors`.
+
 Auto Compaction
 ~~~~~~~~~~~~~~~
 During each flush, the writer triggers a best-effort auto compaction. The

docs/source/user_guide/read.rst

@@ -308,3 +308,45 @@ Implementation Guidance
 - Expect per-file schema variability; design readers to align by field IDs rather than positional indices.
 - Do not assume identical overflow semantics across C++ and Java; tests should validate acceptable ranges and nullability.
 - For timestamp handling, consider precision/range constraints in C++ when interoperating with Java-produced data splits.
+
+.. _data-evolution-deletion-vectors:
+
+Deletion Vectors on Data-Evolution Tables
+-----------------------------------------
+
+A data-evolution table (``data-evolution.enabled = true``) may enable
+``deletion-vectors.enabled``. Reading such a table is supported: a deleted row disappears
+from the result, including from the columns merged out of the other files that cover it.
+
+How the Deletion Vector Is Located
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+A data-evolution split holds several files per row id range, one per group of columns. The
+deletion vector of such a row range group is not per file: it is maintained against the
+group's *anchor file*, the oldest normal file of the group, compared by
+``(max_sequence_number, file_name)`` and skipping blob and vector-store files. Its positions
+are therefore relative to the anchor file's row id range.
+
+Reading applies that one vector to every file of the group, shifted by the file's offset
+inside the anchor range, so all the readers being merged drop the same rows and stay
+positionally aligned. The blob fallback path has no file reader to wrap for the placeholder
+gaps it pads uncovered row ids with, so it removes the deleted row ids from those gap ranges
+instead.
+
+The rule that picks the anchor has to stay identical to the engine that writes the vectors:
+a vector keyed by any other file of the group is never found, and its deleted rows silently
+come back.
+
+Limitations
+~~~~~~~~~~~
+
+- 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++ does not write deletion vectors for data-evolution tables, so the deletes
+  themselves have to be issued by another engine.
+- A commit that drops data files from such a table, an overwrite for instance, is refused.
+  Whether it conflicts with a concurrent commit rewriting those files' deletion vectors cannot
+  be decided yet, so it fails rather than committing against a stale state. Appending is
+  unaffected, and so is another engine replacing a deletion vector.
+- Such a table is never compacted; see
+  :ref:`the compaction note <data-evolution-deletion-vectors-compaction>`.

include/paimon/defs.h

@@ -358,6 +358,9 @@ struct PAIMON_EXPORT Options {
     /// files containing deletion vectors are generated when data is written, which marks the data
     /// for deletion. During read operations, by applying these index files, merging can be avoided.
     /// Default value is false.
+    /// @note On a data-evolution table (`DATA_EVOLUTION_ENABLED`), Paimon C++ reads deletion
+    /// vectors but does not write them: the deletes have to be issued by another engine, and such
+    /// a table is never compacted.
     static const char DELETION_VECTORS_ENABLED[];
 
     /// "deletion-vector.index-file.target-size" - The target size of deletion vector index file.

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/common/reader/blob_fallback_batch_reader.h

@@ -50,7 +50,9 @@ namespace paimon {
 ///    first) and, inside each group, orders them by first row id. Row id ranges the group's
 ///    files do not cover are represented by gap segments, which stand for all-placeholder rows.
 /// 2. All groups span the same overall row id range, so with the same row-ranges selection
-///    applied they yield the same number of rows and can be stepped in lockstep.
+///    applied they yield the same number of rows and can be stepped in lockstep. A deletion
+///    vector has to reach every group the same way, through the file segments' readers and
+///    through the row ids the caller leaves in a gap segment's `gap_selected_ranges`.
 /// 3. Each output row takes the first group, in max-sequence order, whose row is not a
 ///    placeholder. Placeholder rows are identified by exact equality with the
 ///    BlobDefs::kPlaceholderSentinel bytes, emitted by the blob format reader when

src/paimon/core/manifest/index_manifest_file_handler.cpp

@@ -19,15 +19,19 @@
 #include "paimon/core/manifest/index_manifest_file_handler.h"
 
 #include <set>
+#include <string>
 #include <unordered_map>
 #include <utility>
 
+#include "fmt/format.h"
+#include "paimon/common/utils/linked_hash_map.h"
 #include "paimon/core/deletionvectors/deletion_vectors_index_file.h"
+#include "paimon/core/index/index_file_meta.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 +71,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 +108,92 @@ std::vector<IndexManifestEntry> IndexManifestFileHandler::GlobalFileNameCombiner
     return result_entries;
 }
 
+namespace {
+using DeletionVectorRanges = LinkedHashMap<std::string, DeletionVectorMeta>;
+
+/// The deletion vectors an index file holds, keyed by the data file each covers. Null when the
+/// entry holds none.
+const DeletionVectorRanges* GetDeletionVectorRanges(const IndexManifestEntry& entry) {
+    const std::optional<DeletionVectorRanges>& dv_ranges = entry.index_file->DvRanges();
+    return dv_ranges == std::nullopt ? nullptr : &dv_ranges.value();
+}
+}  // 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);
+        const DeletionVectorRanges* dv_ranges = GetDeletionVectorRanges(entry);
+        if (dv_ranges == nullptr) {
+            continue;
+        }
+        for (const auto& [data_file, _] : *dv_ranges) {
+            covered_data_files.insert(data_file);
+        }
+    }
+
+    std::vector<const IndexManifestEntry*> removed;
+    std::vector<const IndexManifestEntry*> added;
+    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. Paimon Java's
+    // GlobalCombiner is order sensitive here; its two sibling combiners are not.
+    for (const IndexManifestEntry* 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));
+        }
+        const DeletionVectorRanges* dv_ranges = GetDeletionVectorRanges(*entry);
+        if (dv_ranges == nullptr) {
+            continue;
+        }
+        for (const auto& [data_file, _] : *dv_ranges) {
+            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 IndexManifestEntry* 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));
+        }
+        const DeletionVectorRanges* dv_ranges = GetDeletionVectorRanges(*entry);
+        if (dv_ranges != nullptr) {
+            for (const auto& [data_file, _] : *dv_ranges) {
+                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 +228,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());
     }
@@ -167,8 +257,14 @@ IndexManifestFileHandler::GetIndexManifestFileCombine(const std::string& index_t
     if (index_type != DeletionVectorsIndexFile::DELETION_VECTORS_INDEX && index_type != "HASH") {
         return std::make_unique<GlobalFileNameCombiner>();
     }
+    // `bucket_mode` is the configured bucket, not a resolved BucketMode, standing in for Paimon
+    // Java's BucketMode.BUCKET_UNAWARE check. The two agree on every table that can exist:
+    // SchemaValidation rejects the other unaware bucket, 0, and BucketIdCalculator refuses to
+    // write the primary key table on which -1 means HASH_DYNAMIC instead. Lifting either
+    // restriction means passing the resolved BucketMode here, or a dynamic bucket table would
+    // combine its per-bucket deletion vectors by index file name.
     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;
     };