refactor(core): simplify merged key-value reader state

Author: zjw1111

src/paimon/core/io/merged_key_value_record_reader.cpp

@@ -92,7 +92,7 @@ Result<bool> MergedKeyValueRecordReader::Iterator::MergeNextKey() const {
 }
 
 Status MergedKeyValueRecordReader::Iterator::LoadNextRawKeyValue() const {
-    if (next_raw_key_value_.has_value() || eof_) {
+    if (next_raw_key_value_.has_value()) {
         return Status::OK();
     }
 
@@ -109,7 +109,6 @@ Status MergedKeyValueRecordReader::Iterator::LoadNextRawKeyValue() const {
         current_iterator_.reset();
         PAIMON_ASSIGN_OR_RAISE(current_iterator_, reader_->reader_->NextBatch());
         if (current_iterator_ == nullptr) {
-            eof_ = true;
             return Status::OK();
         }
     }

src/paimon/core/io/merged_key_value_record_reader.h

@@ -25,6 +25,9 @@
 namespace paimon {
 class FieldsComparator;
 
+/// Merges consecutive key-value records with the same primary key from the wrapped reader.
+/// The wrapped reader must return records ordered by primary key, so all records for the same
+/// primary key are contiguous.
 class MergedKeyValueRecordReader : public KeyValueRecordReader {
  public:
     MergedKeyValueRecordReader(
@@ -52,7 +55,6 @@ class MergedKeyValueRecordReader : public KeyValueRecordReader {
         mutable std::optional<KeyValue> next_raw_key_value_;
         // Merged kv prepared by HasNext() and consumed by Next().
         mutable std::optional<KeyValue> merged_key_value_;
-        mutable bool eof_ = false;
     };
 
     Result<std::unique_ptr<KeyValueRecordReader::Iterator>> NextBatch() override;