fix review

Author: zjw1111

src/paimon/core/append/append_only_writer.h

@@ -69,6 +69,9 @@ class AppendOnlyWriter : public BatchWriter {
         return Flush(/*wait_for_latest_compaction=*/true, full_compaction);
     }
 
+    uint64_t GetMemoryUsage() const override {
+        return 0;
+    }
     Status FlushMemory() override {
         return Flush(/*wait_for_latest_compaction=*/false, /*forced_full_compaction=*/false);
     }

src/paimon/core/io/merged_key_value_record_reader.cpp

@@ -86,7 +86,6 @@ Status MergedKeyValueRecordReader::Iterator::LoadNextKeyValue() {
         current_iterator_.reset();
         PAIMON_ASSIGN_OR_RAISE(current_iterator_, reader_->reader_->NextBatch());
         if (current_iterator_ == nullptr) {
-            reader_->reader_->Close();
             return Status::OK();
         }
     }
@@ -110,6 +109,7 @@ std::shared_ptr<Metrics> MergedKeyValueRecordReader::GetReaderMetrics() const {
 }
 
 void MergedKeyValueRecordReader::Close() {
+    visited_ = true;
     reader_->Close();
 }
 

src/paimon/core/mergetree/merge_tree_writer.h

@@ -75,6 +75,10 @@ class MergeTreeWriter : public BatchWriter {
 
     Result<CommitIncrement> PrepareCommit(bool wait_compaction) override;
 
+    uint64_t GetMemoryUsage() const override {
+        return 0;
+    }
+
     Status FlushMemory() override {
         return Flush(/*wait_for_latest_compaction=*/false, /*forced_full_compaction=*/false);
     }

src/paimon/core/mergetree/write_buffer.cpp

@@ -16,6 +16,7 @@
 
 #include "paimon/core/mergetree/write_buffer.h"
 
+#include <limits>
 #include <memory>
 #include <utility>
 
@@ -34,6 +35,7 @@ WriteBuffer::WriteBuffer(
     const std::shared_ptr<MergeFunctionWrapper<KeyValue>>& merge_function_wrapper,
     const std::shared_ptr<MemoryPool>& pool)
     : key_comparator_(key_comparator), merge_function_wrapper_(merge_function_wrapper) {
+    // TODO(jinli.zjw): input sequence_fields_ascending as parameter
     sort_buffer_ = std::make_unique<BinaryInMemorySortBuffer>(
         last_sequence_number, value_type, trimmed_primary_keys, user_defined_sequence_fields,
         /*sequence_fields_ascending=*/true, key_comparator,

src/paimon/core/postpone/postpone_bucket_writer.h

@@ -72,6 +72,10 @@ class PostponeBucketWriter : public BatchWriter {
         return Status::NotImplemented("not implemented");
     }
 
+    uint64_t GetMemoryUsage() const override {
+        return 0;
+    }
+
     Status FlushMemory() override {
         return Flush();
     }

src/paimon/core/utils/batch_writer.h

@@ -37,9 +37,7 @@ class BatchWriter {
     virtual ~BatchWriter() = default;
 
     /// Return current in-memory data usage in bytes
-    virtual uint64_t GetMemoryUsage() const {
-        return 0;
-    }
+    virtual uint64_t GetMemoryUsage() const = 0;
 
     /// Flush in-memory data.
     virtual Status FlushMemory() = 0;