feat: add scan io metrics and auto increment helper

Author: BohuTANG

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

@@ -35,8 +35,10 @@ pub enum ProfileStatisticsName {
     OutputRows,
     OutputBytes,
     ScanBytes,
-    ScanCacheBytes,
     ScanPartitions,
+    ScanBytesFromRemote,
+    ScanBytesFromLocal,
+    ScanBytesFromMemory,
 
     RemoteSpillWriteCount,
     RemoteSpillWriteBytes,
@@ -190,20 +192,34 @@ 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: "bytes scanned from remote",
+                desc: "The bytes scanned from remote storage (compressed)",
+                index: ProfileStatisticsName::ScanBytesFromRemote as usize,
+                unit: StatisticsUnit::Bytes,
+                plain_statistics: true,
+            }),
+            (ProfileStatisticsName::ScanBytesFromLocal, ProfileDesc {
+                display_name: "bytes scanned from local cache",
+                desc: "The bytes scanned from local disk cache (compressed)",
+                index: ProfileStatisticsName::ScanBytesFromLocal as usize,
+                unit: StatisticsUnit::Bytes,
+                plain_statistics: true,
+            }),
+            (ProfileStatisticsName::ScanBytesFromMemory, ProfileDesc {
+                display_name: "bytes scanned from memory cache",
+                desc: "The bytes scanned from memory cache (compressed)",
+                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

@@ -17,6 +17,7 @@ use std::collections::HashMap;
 use databend_common_ast::ast::FormatTreeNode;
 use databend_common_base::base::format_byte_size;
 use databend_common_base::runtime::profile::get_statistics_desc;
+use databend_common_base::runtime::profile::ProfileStatisticsName;
 use databend_common_catalog::plan::PartStatistics;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
@@ -612,14 +613,46 @@ fn append_profile_info(
     plan_id: u32,
 ) {
     if let Some(prof) = profs.get(&plan_id) {
-        for (_, desc) in get_statistics_desc().iter() {
-            if prof.statistics[desc.index] != 0 {
-                children.push(FormatTreeNode::new(format!(
+        // Calculate total scan IO bytes for percentage
+        let total_scan_io = prof.statistics[ProfileStatisticsName::ScanBytesFromRemote as usize]
+            + prof.statistics[ProfileStatisticsName::ScanBytesFromLocal as usize]
+            + prof.statistics[ProfileStatisticsName::ScanBytesFromMemory as usize];
+
+        for (stat_name, desc) in get_statistics_desc().iter() {
+            let value = prof.statistics[desc.index];
+            if value == 0 {
+                continue;
+            }
+
+            // Add percentage for cache-related statistics
+            let display_text = if total_scan_io > 0 {
+                match stat_name {
+                    ProfileStatisticsName::ScanBytesFromRemote
+                    | ProfileStatisticsName::ScanBytesFromLocal
+                    | ProfileStatisticsName::ScanBytesFromMemory => {
+                        let percentage = (value as f64 / total_scan_io as f64) * 100.0;
+                        format!(
+                            "{}: {} ({:.2}%)",
+                            desc.display_name.to_lowercase(),
+                            desc.human_format(value),
+                            percentage
+                        )
+                    }
+                    _ => format!(
+                        "{}: {}",
+                        desc.display_name.to_lowercase(),
+                        desc.human_format(value)
+                    ),
+                }
+            } else {
+                format!(
                     "{}: {}",
                     desc.display_name.to_lowercase(),
-                    desc.human_format(prof.statistics[desc.index])
-                )));
-            }
+                    desc.human_format(value)
+                )
+            };
+
+            children.push(FormatTreeNode::new(display_text));
         }
     }
 }

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::ScanBytesFromLocal, size);
             metrics_inc_cache_hit_count(1, &self.name);
             Some(item)
         } else {

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

@@ -152,10 +152,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,7 +173,7 @@ where
                 }
             }
         } else {
-            // Cache Miss
+            // Cache Miss - will need to read from remote
             None
         }
     }

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

@@ -79,6 +79,9 @@ impl<V: Into<CacheValue<V>>> InMemoryLruCache<V> {
 mod impls {
     use std::sync::Arc;
 
+    use databend_common_base::runtime::profile::Profile;
+    use databend_common_base::runtime::profile::ProfileStatisticsName;
+    use databend_common_cache::MemSized;
     use databend_common_metrics::cache::metrics_inc_cache_access_count;
     use databend_common_metrics::cache::metrics_inc_cache_hit_count;
     use databend_common_metrics::cache::metrics_inc_cache_miss_bytes;
@@ -93,16 +96,24 @@ mod impls {
 
         fn get<Q: AsRef<str>>(&self, k: Q) -> Option<Arc<V>> {
             metrics_inc_cache_access_count(1, self.name());
-            let v = {
+            let (v, mem_bytes) = {
                 let mut guard = self.inner.write();
-                guard
-                    .get(k.as_ref())
-                    .map(|cache_value: &CacheValue<V>| cache_value.get_inner())
+                match guard.get(k.as_ref()) {
+                    Some(cache_value) => {
+                        (Some(cache_value.get_inner()), Some(cache_value.mem_bytes()))
+                    }
+                    None => (None, None),
+                }
             };
+
             if v.is_none() {
                 metrics_inc_cache_miss_count(1, &self.name);
             } else {
                 metrics_inc_cache_hit_count(1, &self.name);
+                // Record bytes scanned from memory cache
+                if let Some(size) = mem_bytes {
+                    Profile::record_usize_profile(ProfileStatisticsName::ScanBytesFromMemory, size);
+                }
             }
             v
         }

src/query/storages/fuse/src/operations/replace_into/auto_increment.rs

@@ -0,0 +1,105 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed 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.
+
+use std::sync::Arc;
+
+use ahash::AHashMap;
+use databend_common_exception::Result;
+use databend_common_expression::Scalar;
+use databend_common_expression::ScalarRef;
+use databend_common_expression::Value;
+use parking_lot::RwLock;
+
+use super::meta::UniqueKeyDigest;
+
+#[derive(Clone, Debug)]
+pub enum AutoIncValueSource {
+    Key(usize),
+    Remain(usize),
+}
+
+#[derive(Clone, Debug)]
+pub struct AutoIncColumn {
+    pub field_index: usize,
+    pub source: AutoIncValueSource,
+}
+
+#[derive(Clone, Debug)]
+pub struct AutoIncrementReplacer {
+    columns: Vec<AutoIncColumn>,
+    values: Arc<RwLock<AHashMap<UniqueKeyDigest, Vec<Scalar>>>>,
+}
+
+impl AutoIncrementReplacer {
+    pub fn try_new(columns: Vec<AutoIncColumn>) -> Option<Arc<Self>> {
+        if columns.is_empty() {
+            return None;
+        }
+        Some(Arc::new(Self {
+            columns,
+            values: Arc::new(RwLock::new(AHashMap::default())),
+        }))
+    }
+
+    pub fn columns(&self) -> &[AutoIncColumn] {
+        &self.columns
+    }
+
+    pub fn record(&self, hash: UniqueKeyDigest, row_values: Vec<Scalar>) -> Result<()> {
+        if row_values.len() != self.columns.len() {
+            return Err(databend_common_exception::ErrorCode::Internal(format!(
+                "auto increment replacement expects {} values, got {}",
+                self.columns.len(),
+                row_values.len()
+            )));
+        }
+        self.values.write().insert(hash, row_values);
+        Ok(())
+    }
+
+    pub fn take(&self, hash: &UniqueKeyDigest) -> Option<Vec<Scalar>> {
+        self.values.write().remove(hash)
+    }
+}
+
+pub fn scalar_ref_to_owned(scalar_ref: ScalarRef) -> Scalar {
+    match scalar_ref {
+        ScalarRef::Null => Scalar::Null,
+        ScalarRef::EmptyArray => Scalar::EmptyArray,
+        ScalarRef::EmptyMap => Scalar::EmptyMap,
+        ScalarRef::Number(n) => Scalar::Number(n),
+        ScalarRef::Decimal(d) => Scalar::Decimal(d),
+        ScalarRef::Boolean(b) => Scalar::Boolean(b),
+        ScalarRef::Binary(v) => Scalar::Binary(v.to_vec()),
+        ScalarRef::String(v) => Scalar::String(v.to_string()),
+        ScalarRef::Timestamp(t) => Scalar::Timestamp(t),
+        ScalarRef::TimestampTz(t) => Scalar::TimestampTz(t),
+        ScalarRef::Date(d) => Scalar::Date(d),
+        ScalarRef::Interval(val) => Scalar::Interval(val),
+        ScalarRef::Array(col) => Scalar::Array(col),
+        ScalarRef::Map(col) => Scalar::Map(col),
+        ScalarRef::Bitmap(v) => Scalar::Bitmap(v.to_vec()),
+        ScalarRef::Tuple(items) => Scalar::Tuple(items.into_iter().map(scalar_ref_to_owned).collect()),
+        ScalarRef::Variant(v) => Scalar::Variant(v.to_vec()),
+        ScalarRef::Geometry(v) => Scalar::Geometry(v.to_vec()),
+        ScalarRef::Geography(v) => Scalar::Geography(v.into()),
+        ScalarRef::Vector(v) => Scalar::Vector(v.to_owned()),
+        ScalarRef::Opaque(v) => Scalar::Opaque(v.to_owned()),
+    }
+}
+
+pub fn value_row_scalar(value: &Value<databend_common_expression::types::AnyType>, row: usize) -> Result<Scalar> {
+    let scalar_ref = value.row_scalar(row)?;
+    Ok(scalar_ref_to_owned(scalar_ref))
+}