This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 71e415cf feat(metrics): expose scan and prefetch metrics (#211)
71e415cf is described below
commit 71e415cf9c281da641a7f0452d820b836c0c27ae
Author: gripleaf <[email protected]>
AuthorDate: Fri Aug 28 16:47:00 2026 +0800
feat(metrics): expose scan and prefetch metrics (#211)
---
docs/source/user_guide.rst | 1 +
docs/source/user_guide/metrics.rst | 116 ++++++++
include/paimon/defs.h | 4 +
include/paimon/reader/prefetch_file_batch_reader.h | 41 +++
include/paimon/table/source/scan_metrics.h | 54 ++++
include/paimon/table/source/table_scan.h | 7 +
src/paimon/common/defs.cpp | 1 +
.../apply_bitmap_index_batch_reader_test.cpp | 3 +-
.../common/reader/delegating_prefetch_reader.h | 2 +-
.../late_materializing_file_batch_reader_test.cpp | 6 +-
.../reader/prefetch_file_batch_reader_impl.cpp | 293 +++++++++++++++++++--
.../reader/prefetch_file_batch_reader_impl.h | 7 +-
.../prefetch_file_batch_reader_impl_test.cpp | 235 ++++++++++++++---
src/paimon/core/core_options.cpp | 7 +
src/paimon/core/core_options.h | 2 +
src/paimon/core/core_options_test.cpp | 3 +
.../apply_deletion_vector_batch_reader_test.cpp | 3 +-
src/paimon/core/operation/abstract_split_read.cpp | 2 +-
.../operation/append_only_file_store_scan_test.cpp | 43 ++-
src/paimon/core/operation/file_store_scan.cpp | 50 +++-
src/paimon/core/operation/file_store_scan.h | 7 +-
.../operation/key_value_file_store_scan_test.cpp | 2 +-
src/paimon/core/operation/metrics/scan_metrics.h | 35 ---
src/paimon/core/table/source/abstract_table_scan.h | 4 +
src/paimon/core/table/source/realtime_table_scan.h | 4 +
.../core/table/source/snapshot/snapshot_reader.h | 4 +
src/paimon/core/table/source/table_scan.cpp | 5 +
src/paimon/core/table/source/table_scan_test.cpp | 46 ++++
src/paimon/core/table/system/system_table_test.cpp | 13 +
29 files changed, 895 insertions(+), 105 deletions(-)
diff --git a/docs/source/user_guide.rst b/docs/source/user_guide.rst
index 38ef0fc6..fdee89d9 100644
--- a/docs/source/user_guide.rst
+++ b/docs/source/user_guide.rst
@@ -38,6 +38,7 @@ User Guide
user_guide/commit
user_guide/compaction
user_guide/read
+ user_guide/metrics
user_guide/clean
user_guide/prefetch
user_guide/arrow
diff --git a/docs/source/user_guide/metrics.rst
b/docs/source/user_guide/metrics.rst
new file mode 100644
index 00000000..72ee2a17
--- /dev/null
+++ b/docs/source/user_guide/metrics.rst
@@ -0,0 +1,116 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+.. or more contributor license agreements. See the NOTICE file
+.. distributed with this work for additional information
+.. regarding copyright ownership. The ASF licenses this file
+.. to you under the Apache License, Version 2.0 (the
+.. "License"); you may not use this file except in compliance
+.. with the License. You may obtain a copy of the License at
+
+.. http://www.apache.org/licenses/LICENSE-2.0
+
+.. Unless required by applicable law or agreed to in writing,
+.. software distributed under the License is distributed on an
+.. "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+.. KIND, either express or implied. See the License for the
+.. specific language governing permissions and limitations
+.. under the License.
+
+Metrics reference
+=================
+
+``Metrics`` contains counters, gauges, and histogram snapshots. A counter is a
non-negative
+integer, a gauge represents current state, and a histogram records a
distribution of observed
+values. The scan and prefetch-reader metrics described below are returned as
point-in-time
+snapshots. Modifying a returned snapshot does not modify the component that
produced it.
+
+Scan planning
+-------------
+
+The names below are declared by ``ScanMetrics``. ``last*`` counters are
replaced after each
+successful ``CreatePlan()`` call. Histograms and the cache hit/miss counters
accumulate for the
+lifetime of the scan. They are unrelated to ``SetReadSchema()`` and
``ReadAheadCache::Reset()``.
+The first six names match Java ``ScanMetrics``; the remaining names are
C++-only.
+
+.. csv-table::
+ :header: "Name", "Type", "Unit", "Meaning"
+ :widths: 34, 12, 12, 52
+
+ "lastScanDuration", "counter", "milliseconds", "Duration of the last
successful plan"
+ "scanDuration", "histogram", "milliseconds", "Distribution of successful
plan durations"
+ "lastScannedSnapshotId", "counter", "snapshot ID", "Snapshot used by the
last plan, or 0"
+ "lastScannedManifests", "counter", "files", "Manifest files selected by the
last plan"
+ "lastScanSkippedTableFiles", "counter", "files", "Table files skipped by
the last plan"
+ "lastScanResultedTableFiles", "counter", "files", "Table files returned by
the last plan"
+ "lastManifestReadDuration", "counter", "milliseconds", "Manifest-list and
entry read time for the last plan"
+ "manifestReadDuration", "histogram", "milliseconds", "Distribution of
manifest read times"
+ "lastSnapshotCacheEnabled", "counter", "boolean", "Whether snapshot
manifest-entry cache was eligible"
+ "lastSnapshotCacheHit", "counter", "boolean", "Whether the last eligible
lookup hit"
+ "snapshotCacheHits", "counter", "lookups", "Cumulative exact-snapshot cache
hits"
+ "snapshotCacheMisses", "counter", "lookups", "Cumulative eligible cache
misses"
+ "lastSnapshotCacheLoadDuration", "counter", "milliseconds", "Cache load
time for the last plan"
+ "snapshotCacheLoadDuration", "histogram", "milliseconds", "Distribution of
cache load times"
+ "lastSnapshotCacheStoreDuration", "counter", "milliseconds", "Cache store
time for the last plan; 0 when not stored"
+ "snapshotCacheStoreDuration", "histogram", "milliseconds", "Distribution of
cache store times"
+ "lastLazyDecodeScannedRows", "counter", "manifest rows", "Candidate
manifest rows inspected by the last plan"
+ "lastLazyDecodeMaterializedRows", "counter", "manifest rows", "Manifest
rows retained after lazy filtering"
+
+Prefetch reader
+---------------
+
+The names below are declared by ``PrefetchMetrics``. Counters and histograms
accumulate for the
+lifetime of the prefetch reader, including across ``SetReadSchema()``.
``enabled`` and
+``parallelism`` describe the most recently initialized schema. ``queue-depth``
is reset by
+``SetReadSchema()`` and ``Close()``; ``queue-depth.max`` remains the lifetime
maximum.
+These metrics are C++-only and have no counterparts in Java Paimon.
+
+.. csv-table::
+ :header: "Name", "Type", "Unit", "Meaning"
+ :widths: 38, 12, 12, 48
+
+ "prefetch.enabled", "gauge", "boolean", "Whether the most recently
initialized schema selected prefetch"
+ "prefetch.parallelism", "gauge", "readers", "Effective reader parallelism"
+ "prefetch.read-ranges.total", "counter", "ranges", "Generated ranges before
bitmap filtering"
+ "prefetch.read-ranges.after-bitmap", "counter", "ranges", "Ranges retained
after bitmap filtering"
+ "prefetch.seek.count", "counter", "operations", "Underlying reader seek
operations"
+ "prefetch.produced-batches", "counter", "batches", "Data batches placed
into prefetch queues"
+ "prefetch.consumed-batches", "counter", "batches", "Data batches returned
to the consumer"
+ "prefetch.discarded-batches", "counter", "batches", "Data batches released
without consumption, plus EOF entries released during cleanup"
+ "prefetch.errors", "counter", "errors", "Errors recorded by the background
prefetch loop"
+ "prefetch.adaptive-disabled-count", "counter", "decisions", "Times adaptive
strategy disabled prefetch"
+ "prefetch.queue-full-count", "counter", "events", "Times production found a
full queue"
+ "prefetch.queue-depth", "gauge", "queue entries", "Current queued entries,
including retained EOF markers"
+ "prefetch.queue-depth.max", "gauge", "queue entries", "Maximum queued
entries in the reader lifetime"
+ "prefetch.reader-read-latency-us", "histogram", "microseconds", "Underlying
reader batch latency"
+ "prefetch.consumer-wait-latency-us", "histogram", "microseconds", "Consumer
wait latency per returned batch or EOF"
+
+Prefetch I/O
+------------
+
+``PrefetchIoMetrics`` describes only I/O that passes through the prefetch
reader's instrumented
+input streams. It is not a whole-query or whole-table I/O total. All counters
accumulate for the
+reader lifetime and are retained across ``SetReadSchema()`` and cache reset.
Latency uses relaxed
+atomic count and sum counters instead of per-I/O histograms to reduce hot-path
cost. Collection is
+disabled by default; set ``prefetch.io-metrics.enabled`` to ``true`` in the
read options to enable
+it. When disabled, these per-I/O metrics are absent and the input streams have
no metrics
+instrumentation.
+``io.async.pending`` is current state and returns to zero when all callbacks
complete.
+These metrics are C++-only and have no counterparts in Java Paimon.
+
+.. csv-table::
+ :header: "Name", "Type", "Unit", "Meaning"
+ :widths: 34, 12, 12, 52
+
+ "io.read.requests", "counter", "requests", "Synchronous read requests"
+ "io.read.requested-bytes", "counter", "bytes", "Bytes requested by
synchronous reads"
+ "io.read.physical-bytes", "counter", "bytes", "Bytes returned by successful
synchronous reads"
+ "io.read.failed", "counter", "requests", "Failed synchronous reads"
+ "io.read.latency.count", "counter", "requests", "Completed synchronous read
latency samples"
+ "io.read.latency.sum-us", "counter", "microseconds", "Sum of synchronous
read latency"
+ "io.async.requests", "counter", "requests", "Asynchronous read requests"
+ "io.async.requested-bytes", "counter", "bytes", "Bytes requested by
asynchronous reads"
+ "io.async.physical-bytes", "counter", "bytes", "Bytes attributed to
successful asynchronous reads"
+ "io.async.completed", "counter", "requests", "Successful asynchronous reads"
+ "io.async.failed", "counter", "requests", "Failed asynchronous reads"
+ "io.async.pending", "gauge", "requests", "Asynchronous callbacks not yet
completed"
+ "io.async.latency.count", "counter", "requests", "Completed asynchronous
callback latency samples"
+ "io.async.latency.sum-us", "counter", "microseconds", "Sum of asynchronous
callback latency"
diff --git a/include/paimon/defs.h b/include/paimon/defs.h
index c96cf5d4..0ca67e3a 100644
--- a/include/paimon/defs.h
+++ b/include/paimon/defs.h
@@ -208,6 +208,10 @@ struct PAIMON_EXPORT Options {
/// for the target bucket when rebuilding the cache. Default value is true.
static const char SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED[];
+ /// "prefetch.io-metrics.enabled" - Whether to collect per-I/O metrics for
prefetch reads.
+ /// Default value is false.
+ static const char PREFETCH_IO_METRICS_ENABLED[];
+
/// "read.batch-size" - Read batch size for any file format if it supports.
/// The default value is 1024.
static const char READ_BATCH_SIZE[];
diff --git a/include/paimon/reader/prefetch_file_batch_reader.h
b/include/paimon/reader/prefetch_file_batch_reader.h
index 5e6313e8..84227995 100644
--- a/include/paimon/reader/prefetch_file_batch_reader.h
+++ b/include/paimon/reader/prefetch_file_batch_reader.h
@@ -27,6 +27,47 @@
namespace paimon {
+/// C++-only prefetch reader metrics. Java Paimon has no corresponding metrics.
+class PAIMON_EXPORT PrefetchMetrics {
+ public:
+ static constexpr char ENABLED[] = "prefetch.enabled";
+ static constexpr char PARALLELISM[] = "prefetch.parallelism";
+ static constexpr char READ_RANGES_TOTAL[] = "prefetch.read-ranges.total";
+ static constexpr char READ_RANGES_AFTER_BITMAP[] =
"prefetch.read-ranges.after-bitmap";
+ static constexpr char SEEK_COUNT[] = "prefetch.seek.count";
+ static constexpr char PRODUCED_BATCHES[] = "prefetch.produced-batches";
+ static constexpr char CONSUMED_BATCHES[] = "prefetch.consumed-batches";
+ static constexpr char DISCARDED_BATCHES[] = "prefetch.discarded-batches";
+ static constexpr char ERRORS[] = "prefetch.errors";
+ static constexpr char ADAPTIVE_DISABLED_COUNT[] =
"prefetch.adaptive-disabled-count";
+ static constexpr char QUEUE_FULL_COUNT[] = "prefetch.queue-full-count";
+ static constexpr char QUEUE_DEPTH[] = "prefetch.queue-depth";
+ static constexpr char QUEUE_DEPTH_MAX[] = "prefetch.queue-depth.max";
+ static constexpr char READER_READ_LATENCY_US[] =
"prefetch.reader-read-latency-us";
+ static constexpr char CONSUMER_WAIT_LATENCY_US[] =
"prefetch.consumer-wait-latency-us";
+};
+
+/// C++-only metric names for I/O observed by the prefetch reader's
instrumented input streams.
+/// Java Paimon has no corresponding metrics.
+/// These metrics do not represent whole-query or whole-table I/O.
+class PAIMON_EXPORT PrefetchIoMetrics {
+ public:
+ static constexpr char READ_REQUESTS[] = "io.read.requests";
+ static constexpr char READ_REQUESTED_BYTES[] = "io.read.requested-bytes";
+ static constexpr char READ_PHYSICAL_BYTES[] = "io.read.physical-bytes";
+ static constexpr char READ_FAILED[] = "io.read.failed";
+ static constexpr char READ_LATENCY_COUNT[] = "io.read.latency.count";
+ static constexpr char READ_LATENCY_SUM_US[] = "io.read.latency.sum-us";
+ static constexpr char ASYNC_REQUESTS[] = "io.async.requests";
+ static constexpr char ASYNC_REQUESTED_BYTES[] = "io.async.requested-bytes";
+ static constexpr char ASYNC_PHYSICAL_BYTES[] = "io.async.physical-bytes";
+ static constexpr char ASYNC_COMPLETED[] = "io.async.completed";
+ static constexpr char ASYNC_FAILED[] = "io.async.failed";
+ static constexpr char ASYNC_PENDING[] = "io.async.pending";
+ static constexpr char ASYNC_LATENCY_COUNT[] = "io.async.latency.count";
+ static constexpr char ASYNC_LATENCY_SUM_US[] = "io.async.latency.sum-us";
+};
+
/// The prefetch file batch reader extends the basic FileBatchReader interface
for prefetch read,
/// if a format implementation inherits from this class, it will automatically
support the C++
/// Paimon prefetch capability and integrate with the Paimon prefetch
framework.
diff --git a/include/paimon/table/source/scan_metrics.h
b/include/paimon/table/source/scan_metrics.h
new file mode 100644
index 00000000..15bc41ff
--- /dev/null
+++ b/include/paimon/table/source/scan_metrics.h
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// Metric names for scan planning operations.
+class PAIMON_EXPORT ScanMetrics {
+ public:
+ static constexpr char LAST_SCAN_DURATION[] = "lastScanDuration";
+ // Histogram metric for scan plan duration (milliseconds).
+ static constexpr char SCAN_DURATION[] = "scanDuration";
+ static constexpr char LAST_SCANNED_SNAPSHOT_ID[] = "lastScannedSnapshotId";
+ static constexpr char LAST_SCANNED_MANIFESTS[] = "lastScannedManifests";
+ static constexpr char LAST_SCAN_SKIPPED_TABLE_FILES[] =
"lastScanSkippedTableFiles";
+ static constexpr char LAST_SCAN_RESULTED_TABLE_FILES[] =
"lastScanResultedTableFiles";
+
+ // The metrics below are C++-only and do not have counterparts in Java
ScanMetrics.
+ static constexpr char LAST_MANIFEST_READ_DURATION[] =
"lastManifestReadDuration";
+ // Histogram metric for manifest-list and manifest-entry read duration
(milliseconds).
+ static constexpr char MANIFEST_READ_DURATION[] = "manifestReadDuration";
+ static constexpr char LAST_SNAPSHOT_CACHE_ENABLED[] =
"lastSnapshotCacheEnabled";
+ static constexpr char LAST_SNAPSHOT_CACHE_HIT[] = "lastSnapshotCacheHit";
+ static constexpr char SNAPSHOT_CACHE_HITS[] = "snapshotCacheHits";
+ static constexpr char SNAPSHOT_CACHE_MISSES[] = "snapshotCacheMisses";
+ static constexpr char LAST_SNAPSHOT_CACHE_LOAD_DURATION[] =
"lastSnapshotCacheLoadDuration";
+ static constexpr char SNAPSHOT_CACHE_LOAD_DURATION[] =
"snapshotCacheLoadDuration";
+ static constexpr char LAST_SNAPSHOT_CACHE_STORE_DURATION[] =
"lastSnapshotCacheStoreDuration";
+ static constexpr char SNAPSHOT_CACHE_STORE_DURATION[] =
"snapshotCacheStoreDuration";
+ // Candidate manifest-entry rows inspected by lazy scan filtering.
+ static constexpr char LAST_LAZY_DECODE_SCANNED_ROWS[] =
"lastLazyDecodeScannedRows";
+ // Full manifest entries retained after lazy scan filtering.
+ static constexpr char LAST_LAZY_DECODE_MATERIALIZED_ROWS[] =
"lastLazyDecodeMaterializedRows";
+};
+
+} // namespace paimon
diff --git a/include/paimon/table/source/table_scan.h
b/include/paimon/table/source/table_scan.h
index c9b42915..14c44756 100644
--- a/include/paimon/table/source/table_scan.h
+++ b/include/paimon/table/source/table_scan.h
@@ -23,6 +23,7 @@
#include "paimon/result.h"
#include "paimon/table/source/plan.h"
+#include "paimon/table/source/scan_metrics.h"
#include "paimon/type_fwd.h"
#include "paimon/visibility.h"
@@ -44,5 +45,11 @@ class PAIMON_EXPORT TableScan {
///
/// @return A Result containing a shared pointer to the created `Plan` or
an error status.
virtual Result<std::shared_ptr<Plan>> CreatePlan() = 0;
+
+ /// Retrieve metrics related to scan planning operations.
+ ///
+ /// @return A point-in-time snapshot of scan metrics. Mutating the
returned object does not
+ /// affect metrics collected by this scan.
+ virtual std::shared_ptr<Metrics> GetMetrics() const;
};
} // namespace paimon
diff --git a/src/paimon/common/defs.cpp b/src/paimon/common/defs.cpp
index 45496146..39e6d494 100644
--- a/src/paimon/common/defs.cpp
+++ b/src/paimon/common/defs.cpp
@@ -62,6 +62,7 @@ const char Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS[]
=
"scan.manifest-entry-cache.max-snapshots";
const char Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED[] =
"scan.manifest-entry.lazy-decode.enabled";
+const char Options::PREFETCH_IO_METRICS_ENABLED[] =
"prefetch.io-metrics.enabled";
const char Options::READ_BATCH_SIZE[] = "read.batch-size";
const char Options::WRITE_BATCH_SIZE[] = "write.batch-size";
const char Options::WRITE_BUFFER_SIZE[] = "write-buffer-size";
diff --git
a/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
b/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
index 1082e669..ea45c979 100644
---
a/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
+++
b/src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp
@@ -97,7 +97,8 @@ class ApplyBitmapIndexBatchReaderTest : public
::testing::Test,
prefetch_batch_count, batch_size, prefetch_batch_count
* 2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
- /*read_ahead_cache_enabled=*/true, CacheConfig(),
pool_));
+ /*read_ahead_cache_enabled=*/true, CacheConfig(),
+ /*enable_io_metrics=*/false, pool_));
} else {
file_batch_reader =
std::make_unique<MockFileBatchReader>(data, target_type_,
batch_size);
diff --git a/src/paimon/common/reader/delegating_prefetch_reader.h
b/src/paimon/common/reader/delegating_prefetch_reader.h
index 3cbcb08b..78396e20 100644
--- a/src/paimon/common/reader/delegating_prefetch_reader.h
+++ b/src/paimon/common/reader/delegating_prefetch_reader.h
@@ -45,7 +45,7 @@ class DelegatingPrefetchReader : public FileBatchReader {
}
std::shared_ptr<Metrics> GetReaderMetrics() const override {
- return GetReader()->GetReaderMetrics();
+ return prefetch_reader_->GetReaderMetrics();
}
Result<std::unique_ptr<::ArrowSchema>> GetFileSchema() const override {
diff --git
a/src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp
b/src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp
index ff571a28..075fbe10 100644
--- a/src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp
+++ b/src/paimon/common/reader/late_materializing_file_batch_reader_test.cpp
@@ -532,7 +532,7 @@ TEST_F(LateMaterializingFileBatchReaderTest,
WorksAsInnerOfPrefetchReader) {
/*prefetch_max_parallel_num=*/1, /*batch_size=*/3,
/*prefetch_batch_count=*/2,
/*enable_adaptive_prefetch_strategy=*/false, executor,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/false, CacheConfig(),
- GetDefaultPool()));
+ /*enable_io_metrics=*/false, GetDefaultPool()));
auto predicate =
PredicateBuilder::GreaterOrEqual(/*field_index=*/0, "k",
FieldType::BIGINT, Literal(4l));
::ArrowSchema c_schema;
@@ -568,7 +568,7 @@ TEST_F(LateMaterializingFileBatchReaderTest,
PrefetchInnerReentrantSetReadSchema
/*prefetch_max_parallel_num=*/1, /*batch_size=*/3,
/*prefetch_batch_count=*/2,
/*enable_adaptive_prefetch_strategy=*/false, executor,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/false, CacheConfig(),
- GetDefaultPool()));
+ /*enable_io_metrics=*/false, GetDefaultPool()));
auto full_schema = arrow::schema(full_fields_);
auto predicate1 =
@@ -631,7 +631,7 @@ TEST_F(LateMaterializingFileBatchReaderTest,
PrefetchInnerParallelReadersWithSee
/*prefetch_max_parallel_num=*/3, /*batch_size=*/3,
/*prefetch_batch_count=*/6,
/*enable_adaptive_prefetch_strategy=*/false, executor,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/false, CacheConfig(),
- GetDefaultPool()));
+ /*enable_io_metrics=*/false, GetDefaultPool()));
auto predicate =
PredicateBuilder::GreaterOrEqual(/*field_index=*/0, "k",
FieldType::BIGINT, Literal(5l));
::ArrowSchema c_schema;
diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp
b/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp
index 4b6411b7..5285a2d8 100644
--- a/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp
+++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp
@@ -21,6 +21,7 @@
#include <algorithm>
#include <chrono>
#include <future>
+#include <limits>
#include <thread>
#include "arrow/array/array_base.h"
@@ -44,8 +45,149 @@ class Schema;
namespace paimon {
+struct PrefetchMetricsState {
+ std::atomic<uint64_t> read_ranges_total{0};
+ std::atomic<uint64_t> read_ranges_after_bitmap{0};
+ std::atomic<uint64_t> seek_count{0};
+ std::atomic<uint64_t> produced_batches{0};
+ std::atomic<uint64_t> consumed_batches{0};
+ std::atomic<uint64_t> discarded_batches{0};
+ std::atomic<uint64_t> errors{0};
+ std::atomic<uint64_t> adaptive_disabled_count{0};
+ std::atomic<uint64_t> queue_full_count{0};
+ std::atomic<uint64_t> queue_depth{0};
+ std::atomic<uint64_t> queue_depth_max{0};
+ std::atomic<bool> enabled{false};
+ std::shared_ptr<MetricsImpl> histograms = std::make_shared<MetricsImpl>();
+};
+
+struct PrefetchIoMetricsState {
+ std::atomic<uint64_t> read_requests{0};
+ std::atomic<uint64_t> read_requested_bytes{0};
+ std::atomic<uint64_t> read_physical_bytes{0};
+ std::atomic<uint64_t> read_failed{0};
+ std::atomic<uint64_t> read_latency_count{0};
+ std::atomic<uint64_t> read_latency_sum_us{0};
+ std::atomic<uint64_t> async_requests{0};
+ std::atomic<uint64_t> async_requested_bytes{0};
+ std::atomic<uint64_t> async_physical_bytes{0};
+ std::atomic<uint64_t> async_completed{0};
+ std::atomic<uint64_t> async_failed{0};
+ std::atomic<uint64_t> async_pending{0};
+ std::atomic<uint64_t> async_latency_count{0};
+ std::atomic<uint64_t> async_latency_sum_us{0};
+};
+
namespace {
+// Metrics do not synchronize reader state. A concurrently collected snapshot
may be approximate.
+constexpr std::memory_order kMetricsMemoryOrder = std::memory_order_relaxed;
+
+uint64_t ElapsedMicros(const std::chrono::steady_clock::time_point& start) {
+ return
static_cast<uint64_t>(std::chrono::duration_cast<std::chrono::microseconds>(
+ std::chrono::steady_clock::now() - start)
+ .count());
+}
+
+void UpdateMax(std::atomic<uint64_t>* target, uint64_t value) {
+ uint64_t current = target->load(kMetricsMemoryOrder);
+ while (current < value && !target->compare_exchange_weak(current, value,
kMetricsMemoryOrder,
+
kMetricsMemoryOrder)) {
+ }
+}
+
+void RecordLatency(uint64_t latency_us, std::atomic<uint64_t>* count,
+ std::atomic<uint64_t>* sum_us) {
+ count->fetch_add(1, kMetricsMemoryOrder);
+ sum_us->fetch_add(latency_us, kMetricsMemoryOrder);
+}
+
+class MetricsInputStream : public InputStream {
+ public:
+ MetricsInputStream(const std::shared_ptr<InputStream>& stream,
+ const std::shared_ptr<PrefetchIoMetricsState>& metrics)
+ : stream_(stream), metrics_(metrics) {}
+
+ MetricsInputStream(std::unique_ptr<InputStream>&& stream,
+ const std::shared_ptr<PrefetchIoMetricsState>& metrics)
+ : stream_(std::move(stream)), metrics_(metrics) {}
+
+ Status Seek(int64_t offset, SeekOrigin origin) override {
+ return stream_->Seek(offset, origin);
+ }
+
+ Result<int64_t> GetPos() const override {
+ return stream_->GetPos();
+ }
+
+ Result<int64_t> Read(char* buffer, int64_t size) override {
+ return RecordRead([&]() { return stream_->Read(buffer, size); }, size);
+ }
+
+ Result<int64_t> Read(char* buffer, int64_t size, int64_t offset) override {
+ return RecordRead([&]() { return stream_->Read(buffer, size, offset);
}, size);
+ }
+
+ void ReadAsync(char* buffer, int64_t size, int64_t offset,
+ std::function<void(Status)>&& callback) override {
+ metrics_->async_requests.fetch_add(1, kMetricsMemoryOrder);
+
metrics_->async_requested_bytes.fetch_add(static_cast<uint64_t>(std::max<int64_t>(0,
size)),
+ kMetricsMemoryOrder);
+ metrics_->async_pending.fetch_add(1, kMetricsMemoryOrder);
+ std::shared_ptr<PrefetchIoMetricsState> metrics = metrics_;
+ const auto start = std::chrono::steady_clock::now();
+ stream_->ReadAsync(
+ buffer, size, offset,
+ [metrics, size, start, callback = std::move(callback)](Status
status) mutable {
+ metrics->async_pending.fetch_sub(1, kMetricsMemoryOrder);
+ if (status.ok()) {
+ metrics->async_completed.fetch_add(1, kMetricsMemoryOrder);
+ metrics->async_physical_bytes.fetch_add(
+ static_cast<uint64_t>(std::max<int64_t>(0, size)),
kMetricsMemoryOrder);
+ } else {
+ metrics->async_failed.fetch_add(1, kMetricsMemoryOrder);
+ }
+ RecordLatency(ElapsedMicros(start),
&metrics->async_latency_count,
+ &metrics->async_latency_sum_us);
+ callback(status);
+ });
+ }
+
+ Status Close() override {
+ return stream_->Close();
+ }
+
+ Result<std::string> GetUri() const override {
+ return stream_->GetUri();
+ }
+
+ Result<int64_t> Length() const override {
+ return stream_->Length();
+ }
+
+ private:
+ template <typename ReadFunction>
+ Result<int64_t> RecordRead(ReadFunction&& read, int64_t size) {
+ metrics_->read_requests.fetch_add(1, kMetricsMemoryOrder);
+
metrics_->read_requested_bytes.fetch_add(static_cast<uint64_t>(std::max<int64_t>(0,
size)),
+ kMetricsMemoryOrder);
+ const auto start = std::chrono::steady_clock::now();
+ Result<int64_t> result = read();
+ if (result.ok()) {
+ metrics_->read_physical_bytes.fetch_add(
+ static_cast<uint64_t>(std::max<int64_t>(0, result.value())),
kMetricsMemoryOrder);
+ } else {
+ metrics_->read_failed.fetch_add(1, kMetricsMemoryOrder);
+ }
+ RecordLatency(ElapsedMicros(start), &metrics_->read_latency_count,
+ &metrics_->read_latency_sum_us);
+ return result;
+ }
+
+ std::shared_ptr<InputStream> stream_;
+ std::shared_ptr<PrefetchIoMetricsState> metrics_;
+};
+
std::pair<int64_t, int64_t> ComputeBatchSliceByReadRange(
const std::vector<uint64_t>& global_row_ids, const std::pair<uint64_t,
uint64_t>& read_range) {
auto begin_it =
@@ -62,7 +204,7 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>>
PrefetchFileBatchReaderImpl
const std::shared_ptr<FileSystem>& fs, uint32_t prefetch_max_parallel_num,
int32_t batch_size,
uint32_t prefetch_batch_count, bool enable_adaptive_prefetch_strategy,
const std::shared_ptr<Executor>& executor, bool initialize_read_ranges,
- bool read_ahead_cache_enabled, const CacheConfig& cache_config,
+ bool read_ahead_cache_enabled, const CacheConfig& cache_config, bool
enable_io_metrics,
const std::shared_ptr<MemoryPool>& pool) {
if (prefetch_max_parallel_num == 0) {
return Status::Invalid("prefetch max parallel num should be greater
than 0.");
@@ -83,24 +225,35 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>>
PrefetchFileBatchReaderImpl
return Status::Invalid("executor should not be nullptr.");
}
+ std::shared_ptr<PrefetchIoMetricsState> io_metrics;
+ if (enable_io_metrics) {
+ io_metrics = std::make_shared<PrefetchIoMetricsState>();
+ }
std::shared_ptr<ReadAheadCache> cache;
if (read_ahead_cache_enabled) {
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
fs->Open(FileStatus(data_file_path,
data_file_size)));
+ if (io_metrics) {
+ input_stream = std::make_shared<MetricsInputStream>(input_stream,
io_metrics);
+ }
cache = std::make_shared<ReadAheadCache>(input_stream, cache_config,
pool);
}
std::vector<std::future<Result<std::unique_ptr<FileBatchReader>>>> futures;
for (uint32_t i = 0; i < prefetch_max_parallel_num; i++) {
- futures.push_back(
- Via(executor.get(),
- [&fs, &data_file_path, data_file_size, &reader_builder,
- &cache]() -> Result<std::unique_ptr<FileBatchReader>> {
- PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream>
input_stream,
- fs->Open(FileStatus(data_file_path,
data_file_size)));
- auto cache_input_stream =
-
std::make_shared<CacheInputStream>(std::move(input_stream), cache);
- return reader_builder->Build(cache_input_stream);
- }));
+ futures.push_back(Via(
+ executor.get(),
+ [&fs, &data_file_path, data_file_size, &reader_builder, &cache,
+ io_metrics]() -> Result<std::unique_ptr<FileBatchReader>> {
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream>
input_stream,
+ fs->Open(FileStatus(data_file_path,
data_file_size)));
+ if (io_metrics) {
+ input_stream =
+
std::make_unique<MetricsInputStream>(std::move(input_stream), io_metrics);
+ }
+ auto cache_input_stream =
+
std::make_shared<CacheInputStream>(std::move(input_stream), cache);
+ return reader_builder->Build(cache_input_stream);
+ }));
}
std::vector<std::shared_ptr<PrefetchFileBatchReader>> readers;
for (auto& file_batch_reader : CollectAll(futures)) {
@@ -121,9 +274,9 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>>
PrefetchFileBatchReaderImpl
}
uint32_t prefetch_queue_capacity = prefetch_batch_count / readers.size();
- auto reader = std::unique_ptr<PrefetchFileBatchReaderImpl>(
- new PrefetchFileBatchReaderImpl(readers, batch_size,
prefetch_queue_capacity,
- enable_adaptive_prefetch_strategy,
executor, cache, pool));
+ auto reader = std::unique_ptr<PrefetchFileBatchReaderImpl>(new
PrefetchFileBatchReaderImpl(
+ readers, batch_size, prefetch_queue_capacity,
enable_adaptive_prefetch_strategy, executor,
+ cache, io_metrics, pool));
if (initialize_read_ranges) {
// normally initialize read ranges should be false, as set read schema
will refresh read
// ranges, and set read schema will always be called before read.
@@ -136,6 +289,7 @@ PrefetchFileBatchReaderImpl::PrefetchFileBatchReaderImpl(
const std::vector<std::shared_ptr<PrefetchFileBatchReader>>& readers,
int32_t batch_size,
uint32_t prefetch_queue_capacity, bool enable_adaptive_prefetch_strategy,
const std::shared_ptr<Executor>& executor, const
std::shared_ptr<ReadAheadCache>& cache,
+ const std::shared_ptr<PrefetchIoMetricsState>& io_metrics,
const std::shared_ptr<MemoryPool>& pool)
: readers_(std::move(readers)),
batch_size_(batch_size),
@@ -143,7 +297,9 @@ PrefetchFileBatchReaderImpl::PrefetchFileBatchReaderImpl(
cache_(cache),
arrow_pool_(GetArrowPool(pool)),
prefetch_queue_capacity_(prefetch_queue_capacity),
- enable_adaptive_prefetch_strategy_(enable_adaptive_prefetch_strategy) {
+ enable_adaptive_prefetch_strategy_(enable_adaptive_prefetch_strategy),
+ prefetch_metrics_(std::make_shared<PrefetchMetricsState>()),
+ io_metrics_(io_metrics) {
for (size_t i = 0; i < readers_.size(); i++) {
prefetch_queues_.emplace_back(std::make_unique<ThreadsafeQueue<PrefetchBatch>>());
readers_pos_.emplace_back(std::make_unique<std::atomic<uint64_t>>(0));
@@ -186,6 +342,7 @@ Status PrefetchFileBatchReaderImpl::RefreshReadRanges() {
Status PrefetchFileBatchReaderImpl::RefreshReadRangesAfterCleanUp() {
bool need_prefetch;
PAIMON_ASSIGN_OR_RAISE(auto read_ranges,
readers_[0]->GenReadRanges(&need_prefetch));
+ const bool format_requested_prefetch = need_prefetch;
if (!enable_adaptive_prefetch_strategy_) {
need_prefetch = true;
@@ -197,8 +354,17 @@ Status
PrefetchFileBatchReaderImpl::RefreshReadRangesAfterCleanUp() {
}
}
+ if (format_requested_prefetch && !need_prefetch) {
+ prefetch_metrics_->adaptive_disabled_count.fetch_add(1,
kMetricsMemoryOrder);
+ }
need_prefetch_ = need_prefetch;
- PAIMON_RETURN_NOT_OK(SetReadRanges(FilterReadRanges(read_ranges,
selection_bitmap_)));
+ prefetch_metrics_->enabled.store(need_prefetch_, kMetricsMemoryOrder);
+ prefetch_metrics_->read_ranges_total.fetch_add(read_ranges.size(),
kMetricsMemoryOrder);
+ std::vector<std::pair<uint64_t, uint64_t>> filtered_ranges =
+ FilterReadRanges(read_ranges, selection_bitmap_);
+
prefetch_metrics_->read_ranges_after_bitmap.fetch_add(filtered_ranges.size(),
+ kMetricsMemoryOrder);
+ PAIMON_RETURN_NOT_OK(SetReadRanges(filtered_ranges));
return Status::OK();
}
@@ -272,6 +438,7 @@ Status PrefetchFileBatchReaderImpl::CleanUp() {
if (batch == std::nullopt) {
break;
}
+ prefetch_metrics_->discarded_batches.fetch_add(1,
kMetricsMemoryOrder);
ReaderUtils::ReleaseReadBatch(std::move(batch.value().batch.first));
}
}
@@ -297,6 +464,7 @@ Status PrefetchFileBatchReaderImpl::CleanUp() {
current_batch_global_row_ids_.clear();
read_ranges_freshed_ = false;
clean_prefetch_queue();
+ prefetch_metrics_->queue_depth.store(0, kMetricsMemoryOrder);
for (size_t i = 0; i < readers_pos_.size(); i++) {
readers_pos_[i]->store(0);
reader_is_working_[i] = false;
@@ -357,6 +525,7 @@ void PrefetchFileBatchReaderImpl::Workloop() {
}
if (prefetch_queues_[reader_idx]->size() >=
prefetch_queue_capacity_) {
// queue is full, skip
+ prefetch_metrics_->queue_full_count.fetch_add(1,
kMetricsMemoryOrder);
continue;
}
if (readers_pos_[reader_idx]->load() !=
std::numeric_limits<uint64_t>::max()) {
@@ -417,6 +586,7 @@ Status PrefetchFileBatchReaderImpl::EnsureReaderPosition(
uint64_t pos = std::max(readers_pos_[reader_idx]->load(),
current_read_range.first);
PAIMON_ASSIGN_OR_RAISE(uint64_t next_row_to_read,
readers_[reader_idx]->GetNextRowToRead());
if (next_row_to_read != pos) {
+ prefetch_metrics_->seek_count.fetch_add(1, kMetricsMemoryOrder);
return readers_[reader_idx]->SeekToRow(pos);
}
return Status::OK();
@@ -447,6 +617,7 @@ Status PrefetchFileBatchReaderImpl::HandleReadResult(
global_row_ids.push_back(global_row_id);
}
if (global_row_ids.empty()) {
+ prefetch_metrics_->discarded_batches.fetch_add(1,
kMetricsMemoryOrder);
ReaderUtils::ReleaseReadBatch(std::move(read_batch));
return Status::OK();
}
@@ -464,6 +635,7 @@ Status PrefetchFileBatchReaderImpl::HandleReadResult(
FindReadRangeContaining(reader_idx, global_row_ids[0]);
if (owner_range == std::nullopt) {
readers_pos_[reader_idx]->store(global_row_ids[0]);
+ prefetch_metrics_->discarded_batches.fetch_add(1,
kMetricsMemoryOrder);
ReaderUtils::ReleaseReadBatch(std::move(read_batch));
return Status::OK();
}
@@ -493,15 +665,23 @@ Status PrefetchFileBatchReaderImpl::HandleReadResult(
readers_pos_[reader_idx]->store(next_row_to_read);
}
if (bitmap.IsEmpty()) {
+ prefetch_metrics_->discarded_batches.fetch_add(1,
kMetricsMemoryOrder);
ReaderUtils::ReleaseReadBatch(std::move(read_batch));
return Status::OK();
}
prefetch_queue->push(
{read_range, std::move(read_batch_with_bitmap),
std::move(global_row_ids)});
+ prefetch_metrics_->produced_batches.fetch_add(1, kMetricsMemoryOrder);
+ const uint64_t queue_depth =
+ prefetch_metrics_->queue_depth.fetch_add(1, kMetricsMemoryOrder) +
1;
+ UpdateMax(&prefetch_metrics_->queue_depth_max, queue_depth);
} else {
std::pair<uint64_t, uint64_t> eof_range;
PAIMON_ASSIGN_OR_RAISE(eof_range, EofRange());
prefetch_queue->push({eof_range, std::move(read_batch_with_bitmap),
{}});
+ const uint64_t queue_depth =
+ prefetch_metrics_->queue_depth.fetch_add(1, kMetricsMemoryOrder) +
1;
+ UpdateMax(&prefetch_metrics_->queue_depth_max, queue_depth);
readers_pos_[reader_idx]->store(std::numeric_limits<uint64_t>::max());
}
return Status::OK();
@@ -532,8 +712,11 @@ Status PrefetchFileBatchReaderImpl::DoReadBatch(size_t
reader_idx) {
FileBatchReader* reader = readers_[reader_idx].get();
PAIMON_RETURN_NOT_OK(EnsureReaderPosition(reader_idx, read_range));
- PAIMON_ASSIGN_OR_RAISE(ReadBatchWithBitmap read_batch_with_bitmap,
- reader->NextBatchWithBitmap());
+ const auto read_start = std::chrono::steady_clock::now();
+ Result<ReadBatchWithBitmap> read_result = reader->NextBatchWithBitmap();
+
prefetch_metrics_->histograms->ObserveHistogram(PrefetchMetrics::READER_READ_LATENCY_US,
+ ElapsedMicros(read_start));
+ PAIMON_ASSIGN_OR_RAISE(ReadBatchWithBitmap read_batch_with_bitmap,
std::move(read_result));
return HandleReadResult(reader_idx, read_range,
std::move(read_batch_with_bitmap));
}
@@ -547,6 +730,7 @@ Result<BatchReader::ReadBatchWithBitmap>
PrefetchFileBatchReaderImpl::NextBatchW
std::make_unique<std::thread>(&PrefetchFileBatchReaderImpl::Workloop, this);
}
+ const auto wait_start = std::chrono::steady_clock::now();
while (true) {
PAIMON_RETURN_NOT_OK(GetReadStatus());
if (is_shutdown_) {
@@ -585,6 +769,10 @@ Result<BatchReader::ReadBatchWithBitmap>
PrefetchFileBatchReaderImpl::NextBatchW
cv_.notify_one();
}
current_batch_global_row_ids_ =
std::move(prefetch_batch.value().global_row_ids);
+ prefetch_metrics_->consumed_batches.fetch_add(1,
kMetricsMemoryOrder);
+ prefetch_metrics_->queue_depth.fetch_sub(1,
kMetricsMemoryOrder);
+ prefetch_metrics_->histograms->ObserveHistogram(
+ PrefetchMetrics::CONSUMER_WAIT_LATENCY_US,
ElapsedMicros(wait_start));
return std::move(prefetch_batch).value().batch;
}
}
@@ -595,6 +783,8 @@ Result<BatchReader::ReadBatchWithBitmap>
PrefetchFileBatchReaderImpl::NextBatchW
return Status::Invalid("peek batch not suppose to be nullptr");
}
current_batch_global_row_ids_.clear();
+ prefetch_metrics_->histograms->ObserveHistogram(
+ PrefetchMetrics::CONSUMER_WAIT_LATENCY_US,
ElapsedMicros(wait_start));
return BatchReader::MakeEofBatchWithBitmap();
}
if (value_count == prefetch_queues_.size()) {
@@ -620,15 +810,69 @@ Status PrefetchFileBatchReaderImpl::SeekToRow(uint64_t
row_number) {
}
std::shared_ptr<Metrics> PrefetchFileBatchReaderImpl::GetReaderMetrics() const
{
- auto res_metrics = MetricsImpl::CollectReadMetrics(readers_);
+ auto result = std::make_shared<MetricsImpl>();
+ if (need_prefetch_) {
+ result->Merge(MetricsImpl::CollectReadMetrics(readers_));
+ } else if (!readers_.empty()) {
+ result->Merge(readers_[0]->GetReaderMetrics());
+ }
+
+ auto set_prefetch_counter = [&result](const char* name, const
std::atomic<uint64_t>& value) {
+ result->SetCounter(name, value.load(kMetricsMemoryOrder));
+ };
+ set_prefetch_counter(PrefetchMetrics::READ_RANGES_TOTAL,
prefetch_metrics_->read_ranges_total);
+ set_prefetch_counter(PrefetchMetrics::READ_RANGES_AFTER_BITMAP,
+ prefetch_metrics_->read_ranges_after_bitmap);
+ set_prefetch_counter(PrefetchMetrics::SEEK_COUNT,
prefetch_metrics_->seek_count);
+ set_prefetch_counter(PrefetchMetrics::PRODUCED_BATCHES,
prefetch_metrics_->produced_batches);
+ set_prefetch_counter(PrefetchMetrics::CONSUMED_BATCHES,
prefetch_metrics_->consumed_batches);
+ set_prefetch_counter(PrefetchMetrics::DISCARDED_BATCHES,
prefetch_metrics_->discarded_batches);
+ set_prefetch_counter(PrefetchMetrics::ERRORS, prefetch_metrics_->errors);
+ set_prefetch_counter(PrefetchMetrics::ADAPTIVE_DISABLED_COUNT,
+ prefetch_metrics_->adaptive_disabled_count);
+ set_prefetch_counter(PrefetchMetrics::QUEUE_FULL_COUNT,
prefetch_metrics_->queue_full_count);
+ result->SetGauge(PrefetchMetrics::ENABLED,
+ prefetch_metrics_->enabled.load(kMetricsMemoryOrder) ?
1.0 : 0.0);
+ result->SetGauge(PrefetchMetrics::PARALLELISM,
+ prefetch_metrics_->enabled.load(kMetricsMemoryOrder)
+ ? static_cast<double>(parallel_num_)
+ : 1.0);
+ result->SetGauge(PrefetchMetrics::QUEUE_DEPTH,
+
static_cast<double>(prefetch_metrics_->queue_depth.load(kMetricsMemoryOrder)));
+ result->SetGauge(
+ PrefetchMetrics::QUEUE_DEPTH_MAX,
+
static_cast<double>(prefetch_metrics_->queue_depth_max.load(kMetricsMemoryOrder)));
+ result->Merge(prefetch_metrics_->histograms);
if (cache_) {
- // The shared read-ahead cache serves reads of all sub-readers, so its
- // hit/miss counters are file-level and merge into the reader metrics.
+ // PR #209 owns the read-ahead cache metrics. Keep collecting its
file-level
+ // hit/miss counters without defining another C++ metrics surface here.
std::shared_ptr<Metrics> cache_metrics =
std::make_shared<MetricsImpl>();
cache_->CollectMetrics(&cache_metrics);
- res_metrics->Merge(cache_metrics);
+ result->Merge(cache_metrics);
}
- return res_metrics;
+
+ if (!io_metrics_) {
+ return result;
+ }
+ auto set_io_counter = [&result](const char* name, const
std::atomic<uint64_t>& value) {
+ result->SetCounter(name, value.load(kMetricsMemoryOrder));
+ };
+ set_io_counter(PrefetchIoMetrics::READ_REQUESTS,
io_metrics_->read_requests);
+ set_io_counter(PrefetchIoMetrics::READ_REQUESTED_BYTES,
io_metrics_->read_requested_bytes);
+ set_io_counter(PrefetchIoMetrics::READ_PHYSICAL_BYTES,
io_metrics_->read_physical_bytes);
+ set_io_counter(PrefetchIoMetrics::READ_FAILED, io_metrics_->read_failed);
+ set_io_counter(PrefetchIoMetrics::READ_LATENCY_COUNT,
io_metrics_->read_latency_count);
+ set_io_counter(PrefetchIoMetrics::READ_LATENCY_SUM_US,
io_metrics_->read_latency_sum_us);
+ set_io_counter(PrefetchIoMetrics::ASYNC_REQUESTS,
io_metrics_->async_requests);
+ set_io_counter(PrefetchIoMetrics::ASYNC_REQUESTED_BYTES,
io_metrics_->async_requested_bytes);
+ set_io_counter(PrefetchIoMetrics::ASYNC_PHYSICAL_BYTES,
io_metrics_->async_physical_bytes);
+ set_io_counter(PrefetchIoMetrics::ASYNC_COMPLETED,
io_metrics_->async_completed);
+ set_io_counter(PrefetchIoMetrics::ASYNC_FAILED, io_metrics_->async_failed);
+ set_io_counter(PrefetchIoMetrics::ASYNC_LATENCY_COUNT,
io_metrics_->async_latency_count);
+ set_io_counter(PrefetchIoMetrics::ASYNC_LATENCY_SUM_US,
io_metrics_->async_latency_sum_us);
+ result->SetGauge(PrefetchIoMetrics::ASYNC_PENDING,
+
static_cast<double>(io_metrics_->async_pending.load(kMetricsMemoryOrder)));
+ return result;
}
Result<std::unique_ptr<::ArrowSchema>>
PrefetchFileBatchReaderImpl::GetFileSchema() const {
@@ -662,6 +906,9 @@ Result<uint64_t>
PrefetchFileBatchReaderImpl::GetNextRowToRead() const {
}
void PrefetchFileBatchReaderImpl::SetReadStatus(const Status& status) {
+ if (!status.ok()) {
+ prefetch_metrics_->errors.fetch_add(1, kMetricsMemoryOrder);
+ }
std::unique_lock<std::shared_mutex> lock(rw_mutex_);
read_status_ = status;
}
diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl.h
b/src/paimon/common/reader/prefetch_file_batch_reader_impl.h
index 08d3bd37..c9932369 100644
--- a/src/paimon/common/reader/prefetch_file_batch_reader_impl.h
+++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl.h
@@ -57,6 +57,8 @@ class FileSystem;
class Executor;
class Predicate;
class Metrics;
+struct PrefetchMetricsState;
+struct PrefetchIoMetricsState;
class PrefetchFileBatchReaderImpl : public PrefetchFileBatchReader {
public:
@@ -66,7 +68,7 @@ class PrefetchFileBatchReaderImpl : public
PrefetchFileBatchReader {
uint32_t prefetch_max_parallel_num, int32_t batch_size, uint32_t
prefetch_batch_count,
bool enable_adaptive_prefetch_strategy, const
std::shared_ptr<Executor>& executor,
bool initialize_read_ranges, bool read_ahead_cache_enabled, const
CacheConfig& cache_config,
- const std::shared_ptr<MemoryPool>& pool);
+ bool enable_io_metrics, const std::shared_ptr<MemoryPool>& pool);
~PrefetchFileBatchReaderImpl() override;
@@ -119,6 +121,7 @@ class PrefetchFileBatchReaderImpl : public
PrefetchFileBatchReader {
const std::vector<std::shared_ptr<PrefetchFileBatchReader>>& readers,
int32_t batch_size,
uint32_t prefetch_queue_capacity, bool
enable_adaptive_prefetch_strategy,
const std::shared_ptr<Executor>& executor, const
std::shared_ptr<ReadAheadCache>& cache,
+ const std::shared_ptr<PrefetchIoMetricsState>& io_metrics,
const std::shared_ptr<MemoryPool>& pool);
Status CleanUp();
@@ -178,5 +181,7 @@ class PrefetchFileBatchReaderImpl : public
PrefetchFileBatchReader {
const uint32_t prefetch_queue_capacity_;
const bool enable_adaptive_prefetch_strategy_;
int32_t parallel_num_;
+ std::shared_ptr<PrefetchMetricsState> prefetch_metrics_;
+ std::shared_ptr<PrefetchIoMetricsState> io_metrics_;
};
} // namespace paimon
diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
b/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
index 192028ac..514e7e49 100644
--- a/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
+++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
@@ -34,6 +34,7 @@
#include "paimon/format/format_writer.h"
#include "paimon/fs/file_system_factory.h"
#include "paimon/fs/local/local_file_system.h"
+#include "paimon/metrics.h"
#include "paimon/predicate/predicate_builder.h"
#include "paimon/testing/mock/mock_file_batch_reader.h"
#include "paimon/testing/mock/mock_file_system.h"
@@ -113,6 +114,62 @@ class ControlledMockFormatReaderBuilder : public
ReaderBuilder {
mutable std::atomic<size_t> build_count_{0};
};
+class FailingInputStream : public MockInputStream {
+ public:
+ Result<int64_t> Read(char* buffer, int64_t size, int64_t offset) override {
+ return Status::IOError("injected synchronous read failure");
+ }
+};
+
+class FailingFileSystem : public MockFileSystem {
+ public:
+ Result<std::unique_ptr<InputStream>> Open(const std::string& path) const
override {
+ return std::make_unique<FailingInputStream>();
+ }
+};
+
+class IoReadingMockFileBatchReader : public MockFileBatchReader {
+ public:
+ IoReadingMockFileBatchReader(const std::shared_ptr<arrow::Array>& data,
+ const std::shared_ptr<arrow::DataType>&
schema,
+ int32_t read_batch_size,
+ const std::shared_ptr<InputStream>&
input_stream)
+ : MockFileBatchReader(data, schema, read_batch_size),
input_stream_(input_stream) {}
+
+ Result<ReadBatchWithBitmap> NextBatchWithBitmap() override {
+ char value = 0;
+ PAIMON_ASSIGN_OR_RAISE(int64_t read_size, input_stream_->Read(&value,
1, 0));
+ (void)read_size;
+ return MockFileBatchReader::NextBatchWithBitmap();
+ }
+
+ private:
+ std::shared_ptr<InputStream> input_stream_;
+};
+
+class IoReadingMockFormatReaderBuilder : public ReaderBuilder {
+ public:
+ IoReadingMockFormatReaderBuilder(const std::shared_ptr<arrow::Array>& data,
+ const std::shared_ptr<arrow::DataType>&
schema,
+ int32_t read_batch_size)
+ : data_(data), schema_(schema), read_batch_size_(read_batch_size) {}
+
+ ReaderBuilder* WithMemoryPool(const std::shared_ptr<MemoryPool>& pool)
override {
+ return this;
+ }
+
+ Result<std::unique_ptr<FileBatchReader>> Build(
+ const std::shared_ptr<InputStream>& input_stream) const override {
+ return std::make_unique<IoReadingMockFileBatchReader>(data_, schema_,
read_batch_size_,
+ input_stream);
+ }
+
+ private:
+ std::shared_ptr<arrow::Array> data_;
+ std::shared_ptr<arrow::DataType> schema_;
+ int32_t read_batch_size_ = 0;
+};
+
struct TestParam {
std::string file_format;
bool read_ahead_cache_enabled;
@@ -210,7 +267,7 @@ class PrefetchFileBatchReaderImplTest : public
::testing::Test,
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false, executor,
/*initialize_read_ranges=*/false, read_ahead_cache_enabled,
CacheConfig(),
- GetDefaultPool()));
+ /*enable_io_metrics=*/true, GetDefaultPool()));
std::unique_ptr<ArrowSchema> c_schema =
std::make_unique<ArrowSchema>();
auto arrow_status = arrow::ExportSchema(*read_schema, c_schema.get());
EXPECT_TRUE(arrow_status.ok());
@@ -299,7 +356,11 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestSimple) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
+ if (prefetch_max_parallel_num == 1) {
+ ASSERT_NOK(
+
reader->GetReaderMetrics()->GetCounter(PrefetchIoMetrics::READ_LATENCY_COUNT));
+ }
ASSERT_NOK(reader->GetPreviousBatchFileRowId(0));
ASSERT_OK_AND_ASSIGN(auto array_and_row_ids,
CollectResultAndRowIds(reader.get()));
@@ -322,7 +383,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestReadWithLimits)
{
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/true,
GetDefaultPool()));
// simulate read limits, only read 8 batches
for (int32_t i = 0; i < 8; i++) {
ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatchWithBitmap
batch_with_bitmap,
@@ -337,6 +398,30 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestReadWithLimits) {
// test metrics
auto read_metrics = reader->GetReaderMetrics();
ASSERT_TRUE(read_metrics);
+ ASSERT_OK_AND_ASSIGN(double prefetch_enabled,
read_metrics->GetGauge(PrefetchMetrics::ENABLED));
+ ASSERT_OK_AND_ASSIGN(uint64_t produced_batches,
+
read_metrics->GetCounter(PrefetchMetrics::PRODUCED_BATCHES));
+ ASSERT_OK_AND_ASSIGN(uint64_t consumed_batches,
+
read_metrics->GetCounter(PrefetchMetrics::CONSUMED_BATCHES));
+ ASSERT_OK_AND_ASSIGN(uint64_t read_ranges,
+
read_metrics->GetCounter(PrefetchMetrics::READ_RANGES_TOTAL));
+ ASSERT_EQ(prefetch_enabled, 1.0);
+ ASSERT_GT(produced_batches, 0);
+ ASSERT_EQ(consumed_batches, 8);
+ ASSERT_GT(read_ranges, 0);
+ ASSERT_OK(read_metrics->GetGauge(PrefetchMetrics::QUEUE_DEPTH));
+ ASSERT_OK_AND_ASSIGN(uint64_t async_completed,
+
read_metrics->GetCounter(PrefetchIoMetrics::ASYNC_COMPLETED));
+ ASSERT_OK_AND_ASSIGN(uint64_t async_failed,
+
read_metrics->GetCounter(PrefetchIoMetrics::ASYNC_FAILED));
+ ASSERT_OK_AND_ASSIGN(uint64_t async_latency_count,
+
read_metrics->GetCounter(PrefetchIoMetrics::ASYNC_LATENCY_COUNT));
+ ASSERT_EQ(async_latency_count, async_completed + async_failed);
+
+ std::shared_ptr<Metrics> second_metrics = reader->GetReaderMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t second_consumed_batches,
+
second_metrics->GetCounter(PrefetchMetrics::CONSUMED_BATCHES));
+ ASSERT_EQ(second_consumed_batches, consumed_batches);
}
TEST_F(PrefetchFileBatchReaderImplTest, TestReadWithoutInitializeReadRanges) {
@@ -351,13 +436,43 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestReadWithoutInitializeReadRanges) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
// simulate read limits, only read 8 batches
ASSERT_NOK_WITH_MSG(reader->NextBatchWithBitmap(),
"prefetch reader read ranges are not initialized");
reader->Close();
}
+TEST_F(PrefetchFileBatchReaderImplTest, TestFailedIoMetrics) {
+ auto data_array = PrepareArray(10);
+ IoReadingMockFormatReaderBuilder reader_builder(data_array, data_type_,
/*read_batch_size=*/10);
+ auto failing_fs = std::make_shared<FailingFileSystem>();
+ ASSERT_OK_AND_ASSIGN(
+ auto reader,
+ PrefetchFileBatchReaderImpl::Create(
+ /*data_file_path=*/"", /*data_file_size=*/0, &reader_builder,
failing_fs,
+ /*prefetch_max_parallel_num=*/1, /*batch_size=*/10,
+ /*prefetch_batch_count=*/2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
+ /*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/false, CacheConfig(),
+ /*enable_io_metrics=*/true, GetDefaultPool()));
+
+ ASSERT_NOK_WITH_MSG(reader->NextBatchWithBitmap(), "injected synchronous
read failure");
+ std::shared_ptr<Metrics> metrics = reader->GetReaderMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t read_requests,
+
metrics->GetCounter(PrefetchIoMetrics::READ_REQUESTS));
+ ASSERT_OK_AND_ASSIGN(uint64_t failed_requests,
+ metrics->GetCounter(PrefetchIoMetrics::READ_FAILED));
+ ASSERT_OK_AND_ASSIGN(uint64_t physical_bytes,
+
metrics->GetCounter(PrefetchIoMetrics::READ_PHYSICAL_BYTES));
+ ASSERT_EQ(read_requests, 1);
+ ASSERT_EQ(failed_requests, 1);
+ ASSERT_EQ(physical_bytes, 0);
+ ASSERT_OK_AND_ASSIGN(uint64_t latency_count,
+
metrics->GetCounter(PrefetchIoMetrics::READ_LATENCY_COUNT));
+ ASSERT_OK(metrics->GetCounter(PrefetchIoMetrics::READ_LATENCY_SUM_US));
+ ASSERT_EQ(latency_count, read_requests);
+}
+
TEST_F(PrefetchFileBatchReaderImplTest, FilterReadRangesWithoutBitmap) {
std::vector<std::pair<uint64_t, uint64_t>> read_ranges = {
{0, 1000}, {1000, 2000}, {2000, 3000}, {3000, 4000}, {4000, 5000},
@@ -427,7 +542,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, RefreshReadRanges) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
ASSERT_OK(prefetch_reader->RefreshReadRanges());
std::vector<std::pair<uint64_t, uint64_t>> read_ranges_0 = {{0, 30}, {90,
101}};
@@ -457,9 +572,17 @@ TEST_F(PrefetchFileBatchReaderImplTest,
RefreshReadRangesDisablePrefetchByAdapti
/*prefetch_batch_count=*/2,
/*enable_adaptive_prefetch_strategy=*/true,
executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
ASSERT_FALSE(reader->NeedPrefetch());
+ std::shared_ptr<Metrics> metrics = reader->GetReaderMetrics();
+ ASSERT_OK_AND_ASSIGN(double enabled,
metrics->GetGauge(PrefetchMetrics::ENABLED));
+ ASSERT_OK_AND_ASSIGN(double parallelism,
metrics->GetGauge(PrefetchMetrics::PARALLELISM));
+ ASSERT_OK_AND_ASSIGN(uint64_t adaptive_disabled_count,
+
metrics->GetCounter(PrefetchMetrics::ADAPTIVE_DISABLED_COUNT));
+ ASSERT_EQ(enabled, 0.0);
+ ASSERT_EQ(parallelism, 1.0);
+ ASSERT_EQ(adaptive_disabled_count, 1);
}
TEST_F(PrefetchFileBatchReaderImplTest, SetReadRanges) {
@@ -473,7 +596,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, SetReadRanges) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
ASSERT_FALSE(prefetch_reader->need_prefetch_);
prefetch_reader->need_prefetch_ = true;
@@ -516,7 +639,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
SetReadRangesReturnErrorWhenPushDownFail
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
prefetch_reader->need_prefetch_ = true;
@@ -542,7 +665,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
WorkloopSetReadStatusWhenCacheInitFailed
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true,
- invalid_cache_config, GetDefaultPool()));
+ invalid_cache_config,
/*enable_io_metrics=*/false, GetDefaultPool()));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
prefetch_reader->Workloop();
@@ -562,7 +685,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
DoReadBatchReturnOkWhenShutdown) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
prefetch_reader->is_shutdown_ = true;
@@ -580,7 +703,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
DoReadBatchReturnOkWhenNoCurrentReadRang
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/false,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
prefetch_reader->read_ranges_in_group_ = {{}};
@@ -598,7 +721,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestReadWithLargeBatchSize) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
ASSERT_NOK(reader->GetPreviousBatchFileRowId(0));
ASSERT_OK_AND_ASSIGN(auto array_and_row_ids,
CollectResultAndRowIds(reader.get()));
auto row_ids = array_and_row_ids.second;
@@ -618,7 +741,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestPartialReaderSuccessRead) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
for (int32_t i = 0; i < prefetch_max_parallel_num; i++) {
dynamic_cast<MockFileBatchReader*>(prefetch_reader->readers_[i].get())
@@ -663,7 +786,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestAllReaderFailedWithIOError) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
auto prefetch_reader =
dynamic_cast<PrefetchFileBatchReaderImpl*>(reader.get());
for (int32_t i = 0; i < prefetch_max_parallel_num; i++) {
@@ -679,6 +802,9 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestAllReaderFailedWithIOError) {
ASSERT_FALSE(prefetch_reader->is_shutdown_);
ASSERT_NOK(prefetch_reader->GetReadStatus());
ASSERT_FALSE(HasValue(prefetch_reader->prefetch_queues_));
+ std::shared_ptr<Metrics> metrics = reader->GetReaderMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t prefetch_errors,
metrics->GetCounter(PrefetchMetrics::ERRORS));
+ ASSERT_GT(prefetch_errors, 0);
// call NextBatch again, will still return error status
auto batch_result2 = reader->NextBatchWithBitmap();
@@ -698,7 +824,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestPrefetchWithEmptyData) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
ASSERT_NOK(reader->GetPreviousBatchFileRowId(0));
ASSERT_OK_AND_ASSIGN(auto array_and_row_ids,
CollectResultAndRowIds(reader.get()));
auto row_ids = array_and_row_ids.second;
@@ -717,7 +843,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestCallNextBatchAfterReadingEof) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
ASSERT_NOK(reader->GetPreviousBatchFileRowId(0));
ASSERT_OK_AND_ASSIGN(auto array_and_row_ids,
CollectResultAndRowIds(reader.get()));
auto row_ids = array_and_row_ids.second;
@@ -725,10 +851,28 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestCallNextBatchAfterReadingEof) {
auto expected_array = std::make_shared<arrow::ChunkedArray>(data_array);
ASSERT_TRUE(array_and_row_ids.first->Equals(expected_array));
+ std::shared_ptr<Metrics> eof_metrics = reader->GetReaderMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t produced_batches,
+
eof_metrics->GetCounter(PrefetchMetrics::PRODUCED_BATCHES));
+ ASSERT_OK_AND_ASSIGN(uint64_t consumed_batches,
+
eof_metrics->GetCounter(PrefetchMetrics::CONSUMED_BATCHES));
+ ASSERT_OK_AND_ASSIGN(uint64_t discarded_batches,
+
eof_metrics->GetCounter(PrefetchMetrics::DISCARDED_BATCHES));
+
// continue to call NextBatch() after reading eof
ASSERT_OK_AND_ASSIGN(auto batch_with_bitmap,
reader->NextBatchWithBitmap());
ASSERT_NOK(reader->GetPreviousBatchFileRowId(0));
ASSERT_TRUE(BatchReader::IsEofBatch(batch_with_bitmap));
+ std::shared_ptr<Metrics> repeated_eof_metrics = reader->GetReaderMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t produced_batches_after,
+
repeated_eof_metrics->GetCounter(PrefetchMetrics::PRODUCED_BATCHES));
+ ASSERT_OK_AND_ASSIGN(uint64_t consumed_batches_after,
+
repeated_eof_metrics->GetCounter(PrefetchMetrics::CONSUMED_BATCHES));
+ ASSERT_OK_AND_ASSIGN(uint64_t discarded_batches_after,
+
repeated_eof_metrics->GetCounter(PrefetchMetrics::DISCARDED_BATCHES));
+ ASSERT_EQ(produced_batches_after, produced_batches);
+ ASSERT_EQ(consumed_batches_after, consumed_batches);
+ ASSERT_EQ(discarded_batches_after, discarded_batches);
}
TEST_F(PrefetchFileBatchReaderImplTest, TestCreateReaderWithoutNextBatch) {
@@ -742,7 +886,7 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestCreateReaderWithoutNextBatch) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
}
TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
@@ -757,7 +901,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
/*prefetch_max_parallel_num=*/0, batch_size, 2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- GetDefaultPool()));
+ /*enable_io_metrics=*/false, GetDefaultPool()));
}
{
ASSERT_NOK(PrefetchFileBatchReaderImpl::Create(
@@ -765,7 +909,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
prefetch_max_parallel_num, /*batch_size=*/-1,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- GetDefaultPool()));
+ /*enable_io_metrics=*/false, GetDefaultPool()));
}
{
ASSERT_NOK(PrefetchFileBatchReaderImpl::Create(
@@ -773,7 +917,8 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false,
/*executor=*/nullptr, /*initialize_read_ranges=*/true,
- /*read_ahead_cache_enabled=*/true, CacheConfig(),
GetDefaultPool()));
+ /*read_ahead_cache_enabled=*/true, CacheConfig(),
/*enable_io_metrics=*/false,
+ GetDefaultPool()));
}
{
ASSERT_NOK(PrefetchFileBatchReaderImpl::Create(
@@ -781,7 +926,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
prefetch_max_parallel_num, batch_size, prefetch_max_parallel_num *
2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- GetDefaultPool()));
+ /*enable_io_metrics=*/false, GetDefaultPool()));
}
{
ASSERT_NOK(PrefetchFileBatchReaderImpl::Create(
@@ -789,7 +934,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
/*fs=*/nullptr, prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true, CacheConfig(),
- GetDefaultPool()));
+ /*enable_io_metrics=*/false, GetDefaultPool()));
}
{
ASSERT_OK_AND_ASSIGN(
@@ -798,7 +943,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestInvalidCase) {
prefetch_max_parallel_num, batch_size,
prefetch_max_parallel_num * 2,
/*enable_adaptive_prefetch_strategy=*/false,
executor_,
/*initialize_read_ranges=*/true,
/*read_ahead_cache_enabled=*/true,
- CacheConfig(), GetDefaultPool()));
+ CacheConfig(), /*enable_io_metrics=*/false,
GetDefaultPool()));
ASSERT_NOK_WITH_MSG(reader->SeekToRow(/*row_number=*/101),
"not support seek to row for prefetch reader");
}
@@ -834,6 +979,18 @@ TEST_P(PrefetchFileBatchReaderImplTest,
TestPrefetchWithPredicatePushdownWithCom
auto expected_array =
std::make_shared<arrow::ChunkedArray>(expected_array_vector);
ASSERT_TRUE(expected_array->Equals(array_and_row_ids.first));
ASSERT_EQ(expected_row_ids, array_and_row_ids.second);
+
+ std::shared_ptr<Metrics> metrics = reader->GetReaderMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t sync_requests,
+
metrics->GetCounter(PrefetchIoMetrics::READ_REQUESTS));
+ ASSERT_OK_AND_ASSIGN(uint64_t async_requests,
+
metrics->GetCounter(PrefetchIoMetrics::ASYNC_REQUESTS));
+ ASSERT_OK_AND_ASSIGN(uint64_t sync_physical_bytes,
+
metrics->GetCounter(PrefetchIoMetrics::READ_PHYSICAL_BYTES));
+ ASSERT_OK_AND_ASSIGN(uint64_t async_physical_bytes,
+
metrics->GetCounter(PrefetchIoMetrics::ASYNC_PHYSICAL_BYTES));
+ ASSERT_GT(sync_requests + async_requests, 0);
+ ASSERT_GT(sync_physical_bytes + async_physical_bytes, 0);
}
/// There are three stripes: [0,30), [30,60), [60,90). Each stripe has 3 row
groups.
@@ -883,14 +1040,14 @@ TEST_F(PrefetchFileBatchReaderImplTest,
TestPrefetchWithBitmap) {
MockFormatReaderBuilder reader_builder(data_array, data_type_, bitmap,
/*read_batch_size=*/100);
int32_t prefetch_max_parallel_num = 3;
- ASSERT_OK_AND_ASSIGN(auto reader,
- PrefetchFileBatchReaderImpl::Create(
- /*data_file_path=*/"", /*data_file_size=*/0,
&reader_builder, mock_fs_,
- prefetch_max_parallel_num,
- /*batch_size=*/100, prefetch_max_parallel_num * 2,
- /*enable_adaptive_prefetch_strategy=*/false,
executor_,
- /*initialize_read_ranges=*/true,
- /*read_ahead_cache_enabled=*/true, CacheConfig(),
GetDefaultPool()));
+ ASSERT_OK_AND_ASSIGN(auto reader, PrefetchFileBatchReaderImpl::Create(
+ /*data_file_path=*/"",
/*data_file_size=*/0,
+ &reader_builder, mock_fs_,
prefetch_max_parallel_num,
+ /*batch_size=*/100,
prefetch_max_parallel_num * 2,
+
/*enable_adaptive_prefetch_strategy=*/false, executor_,
+ /*initialize_read_ranges=*/true,
+ /*read_ahead_cache_enabled=*/true,
CacheConfig(),
+ /*enable_io_metrics=*/false,
GetDefaultPool()));
ASSERT_OK_AND_ASSIGN(auto result_chunk_array,
ReadResultCollector::CollectResult(reader.get()));
ASSERT_OK_AND_ASSIGN(auto data_batch,
ReadResultCollector::GetReadBatch(data_array));
@@ -934,12 +1091,28 @@ TEST_P(PrefetchFileBatchReaderImplTest, TestRowMapping) {
ASSERT_EQ(reader->GetPreviousBatchFileRowId(i).value(), 70 + i);
}
+ std::shared_ptr<Metrics> metrics_before_schema =
reader->GetReaderMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t consumed_before_schema,
+
metrics_before_schema->GetCounter(PrefetchMetrics::CONSUMED_BATCHES));
+ ASSERT_OK_AND_ASSIGN(uint64_t ranges_before_schema,
+
metrics_before_schema->GetCounter(PrefetchMetrics::READ_RANGES_TOTAL));
+
// Set read schema again
std::unique_ptr<ArrowSchema> c_schema = std::make_unique<ArrowSchema>();
ASSERT_TRUE(arrow::ExportSchema(*schema, c_schema.get()).ok());
predicate = PredicateBuilder::Between(/*field_index=*/1,
/*field_name=*/"f1", FieldType::BIGINT,
Literal(30l), Literal(49l));
ASSERT_OK(reader->SetReadSchema(c_schema.get(), predicate, std::nullopt));
+ std::shared_ptr<Metrics> metrics_after_schema = reader->GetReaderMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t consumed_after_schema,
+
metrics_after_schema->GetCounter(PrefetchMetrics::CONSUMED_BATCHES));
+ ASSERT_OK_AND_ASSIGN(uint64_t ranges_after_schema,
+
metrics_after_schema->GetCounter(PrefetchMetrics::READ_RANGES_TOTAL));
+ ASSERT_OK_AND_ASSIGN(double queue_depth_after_schema,
+
metrics_after_schema->GetGauge(PrefetchMetrics::QUEUE_DEPTH));
+ ASSERT_EQ(consumed_after_schema, consumed_before_schema);
+ ASSERT_GT(ranges_after_schema, ranges_before_schema);
+ ASSERT_EQ(queue_depth_after_schema, 0.0);
ASSERT_NOK(reader->GetPreviousBatchFileRowId(0));
ASSERT_OK_AND_ASSIGN(batch,
diff --git a/src/paimon/core/core_options.cpp b/src/paimon/core/core_options.cpp
index a5b32b09..dfa0aa7f 100644
--- a/src/paimon/core/core_options.cpp
+++ b/src/paimon/core/core_options.cpp
@@ -421,6 +421,7 @@ struct CoreOptions::Impl {
bool key_value_sequence_number_enabled = false;
bool file_index_read_enabled = true;
bool enable_adaptive_prefetch_strategy = true;
+ bool prefetch_io_metrics_enabled = false;
bool index_file_in_data_file_dir = false;
bool row_tracking_enabled = false;
bool row_tracking_partition_group_on_commit = true;
@@ -779,6 +780,8 @@ struct CoreOptions::Impl {
}
PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED,
&scan_manifest_entry_lazy_decode_enabled));
+ PAIMON_RETURN_NOT_OK(
+ parser.Parse<bool>(Options::PREFETCH_IO_METRICS_ENABLED,
&prefetch_io_metrics_enabled));
// Parse scan.fallback-branch - fallback branch when partition not
found
PAIMON_RETURN_NOT_OK(parser.Parse(Options::SCAN_FALLBACK_BRANCH,
&scan_fallback_branch));
// Parse branch - branch name, default "main"
@@ -1376,6 +1379,10 @@ bool CoreOptions::EnableAdaptivePrefetchStrategy() const
{
return impl_->enable_adaptive_prefetch_strategy;
}
+bool CoreOptions::PrefetchIoMetricsEnabled() const {
+ return impl_->prefetch_io_metrics_enabled;
+}
+
Result<std::optional<std::string>> CoreOptions::GetFieldAggFunc(
const std::string& field_name) const {
ConfigParser parser(impl_->raw_options);
diff --git a/src/paimon/core/core_options.h b/src/paimon/core/core_options.h
index 85a4a7fd..f4e7964f 100644
--- a/src/paimon/core/core_options.h
+++ b/src/paimon/core/core_options.h
@@ -230,6 +230,8 @@ class PAIMON_EXPORT CoreOptions {
std::string DataFilePrefix() const;
+ bool PrefetchIoMetricsEnabled() const;
+
bool IndexFileInDataFileDir() const;
bool RowTrackingEnabled() const;
diff --git a/src/paimon/core/core_options_test.cpp
b/src/paimon/core/core_options_test.cpp
index c110623d..b0a3274b 100644
--- a/src/paimon/core/core_options_test.cpp
+++ b/src/paimon/core/core_options_test.cpp
@@ -142,6 +142,7 @@ TEST(CoreOptionsTest, TestDefaultValue) {
ASSERT_EQ(std::nullopt, core_options.GetDataFileExternalPaths());
ASSERT_EQ(ExternalPathStrategy::NONE,
core_options.GetExternalPathStrategy());
ASSERT_TRUE(core_options.EnableAdaptivePrefetchStrategy());
+ ASSERT_FALSE(core_options.PrefetchIoMetricsEnabled());
ASSERT_EQ(core_options.DataFilePrefix(), "data-");
ASSERT_FALSE(core_options.IndexFileInDataFileDir());
ASSERT_FALSE(core_options.RowTrackingEnabled());
@@ -223,6 +224,7 @@ TEST(CoreOptionsTest, TestFromMap) {
{Options::SCAN_MODE, "from-snapshot-full"},
{Options::SCAN_MANIFEST_ENTRY_CACHE_MAX_SNAPSHOTS, "7"},
{Options::SCAN_MANIFEST_ENTRY_LAZY_DECODE_ENABLED, "false"},
+ {Options::PREFETCH_IO_METRICS_ENABLED, "true"},
{Options::SNAPSHOT_NUM_RETAINED_MIN, "15"},
{Options::SNAPSHOT_NUM_RETAINED_MAX, "30"},
{Options::SNAPSHOT_EXPIRE_LIMIT, "20"},
@@ -362,6 +364,7 @@ TEST(CoreOptionsTest, TestFromMap) {
ASSERT_EQ(5, core_options.GetScanSnapshotId().value_or(-1));
ASSERT_EQ(7, core_options.GetScanManifestEntryCacheMaxSnapshots());
ASSERT_FALSE(core_options.ScanManifestEntryLazyDecodeEnabled());
+ ASSERT_TRUE(core_options.PrefetchIoMetricsEnabled());
ExpireConfig expire_config = core_options.GetExpireConfig();
ASSERT_EQ(15, expire_config.GetSnapshotRetainMin());
ASSERT_EQ(30, expire_config.GetSnapshotRetainMax());
diff --git
a/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
b/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
index 5451c9d2..6c0d23be 100644
---
a/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
+++
b/src/paimon/core/deletionvectors/apply_deletion_vector_batch_reader_test.cpp
@@ -87,7 +87,8 @@ class ApplyDeletionVectorBatchReaderTest : public
::testing::Test,
prefetch_batch_count, batch_size, prefetch_batch_count
* 2,
/*enable_adaptive_prefetch_strategy=*/false, executor_,
/*initialize_read_ranges=*/true,
- /*read_ahead_cache_enabled=*/true, CacheConfig(),
pool_));
+ /*read_ahead_cache_enabled=*/true, CacheConfig(),
+ /*enable_io_metrics=*/false, pool_));
} else {
file_batch_reader =
std::make_unique<MockFileBatchReader>(data, target_type_,
batch_size);
diff --git a/src/paimon/core/operation/abstract_split_read.cpp
b/src/paimon/core/operation/abstract_split_read.cpp
index 91157f71..5beee34b 100644
--- a/src/paimon/core/operation/abstract_split_read.cpp
+++ b/src/paimon/core/operation/abstract_split_read.cpp
@@ -168,7 +168,7 @@ Result<std::unique_ptr<FileBatchReader>>
AbstractSplitRead::CreateFileBatchReade
context_->GetPrefetchBatchCount(),
options_.EnableAdaptivePrefetchStrategy(),
executor_,
/*initialize_read_ranges=*/false,
context_->ReadAheadCacheEnabled(),
- context_->GetCacheConfig(), pool_));
+ context_->GetCacheConfig(),
options_.PrefetchIoMetricsEnabled(), pool_));
return
std::make_unique<DelegatingPrefetchReader>(std::move(prefetch_reader));
} else {
PAIMON_ASSIGN_OR_RAISE(
diff --git a/src/paimon/core/operation/append_only_file_store_scan_test.cpp
b/src/paimon/core/operation/append_only_file_store_scan_test.cpp
index e1fb5a43..d72dcaf0 100644
--- a/src/paimon/core/operation/append_only_file_store_scan_test.cpp
+++ b/src/paimon/core/operation/append_only_file_store_scan_test.cpp
@@ -31,7 +31,6 @@
#include "paimon/common/io/cache/lru_cache.h"
#include "paimon/core/manifest/manifest_entry.h"
#include "paimon/core/manifest/partition_entry.h"
-#include "paimon/core/operation/metrics/scan_metrics.h"
#include "paimon/core/schema/schema_manager.h"
#include "paimon/core/schema/table_schema.h"
#include "paimon/core/stats/simple_stats_evolution.h"
@@ -45,6 +44,7 @@
#include "paimon/predicate/predicate_builder.h"
#include "paimon/scan_context.h"
#include "paimon/status.h"
+#include "paimon/table/source/scan_metrics.h"
#include "paimon/table/source/table_scan.h"
#include "paimon/testing/utils/testharness.h"
#include "paimon/testing/utils/timezone_guard.h"
@@ -173,7 +173,20 @@ TEST(AppendOnlyFileStoreScanTest, TestScanDurationMetric) {
metrics->GetCounter(ScanMetrics::LAST_SCAN_DURATION));
ASSERT_OK_AND_ASSIGN(HistogramStats stats,
metrics->GetHistogramStats(ScanMetrics::SCAN_DURATION));
+ ASSERT_OK_AND_ASSIGN(uint64_t manifest_read_duration,
+
metrics->GetCounter(ScanMetrics::LAST_MANIFEST_READ_DURATION));
+ ASSERT_OK_AND_ASSIGN(HistogramStats manifest_stats,
+
metrics->GetHistogramStats(ScanMetrics::MANIFEST_READ_DURATION));
+ ASSERT_OK_AND_ASSIGN(uint64_t scanned_rows,
+
metrics->GetCounter(ScanMetrics::LAST_LAZY_DECODE_SCANNED_ROWS));
+ ASSERT_OK_AND_ASSIGN(uint64_t materialized_rows,
+
metrics->GetCounter(ScanMetrics::LAST_LAZY_DECODE_MATERIALIZED_ROWS));
ASSERT_EQ(stats.count, kPlanCount);
+ ASSERT_EQ(manifest_stats.count, kPlanCount);
+ ASSERT_LE(manifest_stats.min, static_cast<double>(manifest_read_duration));
+ ASSERT_LE(static_cast<double>(manifest_read_duration), manifest_stats.max);
+ ASSERT_GT(scanned_rows, 0);
+ ASSERT_GE(scanned_rows, materialized_rows);
ASSERT_LE(stats.min, stats.max);
ASSERT_LE(stats.min, static_cast<double>(last_scan_duration));
ASSERT_LE(static_cast<double>(last_scan_duration), stats.max);
@@ -268,12 +281,40 @@ TEST(AppendOnlyFileStoreScanTest,
TestSnapshotLiveManifestCachePath) {
scan_first->WithSnapshot(snapshot_5);
ASSERT_OK_AND_ASSIGN(auto plan_first, scan_first->CreatePlan());
std::vector<std::string> first_file_names =
SortedFileNames(plan_first->Files());
+ std::shared_ptr<Metrics> first_metrics = scan_first->GetScanMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t first_cache_enabled,
+
first_metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_ENABLED));
+ ASSERT_OK_AND_ASSIGN(uint64_t first_cache_hit,
+
first_metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+ ASSERT_OK_AND_ASSIGN(uint64_t first_cache_misses,
+
first_metrics->GetCounter(ScanMetrics::SNAPSHOT_CACHE_MISSES));
+
ASSERT_OK(first_metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_LOAD_DURATION));
+
ASSERT_OK(first_metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_STORE_DURATION));
+
ASSERT_OK(first_metrics->GetHistogramStats(ScanMetrics::SNAPSHOT_CACHE_LOAD_DURATION));
+
ASSERT_OK(first_metrics->GetHistogramStats(ScanMetrics::SNAPSHOT_CACHE_STORE_DURATION));
+ ASSERT_EQ(first_cache_enabled, 1);
+ ASSERT_EQ(first_cache_hit, 0);
+ ASSERT_EQ(first_cache_misses, 1);
// Second scan on the same snapshot should read the same bucket live
entries from cache.
auto scan_second = BuildScan(table_path, cache, /*bucket=*/0);
scan_second->WithSnapshot(snapshot_5);
ASSERT_OK_AND_ASSIGN(auto plan_second, scan_second->CreatePlan());
ASSERT_EQ(first_file_names, SortedFileNames(plan_second->Files()));
+ std::shared_ptr<Metrics> second_metrics = scan_second->GetScanMetrics();
+ ASSERT_OK_AND_ASSIGN(uint64_t second_cache_hit,
+
second_metrics->GetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_HIT));
+ ASSERT_OK_AND_ASSIGN(uint64_t second_cache_hits,
+
second_metrics->GetCounter(ScanMetrics::SNAPSHOT_CACHE_HITS));
+ ASSERT_OK_AND_ASSIGN(uint64_t scanned_rows,
+
second_metrics->GetCounter(ScanMetrics::LAST_LAZY_DECODE_SCANNED_ROWS));
+ ASSERT_OK_AND_ASSIGN(
+ uint64_t materialized_rows,
+
second_metrics->GetCounter(ScanMetrics::LAST_LAZY_DECODE_MATERIALIZED_ROWS));
+ ASSERT_EQ(second_cache_hit, 1);
+ ASSERT_EQ(second_cache_hits, 1);
+ ASSERT_GE(scanned_rows, materialized_rows);
+
ASSERT_OK(second_metrics->GetHistogramStats(ScanMetrics::SNAPSHOT_CACHE_LOAD_DURATION));
}
TEST(AppendOnlyFileStoreScanTest, TestSnapshotLiveManifestCacheRebuildOnMiss) {
diff --git a/src/paimon/core/operation/file_store_scan.cpp
b/src/paimon/core/operation/file_store_scan.cpp
index f21b0bb7..467ba6b2 100644
--- a/src/paimon/core/operation/file_store_scan.cpp
+++ b/src/paimon/core/operation/file_store_scan.cpp
@@ -43,7 +43,6 @@
#include "paimon/core/manifest/manifest_file_meta.h"
#include "paimon/core/manifest/manifest_list.h"
#include "paimon/core/manifest/snapshot_live_manifest_entries.h"
-#include "paimon/core/operation/metrics/scan_metrics.h"
#include "paimon/core/partition/partition_info.h"
#include "paimon/core/stats/simple_stats.h"
#include "paimon/core/stats/simple_stats_evolution.h"
@@ -58,6 +57,7 @@
#include "paimon/predicate/predicate_builder.h"
#include "paimon/predicate/predicate_utils.h"
#include "paimon/scan_context.h"
+#include "paimon/table/source/scan_metrics.h"
namespace paimon {
enum class FieldType;
@@ -136,6 +136,9 @@ Result<std::vector<PartitionEntry>>
FileStoreScan::ReadPartitionEntries() const
Result<std::shared_ptr<FileStoreScan::RawPlan>> FileStoreScan::CreatePlan()
const {
Duration duration;
+ Duration manifest_read_duration;
+ metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_LOAD_DURATION, 0);
+ metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_STORE_DURATION, 0);
std::optional<Snapshot> snapshot;
std::vector<ManifestFileMeta> all_manifest_file_metas;
std::vector<ManifestFileMeta> filtered_manifest_file_metas;
@@ -148,9 +151,13 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>>
FileStoreScan::CreatePlan() cons
core_options_.GetScanManifestEntryCacheMaxSnapshots() > 0 &&
core_options_.GetCache() != nullptr && !table_path_.empty() &&
!row_range_index_.has_value() && bucket_filter_.has_value();
+ uint64_t lazy_decode_scanned_rows = 0;
+ bool snapshot_cache_hit = false;
if (use_snapshot_live_manifest_cache) {
- PAIMON_RETURN_NOT_OK(ReadManifestEntriesWithCache(
- snapshot.value(), all_manifest_file_metas, bucket_filter_.value(),
&manifest_entries));
+ PAIMON_RETURN_NOT_OK(ReadManifestEntriesWithCache(snapshot.value(),
all_manifest_file_metas,
+
bucket_filter_.value(), &manifest_entries,
+
&snapshot_cache_hit));
+ lazy_decode_scanned_rows = manifest_entries.size();
std::vector<ManifestEntry> filtered_entries;
filtered_entries.reserve(manifest_entries.size());
for (auto& entry : manifest_entries) {
@@ -161,8 +168,15 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>>
FileStoreScan::CreatePlan() cons
}
manifest_entries = std::move(filtered_entries);
} else {
+ lazy_decode_scanned_rows = std::accumulate(
+ filtered_manifest_file_metas.begin(),
filtered_manifest_file_metas.end(), uint64_t{0},
+ [](uint64_t sum, const ManifestFileMeta& meta) {
+ return sum + static_cast<uint64_t>(meta.NumAddedFiles() +
meta.NumDeletedFiles());
+ });
PAIMON_RETURN_NOT_OK(ReadManifestEntries(filtered_manifest_file_metas,
&manifest_entries));
}
+ const uint64_t lazy_decode_materialized_rows = manifest_entries.size();
+ const uint64_t manifest_read_duration_ms = manifest_read_duration.Get();
PAIMON_ASSIGN_OR_RAISE(manifest_entries,
PostFilterManifestEntries(std::move(manifest_entries)));
@@ -208,6 +222,22 @@ Result<std::shared_ptr<FileStoreScan::RawPlan>>
FileStoreScan::CreatePlan() cons
ScanMetrics::LAST_SCAN_SKIPPED_TABLE_FILES,
std::max(int64_t{0}, all_data_files -
static_cast<int64_t>(manifest_entries.size())));
metrics_->SetCounter(ScanMetrics::LAST_SCAN_RESULTED_TABLE_FILES,
manifest_entries.size());
+ metrics_->SetCounter(ScanMetrics::LAST_MANIFEST_READ_DURATION,
manifest_read_duration_ms);
+ metrics_->ObserveHistogram(ScanMetrics::MANIFEST_READ_DURATION,
+ static_cast<double>(manifest_read_duration_ms));
+ metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_ENABLED,
+ use_snapshot_live_manifest_cache ? 1 : 0);
+ metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_HIT,
snapshot_cache_hit ? 1 : 0);
+ Result<uint64_t> cache_hits =
metrics_->GetCounter(ScanMetrics::SNAPSHOT_CACHE_HITS);
+ Result<uint64_t> cache_misses =
metrics_->GetCounter(ScanMetrics::SNAPSHOT_CACHE_MISSES);
+ metrics_->SetCounter(ScanMetrics::SNAPSHOT_CACHE_HITS,
+ (cache_hits.ok() ? cache_hits.value() : 0) +
(snapshot_cache_hit ? 1 : 0));
+ metrics_->SetCounter(ScanMetrics::SNAPSHOT_CACHE_MISSES,
+ (cache_misses.ok() ? cache_misses.value() : 0) +
+ (use_snapshot_live_manifest_cache &&
!snapshot_cache_hit ? 1 : 0));
+ metrics_->SetCounter(ScanMetrics::LAST_LAZY_DECODE_SCANNED_ROWS,
lazy_decode_scanned_rows);
+ metrics_->SetCounter(ScanMetrics::LAST_LAZY_DECODE_MATERIALIZED_ROWS,
+ lazy_decode_materialized_rows);
return std::make_shared<FileStoreScan::RawPlan>(scan_mode_, snapshot,
std::move(manifest_entries));
}
@@ -298,15 +328,22 @@ Status FileStoreScan::ReadManifestEntries(const
std::vector<ManifestFileMeta>& m
// snapshot's data manifests.
Status FileStoreScan::ReadManifestEntriesWithCache(
const Snapshot& snapshot, const std::vector<ManifestFileMeta>&
all_manifest_metas,
- int32_t bucket, std::vector<ManifestEntry>* manifest_entries) const {
+ int32_t bucket, std::vector<ManifestEntry>* manifest_entries, bool*
cache_hit) const {
+ Duration cache_load_duration;
PAIMON_ASSIGN_OR_RAISE(SnapshotLiveManifestEntries cached_entries,
LoadSnapshotLiveManifestEntries(bucket));
+ const uint64_t cache_load_duration_ms = cache_load_duration.Get();
+ metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_LOAD_DURATION,
cache_load_duration_ms);
+ metrics_->ObserveHistogram(ScanMetrics::SNAPSHOT_CACHE_LOAD_DURATION,
+ static_cast<double>(cache_load_duration_ms));
std::optional<SnapshotLiveManifestEntries::Entry> cached =
cached_entries.LatestBeforeOrEqual(snapshot.Id());
if (cached && cached->snapshot_id == snapshot.Id()) {
+ *cache_hit = true;
*manifest_entries = *cached->entries;
return Status::OK();
}
+ *cache_hit = false;
// Rebuild the target snapshot bucket from all manifests and write the
live entries back to the
// cache.
@@ -320,7 +357,12 @@ Status FileStoreScan::ReadManifestEntriesWithCache(
ReadAndMergeBucketFileEntries(bucket_manifest_metas, bucket,
manifest_entries));
std::vector<ManifestEntry> cache_entries = *manifest_entries;
cached_entries.Put(snapshot.Id(), std::move(cache_entries));
+ Duration cache_store_duration;
PAIMON_RETURN_NOT_OK(StoreSnapshotLiveManifestEntries(bucket,
cached_entries));
+ const uint64_t cache_store_duration_ms = cache_store_duration.Get();
+ metrics_->SetCounter(ScanMetrics::LAST_SNAPSHOT_CACHE_STORE_DURATION,
cache_store_duration_ms);
+ metrics_->ObserveHistogram(ScanMetrics::SNAPSHOT_CACHE_STORE_DURATION,
+ static_cast<double>(cache_store_duration_ms));
return Status::OK();
}
diff --git a/src/paimon/core/operation/file_store_scan.h
b/src/paimon/core/operation/file_store_scan.h
index 155f36e7..ff53e0c9 100644
--- a/src/paimon/core/operation/file_store_scan.h
+++ b/src/paimon/core/operation/file_store_scan.h
@@ -161,7 +161,9 @@ class FileStoreScan {
}
std::shared_ptr<Metrics> GetScanMetrics() const {
- return metrics_;
+ auto snapshot = std::make_shared<MetricsImpl>();
+ snapshot->Overwrite(metrics_);
+ return snapshot;
}
static Result<std::shared_ptr<PredicateFilter>> CreatePartitionPredicate(
@@ -260,7 +262,8 @@ class FileStoreScan {
Status ReadManifestEntriesWithCache(const Snapshot& snapshot,
const std::vector<ManifestFileMeta>&
bucket_manifest_metas,
int32_t bucket,
- std::vector<ManifestEntry>*
manifest_entries) const;
+ std::vector<ManifestEntry>*
manifest_entries,
+ bool* cache_hit) const;
std::shared_ptr<CacheKey> SnapshotLiveManifestEntriesCacheKey(int32_t
bucket) const;
Result<SnapshotLiveManifestEntries>
LoadSnapshotLiveManifestEntries(int32_t bucket) const;
Status StoreSnapshotLiveManifestEntries(int32_t bucket,
diff --git a/src/paimon/core/operation/key_value_file_store_scan_test.cpp
b/src/paimon/core/operation/key_value_file_store_scan_test.cpp
index 703d7e49..f9fd2ea2 100644
--- a/src/paimon/core/operation/key_value_file_store_scan_test.cpp
+++ b/src/paimon/core/operation/key_value_file_store_scan_test.cpp
@@ -33,7 +33,6 @@
#include "paimon/core/manifest/file_source.h"
#include "paimon/core/manifest/manifest_file.h"
#include "paimon/core/manifest/manifest_list.h"
-#include "paimon/core/operation/metrics/scan_metrics.h"
#include "paimon/core/schema/schema_manager.h"
#include "paimon/core/schema/table_schema.h"
#include "paimon/core/snapshot.h"
@@ -50,6 +49,7 @@
#include "paimon/predicate/literal.h"
#include "paimon/predicate/predicate_builder.h"
#include "paimon/scan_context.h"
+#include "paimon/table/source/scan_metrics.h"
#include "paimon/testing/utils/binary_row_generator.h"
#include "paimon/testing/utils/testharness.h"
diff --git a/src/paimon/core/operation/metrics/scan_metrics.h
b/src/paimon/core/operation/metrics/scan_metrics.h
deleted file mode 100644
index 483ab7af..00000000
--- a/src/paimon/core/operation/metrics/scan_metrics.h
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-#pragma once
-
-namespace paimon {
-
-/// Metrics to measure scan operation.
-class ScanMetrics {
- public:
- static constexpr char LAST_SCAN_DURATION[] = "lastScanDuration";
- // Histogram metric for scan plan duration (milliseconds).
- static constexpr char SCAN_DURATION[] = "scanDuration";
- static constexpr char LAST_SCANNED_SNAPSHOT_ID[] = "lastScannedSnapshotId";
- static constexpr char LAST_SCANNED_MANIFESTS[] = "lastScannedManifests";
- static constexpr char LAST_SCAN_SKIPPED_TABLE_FILES[] =
"lastScanSkippedTableFiles";
- static constexpr char LAST_SCAN_RESULTED_TABLE_FILES[] =
"lastScanResultedTableFiles";
-};
-
-} // namespace paimon
diff --git a/src/paimon/core/table/source/abstract_table_scan.h
b/src/paimon/core/table/source/abstract_table_scan.h
index 363fe0dc..f8dd15aa 100644
--- a/src/paimon/core/table/source/abstract_table_scan.h
+++ b/src/paimon/core/table/source/abstract_table_scan.h
@@ -39,6 +39,10 @@ class AbstractTableScan : public TableScan {
const std::shared_ptr<SnapshotReader>& snapshot_reader)
: core_options_(core_options), snapshot_reader_(snapshot_reader) {}
+ std::shared_ptr<Metrics> GetMetrics() const override {
+ return snapshot_reader_->GetMetrics();
+ }
+
protected:
Result<std::shared_ptr<StartingScanner>> CreateStartingScanner(bool
is_streaming) const {
const auto& snapshot_manager = snapshot_reader_->GetSnapshotManager();
diff --git a/src/paimon/core/table/source/realtime_table_scan.h
b/src/paimon/core/table/source/realtime_table_scan.h
index 7d036d42..2b635faf 100644
--- a/src/paimon/core/table/source/realtime_table_scan.h
+++ b/src/paimon/core/table/source/realtime_table_scan.h
@@ -47,6 +47,10 @@ class RealtimeTableScan : public TableScan {
Result<std::shared_ptr<Plan>> CreatePlan() override;
+ std::shared_ptr<Metrics> GetMetrics() const override {
+ return disk_scan_->GetMetrics();
+ }
+
private:
using MemoryViewMap = std::map<RealtimePartitionBucket,
RealtimePartitionBucketView>;
diff --git a/src/paimon/core/table/source/snapshot/snapshot_reader.h
b/src/paimon/core/table/source/snapshot/snapshot_reader.h
index c425fd86..35134d22 100644
--- a/src/paimon/core/table/source/snapshot/snapshot_reader.h
+++ b/src/paimon/core/table/source/snapshot/snapshot_reader.h
@@ -108,6 +108,10 @@ class SnapshotReader {
return scan_->GetPartitionPredicate();
}
+ std::shared_ptr<Metrics> GetMetrics() const {
+ return scan_->GetScanMetrics();
+ }
+
/// Get splits from `FileKind::ADD` files.
Result<std::shared_ptr<Plan>> Read() const;
diff --git a/src/paimon/core/table/source/table_scan.cpp
b/src/paimon/core/table/source/table_scan.cpp
index 2dda955a..8111a94e 100644
--- a/src/paimon/core/table/source/table_scan.cpp
+++ b/src/paimon/core/table/source/table_scan.cpp
@@ -27,6 +27,7 @@
#include <vector>
#include "fmt/format.h"
+#include "paimon/common/metrics/metrics_impl.h"
#include "paimon/common/predicate/predicate_validator.h"
#include "paimon/common/types/data_field.h"
#include "paimon/common/utils/fields_comparator.h"
@@ -187,6 +188,10 @@ Result<std::unique_ptr<TableScan>> NewDataTableScan(const
std::shared_ptr<ScanCo
} // namespace
+std::shared_ptr<Metrics> TableScan::GetMetrics() const {
+ return std::make_shared<MetricsImpl>();
+}
+
Result<std::unique_ptr<TableScan>>
TableScan::Create(std::unique_ptr<ScanContext> context) {
if (context == nullptr) {
return Status::Invalid("scan context is null pointer");
diff --git a/src/paimon/core/table/source/table_scan_test.cpp
b/src/paimon/core/table/source/table_scan_test.cpp
index 358dfd46..dbc5b992 100644
--- a/src/paimon/core/table/source/table_scan_test.cpp
+++ b/src/paimon/core/table/source/table_scan_test.cpp
@@ -26,11 +26,36 @@
#include "gtest/gtest.h"
#include "paimon/defs.h"
+#include "paimon/metrics.h"
#include "paimon/scan_context.h"
#include "paimon/status.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
+namespace {
+
+class DefaultMetricsTableScan : public TableScan {
+ public:
+ Result<std::shared_ptr<Plan>> CreatePlan() override {
+ return Status::NotImplemented("not implemented");
+ }
+};
+
+} // namespace
+
+TEST(TableScanTest, TestDefaultMetricsSnapshot) {
+ DefaultMetricsTableScan table_scan;
+ std::shared_ptr<Metrics> metrics = table_scan.GetMetrics();
+ ASSERT_TRUE(metrics);
+ metrics->SetCounter("external", 1);
+
+ std::shared_ptr<Metrics> second_metrics = table_scan.GetMetrics();
+ ASSERT_TRUE(second_metrics);
+ Result<uint64_t> external_counter = second_metrics->GetCounter("external");
+ ASSERT_FALSE(external_counter.ok());
+ ASSERT_EQ(external_counter.status().code(), StatusCode::KeyError);
+}
+
TEST(TableScanTest, TestNoSnapshot) {
std::string path = paimon::test::GetDataDir() +
"/orc/append_table_with_nested_type.db/append_table_with_nested_type/";
@@ -61,6 +86,27 @@ TEST(TableScanTest, TestPkSchemaEvolutionScan) {
ASSERT_OK_AND_ASSIGN(auto plan, table_scan->CreatePlan());
ASSERT_TRUE(plan->SnapshotId());
ASSERT_FALSE(plan->Splits().empty());
+
+ std::shared_ptr<Metrics> metrics = table_scan->GetMetrics();
+ ASSERT_TRUE(metrics);
+ ASSERT_OK_AND_ASSIGN(uint64_t scanned_snapshot_id,
+
metrics->GetCounter(ScanMetrics::LAST_SCANNED_SNAPSHOT_ID));
+ ASSERT_EQ(scanned_snapshot_id,
static_cast<uint64_t>(plan->SnapshotId().value()));
+ ASSERT_OK_AND_ASSIGN(uint64_t resulted_table_files,
+
metrics->GetCounter(ScanMetrics::LAST_SCAN_RESULTED_TABLE_FILES));
+ ASSERT_GT(resulted_table_files, 0);
+ ASSERT_OK(metrics->GetCounter(ScanMetrics::LAST_MANIFEST_READ_DURATION));
+ ASSERT_OK(metrics->GetHistogramStats(ScanMetrics::MANIFEST_READ_DURATION));
+ ASSERT_OK_AND_ASSIGN(uint64_t lazy_decode_scanned_rows,
+
metrics->GetCounter(ScanMetrics::LAST_LAZY_DECODE_SCANNED_ROWS));
+ ASSERT_OK_AND_ASSIGN(uint64_t lazy_decode_materialized_rows,
+
metrics->GetCounter(ScanMetrics::LAST_LAZY_DECODE_MATERIALIZED_ROWS));
+ ASSERT_GE(lazy_decode_scanned_rows, lazy_decode_materialized_rows);
+
+ metrics->SetCounter(ScanMetrics::LAST_SCANNED_SNAPSHOT_ID, 0);
+ ASSERT_OK_AND_ASSIGN(uint64_t internal_snapshot_id,
table_scan->GetMetrics()->GetCounter(
+
ScanMetrics::LAST_SCANNED_SNAPSHOT_ID));
+ ASSERT_EQ(internal_snapshot_id,
static_cast<uint64_t>(plan->SnapshotId().value()));
}
TEST(TableScanTest, TestReadOptimizedPrimaryKeyStreamingScanUnsupported) {
diff --git a/src/paimon/core/table/system/system_table_test.cpp
b/src/paimon/core/table/system/system_table_test.cpp
index 0c67ae3a..7e5847db 100644
--- a/src/paimon/core/table/system/system_table_test.cpp
+++ b/src/paimon/core/table/system/system_table_test.cpp
@@ -33,10 +33,12 @@
#include "paimon/core/table/system/audit_log_system_table.h"
#include "paimon/core/table/system/binlog_system_table.h"
#include "paimon/core/table/system/read_optimized_system_table.h"
+#include "paimon/core/table/system/system_table_scan.h"
#include "paimon/defs.h"
#include "paimon/fs/file_system.h"
#include "paimon/fs/file_system_factory.h"
#include "paimon/memory/memory_pool.h"
+#include "paimon/metrics.h"
#include "paimon/reader/batch_reader.h"
#include "paimon/result.h"
#include "paimon/status.h"
@@ -199,4 +201,15 @@ TEST(SystemTableTest,
TestGlobalSystemTableWithoutCatalogReturnsNotImplemented)
"global system table requires catalog context:
tables");
}
+TEST(SystemTableTest, TestScanMetricsAreSnapshots) {
+ SystemTableScan scan("/tmp/table");
+ std::shared_ptr<Metrics> metrics = scan.GetMetrics();
+ ASSERT_TRUE(metrics);
+ metrics->SetCounter("external", 1);
+
+ std::shared_ptr<Metrics> second_metrics = scan.GetMetrics();
+ ASSERT_TRUE(second_metrics);
+ ASSERT_NOK_WITH_MSG(second_metrics->GetCounter("external"), "metric
'external' not found");
+}
+
} // namespace paimon::test
