perf(metrics): reduce prefetch I/O metrics overhead

Author: gripleaf

docs/source/user_guide/metrics.rst

@@ -87,8 +87,12 @@ 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 and histograms
-accumulate for the reader lifetime and are retained across ``SetReadSchema()`` and cache reset.
+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.
 
@@ -100,11 +104,13 @@ These metrics are C++-only and have no counterparts in Java Paimon.
    "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-us", "histogram", "microseconds", "Synchronous read latency"
+   "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-us", "histogram", "microseconds", "Asynchronous callback latency"
+   "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

@@ -204,6 +204,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

@@ -56,14 +56,16 @@ class PAIMON_EXPORT PrefetchIoMetrics {
     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_US[] = "io.read.latency-us";
+    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_US[] = "io.async.latency-us";
+    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,

src/paimon/common/defs.cpp

@@ -61,6 +61,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);