From 3dfe4b28ed2b149626239c3f67c87ccb9446a29d Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Tue, 21 Jul 2026 18:24:30 +0800 Subject: [PATCH] feat: add metrics raw-input-bytes/storage-read-bytes for parquet and fix clean inte test --- .../arrow/arrow_input_stream_adapter.cpp | 32 ++++++++---- .../utils/arrow/arrow_input_stream_adapter.h | 10 ++++ .../core/operation/file_store_commit_impl.cpp | 49 ++++++------------- .../core/operation/file_store_commit_impl.h | 7 +-- .../operation/file_store_commit_impl_test.cpp | 17 +++---- .../page_filtered_row_group_reader_test.cpp | 6 ++- .../parquet/parquet_file_batch_reader.cpp | 10 ++-- .../parquet/parquet_file_batch_reader.h | 15 +++++- .../parquet_file_batch_reader_test.cpp | 31 ++++++++---- .../format/parquet/parquet_format_defs.h | 6 +++ .../format/parquet/parquet_reader_builder.h | 4 +- .../parquet/predicate_pushdown_test.cpp | 3 +- test/inte/clean_inte_test.cpp | 2 +- 13 files changed, 112 insertions(+), 80 deletions(-) diff --git a/src/paimon/common/utils/arrow/arrow_input_stream_adapter.cpp b/src/paimon/common/utils/arrow/arrow_input_stream_adapter.cpp index bf44f01a0..a192432a9 100644 --- a/src/paimon/common/utils/arrow/arrow_input_stream_adapter.cpp +++ b/src/paimon/common/utils/arrow/arrow_input_stream_adapter.cpp @@ -45,7 +45,10 @@ arrow::Status ValidateArrowIoRange(int64_t value, const char* name) { ArrowInputStreamAdapter::ArrowInputStreamAdapter( const std::shared_ptr& input_stream, const std::shared_ptr& 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>(0)) { assert(file_size >= 0); } @@ -63,6 +66,9 @@ arrow::Result 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(read_bytes.value())); + } return read_bytes.value(); } @@ -84,6 +90,9 @@ arrow::Result 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(read_bytes.value())); + } return read_bytes.value(); } @@ -119,14 +128,19 @@ arrow::Future> ArrowInputStreamAdapter::ReadAsync return fut; } std::shared_ptr buffer = std::move(buffer_result).ValueUnsafe(); - input_stream_->ReadAsync(reinterpret_cast(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> raw_input_bytes = raw_input_bytes_; + input_stream_->ReadAsync( + reinterpret_cast(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(nbytes)); + } + fut.MarkFinished(std::move(buffer)); + } else { + fut.MarkFinished(ToArrowStatus(callback_status)); + } + }); return fut; } diff --git a/src/paimon/common/utils/arrow/arrow_input_stream_adapter.h b/src/paimon/common/utils/arrow/arrow_input_stream_adapter.h index 5ad7b41a8..524139503 100644 --- a/src/paimon/common/utils/arrow/arrow_input_stream_adapter.h +++ b/src/paimon/common/utils/arrow/arrow_input_stream_adapter.h @@ -16,6 +16,7 @@ #pragma once +#include #include #include @@ -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>& RawInputBytes() const { + return raw_input_bytes_; + } + private: arrow::Status DoClose(); std::shared_ptr input_stream_; std::shared_ptr pool_; int64_t file_size_; + // Accumulates the number of bytes handed back to the reader (rawInputBytes). + std::shared_ptr> raw_input_bytes_; bool closed_ = false; }; diff --git a/src/paimon/core/operation/file_store_commit_impl.cpp b/src/paimon/core/operation/file_store_commit_impl.cpp index c7e2d247f..3f29d866b 100644 --- a/src/paimon/core/operation/file_store_commit_impl.cpp +++ b/src/paimon/core/operation/file_store_commit_impl.cpp @@ -892,7 +892,6 @@ Result FileStoreCommitImpl::TryCommit( Snapshot::CommitKind commit_kind, bool detect_conflicts) { int32_t retry_count = 0; int64_t start_millis = DateTimeUtils::GetCurrentUTCTimeUs() / 1000; - std::optional retry_start_snapshot_id; while (true) { PAIMON_ASSIGN_OR_RAISE(std::optional latest_snapshot, snapshot_manager_->LatestSnapshot()); @@ -902,13 +901,10 @@ Result 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(latest_snapshot.value().Id() + 1) - : std::optional(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 FileStoreCommitImpl::TryCommit( return retry_count + 1; } -Result FileStoreCommitImpl::CheckCommitted(const std::optional& latest_snapshot, - std::optional 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& delta_entries, const std::optional& latest_snapshot) const { @@ -1015,13 +991,7 @@ Result FileStoreCommitImpl::TryCommitOnce( const std::vector& index_entries, int64_t identifier, std::optional watermark, const std::map& properties, Snapshot::CommitKind commit_kind, const std::optional& latest_snapshot, - bool detect_conflicts, std::optional 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 delta_files = delta_entries; int64_t start_millis = DateTimeUtils::GetCurrentUTCTimeUs() / 1000; @@ -1250,11 +1220,20 @@ Result FileStoreCommitImpl::TryCommitOnce( Result 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()); + + // 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()); } bool commit_success = commit_result.value(); if (commit_success) { diff --git a/src/paimon/core/operation/file_store_commit_impl.h b/src/paimon/core/operation/file_store_commit_impl.h index b171e2150..2e9d5e775 100644 --- a/src/paimon/core/operation/file_store_commit_impl.h +++ b/src/paimon/core/operation/file_store_commit_impl.h @@ -196,8 +196,7 @@ class FileStoreCommitImpl : public FileStoreCommit { const std::map& properties, Snapshot::CommitKind commit_kind, const std::optional& latest_snapshot, - bool detect_conflicts, - std::optional retry_start_snapshot_id); + bool detect_conflicts); Result CommitSnapshotImpl(const Snapshot& new_snapshot, const std::vector& delta_statistics); @@ -211,10 +210,6 @@ class FileStoreCommitImpl : public FileStoreCommit { const std::optional& old_index_manifest, const std::optional& new_index_manifest); - Result CheckCommitted(const std::optional& latest_snapshot, - std::optional retry_start_snapshot_id, int64_t identifier, - const Snapshot::CommitKind& commit_kind) const; - Status CheckSameBucketFromSnapshot(const std::vector& delta_entries, const std::optional& latest_snapshot) const; diff --git a/src/paimon/core/operation/file_store_commit_impl_test.cpp b/src/paimon/core/operation/file_store_commit_impl_test.cpp index 96e3363f1..9757e9bd8 100644 --- a/src/paimon/core/operation/file_store_commit_impl_test.cpp +++ b/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 = 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) { diff --git a/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp b/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp index 666587af9..03e162bd0 100644 --- a/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp +++ b/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp @@ -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(); ASSERT_TRUE(arrow::ExportSchema(*read_schema, c_schema.get()).ok()); ASSERT_OK(batch_reader->SetReadSchema(c_schema.get(), predicate, @@ -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(); ASSERT_TRUE(arrow::ExportSchema(*read_schema, c_schema.get()).ok()); ASSERT_OK(batch_reader->SetReadSchema(c_schema.get(), predicate, bitmap)); diff --git a/src/paimon/format/parquet/parquet_file_batch_reader.cpp b/src/paimon/format/parquet/parquet_file_batch_reader.cpp index 366edcf01..2ad4dd5bc 100644 --- a/src/paimon/format/parquet/parquet_file_batch_reader.cpp +++ b/src/paimon/format/parquet/parquet_file_batch_reader.cpp @@ -65,19 +65,22 @@ namespace paimon::parquet { ParquetFileBatchReader::ParquetFileBatchReader( std::shared_ptr&& input_stream, std::unique_ptr&& reader, const std::map& options, - const std::shared_ptr& arrow_pool) + const std::shared_ptr& arrow_pool, + std::shared_ptr> raw_input_bytes) : options_(options), arrow_pool_(arrow_pool), input_stream_(std::move(input_stream)), reader_(std::move(reader)), metrics_(std::make_shared()), + raw_input_bytes_(std::move(raw_input_bytes)), logger_(Logger::GetLogger("ParquetFileBatchReader")) {} Result> ParquetFileBatchReader::Create( std::shared_ptr&& input_stream, const std::map& options, int32_t batch_size, std::shared_ptr<::parquet::FileMetaData> file_metadata, - const std::shared_ptr& pool) { + const std::shared_ptr& pool, + std::shared_ptr> raw_input_bytes) { try { assert(input_stream); PAIMON_ASSIGN_OR_RAISE(::parquet::ReaderProperties reader_properties, @@ -98,7 +101,8 @@ Result> ParquetFileBatchReader::Create( FileReaderWrapper::Create(std::move(file_reader), static_cast(batch_size), pool)); auto parquet_file_batch_reader = std::unique_ptr( - 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( diff --git a/src/paimon/format/parquet/parquet_file_batch_reader.h b/src/paimon/format/parquet/parquet_file_batch_reader.h index 1d6b7e46d..46fced7ae 100644 --- a/src/paimon/format/parquet/parquet_file_batch_reader.h +++ b/src/paimon/format/parquet/parquet_file_batch_reader.h @@ -18,6 +18,7 @@ #include +#include #include #include #include @@ -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" @@ -71,7 +73,8 @@ class ParquetFileBatchReader : public PrefetchFileBatchReader { std::shared_ptr&& input_stream, const std::map& options, int32_t batch_size, std::shared_ptr<::parquet::FileMetaData> file_metadata, - const std::shared_ptr& pool); + const std::shared_ptr& pool, + std::shared_ptr> raw_input_bytes); static Result<::parquet::ReaderProperties> CreateReaderProperties( const std::shared_ptr& pool, @@ -130,6 +133,11 @@ class ParquetFileBatchReader : public PrefetchFileBatchReader { } std::shared_ptr 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_; } @@ -150,7 +158,8 @@ class ParquetFileBatchReader : public PrefetchFileBatchReader { ParquetFileBatchReader(std::shared_ptr&& input_stream, std::unique_ptr&& reader, const std::map& options, - const std::shared_ptr& arrow_pool); + const std::shared_ptr& arrow_pool, + std::shared_ptr> raw_input_bytes); static Result<::parquet::ArrowReaderProperties> CreateArrowReaderProperties( const std::shared_ptr& pool, @@ -251,6 +260,8 @@ class ParquetFileBatchReader : public PrefetchFileBatchReader { std::shared_ptr read_data_type_; std::shared_ptr metrics_; + // rawInputBytes counter shared with the underlying ArrowInputStreamAdapter. + std::shared_ptr> raw_input_bytes_; std::unique_ptr logger_; uint64_t read_rows_ = 0; diff --git a/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp b/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp index a406ea527..2f2ad73ae 100644 --- a/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp +++ b/src/paimon/format/parquet/parquet_file_batch_reader_test.cpp @@ -16,9 +16,11 @@ #include "paimon/format/parquet/parquet_file_batch_reader.h" +#include #include #include #include +#include #include #include "arrow/api.h" @@ -201,11 +203,13 @@ class ParquetFileBatchReaderTest : public ::testing::Test, auto length = fs_->GetFileStatus(file_name).value()->GetLen(); auto in_stream = std::make_unique(std::move(input_stream), pool_, length); + auto raw_input_bytes = in_stream->RawInputBytes(); std::map options; options[PARQUET_READ_ENABLE_PAGE_INDEX_FILTER] = enable_page_level_filter ? "true" : "false"; return PrepareParquetFileBatchReader(std::move(in_stream), options, read_schema, predicate, - selection_bitmap, batch_size); + selection_bitmap, batch_size, + std::move(raw_input_bytes)); } std::unique_ptr PrepareParquetFileBatchReader( @@ -213,11 +217,12 @@ class ParquetFileBatchReaderTest : public ::testing::Test, const std::map& options, const std::shared_ptr& read_schema, const std::shared_ptr& predicate, - const std::optional& selection_bitmap, int32_t batch_size) const { - EXPECT_OK_AND_ASSIGN( - auto parquet_batch_reader, - ParquetFileBatchReader::Create(std::move(in_stream), options, batch_size, - /*file_metadata=*/nullptr, pool_)); + const std::optional& selection_bitmap, int32_t batch_size, + std::shared_ptr> raw_input_bytes = nullptr) const { + EXPECT_OK_AND_ASSIGN(auto parquet_batch_reader, + ParquetFileBatchReader::Create( + std::move(in_stream), options, batch_size, + /*file_metadata=*/nullptr, pool_, std::move(raw_input_bytes))); std::unique_ptr c_schema = std::make_unique(); auto arrow_status = arrow::ExportSchema(*read_schema, c_schema.get()); EXPECT_TRUE(arrow_status.ok()); @@ -383,7 +388,8 @@ TEST_F(ParquetFileBatchReaderTest, TestSetReadSchema) { std::map options; ASSERT_OK_AND_ASSIGN(auto parquet_batch_reader, ParquetFileBatchReader::Create(std::move(in_stream), options, batch_size_, - /*file_metadata=*/nullptr, pool_)); + /*file_metadata=*/nullptr, pool_, + /*raw_input_bytes=*/nullptr)); // test GetFileSchema() ASSERT_OK_AND_ASSIGN(auto c_file_schema, parquet_batch_reader->GetFileSchema()); auto arrow_file_schema = arrow::ImportSchema(c_file_schema.get()).ValueOrDie(); @@ -472,9 +478,14 @@ TEST_F(ParquetFileBatchReaderTest, TestNextBatchSimple) { // test metrics auto read_metrics = parquet_batch_reader->GetReaderMetrics(); ASSERT_TRUE(read_metrics); - // TODO(jinli.zjw): test metrics - // ASSERT_TRUE(read_metrics->GetCounter(ParquetMetrics::READ_BYTES) > 0); - // ASSERT_TRUE(read_metrics->GetCounter(ParquetMetrics::READ_RAW_BYTES) > 0); + ASSERT_OK_AND_ASSIGN(uint64_t raw_input_bytes, + read_metrics->GetCounter(ParquetMetrics::READ_RAW_INPUT_BYTES)); + ASSERT_OK_AND_ASSIGN(uint64_t storage_read_bytes, + read_metrics->GetCounter(ParquetMetrics::READ_STORAGE_BYTES)); + ASSERT_GT(raw_input_bytes, 0u); + // storageReadBytes is not cache-aware at the Parquet format layer, so it equals + // rawInputBytes. + ASSERT_EQ(storage_read_bytes, raw_input_bytes); } } diff --git a/src/paimon/format/parquet/parquet_format_defs.h b/src/paimon/format/parquet/parquet_format_defs.h index 2804c154e..42c71115b 100644 --- a/src/paimon/format/parquet/parquet_format_defs.h +++ b/src/paimon/format/parquet/parquet_format_defs.h @@ -112,6 +112,12 @@ class ParquetMetrics { "parquet.read.row-groups.after-filter"; static inline const char READ_ROWS[] = "parquet.read.rows"; static inline const char READ_BATCH_COUNT[] = "parquet.read.batch-count"; + // Byte-level read metrics. + // raw-input-bytes: logical bytes requested by the reader (top-of-stack demand). + // storage-read-bytes: physical bytes read from storage. This layer is not cache-aware, + // so it currently equals raw-input-bytes. + static inline const char READ_RAW_INPUT_BYTES[] = "parquet.read.raw-input-bytes"; + static inline const char READ_STORAGE_BYTES[] = "parquet.read.storage-read-bytes"; }; } // namespace paimon::parquet diff --git a/src/paimon/format/parquet/parquet_reader_builder.h b/src/paimon/format/parquet/parquet_reader_builder.h index 440e9c5dd..a54175d7e 100644 --- a/src/paimon/format/parquet/parquet_reader_builder.h +++ b/src/paimon/format/parquet/parquet_reader_builder.h @@ -71,12 +71,14 @@ class ParquetReaderBuilder : public ReaderBuilder { std::shared_ptr arrow_pool = GetArrowPool(pool_); auto unique_input_stream = std::make_unique(path, arrow_pool, file_length); + auto raw_input_bytes = unique_input_stream->RawInputBytes(); std::shared_ptr input_stream( std::move(unique_input_stream)); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<::parquet::FileMetaData> file_metadata, GetCachedParquetMetadata(input_stream, file_uri, arrow_pool)); return ParquetFileBatchReader::Create(std::move(input_stream), options_, batch_size_, - std::move(file_metadata), arrow_pool); + std::move(file_metadata), arrow_pool, + std::move(raw_input_bytes)); } PAIMON_PARQUET_CATCH_AND_RETURN_STATUS("ParquetReaderBuilder::Build") } diff --git a/src/paimon/format/parquet/predicate_pushdown_test.cpp b/src/paimon/format/parquet/predicate_pushdown_test.cpp index dbc47c47b..65551083f 100644 --- a/src/paimon/format/parquet/predicate_pushdown_test.cpp +++ b/src/paimon/format/parquet/predicate_pushdown_test.cpp @@ -112,7 +112,8 @@ class PredicatePushdownTest : public ::testing::Test { std::to_string(predicate_node_count_limit); 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)); std::unique_ptr c_schema = std::make_unique(); auto arrow_status = arrow::ExportSchema(*read_schema, c_schema.get()); ASSERT_TRUE(arrow_status.ok()); diff --git a/test/inte/clean_inte_test.cpp b/test/inte/clean_inte_test.cpp index 4461c92ed..685645fc5 100644 --- a/test/inte/clean_inte_test.cpp +++ b/test/inte/clean_inte_test.cpp @@ -405,7 +405,7 @@ TEST_F(CleanInteTest, TestDropPartitionAndExpireSnapshot) { ASSERT_EQ(3u, manifests[1].NumAddedFiles()); } -TEST_F(CleanInteTest, DISABLED_TestDropPartitionAndExpireSnapshotWithIOException) { +TEST_F(CleanInteTest, TestDropPartitionAndExpireSnapshotWithIOException) { auto string_field = arrow::field("f0", arrow::utf8()); auto int_field = arrow::field("f1", arrow::int32()); auto int_field1 = arrow::field("f2", arrow::int32());