feat(metrics): expose scan and prefetch metrics

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

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"

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[];

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.

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

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

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";

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);

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 {

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;