feat: improve scan IO profile metrics (#18975)

* feat(profile): split scan io metrics and show in explain

* revert: only show non-zero scan io metrics

* fix: always display scan IO metrics in EXPLAIN ANALYZE for consistent output

- Add bytes scanned from local cache to test expectations
- Make scan IO metrics (remote/local/memory) always display even when zero
  to ensure consistent output format for sqllogic tests

* fix: only display scan IO metrics when any of them is non-zero

This ensures:
- Fuse table scans show all three scan IO metrics together (remote/local/memory)
- System tables (like numbers) don't show these metrics since they don't use Fuse storage

* fix: use ProfileStatisticsName enum for scan IO metrics matching

Use enum comparison instead of string comparison for more reliable matching

* fix: display all scan IO metrics together when any is non-zero

The fix was applied to the correct file: physical_format.rs in databend-query.
The previous fix in sql/executor/format.rs was ineffective because that file
is not compiled (not declared in mod.rs).

When any of the three scan IO metrics (remote/local cache/memory cache) is
non-zero, all three will be displayed. This ensures consistent output in
EXPLAIN ANALYZE for Fuse table scans.

Author: BohuTANG

src/common/base/src/runtime/profile/profiles.rs

@@ -35,8 +35,11 @@ pub enum ProfileStatisticsName {
     OutputRows,
     OutputBytes,
     ScanBytes,
-    ScanCacheBytes,
     ScanPartitions,
+    ScanBytesFromRemote,
+    ScanBytesFromLocalDisk,
+    ScanBytesFromDataCache,
+    ScanBytesFromMemory,
 
     RemoteSpillWriteCount,
     RemoteSpillWriteBytes,
@@ -190,20 +193,41 @@ pub fn get_statistics_desc() -> Arc<BTreeMap<ProfileStatisticsName, ProfileDesc>
                 unit: StatisticsUnit::Bytes,
                 plain_statistics: true,
             }),
-            (ProfileStatisticsName::ScanCacheBytes, ProfileDesc {
-                display_name: "bytes scanned from cache",
-                desc: "The bytes scanned from cache of query",
-                index: ProfileStatisticsName::ScanCacheBytes as usize,
-                unit: StatisticsUnit::Bytes,
-                plain_statistics: true,
-            }),
             (ProfileStatisticsName::ScanPartitions, ProfileDesc {
                 display_name: "partitions scanned",
                 desc: "The partitions scanned of query",
                 index: ProfileStatisticsName::ScanPartitions as usize,
                 unit: StatisticsUnit::Count,
                 plain_statistics: true,
             }),
+            (ProfileStatisticsName::ScanBytesFromRemote, ProfileDesc {
+                display_name: "read from remote",
+                desc: "The bytes read from remote storage",
+                index: ProfileStatisticsName::ScanBytesFromRemote as usize,
+                unit: StatisticsUnit::Bytes,
+                plain_statistics: true,
+            }),
+            (ProfileStatisticsName::ScanBytesFromLocalDisk, ProfileDesc {
+                display_name: "read from local disk",
+                desc: "The bytes read from local disk cache",
+                index: ProfileStatisticsName::ScanBytesFromLocalDisk as usize,
+                unit: StatisticsUnit::Bytes,
+                plain_statistics: true,
+            }),
+            (ProfileStatisticsName::ScanBytesFromDataCache, ProfileDesc {
+                display_name: "read from data cache",
+                desc: "The bytes read from data cache memory",
+                index: ProfileStatisticsName::ScanBytesFromDataCache as usize,
+                unit: StatisticsUnit::Bytes,
+                plain_statistics: true,
+            }),
+            (ProfileStatisticsName::ScanBytesFromMemory, ProfileDesc {
+                display_name: "read from memory cache",
+                desc: "The bytes read from memory cache",
+                index: ProfileStatisticsName::ScanBytesFromMemory as usize,
+                unit: StatisticsUnit::Bytes,
+                plain_statistics: true,
+            }),
             (ProfileStatisticsName::RemoteSpillWriteCount, ProfileDesc {
                 display_name: "numbers remote spilled by write",
                 desc: "The number of remote spilled by write",

src/query/sql/src/executor/format.rs

@@ -138,7 +138,10 @@ impl PhysicalPlan {
                 };
 
                 Ok(FormatTreeNode::with_children(
-                    format!("HashJoin: {} estimated_rows: {}", plan.join_type, estimated_rows),
+                    format!(
+                        "HashJoin: {} estimated_rows: {}",
+                        plan.join_type, estimated_rows
+                    ),
                     children,
                 ))
             }

src/query/storages/common/cache/src/providers/disk_cache_builder.rs

@@ -118,7 +118,8 @@ impl CacheAccessor for DiskCacheAccessor {
         metrics_inc_cache_access_count(1, &self.name);
         let k = k.as_ref();
         if let Some(item) = self.lru_disk_cache.get(k) {
-            Profile::record_usize_profile(ProfileStatisticsName::ScanCacheBytes, item.len());
+            let size = item.len();
+            Profile::record_usize_profile(ProfileStatisticsName::ScanBytesFromLocalDisk, size);
             metrics_inc_cache_hit_count(1, &self.name);
             Some(item)
         } else {

src/query/storages/common/cache/src/providers/hybrid_cache.rs

@@ -15,6 +15,8 @@
 use std::sync::Arc;
 
 use bytes::Bytes;
+use databend_common_base::runtime::profile::Profile;
+use databend_common_base::runtime::profile::ProfileStatisticsName;
 use databend_common_exception::ErrorCode;
 use databend_common_metrics::cache::metrics_inc_cache_miss_bytes;
 use log::warn;
@@ -152,10 +154,14 @@ where
 
     fn get<Q: AsRef<str>>(&self, k: Q) -> Option<Arc<Self::V>> {
         if let Some(item) = self.memory_cache.get(k.as_ref()) {
+            // Record memory cache hit
+            // Note: The actual size recording happens in memory_cache.get()
             // try putting it bach to on-disk cache if necessary
             self.insert_to_disk_cache_if_necessary(k.as_ref(), item.as_ref());
             Some(item)
         } else if let Some(bytes) = self.disk_cache.get(k.as_ref()) {
+            // Record disk cache hit
+            // Note: The actual size recording happens in disk_cache.get()
             let bytes = bytes.as_ref().clone();
             match bytes.try_into() {
                 Ok(v) => Some(self.memory_cache.insert(k.as_ref().to_owned(), v)),
@@ -169,21 +175,45 @@ where
                 }
             }
         } else {
-            // Cache Miss
+            // Cache Miss - will need to read from remote
             None
         }
     }
 
     fn get_sized<Q: AsRef<str>>(&self, k: Q, len: u64) -> Option<Arc<Self::V>> {
-        let Some(cached_value) = self.get(k) else {
-            // Both in-memory and on-disk cache are missed, record it
-            let in_memory_cache_name = self.memory_cache.name();
-            let on_disk_cache_name = self.disk_cache.name();
-            metrics_inc_cache_miss_bytes(len, in_memory_cache_name);
-            metrics_inc_cache_miss_bytes(len, on_disk_cache_name);
-            return None;
-        };
-        Some(cached_value)
+        // First, check in-memory cache
+        if let Some(item) = self.memory_cache.get(k.as_ref()) {
+            // Record bytes read from data cache memory
+            Profile::record_usize_profile(
+                ProfileStatisticsName::ScanBytesFromDataCache,
+                len as usize,
+            );
+            // try putting it back to on-disk cache if necessary
+            self.insert_to_disk_cache_if_necessary(k.as_ref(), item.as_ref());
+            return Some(item);
+        }
+
+        // Then, check on-disk cache (ScanBytesFromLocalDisk is recorded inside disk_cache.get)
+        if let Some(bytes) = self.disk_cache.get(k.as_ref()) {
+            let bytes = bytes.as_ref().clone();
+            match bytes.try_into() {
+                Ok(v) => return Some(self.memory_cache.insert(k.as_ref().to_owned(), v)),
+                Err(e) => {
+                    let key = k.as_ref();
+                    // Disk cache crc is correct, but failed to deserialize.
+                    // Likely the serialization format has been changed, evict it.
+                    warn!("failed to decode cache value, key {key}, {}", e);
+                    self.disk_cache.evict(key);
+                }
+            }
+        }
+
+        // Both in-memory and on-disk cache are missed, record it
+        let in_memory_cache_name = self.memory_cache.name();
+        let on_disk_cache_name = self.disk_cache.name();
+        metrics_inc_cache_miss_bytes(len, in_memory_cache_name);
+        metrics_inc_cache_miss_bytes(len, on_disk_cache_name);
+        None
     }
 
     fn insert(&self, key: String, value: Self::V) -> Arc<Self::V> {

src/query/storages/common/cache/src/providers/memory_cache.rs

@@ -97,8 +97,9 @@ mod impls {
                 let mut guard = self.inner.write();
                 guard
                     .get(k.as_ref())
-                    .map(|cache_value: &CacheValue<V>| cache_value.get_inner())
+                    .map(|cache_value| cache_value.get_inner())
             };
+
             if v.is_none() {
                 metrics_inc_cache_miss_count(1, &self.name);
             } else {

src/query/storages/fuse/src/io/read/block/block_reader_merge_io_async.rs

@@ -15,6 +15,8 @@
 use std::collections::HashMap;
 use std::collections::HashSet;
 
+use databend_common_base::runtime::profile::Profile;
+use databend_common_base::runtime::profile::ProfileStatisticsName;
 use databend_common_exception::Result;
 use databend_common_expression::ColumnId;
 use databend_common_metrics::storage::*;
@@ -67,6 +69,11 @@ impl BlockReader {
                 // first, check in memory table data cache
                 // column_array_cache
                 if let Some(cache_array) = column_array_cache.get_sized(&column_cache_key, len) {
+                    // Record bytes scanned from memory cache (table data only)
+                    Profile::record_usize_profile(
+                        ProfileStatisticsName::ScanBytesFromMemory,
+                        len as usize,
+                    );
                     cached_column_array.push((*column_id, cache_array));
                     continue;
                 }
@@ -87,6 +94,12 @@ impl BlockReader {
                     metrics_inc_remote_io_seeks(1);
                     metrics_inc_remote_io_read_bytes(len);
                 }
+
+                // Record bytes scanned from remote storage
+                Profile::record_usize_profile(
+                    ProfileStatisticsName::ScanBytesFromRemote,
+                    len as usize,
+                );
             }
         }
 

tests/sqllogictests/suites/mode/cluster/explain_analyze.test

@@ -128,6 +128,7 @@ Exchange
     ├── output rows: 6
     ├── output bytes: 120.00 B
     ├── bytes scanned: 120.00 B
+    ├── read from remote: <slt:ignore>
     ├── runtime filter inlist/min-max time: <slt:ignore>
     ├── table: default.default.article
     ├── scan id: 0
@@ -159,6 +160,7 @@ Exchange
         ├── output rows: 4
         ├── output bytes: 80.00 B
         ├── bytes scanned: 80.00 B
+        ├── read from remote: <slt:ignore>
         ├── runtime filter inlist/min-max time: <slt:ignore>
         ├── table: default.default.article
         ├── scan id: 0
@@ -208,6 +210,7 @@ Exchange
     │           ├── output rows: 1
     │           ├── output bytes: <slt:ignore>
     │           ├── bytes scanned: <slt:ignore>
+    │           ├── read from remote: <slt:ignore>
     │           ├── runtime filter inlist/min-max time: <slt:ignore>
     │           ├── table: default.default.author
     │           ├── scan id: 1
@@ -225,6 +228,7 @@ Exchange
         ├── output rows: 6
         ├── output bytes: 120.00 B
         ├── bytes scanned: 120.00 B
+        ├── read from remote: <slt:ignore>
         ├── runtime filter inlist/min-max time: <slt:ignore>
         ├── table: default.default.article
         ├── scan id: 0

tests/sqllogictests/suites/mode/standalone/explain/explain_analyze.test

@@ -138,6 +138,7 @@ Filter
     ├── output rows: 4
     ├── output bytes: 80.00 B
     ├── bytes scanned: 80.00 B
+    ├── read from remote: <slt:ignore>
     ├── runtime filter inlist/min-max time: <slt:ignore>
     ├── table: default.default.article
     ├── scan id: 0
@@ -176,6 +177,7 @@ HashJoin
 │       ├── output rows: 1
 │       ├── output bytes: <slt:ignore>
 │       ├── bytes scanned: <slt:ignore>
+│       ├── read from remote: <slt:ignore>
 │       ├── runtime filter inlist/min-max time: <slt:ignore>
 │       ├── table: default.default.author
 │       ├── scan id: 1
@@ -193,6 +195,7 @@ HashJoin
     ├── output rows: 6
     ├── output bytes: 120.00 B
     ├── bytes scanned: 120.00 B
+    ├── read from remote: <slt:ignore>
     ├── runtime filter inlist/min-max time: <slt:ignore>
     ├── table: default.default.article
     ├── scan id: 0