feat: add metrics storage-read-bytes for parquet and fix clean inte test (#439)

Author: lucasfang

src/paimon/common/utils/arrow/arrow_input_stream_adapter.cpp

@@ -43,9 +43,12 @@ arrow::Status ValidateArrowIoRange(int64_t value, const char* name) {
 }  // namespace
 
 ArrowInputStreamAdapter::ArrowInputStreamAdapter(
-    const std::shared_ptr<paimon::InputStream>& input_stream,
-    const std::shared_ptr<arrow::MemoryPool>& pool, int64_t file_size)
-    : input_stream_(input_stream), pool_(pool), file_size_(file_size) {
+    const std::shared_ptr<paimon::InputStream>& input_stream, int64_t file_size,
+    const std::shared_ptr<arrow::MemoryPool>& pool)
+    : input_stream_(input_stream),
+      pool_(pool),
+      file_size_(file_size),
+      storage_read_bytes_(std::make_shared<std::atomic<uint64_t>>(0)) {
     assert(file_size >= 0);
 }
 
@@ -63,6 +66,9 @@ arrow::Result<int64_t> ArrowInputStreamAdapter::Read(int64_t nbytes, void* out)
     if (!read_bytes.ok()) {
         return ToArrowStatus(read_bytes.status());
     }
+    if (storage_read_bytes_) {
+        storage_read_bytes_->fetch_add(static_cast<uint64_t>(read_bytes.value()));
+    }
     return read_bytes.value();
 }
 
@@ -84,6 +90,9 @@ arrow::Result<int64_t> ArrowInputStreamAdapter::ReadAt(int64_t position, int64_t
     if (!read_bytes.ok()) {
         return ToArrowStatus(read_bytes.status());
     }
+    if (storage_read_bytes_) {
+        storage_read_bytes_->fetch_add(static_cast<uint64_t>(read_bytes.value()));
+    }
     return read_bytes.value();
 }
 
@@ -119,14 +128,19 @@ arrow::Future<std::shared_ptr<arrow::Buffer>> ArrowInputStreamAdapter::ReadAsync
         return fut;
     }
     std::shared_ptr<arrow::Buffer> buffer = std::move(buffer_result).ValueUnsafe();
-    input_stream_->ReadAsync(reinterpret_cast<char*>(buffer->mutable_data()), nbytes, position,
-                             [fut, buffer](Status callback_status) mutable {
-                                 if (callback_status.ok()) {
-                                     fut.MarkFinished(std::move(buffer));
-                                 } else {
-                                     fut.MarkFinished(ToArrowStatus(callback_status));
-                                 }
-                             });
+    std::shared_ptr<std::atomic<uint64_t>> storage_read_bytes = storage_read_bytes_;
+    input_stream_->ReadAsync(
+        reinterpret_cast<char*>(buffer->mutable_data()), nbytes, position,
+        [fut, buffer, storage_read_bytes, nbytes](Status callback_status) mutable {
+            if (callback_status.ok()) {
+                if (storage_read_bytes) {
+                    storage_read_bytes->fetch_add(static_cast<uint64_t>(nbytes));
+                }
+                fut.MarkFinished(std::move(buffer));
+            } else {
+                fut.MarkFinished(ToArrowStatus(callback_status));
+            }
+        });
     return fut;
 }
 

src/paimon/common/utils/arrow/arrow_input_stream_adapter.h

@@ -16,6 +16,7 @@
 
 #pragma once
 
+#include <atomic>
 #include <cstdint>
 #include <memory>
 
@@ -30,7 +31,7 @@ class InputStream;
 class PAIMON_EXPORT ArrowInputStreamAdapter : public arrow::io::RandomAccessFile {
  public:
     ArrowInputStreamAdapter(const std::shared_ptr<paimon::InputStream>& input_stream,
-                            const std::shared_ptr<arrow::MemoryPool>& pool, int64_t file_size);
+                            int64_t file_size, const std::shared_ptr<arrow::MemoryPool>& pool);
     ~ArrowInputStreamAdapter() override;
 
     // NOTE: In paimon file system definition, position + nbytes should not exceed file_size_.
@@ -49,12 +50,21 @@ class PAIMON_EXPORT ArrowInputStreamAdapter : public arrow::io::RandomAccessFile
     }
     bool closed() const override;
 
+    // Accumulated bytes read from the underlying stream (storageReadBytes). The counter is owned
+    // by this adapter and initialized to 0; callers may retain the returned shared_ptr to read the
+    // value after the adapter is closed or destroyed.
+    const std::shared_ptr<std::atomic<uint64_t>>& StorageReadBytes() const {
+        return storage_read_bytes_;
+    }
+
  private:
     arrow::Status DoClose();
 
     std::shared_ptr<paimon::InputStream> input_stream_;
     std::shared_ptr<arrow::MemoryPool> pool_;
     int64_t file_size_;
+    // Accumulates the number of bytes read from the underlying stream (storageReadBytes).
+    std::shared_ptr<std::atomic<uint64_t>> storage_read_bytes_;
     bool closed_ = false;
 };
 

src/paimon/common/utils/arrow/arrow_stream_adapter_test.cpp

@@ -59,7 +59,7 @@ TEST(ArrowStreamAdapterTest, TestInputAndOutputStream) {
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, file_system->Open(file_name));
     ASSERT_OK_AND_ASSIGN(int64_t length, in->Length());
     auto in_stream =
-        std::make_unique<ArrowInputStreamAdapter>(in, GetArrowPool(GetDefaultPool()), length);
+        std::make_unique<ArrowInputStreamAdapter>(in, length, GetArrowPool(GetDefaultPool()));
     ASSERT_EQ(in_stream->GetSize().ValueOrDie(), static_cast<int64_t>(data.length()));
     ASSERT_EQ(in_stream->Tell().ValueOrDie(), 0);
     ASSERT_FALSE(in_stream->closed());

src/paimon/core/mergetree/spill_reader.cpp

@@ -54,7 +54,7 @@ Status SpillReader::Open(const FileIOChannel::ID& channel_id) {
     PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStatus> file_status, fs_->GetFileStatus(file_path));
     int64_t file_len = file_status->GetLen();
     arrow_input_stream_adapter_ =
-        std::make_shared<ArrowInputStreamAdapter>(in_stream_, arrow_pool_, file_len);
+        std::make_shared<ArrowInputStreamAdapter>(in_stream_, file_len, arrow_pool_);
     auto ipc_read_options = arrow::ipc::IpcReadOptions::Defaults();
     ipc_read_options.memory_pool = arrow_pool_.get();
     ipc_read_options.use_threads = use_threads_;

src/paimon/core/operation/file_store_commit_impl.cpp

@@ -892,7 +892,6 @@ Result<int32_t> FileStoreCommitImpl::TryCommit(
     Snapshot::CommitKind commit_kind, bool detect_conflicts) {
     int32_t retry_count = 0;
     int64_t start_millis = DateTimeUtils::GetCurrentUTCTimeUs() / 1000;
-    std::optional<int64_t> retry_start_snapshot_id;
     while (true) {
         PAIMON_ASSIGN_OR_RAISE(std::optional<Snapshot> latest_snapshot,
                                snapshot_manager_->LatestSnapshot());
@@ -902,13 +901,10 @@ Result<int32_t> FileStoreCommitImpl::TryCommit(
             bool commit_success,
             TryCommitOnce(commit_changes->delta_files, commit_changes->changelog_files,
                           commit_changes->index_entries, identifier, watermark, properties,
-                          commit_kind, latest_snapshot, detect_conflicts, retry_start_snapshot_id));
+                          commit_kind, latest_snapshot, detect_conflicts));
         if (commit_success) {
             break;
         }
-        retry_start_snapshot_id = latest_snapshot
-                                      ? std::optional<int64_t>(latest_snapshot.value().Id() + 1)
-                                      : std::optional<int64_t>(Snapshot::FIRST_SNAPSHOT_ID);
         int64_t current_millis = DateTimeUtils::GetCurrentUTCTimeUs() / 1000;
         if (current_millis - start_millis > options_.GetCommitTimeout() ||
             retry_count >= options_.GetCommitMaxRetries()) {
@@ -923,26 +919,6 @@ Result<int32_t> FileStoreCommitImpl::TryCommit(
     return retry_count + 1;
 }
 
-Result<bool> FileStoreCommitImpl::CheckCommitted(const std::optional<Snapshot>& latest_snapshot,
-                                                 std::optional<int64_t> retry_start_snapshot_id,
-                                                 int64_t identifier,
-                                                 const Snapshot::CommitKind& commit_kind) const {
-    if (!latest_snapshot || !retry_start_snapshot_id ||
-        retry_start_snapshot_id.value() > latest_snapshot.value().Id()) {
-        return false;
-    }
-
-    for (int64_t snapshot_id = retry_start_snapshot_id.value();
-         snapshot_id <= latest_snapshot.value().Id(); ++snapshot_id) {
-        PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, snapshot_manager_->LoadSnapshot(snapshot_id));
-        if (snapshot.CommitUser() == commit_user_ && snapshot.CommitIdentifier() == identifier &&
-            snapshot.GetCommitKind() == commit_kind) {
-            return true;
-        }
-    }
-    return false;
-}
-
 Status FileStoreCommitImpl::CheckSameBucketFromSnapshot(
     const std::vector<ManifestEntry>& delta_entries,
     const std::optional<Snapshot>& latest_snapshot) const {
@@ -1015,13 +991,7 @@ Result<bool> FileStoreCommitImpl::TryCommitOnce(
     const std::vector<IndexManifestEntry>& index_entries, int64_t identifier,
     std::optional<int64_t> watermark, const std::map<std::string, std::string>& properties,
     Snapshot::CommitKind commit_kind, const std::optional<Snapshot>& latest_snapshot,
-    bool detect_conflicts, std::optional<int64_t> retry_start_snapshot_id) {
-    PAIMON_ASSIGN_OR_RAISE(bool committed, CheckCommitted(latest_snapshot, retry_start_snapshot_id,
-                                                          identifier, commit_kind));
-    if (committed) {
-        return true;
-    }
-
+    bool detect_conflicts) {
     std::vector<ManifestEntry> delta_files = delta_entries;
     int64_t start_millis = DateTimeUtils::GetCurrentUTCTimeUs() / 1000;
 
@@ -1250,11 +1220,21 @@ Result<bool> FileStoreCommitImpl::TryCommitOnce(
 
     Result<bool> commit_result = CommitSnapshotImpl(new_snapshot, delta_statistics);
     if (!commit_result.ok()) {
-        // commit exception is uncertain; retry after checking whether this commit already exists.
-        PAIMON_LOG_WARN(logger_, "Retry commit for exception. %s",
+        // commit exception, not sure about the situation and should not clean up the files.
+        PAIMON_LOG_WARN(logger_,
+                        "You need to call FilterAndCommit to retry commit for exception. %s",
                         commit_result.status().ToString().c_str());
+
+        // To prevent the case where an atomic write times out but actually succeeds,
+        // retrying the commit could lead to the snapshot file being committed multiple times.
+        // Therefore, retries should be handled by the upper layer,
+        // which should call FilterAndCommit to avoid duplicate commits.
+        // Therefore, we should not trigger cleanup here,
+        // as it may delete meta files from a snapshot that was just written by ourselves,
+        // leading to an incomplete or corrupted snapshot.
         guard.Release();
-        return false;
+        return Status::Invalid("You need to call FilterAndCommit to retry commit for exception. ",
+                               commit_result.status().ToString());
     }
     bool commit_success = commit_result.value();
     if (commit_success) {

src/paimon/core/operation/file_store_commit_impl.h

@@ -196,8 +196,7 @@ class FileStoreCommitImpl : public FileStoreCommit {
                                const std::map<std::string, std::string>& properties,
                                Snapshot::CommitKind commit_kind,
                                const std::optional<Snapshot>& latest_snapshot,
-                               bool detect_conflicts,
-                               std::optional<int64_t> retry_start_snapshot_id);
+                               bool detect_conflicts);
 
     Result<bool> CommitSnapshotImpl(const Snapshot& new_snapshot,
                                     const std::vector<PartitionEntry>& delta_statistics);
@@ -211,10 +210,6 @@ class FileStoreCommitImpl : public FileStoreCommit {
                              const std::optional<std::string>& old_index_manifest,
                              const std::optional<std::string>& new_index_manifest);
 
-    Result<bool> CheckCommitted(const std::optional<Snapshot>& latest_snapshot,
-                                std::optional<int64_t> retry_start_snapshot_id, int64_t identifier,
-                                const Snapshot::CommitKind& commit_kind) const;
-
     Status CheckSameBucketFromSnapshot(const std::vector<ManifestEntry>& delta_entries,
                                        const std::optional<Snapshot>& latest_snapshot) const;
 

src/paimon/core/operation/file_store_commit_impl_test.cpp

@@ -276,8 +276,8 @@ class FileStoreCommitImplTest : public testing::Test {
             /*creation_time=*/Timestamp(0, 0),
             /*delete_row_count=*/std::nullopt,
             /*embedded_index=*/nullptr, FileSource::Append(),
-            /*external_path=*/std::nullopt,
-            /*value_stats_cols=*/std::nullopt, /*first_row_id=*/std::nullopt,
+            /*value_stats_cols=*/std::nullopt,
+            /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt,
             /*write_cols=*/std::nullopt);
     }
 
@@ -292,8 +292,8 @@ class FileStoreCommitImplTest : public testing::Test {
             /*creation_time=*/Timestamp(0, 0),
             /*delete_row_count=*/std::nullopt,
             /*embedded_index=*/nullptr, FileSource::Append(),
-            /*external_path=*/std::nullopt,
-            /*value_stats_cols=*/std::nullopt, /*first_row_id=*/std::nullopt,
+            /*value_stats_cols=*/std::nullopt,
+            /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt,
             /*write_cols=*/std::nullopt);
     }
 
@@ -638,12 +638,7 @@ TEST_F(FileStoreCommitImplTest, TestCommitWithAtomicWriteSnapshotTimeoutAndActua
                               "/orc/append_09.db/append_09/commit_messages/commit_messages-01",
                           /*version=*/3);
     ASSERT_GT(msgs.size(), 0);
-    ASSERT_OK(commit->Commit(msgs, /*commit_identifier=*/1));
-    std::shared_ptr<Metrics> metrics = commit->GetCommitMetrics();
-    ASSERT_TRUE(metrics);
-    ASSERT_OK_AND_ASSIGN(uint64_t counter,
-                         metrics->GetCounter(CommitMetrics::LAST_COMMIT_ATTEMPTS));
-    ASSERT_EQ(2u, counter);
+    ASSERT_NOK(commit->Commit(msgs, /*commit_identifier=*/1));
     ASSERT_OK_AND_ASSIGN(
         bool exist, file_system_->Exists(PathUtil::JoinPath(table_path, "snapshot/snapshot-6")));
     ASSERT_TRUE(exist);
@@ -656,6 +651,8 @@ TEST_F(FileStoreCommitImplTest, TestCommitWithAtomicWriteSnapshotTimeoutAndActua
                              .Finish());
 
     ASSERT_OK_AND_ASSIGN(auto commit_2, FileStoreCommit::Create(std::move(commit_context_2)));
+    ASSERT_OK_AND_ASSIGN(int32_t num_committed, commit_2->FilterAndCommit({{1, msgs}}));
+    ASSERT_EQ(0, num_committed);
     std::string new_snapshot_7 = PathUtil::JoinPath(table_path, "snapshot/snapshot-7");
     EXPECT_CALL(*mock_fs, AtomicStore(testing::StrEq(new_snapshot_7), testing::_))
         .WillOnce(testing::Invoke([&](const std::string& path, const std::string& content) {

src/paimon/format/parquet/column_index_filter_test.cpp

@@ -252,7 +252,7 @@ class ColumnIndexFilterTest : public ::testing::Test {
         // Open as raw ParquetFileReader
         ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs_->Open(file_name_));
         ASSERT_OK_AND_ASSIGN(int64_t length, in->Length());
-        auto in_stream = std::make_shared<ArrowInputStreamAdapter>(in, arrow_pool_, length);
+        auto in_stream = std::make_shared<ArrowInputStreamAdapter>(in, length, arrow_pool_);
         parquet_reader_ = ::parquet::ParquetFileReader::Open(in_stream);
         ASSERT_TRUE(parquet_reader_);
 

src/paimon/format/parquet/file_reader_wrapper_test.cpp

@@ -119,7 +119,7 @@ class FileReaderWrapperTest : public ::testing::Test {
         const std::string& file_path, int64_t wrapper_batch_size = 0) {
         PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> in, fs_->Open(file_path));
         PAIMON_ASSIGN_OR_RAISE(int64_t file_length, in->Length());
-        auto input_stream = std::make_unique<ArrowInputStreamAdapter>(in, arrow_pool_, file_length);
+        auto input_stream = std::make_unique<ArrowInputStreamAdapter>(in, file_length, arrow_pool_);
         ::parquet::arrow::FileReaderBuilder file_reader_builder;
         ::parquet::ReaderProperties reader_properties;
         reader_properties.enable_buffered_stream();