Skip to content

feat: add metrics raw-input-bytes/storage-read-bytes for parquet and fix clean inte test #439

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lucasfang wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: add metrics raw-input-bytes/storage-read-bytes for parquet and fix clean inte test #439

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 23 additions & 9 deletions src/paimon/common/utils/arrow/arrow_input_stream_adapter.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,10 @@ arrow::Status ValidateArrowIoRange(int64_t value, const char* name) {
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) {
: input_stream_(input_stream),
pool_(pool),
file_size_(file_size),
raw_input_bytes_(std::make_shared<std::atomic<uint64_t>>(0)) {
assert(file_size >= 0);
}

Expand All @@ -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 (raw_input_bytes_) {
raw_input_bytes_->fetch_add(static_cast<uint64_t>(read_bytes.value()));
}
return read_bytes.value();
}

Expand All @@ -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 (raw_input_bytes_) {
raw_input_bytes_->fetch_add(static_cast<uint64_t>(read_bytes.value()));
}
return read_bytes.value();
}

Expand Down Expand Up @@ -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>> raw_input_bytes = raw_input_bytes_;
input_stream_->ReadAsync(
reinterpret_cast<char*>(buffer->mutable_data()), nbytes, position,
[fut, buffer, raw_input_bytes, nbytes](Status callback_status) mutable {
if (callback_status.ok()) {
if (raw_input_bytes) {
raw_input_bytes->fetch_add(static_cast<uint64_t>(nbytes));
}
fut.MarkFinished(std::move(buffer));
Comment on lines +131 to +139
} else {
fut.MarkFinished(ToArrowStatus(callback_status));
}
});
return fut;
}

Expand Down
10 changes: 10 additions & 0 deletions src/paimon/common/utils/arrow/arrow_input_stream_adapter.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

#pragma once

#include <atomic>
#include <cstdint>
#include <memory>

Expand Down Expand Up @@ -49,12 +50,21 @@ class PAIMON_EXPORT ArrowInputStreamAdapter : public arrow::io::RandomAccessFile
}
bool closed() const override;

// Accumulated bytes handed back to the reader (rawInputBytes). 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>>& RawInputBytes() const {
return raw_input_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 handed back to the reader (rawInputBytes).
std::shared_ptr<std::atomic<uint64_t>> raw_input_bytes_;
bool closed_ = false;
};

Expand Down
49 changes: 14 additions & 35 deletions src/paimon/core/operation/file_store_commit_impl.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());
Expand All @@ -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()) {
Expand All @@ -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 {
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -1250,11 +1220,20 @@ 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 call FilterAndCommit to retry commit for exception. %s",
commit_result.status().ToString().c_str());
Comment on lines +1223 to 1225

// 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 call FilterAndCommit to retry commit for exception. ",
commit_result.status().ToString());
Comment on lines +1235 to +1236
}
bool commit_success = commit_result.value();
if (commit_success) {
Expand Down
7 changes: 1 addition & 6 deletions src/paimon/core/operation/file_store_commit_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand All @@ -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;

Expand Down
17 changes: 7 additions & 10 deletions src/paimon/core/operation/file_store_commit_impl_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}

Expand All @@ -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);
}

Expand Down Expand Up @@ -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);
Expand All @@ -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) {
Expand Down
6 changes: 4 additions & 2 deletions src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,8 @@ class PageFilteredRowGroupReaderTest : public ::testing::Test {
options[PARQUET_READ_ENABLE_PAGE_INDEX_FILTER] = "true";
ASSERT_OK_AND_ASSIGN(auto batch_reader, ParquetFileBatchReader::Create(
std::move(in_stream), options, batch_size,
/*file_metadata=*/nullptr, arrow_pool_));
/*file_metadata=*/nullptr, arrow_pool_,
/*raw_input_bytes=*/nullptr));
auto c_schema = std::make_unique<ArrowSchema>();
ASSERT_TRUE(arrow::ExportSchema(*read_schema, c_schema.get()).ok());
ASSERT_OK(batch_reader->SetReadSchema(c_schema.get(), predicate,
Expand All @@ -144,7 +145,8 @@ class PageFilteredRowGroupReaderTest : public ::testing::Test {

ASSERT_OK_AND_ASSIGN(auto batch_reader,
ParquetFileBatchReader::Create(std::move(in_stream), options,
batch_size, nullptr, arrow_pool_));
batch_size, nullptr, arrow_pool_,
/*raw_input_bytes=*/nullptr));
auto c_schema = std::make_unique<ArrowSchema>();
ASSERT_TRUE(arrow::ExportSchema(*read_schema, c_schema.get()).ok());
ASSERT_OK(batch_reader->SetReadSchema(c_schema.get(), predicate, bitmap));
Expand Down
10 changes: 7 additions & 3 deletions src/paimon/format/parquet/parquet_file_batch_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,19 +65,22 @@ namespace paimon::parquet {
ParquetFileBatchReader::ParquetFileBatchReader(
std::shared_ptr<arrow::io::RandomAccessFile>&& input_stream,
std::unique_ptr<FileReaderWrapper>&& reader, const std::map<std::string, std::string>& options,
const std::shared_ptr<arrow::MemoryPool>& arrow_pool)
const std::shared_ptr<arrow::MemoryPool>& arrow_pool,
std::shared_ptr<std::atomic<uint64_t>> raw_input_bytes)
: options_(options),
arrow_pool_(arrow_pool),
input_stream_(std::move(input_stream)),
reader_(std::move(reader)),
metrics_(std::make_shared<MetricsImpl>()),
raw_input_bytes_(std::move(raw_input_bytes)),
logger_(Logger::GetLogger("ParquetFileBatchReader")) {}

Result<std::unique_ptr<ParquetFileBatchReader>> ParquetFileBatchReader::Create(
std::shared_ptr<arrow::io::RandomAccessFile>&& input_stream,
const std::map<std::string, std::string>& options, int32_t batch_size,
std::shared_ptr<::parquet::FileMetaData> file_metadata,
const std::shared_ptr<arrow::MemoryPool>& pool) {
const std::shared_ptr<arrow::MemoryPool>& pool,
std::shared_ptr<std::atomic<uint64_t>> raw_input_bytes) {
try {
assert(input_stream);
PAIMON_ASSIGN_OR_RAISE(::parquet::ReaderProperties reader_properties,
Expand All @@ -98,7 +101,8 @@ Result<std::unique_ptr<ParquetFileBatchReader>> ParquetFileBatchReader::Create(
FileReaderWrapper::Create(std::move(file_reader),
static_cast<int64_t>(batch_size), pool));
auto parquet_file_batch_reader = std::unique_ptr<ParquetFileBatchReader>(
new ParquetFileBatchReader(std::move(input_stream), std::move(reader), options, pool));
new ParquetFileBatchReader(std::move(input_stream), std::move(reader), options, pool,
std::move(raw_input_bytes)));
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<::ArrowSchema> file_schema,
parquet_file_batch_reader->GetFileSchema());
PAIMON_RETURN_NOT_OK(parquet_file_batch_reader->SetReadSchema(
Expand Down
15 changes: 13 additions & 2 deletions src/paimon/format/parquet/parquet_file_batch_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

#include <fmt/format.h>

#include <atomic>
#include <cassert>
#include <cstdint>
#include <map>
Expand All @@ -38,6 +39,7 @@
#include "paimon/common/metrics/metrics_impl.h"
#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/format/parquet/file_reader_wrapper.h"
#include "paimon/format/parquet/parquet_format_defs.h"
#include "paimon/format/parquet/row_ranges.h"
#include "paimon/format/parquet/target_row_group.h"
#include "paimon/logging.h"
Expand Down Expand Up @@ -71,7 +73,8 @@ class ParquetFileBatchReader : public PrefetchFileBatchReader {
std::shared_ptr<arrow::io::RandomAccessFile>&& input_stream,
const std::map<std::string, std::string>& options, int32_t batch_size,
std::shared_ptr<::parquet::FileMetaData> file_metadata,
const std::shared_ptr<arrow::MemoryPool>& pool);
const std::shared_ptr<arrow::MemoryPool>& pool,

@zjw1111 zjw1111 Jul 21, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

make pool at last

std::shared_ptr<std::atomic<uint64_t>> raw_input_bytes);

static Result<::parquet::ReaderProperties> CreateReaderProperties(
const std::shared_ptr<arrow::MemoryPool>& pool,
Expand Down Expand Up @@ -130,6 +133,11 @@ class ParquetFileBatchReader : public PrefetchFileBatchReader {
}

std::shared_ptr<Metrics> GetReaderMetrics() const override {
// storageReadBytes is not cache-aware at the Parquet format layer, so it mirrors
// rawInputBytes (physical/logical read boundaries are indistinguishable here).
uint64_t raw = raw_input_bytes_ ? raw_input_bytes_->load() : 0;
metrics_->SetCounter(ParquetMetrics::READ_RAW_INPUT_BYTES, raw);
metrics_->SetCounter(ParquetMetrics::READ_STORAGE_BYTES, raw);
return metrics_;
}

Expand All @@ -150,7 +158,8 @@ class ParquetFileBatchReader : public PrefetchFileBatchReader {
ParquetFileBatchReader(std::shared_ptr<arrow::io::RandomAccessFile>&& input_stream,
std::unique_ptr<FileReaderWrapper>&& reader,
const std::map<std::string, std::string>& options,
const std::shared_ptr<arrow::MemoryPool>& arrow_pool);
const std::shared_ptr<arrow::MemoryPool>& arrow_pool,
std::shared_ptr<std::atomic<uint64_t>> raw_input_bytes);

static Result<::parquet::ArrowReaderProperties> CreateArrowReaderProperties(
const std::shared_ptr<arrow::MemoryPool>& pool,
Expand Down Expand Up @@ -251,6 +260,8 @@ class ParquetFileBatchReader : public PrefetchFileBatchReader {
std::shared_ptr<arrow::DataType> read_data_type_;

std::shared_ptr<Metrics> metrics_;
// rawInputBytes counter shared with the underlying ArrowInputStreamAdapter.
std::shared_ptr<std::atomic<uint64_t>> raw_input_bytes_;
std::unique_ptr<Logger> logger_;

uint64_t read_rows_ = 0;
Expand Down
Loading
Loading