diff --git a/src/catalog/src/system_schema/information_schema/region_statistics.rs b/src/catalog/src/system_schema/information_schema/region_statistics.rs index 7d90810f05..b781c27753 100644 --- a/src/catalog/src/system_schema/information_schema/region_statistics.rs +++ b/src/catalog/src/system_schema/information_schema/region_statistics.rs @@ -41,6 +41,8 @@ const TABLE_ID: &str = "table_id"; const REGION_NUMBER: &str = "region_number"; const REGION_ROWS: &str = "region_rows"; const WRITTEN_BYTES: &str = "written_bytes_since_open"; +const QUERY_CPU_TIME_MILLIS: &str = "query_cpu_time_millis"; +const QUERY_SCANNED_BYTES: &str = "query_scanned_bytes"; const DISK_SIZE: &str = "disk_size"; const MEMTABLE_SIZE: &str = "memtable_size"; const MANIFEST_SIZE: &str = "manifest_size"; @@ -59,6 +61,8 @@ const INIT_CAPACITY: usize = 42; /// - `region_number`: The region number. /// - `region_rows`: The number of rows in region. /// - `written_bytes_since_open`: The total bytes written of the region since region opened. +/// - `query_cpu_time_millis`: The total query CPU time of the region since region opened, in milliseconds. +/// - `query_scanned_bytes`: The total bytes scanned by queries since region opened. /// - `memtable_size`: The memtable size in bytes. /// - `disk_size`: The approximate disk size in bytes. /// - `manifest_size`: The manifest size in bytes. @@ -87,6 +91,16 @@ impl InformationSchemaRegionStatistics { ColumnSchema::new(REGION_NUMBER, ConcreteDataType::uint32_datatype(), false), ColumnSchema::new(REGION_ROWS, ConcreteDataType::uint64_datatype(), true), ColumnSchema::new(WRITTEN_BYTES, ConcreteDataType::uint64_datatype(), true), + ColumnSchema::new( + QUERY_CPU_TIME_MILLIS, + ConcreteDataType::uint64_datatype(), + true, + ), + ColumnSchema::new( + QUERY_SCANNED_BYTES, + ConcreteDataType::uint64_datatype(), + true, + ), ColumnSchema::new(DISK_SIZE, ConcreteDataType::uint64_datatype(), true), ColumnSchema::new(MEMTABLE_SIZE, ConcreteDataType::uint64_datatype(), true), ColumnSchema::new(MANIFEST_SIZE, ConcreteDataType::uint64_datatype(), true), @@ -151,6 +165,8 @@ struct InformationSchemaRegionStatisticsBuilder { region_numbers: UInt32VectorBuilder, region_rows: UInt64VectorBuilder, written_bytes: UInt64VectorBuilder, + query_cpu_time_millis: UInt64VectorBuilder, + query_scanned_bytes: UInt64VectorBuilder, disk_sizes: UInt64VectorBuilder, memtable_sizes: UInt64VectorBuilder, manifest_sizes: UInt64VectorBuilder, @@ -171,6 +187,8 @@ impl InformationSchemaRegionStatisticsBuilder { region_numbers: UInt32VectorBuilder::with_capacity(INIT_CAPACITY), region_rows: UInt64VectorBuilder::with_capacity(INIT_CAPACITY), written_bytes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY), + query_cpu_time_millis: UInt64VectorBuilder::with_capacity(INIT_CAPACITY), + query_scanned_bytes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY), disk_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY), memtable_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY), manifest_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY), @@ -203,6 +221,14 @@ impl InformationSchemaRegionStatisticsBuilder { (REGION_NUMBER, &Value::from(region_stat.id.region_number())), (REGION_ROWS, &Value::from(region_stat.num_rows)), (WRITTEN_BYTES, &Value::from(region_stat.written_bytes)), + ( + QUERY_CPU_TIME_MILLIS, + &Value::from(region_stat.query_cpu_time / 1_000_000), + ), + ( + QUERY_SCANNED_BYTES, + &Value::from(region_stat.query_scanned_bytes), + ), (DISK_SIZE, &Value::from(region_stat.approximate_bytes)), (MEMTABLE_SIZE, &Value::from(region_stat.memtable_size)), (MANIFEST_SIZE, &Value::from(region_stat.manifest_size)), @@ -223,6 +249,10 @@ impl InformationSchemaRegionStatisticsBuilder { .push(Some(region_stat.id.region_number())); self.region_rows.push(Some(region_stat.num_rows)); self.written_bytes.push(Some(region_stat.written_bytes)); + self.query_cpu_time_millis + .push(Some(region_stat.query_cpu_time / 1_000_000)); + self.query_scanned_bytes + .push(Some(region_stat.query_scanned_bytes)); self.disk_sizes.push(Some(region_stat.approximate_bytes)); self.memtable_sizes.push(Some(region_stat.memtable_size)); self.manifest_sizes.push(Some(region_stat.manifest_size)); @@ -240,6 +270,8 @@ impl InformationSchemaRegionStatisticsBuilder { Arc::new(self.region_numbers.finish()), Arc::new(self.region_rows.finish()), Arc::new(self.written_bytes.finish()), + Arc::new(self.query_cpu_time_millis.finish()), + Arc::new(self.query_scanned_bytes.finish()), Arc::new(self.disk_sizes.finish()), Arc::new(self.memtable_sizes.finish()), Arc::new(self.manifest_sizes.finish()), diff --git a/src/common/meta/src/datanode.rs b/src/common/meta/src/datanode.rs index f3d3a4bed1..8af8646212 100644 --- a/src/common/meta/src/datanode.rs +++ b/src/common/meta/src/datanode.rs @@ -100,6 +100,14 @@ pub struct RegionStat { pub region_manifest: RegionManifestInfo, /// The total bytes written of the region since region opened. pub written_bytes: u64, + /// The total query CPU time of the region since region opened. + /// + /// Unit: nanoseconds. + #[serde(default)] + pub query_cpu_time: u64, + /// The total scanned bytes of the region since region opened. + #[serde(default)] + pub query_scanned_bytes: u64, /// The latest entry id of topic used by data. /// **Only used by remote WAL prune.** pub data_topic_latest_entry_id: u64, @@ -317,6 +325,8 @@ impl From<&api::v1::meta::RegionStat> for RegionStat { index_size: region_stat.index_size, region_manifest: region_stat.manifest.into(), written_bytes: region_stat.written_bytes, + query_cpu_time: region_stat.query_cpu_time, + query_scanned_bytes: region_stat.query_scanned_bytes, data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id, metadata_topic_latest_entry_id: region_stat.metadata_topic_latest_entry_id, } @@ -555,6 +565,48 @@ mod tests { assert_eq!(100, stat.region_num); } + #[test] + fn test_stat_val_deserializes_without_query_stats() { + let stat = Stat { + region_stats: vec![RegionStat { + id: RegionId::new(1024, 1), + rcus: 0, + wcus: 0, + approximate_bytes: 0, + engine: "mito".to_string(), + role: RegionRole::Leader, + num_rows: 0, + memtable_size: 0, + manifest_size: 0, + sst_size: 0, + sst_num: 0, + index_size: 0, + region_manifest: RegionManifestInfo::Mito { + manifest_version: 0, + flushed_entry_id: 0, + file_removed_cnt: 0, + }, + written_bytes: 0, + query_cpu_time: 10, + query_scanned_bytes: 20, + data_topic_latest_entry_id: 0, + metadata_topic_latest_entry_id: 0, + }], + ..Default::default() + }; + let stat_val = DatanodeStatValue { stats: vec![stat] }; + let mut value = serde_json::to_value(stat_val).unwrap(); + let region_stat = value[0]["region_stats"][0].as_object_mut().unwrap(); + region_stat.remove("query_cpu_time"); + region_stat.remove("query_scanned_bytes"); + + let stat_val: DatanodeStatValue = serde_json::from_value(value).unwrap(); + let region_stat = &stat_val.stats[0].region_stats[0]; + + assert_eq!(region_stat.query_cpu_time, 0); + assert_eq!(region_stat.query_scanned_bytes, 0); + } + #[test] fn test_get_addr_from_stat_val() { let empty = DatanodeStatValue { stats: vec![] }; diff --git a/src/common/recordbatch/src/adapter.rs b/src/common/recordbatch/src/adapter.rs index e6a620ab6c..1a0cc351b8 100644 --- a/src/common/recordbatch/src/adapter.rs +++ b/src/common/recordbatch/src/adapter.rs @@ -18,6 +18,7 @@ use std::marker::PhantomData; use std::pin::Pin; use std::str::FromStr; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use std::task::{Context, Poll}; use common_base::readable_size::ReadableSize; @@ -52,6 +53,8 @@ use crate::{ SendableRecordBatchStream, Stream, }; +const REGION_SCAN_EXEC_NAME: &str = "RegionScanExec"; + type FutureStream = Pin> + Send>>; @@ -218,11 +221,21 @@ pub struct RecordBatchStreamAdapter { /// Aggregated plan-level metrics. Resolved after an [ExecutionPlan] is finished. metrics_2: Metrics, query_load_region_id: Option, + query_stat_counters: Option, /// Display plan and metrics in verbose mode. explain_verbose: bool, span: Span, } +/// Query statistic counters owned by a region. +#[derive(Debug, Clone)] +pub struct RegionQueryStatCounters { + /// The total query CPU time in nanoseconds. + pub query_cpu_time: Arc, + /// The total scanned bytes. + pub query_scanned_bytes: Arc, +} + /// Json encoded metrics. Contains metric from a whole plan tree. enum Metrics { Unavailable, @@ -241,6 +254,7 @@ impl RecordBatchStreamAdapter { metrics: None, metrics_2: Metrics::Unavailable, query_load_region_id: None, + query_stat_counters: None, explain_verbose: false, span: Span::current(), }) @@ -256,6 +270,7 @@ impl RecordBatchStreamAdapter { metrics: None, metrics_2: Metrics::Unavailable, query_load_region_id: None, + query_stat_counters: None, explain_verbose: false, span: subspan, }) @@ -265,16 +280,58 @@ impl RecordBatchStreamAdapter { self.metrics_2 = Metrics::Unresolved(plan) } + fn record_query_stats_on_drop(&self) { + let Some(counters) = &self.query_stat_counters else { + return; + }; + + match &self.metrics_2 { + Metrics::Unresolved(df_plan) | Metrics::PartialResolved(df_plan, _) => { + let mut metric_collector = MetricCollector::new(self.explain_verbose); + accept(df_plan.as_ref(), &mut metric_collector).unwrap(); + metric_collector.record_batch_metrics.query_load_region_id = + self.query_load_region_id; + record_query_stats(counters, &metric_collector.record_batch_metrics); + } + Metrics::Resolved(metrics) => record_query_stats(counters, metrics), + Metrics::Unavailable => {} + } + } + pub fn set_query_load_region_id(&mut self, region_id: Option) { self.query_load_region_id = region_id; } + pub fn set_query_stat_counters(&mut self, counters: Option) { + self.query_stat_counters = counters; + } + /// Set the verbose mode for displaying plan and metrics. pub fn set_explain_verbose(&mut self, verbose: bool) { self.explain_verbose = verbose; } } +/// Extracts total `output_bytes` from region scan plan nodes. +pub fn region_scan_output_bytes(metrics: &RecordBatchMetrics) -> usize { + metrics + .plan_metrics + .iter() + .filter(|pm| pm.plan_name == REGION_SCAN_EXEC_NAME) + .flat_map(|pm| &pm.metrics) + .filter_map(|(name, value)| (name == "output_bytes").then_some(*value)) + .sum() +} + +fn record_query_stats(counters: &RegionQueryStatCounters, metrics: &RecordBatchMetrics) { + counters + .query_cpu_time + .fetch_add(metrics.elapsed_compute as u64, Ordering::Relaxed); + counters + .query_scanned_bytes + .fetch_add(region_scan_output_bytes(metrics) as u64, Ordering::Relaxed); +} + impl RecordBatchStream for RecordBatchStreamAdapter { fn name(&self) -> &str { "RecordBatchStreamAdapter" @@ -352,6 +409,12 @@ impl Stream for RecordBatchStreamAdapter { } } +impl Drop for RecordBatchStreamAdapter { + fn drop(&mut self) { + self.record_query_stats_on_drop(); + } +} + /// An [ExecutionPlanVisitor] to collect metrics from a [ExecutionPlan]. pub struct MetricCollector { current_level: usize, @@ -921,6 +984,73 @@ mod test { } } + #[test] + fn test_record_query_stats_updates_region_counters() { + let counters = RegionQueryStatCounters { + query_cpu_time: Arc::new(AtomicU64::new(10)), + query_scanned_bytes: Arc::new(AtomicU64::new(20)), + }; + let metrics = RecordBatchMetrics { + elapsed_compute: 2_000_000, + plan_metrics: vec![PlanMetrics { + plan: "RegionScanExec: region=1".to_string(), + plan_name: REGION_SCAN_EXEC_NAME.to_string(), + level: 0, + metrics: vec![("output_bytes".to_string(), 42)], + }], + ..Default::default() + }; + + record_query_stats(&counters, &metrics); + + assert_eq!(counters.query_cpu_time.load(Ordering::Relaxed), 2_000_010); + assert_eq!(counters.query_scanned_bytes.load(Ordering::Relaxed), 62); + } + + #[test] + fn test_record_batch_stream_adapter_records_query_stats_on_drop() { + let schema = Arc::new(Schema::new(vec![ColumnSchema::new( + "a", + ConcreteDataType::int32_datatype(), + false, + )])); + let df_stream = Box::pin( + datafusion::physical_plan::stream::RecordBatchStreamAdapter::new( + schema.arrow_schema().clone(), + futures::stream::empty::>(), + ), + ); + let counters = RegionQueryStatCounters { + query_cpu_time: Arc::new(AtomicU64::new(10)), + query_scanned_bytes: Arc::new(AtomicU64::new(20)), + }; + let metrics = RecordBatchMetrics { + elapsed_compute: 2_000_000, + plan_metrics: vec![PlanMetrics { + plan: "RegionScanExec: region=1".to_string(), + plan_name: REGION_SCAN_EXEC_NAME.to_string(), + level: 0, + metrics: vec![("output_bytes".to_string(), 42)], + }], + ..Default::default() + }; + let adapter = RecordBatchStreamAdapter { + schema, + stream: df_stream, + metrics: None, + metrics_2: Metrics::Resolved(metrics), + query_load_region_id: None, + query_stat_counters: Some(counters.clone()), + explain_verbose: false, + span: Span::current(), + }; + + drop(adapter); + + assert_eq!(counters.query_cpu_time.load(Ordering::Relaxed), 2_000_010); + assert_eq!(counters.query_scanned_bytes.load(Ordering::Relaxed), 62); + } + #[test] fn test_recordbatch_metrics_deserializes_without_region_watermarks() { let metrics: RecordBatchMetrics = serde_json::from_value(json!({ diff --git a/src/meta-srv/src/gc/handler.rs b/src/meta-srv/src/gc/handler.rs index c62e927f89..37a71580fa 100644 --- a/src/meta-srv/src/gc/handler.rs +++ b/src/meta-srv/src/gc/handler.rs @@ -484,6 +484,8 @@ fn dropped_region_stat(region_id: RegionId) -> RegionStat { file_removed_cnt: 0, }, written_bytes: 0, + query_cpu_time: 0, + query_scanned_bytes: 0, data_topic_latest_entry_id: 0, metadata_topic_latest_entry_id: 0, } diff --git a/src/meta-srv/src/gc/mock.rs b/src/meta-srv/src/gc/mock.rs index 6c736490ab..ac257576a4 100644 --- a/src/meta-srv/src/gc/mock.rs +++ b/src/meta-srv/src/gc/mock.rs @@ -419,5 +419,7 @@ fn mock_region_stat( data_topic_latest_entry_id: 0, metadata_topic_latest_entry_id: 0, written_bytes: 0, + query_cpu_time: 0, + query_scanned_bytes: 0, } } diff --git a/src/meta-srv/src/handler/collect_leader_region_handler.rs b/src/meta-srv/src/handler/collect_leader_region_handler.rs index 95b03e3341..2b6aa1cb9f 100644 --- a/src/meta-srv/src/handler/collect_leader_region_handler.rs +++ b/src/meta-srv/src/handler/collect_leader_region_handler.rs @@ -89,6 +89,8 @@ mod tests { data_topic_latest_entry_id: 0, metadata_topic_latest_entry_id: 0, written_bytes: 0, + query_cpu_time: 0, + query_scanned_bytes: 0, } } diff --git a/src/meta-srv/src/handler/failure_handler.rs b/src/meta-srv/src/handler/failure_handler.rs index eb79a1c30d..fa089c14c2 100644 --- a/src/meta-srv/src/handler/failure_handler.rs +++ b/src/meta-srv/src/handler/failure_handler.rs @@ -107,6 +107,8 @@ mod tests { data_topic_latest_entry_id: 0, metadata_topic_latest_entry_id: 0, written_bytes: 0, + query_cpu_time: 0, + query_scanned_bytes: 0, } } acc.stat = Some(Stat { diff --git a/src/meta-srv/src/handler/persist_stats_handler.rs b/src/meta-srv/src/handler/persist_stats_handler.rs index d863070225..b1378c39e6 100644 --- a/src/meta-srv/src/handler/persist_stats_handler.rs +++ b/src/meta-srv/src/handler/persist_stats_handler.rs @@ -300,6 +300,8 @@ mod tests { file_removed_cnt: 0, }, written_bytes, + query_cpu_time: 0, + query_scanned_bytes: 0, data_topic_latest_entry_id: 200, metadata_topic_latest_entry_id: 200, } diff --git a/src/meta-srv/src/handler/region_lease_handler.rs b/src/meta-srv/src/handler/region_lease_handler.rs index aa26a037e3..c8d3e75264 100644 --- a/src/meta-srv/src/handler/region_lease_handler.rs +++ b/src/meta-srv/src/handler/region_lease_handler.rs @@ -180,6 +180,8 @@ mod test { data_topic_latest_entry_id: 0, metadata_topic_latest_entry_id: 0, written_bytes: 0, + query_cpu_time: 0, + query_scanned_bytes: 0, } } diff --git a/src/meta-srv/src/selector/weight_compute.rs b/src/meta-srv/src/selector/weight_compute.rs index 6508f78efe..dfbabfa650 100644 --- a/src/meta-srv/src/selector/weight_compute.rs +++ b/src/meta-srv/src/selector/weight_compute.rs @@ -200,6 +200,8 @@ mod tests { data_topic_latest_entry_id: 0, metadata_topic_latest_entry_id: 0, written_bytes: 0, + query_cpu_time: 0, + query_scanned_bytes: 0, }], ..Default::default() } @@ -230,6 +232,8 @@ mod tests { data_topic_latest_entry_id: 0, metadata_topic_latest_entry_id: 0, written_bytes: 0, + query_cpu_time: 0, + query_scanned_bytes: 0, }], ..Default::default() } @@ -260,6 +264,8 @@ mod tests { data_topic_latest_entry_id: 0, metadata_topic_latest_entry_id: 0, written_bytes: 0, + query_cpu_time: 0, + query_scanned_bytes: 0, }], ..Default::default() } diff --git a/src/metric-engine/src/utils.rs b/src/metric-engine/src/utils.rs index f1930b4f63..1620b1bb9e 100644 --- a/src/metric-engine/src/utils.rs +++ b/src/metric-engine/src/utils.rs @@ -48,24 +48,9 @@ pub fn get_region_statistic(mito: &MitoEngine, region_id: RegionId) -> Option Some(RegionStatistic { - num_rows: metadata_stat.num_rows + data_stat.num_rows, - memtable_size: metadata_stat.memtable_size + data_stat.memtable_size, - wal_size: metadata_stat.wal_size + data_stat.wal_size, - manifest_size: metadata_stat.manifest_size + data_stat.manifest_size, - sst_size: metadata_stat.sst_size + data_stat.sst_size, - sst_num: metadata_stat.sst_num + data_stat.sst_num, - index_size: metadata_stat.index_size + data_stat.index_size, - manifest: RegionManifestInfo::Metric { - data_flushed_entry_id: data_stat.manifest.data_flushed_entry_id(), - data_manifest_version: data_stat.manifest.data_manifest_version(), - metadata_flushed_entry_id: metadata_stat.manifest.data_flushed_entry_id(), - metadata_manifest_version: metadata_stat.manifest.data_manifest_version(), - }, - written_bytes: metadata_stat.written_bytes + data_stat.written_bytes, - data_topic_latest_entry_id: data_stat.data_topic_latest_entry_id, - metadata_topic_latest_entry_id: metadata_stat.metadata_topic_latest_entry_id, - }), + (Some(metadata_stat), Some(data_stat)) => { + Some(merge_region_statistic(metadata_stat, data_stat)) + } _ => { warn!( "Failed to get region statistic for region {}, metadata_stat: {:?}, data_stat: {:?}", @@ -76,6 +61,32 @@ pub fn get_region_statistic(mito: &MitoEngine, region_id: RegionId) -> Option RegionStatistic { + RegionStatistic { + num_rows: metadata_stat.num_rows + data_stat.num_rows, + memtable_size: metadata_stat.memtable_size + data_stat.memtable_size, + wal_size: metadata_stat.wal_size + data_stat.wal_size, + manifest_size: metadata_stat.manifest_size + data_stat.manifest_size, + sst_size: metadata_stat.sst_size + data_stat.sst_size, + sst_num: metadata_stat.sst_num + data_stat.sst_num, + index_size: metadata_stat.index_size + data_stat.index_size, + manifest: RegionManifestInfo::Metric { + data_flushed_entry_id: data_stat.manifest.data_flushed_entry_id(), + data_manifest_version: data_stat.manifest.data_manifest_version(), + metadata_flushed_entry_id: metadata_stat.manifest.data_flushed_entry_id(), + metadata_manifest_version: metadata_stat.manifest.data_manifest_version(), + }, + written_bytes: metadata_stat.written_bytes + data_stat.written_bytes, + query_cpu_time: data_stat.query_cpu_time, + query_scanned_bytes: data_stat.query_scanned_bytes, + data_topic_latest_entry_id: data_stat.data_topic_latest_entry_id, + metadata_topic_latest_entry_id: metadata_stat.metadata_topic_latest_entry_id, + } +} + /// Appends the given [RegionId]'s manifest info to the given list. pub(crate) fn append_manifest_info( mito: &MitoEngine, @@ -132,4 +143,25 @@ mod tests { let expected_region_id = RegionId::with_group_and_seq(1, METRIC_DATA_REGION_GROUP, 2); assert_eq!(to_data_region_id(region_id), expected_region_id); } + + #[test] + fn merge_region_statistic_uses_data_region_query_stats() { + let metadata_stat = RegionStatistic { + query_cpu_time: 100, + query_scanned_bytes: 200, + manifest: RegionManifestInfo::mito(1, 2, 0), + ..Default::default() + }; + let data_stat = RegionStatistic { + query_cpu_time: 10, + query_scanned_bytes: 20, + manifest: RegionManifestInfo::mito(3, 4, 0), + ..Default::default() + }; + + let statistic = merge_region_statistic(&metadata_stat, &data_stat); + + assert_eq!(statistic.query_cpu_time, 10); + assert_eq!(statistic.query_scanned_bytes, 20); + } } diff --git a/src/mito2/src/engine.rs b/src/mito2/src/engine.rs index ee679cb225..e0923b8dfa 100644 --- a/src/mito2/src/engine.rs +++ b/src/mito2/src/engine.rs @@ -1076,6 +1076,7 @@ impl EngineInner { request, CacheStrategy::EnableAll(cache_manager), ) + .with_query_stat_counters(region.region_stats.query_stat_counters()) .with_max_concurrent_scan_files(self.config.max_concurrent_scan_files) .with_ignore_inverted_index(self.config.inverted_index.apply_on_query.disabled()) .with_ignore_fulltext_index(self.config.fulltext_index.apply_on_query.disabled()) diff --git a/src/mito2/src/engine/basic_test.rs b/src/mito2/src/engine/basic_test.rs index 72bc00aedd..da36c82766 100644 --- a/src/mito2/src/engine/basic_test.rs +++ b/src/mito2/src/engine/basic_test.rs @@ -131,7 +131,7 @@ async fn test_write_to_region_with_format(flat_format: bool) { }; put_rows(&engine, region_id, rows).await; let region = engine.get_region(region_id).unwrap(); - assert!(region.written_bytes.load(Ordering::Relaxed) > 0); + assert!(region.region_stats.written_bytes.load(Ordering::Relaxed) > 0); } #[apply(multiple_log_store_factories)] @@ -220,7 +220,7 @@ async fn test_region_replay_with_format(factory: Option, flat_f // The replay won't update the write bytes rate meter. let region = engine.get_region(region_id).unwrap(); - assert_eq!(region.written_bytes.load(Ordering::Relaxed), 0); + assert_eq!(region.region_stats.written_bytes.load(Ordering::Relaxed), 0); let request = ScanRequest::default(); let stream = engine.scan_to_stream(region_id, request).await.unwrap(); diff --git a/src/mito2/src/read/scan_region.rs b/src/mito2/src/read/scan_region.rs index 0101b529d1..9e1f3b3acf 100644 --- a/src/mito2/src/read/scan_region.rs +++ b/src/mito2/src/read/scan_region.rs @@ -23,6 +23,7 @@ use std::time::Instant; use api::v1::SemanticType; use common_error::ext::BoxedError; use common_recordbatch::SendableRecordBatchStream; +use common_recordbatch::adapter::RegionQueryStatCounters; use common_recordbatch::filter::SimpleFilterEvaluator; use common_telemetry::tracing::Instrument; use common_telemetry::{debug, error, tracing, warn}; @@ -251,6 +252,8 @@ pub(crate) struct ScanRegion { /// Whether to filter out the deleted rows. /// Usually true for normal read, and false for scan for compaction. filter_deleted: bool, + /// Counters that should receive query-load metrics. + query_stat_counters: Option, #[cfg(feature = "enterprise")] extension_range_provider: Option, } @@ -274,11 +277,19 @@ impl ScanRegion { ignore_bloom_filter: false, start_time: None, filter_deleted: true, + query_stat_counters: None, #[cfg(feature = "enterprise")] extension_range_provider: None, } } + /// Sets counters that should receive query-load metrics. + #[must_use] + pub(crate) fn with_query_stat_counters(mut self, counters: RegionQueryStatCounters) -> Self { + self.query_stat_counters = Some(counters); + self + } + /// Sets maximum number of SST files to scan concurrently. #[must_use] pub(crate) fn with_max_concurrent_scan_files( @@ -592,7 +603,8 @@ impl ScanRegion { .snapshot_on_scan .then_some(self.request.memtable_max_sequence) .flatten(), - ); + ) + .with_query_stat_counters(self.query_stat_counters); #[cfg(feature = "vector_index")] let input = input .with_vector_index_applier(vector_index_applier) @@ -835,6 +847,8 @@ pub struct ScanInput { pub(crate) snapshot_sequence: Option, /// Whether this scan is for compaction. pub(crate) compaction: bool, + /// Counters that should receive query-load metrics. + pub(crate) query_stat_counters: Option, #[cfg(feature = "enterprise")] extension_ranges: Vec, } @@ -871,6 +885,7 @@ impl ScanInput { explain_flat_format: false, snapshot_sequence: None, compaction: false, + query_stat_counters: None, #[cfg(feature = "enterprise")] extension_ranges: Vec::new(), } @@ -991,6 +1006,14 @@ impl ScanInput { self } + pub(crate) fn with_query_stat_counters( + mut self, + counters: Option, + ) -> Self { + self.query_stat_counters = counters; + self + } + /// Sets whether to remove deletion markers during scan. #[must_use] pub(crate) fn with_filter_deleted(mut self, filter_deleted: bool) -> Self { diff --git a/src/mito2/src/read/seq_scan.rs b/src/mito2/src/read/seq_scan.rs index aa4087abde..686325d475 100644 --- a/src/mito2/src/read/seq_scan.rs +++ b/src/mito2/src/read/seq_scan.rs @@ -77,6 +77,9 @@ impl SeqScan { let mut properties = ScannerProperties::default() .with_append_mode(input.append_mode) .with_total_rows(input.total_rows()); + if let Some(counters) = input.query_stat_counters.clone() { + properties.set_query_stat_counters(counters); + } let stream_ctx = Arc::new(StreamContext::seq_scan_ctx(input)); properties.partitions = vec![stream_ctx.partition_ranges()]; diff --git a/src/mito2/src/read/series_scan.rs b/src/mito2/src/read/series_scan.rs index 856b57331a..adefae9117 100644 --- a/src/mito2/src/read/series_scan.rs +++ b/src/mito2/src/read/series_scan.rs @@ -87,6 +87,9 @@ impl SeriesScan { let mut properties = ScannerProperties::default() .with_append_mode(input.append_mode) .with_total_rows(input.total_rows()); + if let Some(counters) = input.query_stat_counters.clone() { + properties.set_query_stat_counters(counters); + } let stream_ctx = Arc::new(StreamContext::seq_scan_ctx(input)); properties.partitions = vec![stream_ctx.partition_ranges()]; diff --git a/src/mito2/src/read/unordered_scan.rs b/src/mito2/src/read/unordered_scan.rs index bb2af12a35..a5c09e5cde 100644 --- a/src/mito2/src/read/unordered_scan.rs +++ b/src/mito2/src/read/unordered_scan.rs @@ -62,6 +62,9 @@ impl UnorderedScan { let mut properties = ScannerProperties::default() .with_append_mode(input.append_mode) .with_total_rows(input.total_rows()); + if let Some(counters) = input.query_stat_counters.clone() { + properties.set_query_stat_counters(counters); + } let stream_ctx = Arc::new(StreamContext::unordered_scan_ctx(input)); properties.partitions = vec![stream_ctx.partition_ranges()]; diff --git a/src/mito2/src/region.rs b/src/mito2/src/region.rs index 23d3347a76..1e173ab62d 100644 --- a/src/mito2/src/region.rs +++ b/src/mito2/src/region.rs @@ -26,6 +26,7 @@ use std::sync::atomic::{AtomicI64, AtomicU64, Ordering}; use std::sync::{Arc, Mutex, RwLock}; use common_base::hash::partition_expr_version; +use common_recordbatch::adapter::RegionQueryStatCounters; use common_telemetry::{error, info, warn}; use crossbeam_utils::atomic::AtomicCell; use partition::expr::PartitionExpr; @@ -173,12 +174,40 @@ pub struct MitoRegion { /// There are no WAL entries in range [flushed_entry_id, topic_latest_entry_id] for current region, /// which means these WAL entries maybe able to be pruned up to `topic_latest_entry_id`. pub(crate) topic_latest_entry_id: AtomicU64, - /// The total bytes written to the region. - pub(crate) written_bytes: Arc, + /// Region stats. + pub(crate) region_stats: RegionStats, /// manifest stats stats: ManifestStats, } +/// Runtime statistics of a region. +#[derive(Debug)] +pub(crate) struct RegionStats { + /// The total bytes written to the region. + pub(crate) written_bytes: Arc, + /// The total query CPU time of the region in nanoseconds. + pub(crate) query_cpu_time: Arc, + /// The total scanned bytes of the region. + pub(crate) query_scanned_bytes: Arc, +} + +impl RegionStats { + pub(crate) fn new() -> Self { + Self { + written_bytes: Arc::new(AtomicU64::new(0)), + query_cpu_time: Arc::new(AtomicU64::new(0)), + query_scanned_bytes: Arc::new(AtomicU64::new(0)), + } + } + + pub(crate) fn query_stat_counters(&self) -> RegionQueryStatCounters { + RegionQueryStatCounters { + query_cpu_time: self.query_cpu_time.clone(), + query_scanned_bytes: self.query_scanned_bytes.clone(), + } + } +} + pub type MitoRegionRef = Arc; #[derive(Debug, Clone)] @@ -644,7 +673,12 @@ impl MitoRegion { let file_removed_cnt = self.stats.file_removed_cnt(); let topic_latest_entry_id = self.topic_latest_entry_id.load(Ordering::Relaxed); - let written_bytes = self.written_bytes.load(Ordering::Relaxed); + let written_bytes = self.region_stats.written_bytes.load(Ordering::Relaxed); + let query_cpu_time = self.region_stats.query_cpu_time.load(Ordering::Relaxed); + let query_scanned_bytes = self + .region_stats + .query_scanned_bytes + .load(Ordering::Relaxed); RegionStatistic { num_rows, @@ -662,6 +696,8 @@ impl MitoRegion { data_topic_latest_entry_id: topic_latest_entry_id, metadata_topic_latest_entry_id: topic_latest_entry_id, written_bytes, + query_cpu_time, + query_scanned_bytes, } } @@ -1747,7 +1783,6 @@ pub fn parse_partition_expr(partition_expr_str: Option<&str>) -> Result RegionWorkerLoop { region.region_id, ®ion.version_control, region.provider.clone(), - Some(region.written_bytes.clone()), + Some(region.region_stats.written_bytes.clone()), ); e.insert(region_ctx); @@ -471,7 +471,7 @@ impl RegionWorkerLoop { region.region_id, ®ion.version_control, region.provider.clone(), - Some(region.written_bytes.clone()), + Some(region.region_stats.written_bytes.clone()), ); e.insert(region_ctx); diff --git a/src/query/src/datafusion.rs b/src/query/src/datafusion.rs index 8ed0c3f3c5..08a401e048 100644 --- a/src/query/src/datafusion.rs +++ b/src/query/src/datafusion.rs @@ -29,7 +29,7 @@ use common_error::ext::BoxedError; use common_function::function::FunctionContext; use common_function::function_factory::ScalarFunctionFactory; use common_query::{Output, OutputData, OutputMeta}; -use common_recordbatch::adapter::RecordBatchStreamAdapter; +use common_recordbatch::adapter::{RecordBatchStreamAdapter, RegionQueryStatCounters}; use common_recordbatch::{EmptyRecordBatchStream, SendableRecordBatchStream}; use common_telemetry::tracing; use datafusion::catalog::TableFunction; @@ -101,6 +101,43 @@ fn query_load_region_id(plan: &Arc) -> Option { region_id } +// Finds the region-owned query statistic counters from the local datanode scan plan. +// +// Unlike the Prometheus read-load reporting in `MergeScanExec`, the heartbeat +// counters must be updated before metrics leave the datanode process. The +// `RecordBatchStreamAdapter` that resolves `RecordBatchMetrics` does not know +// the owning `MitoRegion`, so we extract the counters from `RegionScanExec` and +// pass them to the adapter. If a plan contains scans from different regions, +// return `None` to avoid charging the whole plan metrics to one region. +fn query_stat_counters(plan: &Arc) -> Option { + let mut counters: Option = None; + let mut stack = vec![plan.clone()]; + + while let Some(plan) = stack.pop() { + if plan.name() == REGION_SCAN_EXEC_NAME + && let Some(scan) = plan.as_any().downcast_ref::() + && let Some(scan_counters) = scan.query_stat_counters() + { + match &counters { + Some(counters) + if !Arc::ptr_eq(&counters.query_cpu_time, &scan_counters.query_cpu_time) + || !Arc::ptr_eq( + &counters.query_scanned_bytes, + &scan_counters.query_scanned_bytes, + ) => + { + return None; + } + Some(_) => {} + None => counters = Some(scan_counters), + } + } + stack.extend(plan.children().into_iter().cloned()); + } + + counters +} + pub struct DatafusionQueryEngine { state: Arc, plugins: Plugins, @@ -635,6 +672,7 @@ impl QueryExecutor for DatafusionQueryEngine { .context(QueryExecutionSnafu)?; stream.set_metrics2(plan.clone()); stream.set_query_load_region_id(query_load_region_id(plan)); + stream.set_query_stat_counters(query_stat_counters(plan)); stream.set_explain_verbose(explain_verbose); let stream = OnDone::new(Box::pin(stream), move || { let exec_cost = exec_timer.stop_and_record(); @@ -669,6 +707,7 @@ impl QueryExecutor for DatafusionQueryEngine { .context(QueryExecutionSnafu)?; stream.set_metrics2(plan.clone()); stream.set_query_load_region_id(query_load_region_id(plan)); + stream.set_query_stat_counters(query_stat_counters(plan)); stream.set_explain_verbose(explain_verbose); let stream = OnDone::new(Box::pin(stream), move || { let exec_cost = exec_timer.stop_and_record(); @@ -693,7 +732,7 @@ impl QueryExecutor for DatafusionQueryEngine { mod tests { use std::fmt; use std::sync::Arc; - use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use api::v1::SemanticType; use arrow::array::{ArrayRef, UInt64Array}; @@ -816,6 +855,19 @@ mod tests { fn build_query_load_region_scan( query_load_region_id: Option, + ) -> Arc { + build_region_scan(query_load_region_id, None) + } + + fn build_query_stat_counter_region_scan( + counters: RegionQueryStatCounters, + ) -> Arc { + build_region_scan(None, Some(counters)) + } + + fn build_region_scan( + query_load_region_id: Option, + query_stat_counters: Option, ) -> Arc { let schema = Arc::new(datatypes::schema::Schema::new(vec![ColumnSchema::new( "ts", @@ -846,10 +898,20 @@ mod tests { if let Some(region_id) = query_load_region_id { scanner.set_query_load_region_id(region_id); } + if let Some(counters) = query_stat_counters { + scanner.properties.set_query_stat_counters(counters); + } Arc::new(RegionScanExec::new(Box::new(scanner), ScanRequest::default(), None).unwrap()) } + fn query_stat_counters_for_test() -> RegionQueryStatCounters { + RegionQueryStatCounters { + query_cpu_time: Arc::new(AtomicU64::new(0)), + query_scanned_bytes: Arc::new(AtomicU64::new(0)), + } + } + #[test] fn query_load_region_id_ignores_scans_without_region_id() { let query_load_region_id = RegionId::new(1024, 42); @@ -880,6 +942,67 @@ mod tests { ); } + #[test] + fn query_stat_counters_returns_shared_counter_for_multi_scan_plan() { + let counters = query_stat_counters_for_test(); + let left = build_query_stat_counter_region_scan(counters.clone()); + let right = build_query_stat_counter_region_scan(counters.clone()); + let on: JoinOn = vec![( + Arc::new(Column::new("ts", 0)) as Arc, + Arc::new(Column::new("ts", 0)) as Arc, + )]; + let plan: Arc = Arc::new( + HashJoinExec::try_new( + left, + right, + on, + None, + &JoinType::Inner, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNull, + false, + ) + .unwrap(), + ); + + let actual = super::query_stat_counters(&plan).unwrap(); + assert!(Arc::ptr_eq( + &actual.query_cpu_time, + &counters.query_cpu_time + )); + assert!(Arc::ptr_eq( + &actual.query_scanned_bytes, + &counters.query_scanned_bytes + )); + } + + #[test] + fn query_stat_counters_ignores_mixed_counter_plan() { + let left = build_query_stat_counter_region_scan(query_stat_counters_for_test()); + let right = build_query_stat_counter_region_scan(query_stat_counters_for_test()); + let on: JoinOn = vec![( + Arc::new(Column::new("ts", 0)) as Arc, + Arc::new(Column::new("ts", 0)) as Arc, + )]; + let plan: Arc = Arc::new( + HashJoinExec::try_new( + left, + right, + on, + None, + &JoinType::Inner, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNull, + false, + ) + .unwrap(), + ); + + assert!(super::query_stat_counters(&plan).is_none()); + } + async fn create_test_engine() -> QueryEngineRef { let catalog_manager = catalog::memory::new_memory_catalog_manager().unwrap(); let req = RegisterTableRequest { diff --git a/src/query/src/dist_plan/merge_scan.rs b/src/query/src/dist_plan/merge_scan.rs index 262e391b55..27a0d9196b 100644 --- a/src/query/src/dist_plan/merge_scan.rs +++ b/src/query/src/dist_plan/merge_scan.rs @@ -22,7 +22,7 @@ use async_stream::stream; use common_catalog::parse_catalog_and_schema_from_db_string; use common_plugins::GREPTIME_EXEC_READ_COST; use common_query::request::QueryRequest; -use common_recordbatch::adapter::RecordBatchMetrics; +use common_recordbatch::adapter::{RecordBatchMetrics, region_scan_output_bytes}; use common_telemetry::tracing_context::TracingContext; use datafusion::execution::{SessionState, TaskContext}; use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; @@ -48,7 +48,6 @@ use meter_macros::read_meter; use session::context::QueryContextRef; use store_api::metrics::{REGION_QUERY_CPU_TIME, REGION_QUERY_SCANNED_BYTES}; use store_api::storage::RegionId; -use table::table::scan::REGION_SCAN_EXEC_NAME; use table::table_name::TableName; use tokio::time::Instant; use tracing::{Instrument, Span}; @@ -915,21 +914,10 @@ impl DisplayAs for MergeScanExec { } } -/// Extract total `output_bytes` from [`RegionScanExec`] plan nodes. -fn scan_output_bytes(metrics: &RecordBatchMetrics) -> usize { - metrics - .plan_metrics - .iter() - .filter(|pm| pm.plan_name == REGION_SCAN_EXEC_NAME) - .flat_map(|pm| &pm.metrics) - .filter_map(|(name, value)| (name == "output_bytes").then_some(*value)) - .sum() -} - fn region_scan_load(metrics: &RecordBatchMetrics) -> ReadItem { ReadItem { cpu_time: metrics.elapsed_compute as u64, - table_scan: scan_output_bytes(metrics) as u64, + table_scan: region_scan_output_bytes(metrics) as u64, } } @@ -952,6 +940,9 @@ fn query_load_region_id(default_region_id: RegionId, metrics: &RecordBatchMetric impl Drop for MergeScanExec { fn drop(&mut self) { + // Per-region Prometheus metrics can have high cardinality, so they are + // controlled by `enable_per_region_metrics`. Region-owned counters for + // heartbeat reporting are updated on datanodes when query metrics resolve. if !self.enable_per_region_metrics { return; } @@ -1030,6 +1021,7 @@ mod tests { use session::ReadPreference; use session::context::QueryContext; use session::query_id::QueryId; + use table::table::scan::REGION_SCAN_EXEC_NAME; use table::table_name::TableName; use uuid::Uuid; @@ -1274,7 +1266,7 @@ mod tests { ..Default::default() }; - assert_eq!(scan_output_bytes(&metrics), 42); + assert_eq!(region_scan_output_bytes(&metrics), 42); } #[test] @@ -1289,7 +1281,7 @@ mod tests { ..Default::default() }; - assert_eq!(scan_output_bytes(&metrics), 0); + assert_eq!(region_scan_output_bytes(&metrics), 0); } #[test] @@ -1312,7 +1304,7 @@ mod tests { ..Default::default() }; - assert_eq!(scan_output_bytes(&metrics), 60); + assert_eq!(region_scan_output_bytes(&metrics), 60); } #[test] diff --git a/src/standalone/src/information_extension.rs b/src/standalone/src/information_extension.rs index a7de19e48a..c8be92a6f0 100644 --- a/src/standalone/src/information_extension.rs +++ b/src/standalone/src/information_extension.rs @@ -137,6 +137,8 @@ impl InformationExtension for StandaloneInformationExtension { data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id, metadata_topic_latest_entry_id: region_stat.metadata_topic_latest_entry_id, written_bytes: region_stat.written_bytes, + query_cpu_time: region_stat.query_cpu_time, + query_scanned_bytes: region_stat.query_scanned_bytes, } }) .collect::>(); diff --git a/src/store-api/src/region_engine.rs b/src/store-api/src/region_engine.rs index ce3d1bf1a8..d2d7410257 100644 --- a/src/store-api/src/region_engine.rs +++ b/src/store-api/src/region_engine.rs @@ -23,6 +23,7 @@ use api::greptime_proto::v1::meta::{GrantedRegion as PbGrantedRegion, RegionRole use api::region::RegionResponse; use async_trait::async_trait; use common_error::ext::BoxedError; +use common_recordbatch::adapter::RegionQueryStatCounters; use common_recordbatch::{EmptyRecordBatchStream, QueryMemoryTracker, SendableRecordBatchStream}; use common_time::Timestamp; use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet; @@ -316,6 +317,8 @@ pub struct ScannerProperties { /// Region id that should receive query-load metrics for this scanner. query_load_region_id: Option, + /// Counters that should receive query-load metrics for this scanner. + query_stat_counters: Option, } impl ScannerProperties { @@ -341,6 +344,7 @@ impl ScannerProperties { target_partitions: 0, logical_region: false, query_load_region_id: None, + query_stat_counters: None, } } @@ -394,10 +398,20 @@ impl ScannerProperties { self.query_load_region_id } + /// Returns the counters that should receive query-load metrics. + pub fn query_stat_counters(&self) -> Option { + self.query_stat_counters.clone() + } + /// Sets the region id that should receive query-load metrics. pub fn set_query_load_region_id(&mut self, region_id: RegionId) { self.query_load_region_id = Some(region_id); } + + /// Sets the counters that should receive query-load metrics. + pub fn set_query_stat_counters(&mut self, counters: RegionQueryStatCounters) { + self.query_stat_counters = Some(counters); + } } /// Request to override the scanner properties. @@ -531,6 +545,14 @@ pub struct RegionStatistic { #[serde(default)] /// The total bytes written of the region since region opened. pub written_bytes: u64, + /// The total query CPU time of the region since region opened. + /// + /// Unit: nanoseconds. + #[serde(default)] + pub query_cpu_time: u64, + /// The total scanned bytes of the region since region opened. + #[serde(default)] + pub query_scanned_bytes: u64, /// The latest entry id of the region's remote WAL since last flush. /// For metric engine, there're two latest entry ids, one for data and one for metadata. /// TODO(weny): remove this two fields and use single instead. @@ -684,6 +706,38 @@ impl RegionStatistic { } } +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + #[test] + fn region_statistic_deserializes_without_query_stats() { + let statistic: RegionStatistic = serde_json::from_value(json!({ + "num_rows": 1, + "memtable_size": 2, + "wal_size": 3, + "manifest_size": 4, + "sst_size": 5, + "sst_num": 6, + "index_size": 7, + "manifest": { + "Mito": { + "manifest_version": 8, + "flushed_entry_id": 9, + "file_removed_cnt": 10 + } + }, + "written_bytes": 11 + })) + .unwrap(); + + assert_eq!(statistic.query_cpu_time, 0); + assert_eq!(statistic.query_scanned_bytes, 0); + } +} + /// Request to sync the region from a manifest or a region. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub enum SyncRegionFromRequest { diff --git a/src/table/src/table/scan.rs b/src/table/src/table/scan.rs index a5f06a40d3..825d930a6f 100644 --- a/src/table/src/table/scan.rs +++ b/src/table/src/table/scan.rs @@ -19,6 +19,7 @@ use std::task::{Context, Poll}; use std::time::Instant; use common_error::ext::BoxedError; +use common_recordbatch::adapter::RegionQueryStatCounters; use common_recordbatch::{ DfRecordBatch, DfSendableRecordBatchStream, MemoryTrackedStream, QueryMemoryTracker, SendableRecordBatchStream, @@ -369,6 +370,14 @@ impl RegionScanExec { .query_load_region_id() .map(|region_id| region_id.as_u64()) } + + pub fn query_stat_counters(&self) -> Option { + self.scanner + .lock() + .unwrap() + .properties() + .query_stat_counters() + } } impl ExecutionPlan for RegionScanExec { diff --git a/tests/cases/standalone/common/information_schema/region_statistics.result b/tests/cases/standalone/common/information_schema/region_statistics.result index 8080a61741..e2b89df6cf 100644 --- a/tests/cases/standalone/common/information_schema/region_statistics.result +++ b/tests/cases/standalone/common/information_schema/region_statistics.result @@ -25,15 +25,16 @@ Affected Rows: 3 -- SQLNESS REPLACE (\s+\d+\s+) -- For regions using different WAL implementations, the manifest size may vary. -- The remote WAL implementation additionally stores a flushed entry ID when creating the manifest. -SELECT SUM(region_rows),SUM(written_bytes_since_open), SUM(memtable_size), SUM(sst_size), SUM(index_size) +SELECT SUM(region_rows), SUM(written_bytes_since_open), SUM(query_cpu_time_millis), + SUM(query_scanned_bytes), SUM(memtable_size), SUM(sst_size), SUM(index_size) FROM INFORMATION_SCHEMA.REGION_STATISTICS WHERE table_id IN (SELECT TABLE_ID FROM INFORMATION_SCHEMA.TABLES WHERE table_name = 'test' and table_schema = 'public'); -+-------------------------------------------------------+--------------------------------------------------------------------+---------------------------------------------------------+----------------------------------------------------+------------------------------------------------------+ -| sum(information_schema.region_statistics.region_rows) | sum(information_schema.region_statistics.written_bytes_since_open) | sum(information_schema.region_statistics.memtable_size) | sum(information_schema.region_statistics.sst_size) | sum(information_schema.region_statistics.index_size) | -+-------------------------------------------------------+--------------------------------------------------------------------+---------------------------------------------------------+----------------------------------------------------+------------------------------------------------------+ -|||||| -+-------------------------------------------------------+--------------------------------------------------------------------+---------------------------------------------------------+----------------------------------------------------+------------------------------------------------------+ ++-------------------------------------------------------+--------------------------------------------------------------------+-----------------------------------------------------------------+---------------------------------------------------------------+---------------------------------------------------------+----------------------------------------------------+------------------------------------------------------+ +| sum(information_schema.region_statistics.region_rows) | sum(information_schema.region_statistics.written_bytes_since_open) | sum(information_schema.region_statistics.query_cpu_time_millis) | sum(information_schema.region_statistics.query_scanned_bytes) | sum(information_schema.region_statistics.memtable_size) | sum(information_schema.region_statistics.sst_size) | sum(information_schema.region_statistics.index_size) | ++-------------------------------------------------------+--------------------------------------------------------------------+-----------------------------------------------------------------+---------------------------------------------------------------+---------------------------------------------------------+----------------------------------------------------+------------------------------------------------------+ +|||||||| ++-------------------------------------------------------+--------------------------------------------------------------------+-----------------------------------------------------------------+---------------------------------------------------------------+---------------------------------------------------------+----------------------------------------------------+------------------------------------------------------+ -- SQLNESS REPLACE (\s+\d+\s+) SELECT data_length, index_length, avg_row_length, table_rows FROM INFORMATION_SCHEMA.TABLES WHERE table_name = 'test'; diff --git a/tests/cases/standalone/common/information_schema/region_statistics.sql b/tests/cases/standalone/common/information_schema/region_statistics.sql index 3decb8130e..b5202d051c 100644 --- a/tests/cases/standalone/common/information_schema/region_statistics.sql +++ b/tests/cases/standalone/common/information_schema/region_statistics.sql @@ -20,7 +20,8 @@ INSERT INTO test VALUES -- SQLNESS REPLACE (\s+\d+\s+) -- For regions using different WAL implementations, the manifest size may vary. -- The remote WAL implementation additionally stores a flushed entry ID when creating the manifest. -SELECT SUM(region_rows),SUM(written_bytes_since_open), SUM(memtable_size), SUM(sst_size), SUM(index_size) +SELECT SUM(region_rows), SUM(written_bytes_since_open), SUM(query_cpu_time_millis), + SUM(query_scanned_bytes), SUM(memtable_size), SUM(sst_size), SUM(index_size) FROM INFORMATION_SCHEMA.REGION_STATISTICS WHERE table_id IN (SELECT TABLE_ID FROM INFORMATION_SCHEMA.TABLES WHERE table_name = 'test' and table_schema = 'public'); diff --git a/tests/cases/standalone/common/system/information_schema.result b/tests/cases/standalone/common/system/information_schema.result index cf141afe81..5cb8aa9fbd 100644 --- a/tests/cases/standalone/common/system/information_schema.result +++ b/tests/cases/standalone/common/system/information_schema.result @@ -343,17 +343,19 @@ select * from information_schema.columns order by table_schema, table_name, colu | greptime | information_schema | region_peers | table_catalog | 1 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | | greptime | information_schema | region_peers | table_name | 3 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | | greptime | information_schema | region_peers | table_schema | 2 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | | -| greptime | information_schema | region_statistics | disk_size | 6 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | -| greptime | information_schema | region_statistics | engine | 12 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | YES | string | | | -| greptime | information_schema | region_statistics | index_size | 11 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | -| greptime | information_schema | region_statistics | manifest_size | 8 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | -| greptime | information_schema | region_statistics | memtable_size | 7 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | +| greptime | information_schema | region_statistics | disk_size | 8 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | +| greptime | information_schema | region_statistics | engine | 14 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | YES | string | | | +| greptime | information_schema | region_statistics | index_size | 13 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | +| greptime | information_schema | region_statistics | manifest_size | 10 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | +| greptime | information_schema | region_statistics | memtable_size | 9 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | +| greptime | information_schema | region_statistics | query_cpu_time_millis | 6 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | +| greptime | information_schema | region_statistics | query_scanned_bytes | 7 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | | greptime | information_schema | region_statistics | region_id | 1 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | NO | bigint unsigned | | | | greptime | information_schema | region_statistics | region_number | 3 | | | 10 | 0 | | | | | | select,insert | | UInt32 | int unsigned | FIELD | | NO | int unsigned | | | -| greptime | information_schema | region_statistics | region_role | 13 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | YES | string | | | +| greptime | information_schema | region_statistics | region_role | 15 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | YES | string | | | | greptime | information_schema | region_statistics | region_rows | 4 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | -| greptime | information_schema | region_statistics | sst_num | 10 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | -| greptime | information_schema | region_statistics | sst_size | 9 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | +| greptime | information_schema | region_statistics | sst_num | 12 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | +| greptime | information_schema | region_statistics | sst_size | 11 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | | greptime | information_schema | region_statistics | table_id | 2 | | | 10 | 0 | | | | | | select,insert | | UInt32 | int unsigned | FIELD | | NO | int unsigned | | | | greptime | information_schema | region_statistics | written_bytes_since_open | 5 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | | | greptime | information_schema | routines | character_maximum_length | 7 | | | 19 | 0 | | | | | | select,insert | | Int64 | bigint | FIELD | | NO | bigint | | |