Add ignore_delete coverage for merged key value readers

Author: zjw1111

.githooks/post-checkout

@@ -0,0 +1,3 @@
+#!/bin/sh
+command -v git-lfs >/dev/null 2>&1 || { echo >&2 "\nThis repository is configured for Git LFS but 'git-lfs' was not found on your path. If you no longer wish to use Git LFS, remove this hook by deleting .git/hooks/post-checkout.\n"; exit 2; }
+git lfs post-checkout "$@"

.githooks/post-commit

@@ -0,0 +1,3 @@
+#!/bin/sh
+command -v git-lfs >/dev/null 2>&1 || { echo >&2 "\nThis repository is configured for Git LFS but 'git-lfs' was not found on your path. If you no longer wish to use Git LFS, remove this hook by deleting .git/hooks/post-commit.\n"; exit 2; }
+git lfs post-commit "$@"

.githooks/post-merge

@@ -0,0 +1,3 @@
+#!/bin/sh
+command -v git-lfs >/dev/null 2>&1 || { echo >&2 "\nThis repository is configured for Git LFS but 'git-lfs' was not found on your path. If you no longer wish to use Git LFS, remove this hook by deleting .git/hooks/post-merge.\n"; exit 2; }
+git lfs post-merge "$@"

.githooks/pre-push

@@ -0,0 +1,3 @@
+#!/bin/sh
+command -v git-lfs >/dev/null 2>&1 || { echo >&2 "\nThis repository is configured for Git LFS but 'git-lfs' was not found on your path. If you no longer wish to use Git LFS, remove this hook by deleting .git/hooks/pre-push.\n"; exit 2; }
+git lfs pre-push "$@"

git.diff

@@ -0,0 +1,169 @@
+diff --git a/src/paimon/core/io/merged_key_value_record_reader.cpp b/src/paimon/core/io/merged_key_value_record_reader.cpp
+index 8a8a9cc..5a94236 100644
+--- a/src/paimon/core/io/merged_key_value_record_reader.cpp
++++ b/src/paimon/core/io/merged_key_value_record_reader.cpp
+@@ -17,6 +17,7 @@
+ #include "paimon/core/io/merged_key_value_record_reader.h"
+ 
+ #include <cassert>
++#include <memory>
+ #include <optional>
+ #include <utility>
+ 
+@@ -36,56 +37,77 @@ MergedKeyValueRecordReader::MergedKeyValueRecordReader(
+     assert(merge_function_wrapper_ != nullptr);
+ }
+ 
+-Result<std::unique_ptr<MergedKeyValueRecordReader::Iterator>>
+-MergedKeyValueRecordReader::Iterator::Create(MergedKeyValueRecordReader* reader) {
+-    std::unique_ptr<Iterator> iterator(new Iterator(reader));
+-    PAIMON_RETURN_NOT_OK(iterator->LoadNextKeyValue());
+-    return iterator;
++Result<bool> MergedKeyValueRecordReader::Iterator::HasNext() const {
++    if (next_merged_key_value_.has_value()) {
++        return true;
++    }
++    return PrepareNextMergedKeyValue();
+ }
+ 
+ Result<KeyValue> MergedKeyValueRecordReader::Iterator::Next() {
+-    assert(next_key_value_.has_value());
+-    reader_->merge_function_wrapper_->Reset();
+-    auto current_key = next_key_value_->key;
+-    PAIMON_RETURN_NOT_OK(reader_->merge_function_wrapper_->Add(std::move(*next_key_value_)));
+-    next_key_value_.reset();
++    if (!next_merged_key_value_.has_value()) {
++        return Status::Invalid("No more merged key values in current iterator");
++    }
+ 
++    KeyValue result = std::move(*next_merged_key_value_);
++    next_merged_key_value_.reset();
++    return result;
++}
++
++Result<bool> MergedKeyValueRecordReader::Iterator::PrepareNextMergedKeyValue() const {
+     while (true) {
+-        PAIMON_RETURN_NOT_OK(LoadNextKeyValue());
+-        if (!next_key_value_.has_value()) {
+-            break;
++        PAIMON_ASSIGN_OR_RAISE(bool has_group, MergeNextKeyGroup());
++        if (!has_group) {
++            return false;
+         }
+-        if (reader_->key_comparator_->CompareTo(*current_key, *next_key_value_->key) != 0) {
+-            break;
++        if (next_merged_key_value_.has_value()) {
++            return true;
+         }
+-        PAIMON_RETURN_NOT_OK(reader_->merge_function_wrapper_->Add(std::move(*next_key_value_)));
+-        next_key_value_.reset();
+     }
++}
++
++Result<bool> MergedKeyValueRecordReader::Iterator::MergeNextKeyGroup() const {
++    PAIMON_RETURN_NOT_OK(LoadLookaheadKeyValue());
++    if (!lookahead_key_value_.has_value()) {
++        return false;
++    }
++
++    reader_->merge_function_wrapper_->Reset();
++    auto current_key = lookahead_key_value_->key;
++
++    do {
++        PAIMON_RETURN_NOT_OK(
++            reader_->merge_function_wrapper_->Add(std::move(*lookahead_key_value_)));
++        lookahead_key_value_.reset();
++        PAIMON_RETURN_NOT_OK(LoadLookaheadKeyValue());
++    } while (lookahead_key_value_.has_value() &&
++             reader_->key_comparator_->CompareTo(*current_key, *lookahead_key_value_->key) == 0);
+ 
+     PAIMON_ASSIGN_OR_RAISE(std::optional<KeyValue> result,
+                            reader_->merge_function_wrapper_->GetResult());
+-    // TODO(jinli.zjw): support merge function producing no result (e.g. all rows are filtered out)
+-    if (result == std::nullopt) {
+-        return Status::Invalid("merged key value reader produced empty result");
+-    }
+-    return std::move(result).value();
++    next_merged_key_value_ = std::move(result);
++    return true;
+ }
+ 
+-Status MergedKeyValueRecordReader::Iterator::LoadNextKeyValue() {
+-    if (next_key_value_.has_value()) {
++Status MergedKeyValueRecordReader::Iterator::LoadLookaheadKeyValue() const {
++    if (lookahead_key_value_.has_value() || exhausted_) {
+         return Status::OK();
+     }
+ 
+     while (true) {
+-        if (current_iterator_ != nullptr && current_iterator_->HasNext()) {
+-            PAIMON_ASSIGN_OR_RAISE(KeyValue key_value, current_iterator_->Next());
+-            next_key_value_.emplace(std::move(key_value));
+-            return Status::OK();
++        if (current_iterator_ != nullptr) {
++            PAIMON_ASSIGN_OR_RAISE(bool has_next, current_iterator_->HasNext());
++            if (has_next) {
++                PAIMON_ASSIGN_OR_RAISE(KeyValue key_value, current_iterator_->Next());
++                lookahead_key_value_.emplace(std::move(key_value));
++                return Status::OK();
++            }
+         }
+ 
+         current_iterator_.reset();
+         PAIMON_ASSIGN_OR_RAISE(current_iterator_, reader_->reader_->NextBatch());
+         if (current_iterator_ == nullptr) {
++            exhausted_ = true;
+             return Status::OK();
+         }
+     }
+@@ -97,8 +119,9 @@ Result<std::unique_ptr<KeyValueRecordReader::Iterator>> MergedKeyValueRecordRead
+     }
+     visited_ = true;
+ 
+-    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<Iterator> iterator, Iterator::Create(this));
+-    if (!iterator->HasNext()) {
++    auto iterator = std::make_unique<Iterator>(this);
++    PAIMON_ASSIGN_OR_RAISE(bool has_next, iterator->HasNext());
++    if (!has_next) {
+         return std::unique_ptr<KeyValueRecordReader::Iterator>();
+     }
+     return iterator;
+diff --git a/src/paimon/core/io/merged_key_value_record_reader.h b/src/paimon/core/io/merged_key_value_record_reader.h
+index 4d6ef54..e458c1d 100644
+--- a/src/paimon/core/io/merged_key_value_record_reader.h
++++ b/src/paimon/core/io/merged_key_value_record_reader.h
+@@ -34,23 +34,25 @@ class MergedKeyValueRecordReader : public KeyValueRecordReader {
+ 
+     class Iterator : public KeyValueRecordReader::Iterator {
+      public:
+-        static Result<std::unique_ptr<Iterator>> Create(MergedKeyValueRecordReader* reader);
++        explicit Iterator(MergedKeyValueRecordReader* reader) : reader_(reader) {}
+ 
+-        bool HasNext() const override {
+-            return next_key_value_.has_value();
+-        }
++        Result<bool> HasNext() const override;
+ 
+         Result<KeyValue> Next() override;
+ 
+      private:
+-        explicit Iterator(MergedKeyValueRecordReader* reader) : reader_(reader) {}
+-
+-        Status LoadNextKeyValue();
++        Result<bool> PrepareNextMergedKeyValue() const;
++        Result<bool> MergeNextKeyGroup() const;
++        Status LoadLookaheadKeyValue() const;
+ 
+      private:
+         MergedKeyValueRecordReader* reader_;
+-        std::unique_ptr<KeyValueRecordReader::Iterator> current_iterator_;
+-        std::optional<KeyValue> next_key_value_;
++        mutable std::unique_ptr<KeyValueRecordReader::Iterator> current_iterator_;
++        // Lookahead raw kv used to detect the boundary between two keys.
++        mutable std::optional<KeyValue> lookahead_key_value_;
++        // Merged kv prepared by HasNext() and consumed by Next().
++        mutable std::optional<KeyValue> next_merged_key_value_;
++        mutable bool exhausted_ = false;
+     };
+ 
+     Result<std::unique_ptr<KeyValueRecordReader::Iterator>> NextBatch() override;

src/paimon/core/io/key_value_data_file_record_reader.cpp

@@ -48,7 +48,7 @@ KeyValueDataFileRecordReader::KeyValueDataFileRecordReader(
       value_schema_(value_schema),
       value_names_(value_schema_->field_names()) {}
 
-bool KeyValueDataFileRecordReader::Iterator::HasNext() const {
+Result<bool> KeyValueDataFileRecordReader::Iterator::HasNext() const {
     int64_t array_length = reader_->row_kind_array_->length();
     const auto& selection_bitmap = reader_->selection_bitmap_;
     if (selection_bitmap.Cardinality() == array_length) {
@@ -67,7 +67,6 @@ bool KeyValueDataFileRecordReader::Iterator::HasNext() const {
 }
 
 Result<KeyValue> KeyValueDataFileRecordReader::Iterator::Next() {
-    assert(HasNext());
     // key is only used in merge sort; key context does not hold parent struct array
     auto key = std::make_unique<ColumnarRowRef>(reader_->key_ctx_, cursor_);
     // value is used in merge sort and projection (maybe async and multi-thread), so value context

src/paimon/core/io/key_value_data_file_record_reader.h

@@ -56,7 +56,7 @@ class KeyValueDataFileRecordReader : public KeyValueRecordReader {
      public:
         Iterator(KeyValueDataFileRecordReader* reader, int64_t previous_batch_first_row_number)
             : previous_batch_first_row_number_(previous_batch_first_row_number), reader_(reader) {}
-        bool HasNext() const override;
+        Result<bool> HasNext() const override;
         Result<KeyValue> Next() override;
         Result<std::pair<int64_t, KeyValue>> NextWithFilePos();
 

src/paimon/core/io/key_value_data_file_record_reader_test.cpp

@@ -285,7 +285,11 @@ TEST_F(KeyValueDataFileRecordReaderTest, TestWithSelectedBitmapWithFilePos) {
         auto typed_iter = dynamic_cast<KeyValueDataFileRecordReader::Iterator*>(iter);
         ASSERT_TRUE(typed_iter);
         size_t pos_iter = 0;
-        while (iter->HasNext()) {
+        while (true) {
+            ASSERT_OK_AND_ASSIGN(bool has_next, iter->HasNext());
+            if (!has_next) {
+                break;
+            }
             ASSERT_OK_AND_ASSIGN(auto kv_and_pos, typed_iter->NextWithFilePos());
             const auto& [pos, kv] = kv_and_pos;
             ASSERT_EQ(pos, expected_pos_vector[pos_iter++]);

src/paimon/core/io/key_value_in_memory_record_reader.h

@@ -55,7 +55,7 @@ class KeyValueInMemoryRecordReader : public KeyValueRecordReader {
     class Iterator : public KeyValueRecordReader::Iterator {
      public:
         explicit Iterator(KeyValueInMemoryRecordReader* reader) : reader_(reader) {}
-        bool HasNext() const override {
+        Result<bool> HasNext() const override {
             return cursor_ < reader_->value_struct_array_->length();
         }
         Result<KeyValue> Next() override;

src/paimon/core/io/key_value_record_reader.h

@@ -20,6 +20,7 @@
 
 #include "paimon/core/key_value.h"
 #include "paimon/metrics.h"
+#include "paimon/result.h"
 namespace paimon {
 class KeyValueRecordReader {
  public:
@@ -28,7 +29,7 @@ class KeyValueRecordReader {
     class Iterator {
      public:
         virtual ~Iterator() = default;
-        virtual bool HasNext() const = 0;
+        virtual Result<bool> HasNext() const = 0;
         virtual Result<KeyValue> Next() = 0;
     };