fix(read): correct data evolution selection and blob backfill reads

include/paimon/utils/roaring_bitmap32.h

@@ -65,6 +65,8 @@ class PAIMON_EXPORT RoaringBitmap32 {
         bool operator!=(const Iterator& other) const;
 
      private:
+        Iterator() = default;
+        // Null represents the end sentinel without an underlying iterator allocation.
         void* iterator_ = nullptr;
     };
 
@@ -136,6 +138,7 @@ class PAIMON_EXPORT RoaringBitmap32 {
     std::string ToString() const;
 
     Iterator Begin() const;
+    /// Returns an end sentinel without allocating an underlying iterator.
     Iterator End() const;
     /// @return the iterator moved to the value which is equal or larger than key
     Iterator EqualOrLarger(int32_t key) const;

include/paimon/utils/roaring_bitmap64.h

@@ -65,9 +65,12 @@ class PAIMON_EXPORT RoaringBitmap64 {
         bool operator==(const Iterator& other) const;
         bool operator!=(const Iterator& other) const;
         /// Move the iterator to the value which is equal or larger than input value
+        /// An End() sentinel remains at the end.
         void EqualOrLarger(int64_t value);
 
      private:
+        Iterator() = default;
+        // Null represents the end sentinel without an underlying iterator allocation.
         void* iterator_ = nullptr;
     };
 
@@ -149,6 +152,7 @@ class PAIMON_EXPORT RoaringBitmap64 {
     std::string ToString() const;
 
     Iterator Begin() const;
+    /// Returns an end sentinel without allocating an underlying iterator.
     Iterator End() const;
     /// @return the iterator moved to the value which is equal or larger than key
     Iterator EqualOrLarger(int64_t key) const;

src/paimon/common/reader/blob_fallback_batch_reader.cpp

@@ -42,10 +42,8 @@ Result<std::unique_ptr<BlobFallbackBatchReader>> BlobFallbackBatchReader::Create
     std::vector<std::vector<Segment>>&& sequence_groups,
     const std::shared_ptr<arrow::Schema>& read_schema, int32_t read_batch_size,
     const std::shared_ptr<arrow::MemoryPool>& arrow_pool) {
-    if (sequence_groups.size() < 2) {
-        return Status::Invalid(
-            "Blob fallback needs at least two sequence groups; a single group should be read "
-            "sequentially.");
+    if (sequence_groups.empty()) {
+        return Status::Invalid("Blob fallback needs at least one sequence group.");
     }
     if (read_schema == nullptr) {
         return Status::Invalid("Blob fallback read schema cannot be nullptr.");
@@ -339,6 +337,9 @@ Result<BatchReader::ReadBatch> BlobFallbackBatchReader::NextBatch() {
         return Status::Invalid("blob fallback batch reader is closed");
     }
     std::vector<std::vector<Chunk>> group_chunks(groups_.size());
+    // TODO(xinyu.lxy): Older layers are materialized even when newer values cover them, adding
+    // unnecessary reads and potentially exposing stale payload errors. Add Java-like read/skip
+    // support to materialize older values only when fallback is needed.
     // the first (newest) group defines the window; the others must step in lockstep
     PAIMON_ASSIGN_OR_RAISE(int64_t row_count, FillWindow(0, read_batch_size_, &group_chunks[0]));
     for (size_t g = 1; g < groups_.size(); g++) {

src/paimon/common/reader/blob_fallback_batch_reader.h

@@ -73,9 +73,9 @@ class BlobFallbackBatchReader : public BatchReader {
     };
 
     /// `sequence_groups` must be ordered by descending max sequence number and contain at least
-    /// two groups (a single group needs no fallback). `read_schema` is the schema every file
-    /// reader emits; it must contain exactly one blob field and may additionally contain the
-    /// row-tracking fields _ROW_ID and _SEQUENCE_NUMBER (completed per file by
+    /// one group (a single group can have gaps to fill with NULLs). `read_schema` is the schema
+    /// every file reader emits; it must contain exactly one blob field and may additionally contain
+    /// the row-tracking fields _ROW_ID and _SEQUENCE_NUMBER (completed per file by
     /// CompleteRowTrackingFieldsBatchReader), which stay correct for rows that are a
     /// placeholder in every layer.
     static Result<std::unique_ptr<BlobFallbackBatchReader>> Create(

src/paimon/common/reader/blob_fallback_batch_reader_test.cpp

@@ -145,6 +145,11 @@ TEST_F(BlobFallbackBatchReaderTest, TestBasicFallback) {
         {"b0", "u1", "b2"});
 }
 
+TEST_F(BlobFallbackBatchReaderTest, TestSingleLayerFillsUnwrittenRows) {
+    CheckFallback({{SegmentSpec::Gap(2), SegmentSpec::File({"b2", "PH"}), SegmentSpec::Gap(2)}},
+                  {std::nullopt, std::nullopt, "b2", std::nullopt, std::nullopt, std::nullopt});
+}
+
 TEST_F(BlobFallbackBatchReaderTest, TestGapPadding) {
     // the newer layer only covers rows 2-3; the gaps stand for placeholders
     CheckFallback({{SegmentSpec::Gap(2), SegmentSpec::File({"u2", "PH"})},
@@ -325,12 +330,10 @@ TEST_F(BlobFallbackBatchReaderTest, TestMisalignedGroupsFail) {
 }
 
 TEST_F(BlobFallbackBatchReaderTest, TestCreateValidation) {
-    // a single group needs no fallback
-    std::vector<std::vector<BlobFallbackBatchReader::Segment>> single_group;
-    single_group.push_back(MakeGroup({SegmentSpec::File({"b0"})}, 1024));
-    ASSERT_NOK_WITH_MSG(BlobFallbackBatchReader::Create(std::move(single_group), read_schema_, 1024,
+    std::vector<std::vector<BlobFallbackBatchReader::Segment>> empty_groups;
+    ASSERT_NOK_WITH_MSG(BlobFallbackBatchReader::Create(std::move(empty_groups), read_schema_, 1024,
                                                         GetArrowPool(pool_)),
-                        "at least two sequence groups");
+                        "at least one sequence group");
 
     // the read schema must contain a blob field
     std::vector<std::vector<BlobFallbackBatchReader::Segment>> groups;

src/paimon/common/utils/roaring_bitmap32.cpp

@@ -56,6 +56,11 @@ RoaringBitmap32::Iterator& RoaringBitmap32::Iterator::operator=(
     if (&other == this) {
         return *this;
     }
+    if (!other.iterator_) {
+        delete static_cast<roaring::Roaring::const_iterator*>(iterator_);
+        iterator_ = nullptr;
+        return *this;
+    }
     if (!iterator_) {
         iterator_ = new roaring::Roaring::const_iterator(GetIterator(other.iterator_));
     } else {
@@ -92,7 +97,11 @@ bool RoaringBitmap32::Iterator::operator==(const Iterator& other) const {
     if (&other == this) {
         return true;
     }
-    assert(iterator_ && other.iterator_);
+    bool is_end = !iterator_ || !GetIterator(iterator_).i.has_value;
+    bool other_is_end = !other.iterator_ || !GetIterator(other.iterator_).i.has_value;
+    if (is_end || other_is_end) {
+        return is_end == other_is_end;
+    }
     return GetIterator(iterator_) == GetIterator(other.iterator_);
 }
 bool RoaringBitmap32::Iterator::operator!=(const Iterator& other) const {
@@ -301,9 +310,7 @@ RoaringBitmap32::Iterator RoaringBitmap32::Begin() const {
 }
 
 RoaringBitmap32::Iterator RoaringBitmap32::End() const {
-    RoaringBitmap32::Iterator iter(*this);
-    GetIterator(iter.iterator_) = GetRoaringBitmap(roaring_bitmap_).end();
-    return iter;
+    return Iterator();
 }
 
 RoaringBitmap32::Iterator RoaringBitmap32::EqualOrLarger(int32_t key) const {

src/paimon/common/utils/roaring_bitmap32_test.cpp

@@ -256,6 +256,21 @@ TEST(RoaringBitmap32Test, TestIterator) {
     ASSERT_EQ(iter, roaring.End());
 }
 
+TEST(RoaringBitmap32Test, TestEndIterator) {
+    RoaringBitmap32 empty;
+    ASSERT_EQ(empty.Begin(), empty.End());
+    ASSERT_EQ(empty.EqualOrLarger(0), empty.End());
+
+    auto bitmap = RoaringBitmap32::From({0, 10, 20});
+    auto iter = bitmap.Begin();
+    std::vector<int32_t> values;
+    for (; iter != bitmap.End(); ++iter) {
+        values.push_back(*iter);
+    }
+    ASSERT_EQ(values, (std::vector<int32_t>{0, 10, 20}));
+    ASSERT_EQ(iter, bitmap.End());
+}
+
 TEST(RoaringBitmap32Test, TestNextAndPreviousValue) {
     RoaringBitmap32 roaring = RoaringBitmap32::From({10, 20, 30});
 
@@ -307,6 +322,18 @@ TEST(RoaringBitmap32Test, TestIteratorAssignAndMove) {
     iter3 = std::move(iter2);
     ASSERT_EQ(10, *iter3);
     ASSERT_FALSE(iter2.iterator_);  // NOLINT(bugprone-use-after-move)
+
+    iter3 = roaring.End();
+    auto copied_end = iter3;
+    auto moved_end = std::move(copied_end);
+    ASSERT_EQ(moved_end, roaring.End());
+    iter3 = roaring.Begin();
+    iter3 = moved_end;
+    ASSERT_EQ(iter3, roaring.End());
+    iter3 = roaring.Begin();
+    ASSERT_EQ(*iter3, 10);
+    iter3 = std::move(moved_end);
+    ASSERT_EQ(iter3, roaring.End());
 }
 
 TEST(RoaringBitmap32Test, TestDeserializeFromInputStream) {

src/paimon/common/utils/roaring_bitmap64.cpp

@@ -30,22 +30,32 @@
 
 namespace paimon {
 namespace {
-roaring::Roaring64Map::const_iterator& GetIterator(void* iter) {
-    return *(static_cast<roaring::Roaring64Map::const_iterator*>(iter));
+// Expose exhaustion without constructing another native iterator or modifying CRoaring.
+class Bitmap64Iterator : public roaring::Roaring64Map::const_iterator {
+ public:
+    explicit Bitmap64Iterator(const roaring::Roaring64Map& bitmap)
+        : roaring::Roaring64Map::const_iterator(bitmap) {}
+
+    bool IsEnd() const {
+        return map_iter == map_end;
+    }
+};
+
+Bitmap64Iterator& GetIterator(void* iter) {
+    return *(static_cast<Bitmap64Iterator*>(iter));
 }
 roaring::Roaring64Map& GetRoaringBitmap(void* bitmap) {
     return *(static_cast<roaring::Roaring64Map*>(bitmap));
 }
 }  // namespace
 
 RoaringBitmap64::Iterator::Iterator(const RoaringBitmap64& bitmap) {
-    iterator_ =
-        new roaring::Roaring64Map::const_iterator(GetRoaringBitmap(bitmap.roaring_bitmap_).begin());
+    iterator_ = new Bitmap64Iterator(GetRoaringBitmap(bitmap.roaring_bitmap_));
 }
 
 RoaringBitmap64::Iterator::~Iterator() {
     if (iterator_) {
-        delete static_cast<roaring::Roaring64Map::const_iterator*>(iterator_);
+        delete static_cast<Bitmap64Iterator*>(iterator_);
     }
 }
 
@@ -57,8 +67,13 @@ RoaringBitmap64::Iterator& RoaringBitmap64::Iterator::operator=(
     if (&other == this) {
         return *this;
     }
+    if (!other.iterator_) {
+        delete static_cast<Bitmap64Iterator*>(iterator_);
+        iterator_ = nullptr;
+        return *this;
+    }
     if (!iterator_) {
-        iterator_ = new roaring::Roaring64Map::const_iterator(GetIterator(other.iterator_));
+        iterator_ = new Bitmap64Iterator(GetIterator(other.iterator_));
     } else {
         GetIterator(iterator_) = GetIterator(other.iterator_);
     }
@@ -75,7 +90,7 @@ RoaringBitmap64::Iterator& RoaringBitmap64::Iterator::operator=(
         return *this;
     }
     if (iterator_) {
-        delete static_cast<roaring::Roaring64Map::const_iterator*>(iterator_);
+        delete static_cast<Bitmap64Iterator*>(iterator_);
     }
     iterator_ = other.iterator_;
     other.iterator_ = nullptr;
@@ -93,14 +108,21 @@ bool RoaringBitmap64::Iterator::operator==(const Iterator& other) const {
     if (&other == this) {
         return true;
     }
-    assert(iterator_ && other.iterator_);
+    bool is_end = !iterator_ || GetIterator(iterator_).IsEnd();
+    bool other_is_end = !other.iterator_ || GetIterator(other.iterator_).IsEnd();
+    if (is_end || other_is_end) {
+        return is_end == other_is_end;
+    }
     return GetIterator(iterator_) == GetIterator(other.iterator_);
 }
 bool RoaringBitmap64::Iterator::operator!=(const Iterator& other) const {
     return !(*this == other);
 }
 
 void RoaringBitmap64::Iterator::EqualOrLarger(int64_t value) {
+    if (!iterator_) {
+        return;
+    }
     [[maybe_unused]] bool _ = GetIterator(iterator_).move(value);
 }
 
@@ -372,17 +394,12 @@ RoaringBitmap64::Iterator RoaringBitmap64::Begin() const {
 }
 
 RoaringBitmap64::Iterator RoaringBitmap64::End() const {
-    RoaringBitmap64::Iterator iter(*this);
-    GetIterator(iter.iterator_) = GetRoaringBitmap(roaring_bitmap_).end();
-    return iter;
+    return Iterator();
 }
 
 RoaringBitmap64::Iterator RoaringBitmap64::EqualOrLarger(int64_t key) const {
     RoaringBitmap64::Iterator iter(*this);
-    bool not_end = GetIterator(iter.iterator_).move(key);
-    if (!not_end) {
-        return End();
-    }
+    iter.EqualOrLarger(key);
     return iter;
 }
 

src/paimon/common/utils/roaring_bitmap64_test.cpp

@@ -275,6 +275,22 @@ TEST(RoaringBitmap64Test, TestIterator) {
     ASSERT_EQ(iter, roaring.End());
 }
 
+TEST(RoaringBitmap64Test, TestEndIterator) {
+    RoaringBitmap64 empty;
+    ASSERT_EQ(empty.Begin(), empty.End());
+    ASSERT_EQ(empty.EqualOrLarger(0), empty.End());
+
+    const std::vector<int64_t> expected = {0, 1LL << 32, RoaringBitmap64::MAX_VALUE};
+    auto bitmap = RoaringBitmap64::From(expected);
+    auto iter = bitmap.Begin();
+    std::vector<int64_t> values;
+    for (; iter != bitmap.End(); ++iter) {
+        values.push_back(*iter);
+    }
+    ASSERT_EQ(values, expected);
+    ASSERT_EQ(iter, bitmap.End());
+}
+
 TEST(RoaringBitmap64Test, TestIteratorAssignAndMove) {
     RoaringBitmap64 roaring =
         RoaringBitmap64::From({10000000000010l, 100000000000100l, 100000000000200l});
@@ -308,6 +324,18 @@ TEST(RoaringBitmap64Test, TestIteratorAssignAndMove) {
     iter3 = std::move(iter2);
     ASSERT_EQ(10000000000010l, *iter3);
     ASSERT_FALSE(iter2.iterator_);  // NOLINT(bugprone-use-after-move)
+
+    iter3 = roaring.End();
+    auto copied_end = iter3;
+    auto moved_end = std::move(copied_end);
+    ASSERT_EQ(moved_end, roaring.End());
+    iter3 = roaring.Begin();
+    iter3 = moved_end;
+    ASSERT_EQ(iter3, roaring.End());
+    iter3 = roaring.Begin();
+    ASSERT_EQ(*iter3, 10000000000010l);
+    iter3 = std::move(moved_end);
+    ASSERT_EQ(iter3, roaring.End());
 }
 
 TEST(RoaringBitmap64Test, TestDeserializeFromInputStream) {
@@ -417,6 +445,14 @@ TEST(RoaringBitmap64Test, TestIteratorEqualOrLarger) {
     ASSERT_EQ(*iter, 100l);
     iter.EqualOrLarger(200l);
     ASSERT_EQ(iter, roaring.End());
+    ASSERT_EQ(roaring.End(), iter);
+    iter.EqualOrLarger(3l);
+    ASSERT_EQ(*iter, 3l);
+
+    iter = roaring.EqualOrLarger(200l);
+    ASSERT_EQ(iter, roaring.End());
+    iter.EqualOrLarger(5l);
+    ASSERT_EQ(*iter, 5l);
 }
 
 // Helper function to convert a RoaringBitmap64 to a list of contiguous ranges.

src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader.h

@@ -57,8 +57,7 @@ class ApplyDeletionVectorBatchReader : public FileBatchReader {
                 return batch_with_bitmap;
             }
             auto& [batch, bitmap] = batch_with_bitmap;
-            PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 valid_bitmap, Filter(batch.first->length));
-            bitmap &= valid_bitmap;
+            PAIMON_ASSIGN_OR_RAISE(bitmap, Filter(bitmap));
             if (bitmap.IsEmpty()) {
                 ReaderUtils::ReleaseReadBatch(std::move(batch));
                 continue;
@@ -101,9 +100,12 @@ class ApplyDeletionVectorBatchReader : public FileBatchReader {
     }
 
  private:
-    Result<RoaringBitmap32> Filter(int32_t batch_size) const {
+    Result<RoaringBitmap32> Filter(const RoaringBitmap32& selection) const {
         RoaringBitmap32 is_valid;
-        for (int32_t i = 0; i < batch_size; ++i) {
+        // Unselected rows may lie outside the group's anchor range. Probe only rows that the
+        // format/index selection actually returns, just as ApplyDeletionVectorReader does.
+        for (auto iter = selection.Begin(); iter != selection.End(); ++iter) {
+            int32_t i = *iter;
             PAIMON_ASSIGN_OR_RAISE(uint64_t file_row_id, reader_->GetPreviousBatchFileRowId(i));
             PAIMON_ASSIGN_OR_RAISE(bool is_deleted, deletion_vector_->IsDeleted(file_row_id));
             if (!is_deleted) {