rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-07-06 05:50:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: report region query stats in heartbeat (#8401)

Browse Source 
* feat: report region read load in heartbeat

Signed-off-by: WenyXu <wenymedia@gmail.com>

* feat: expose region query stats in information schema

Signed-off-by: WenyXu <wenymedia@gmail.com>

* chore: update sqlness result

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix: record region query stats on stream drop

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix: keep region query cpu stats in nanoseconds

Signed-off-by: WenyXu <wenymedia@gmail.com>

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>
This commit is contained in:
Weny Xu
2026-07-03 14:11:57 +08:00
committed by GitHub
parent 33902eef3e
commit 2dc6addb66
28 changed files with 583 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
src/catalog/src/system_schema/information_schema/region_statistics.rs
View File

@@ -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()),

52
src/common/meta/src/datanode.rs
View File

@@ -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![] };

130
src/common/recordbatch/src/adapter.rs
View File

@@ -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<Box<dyn std::future::Future<Output = Result<SendableRecordBatchStream>> + 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<u64>,
query_stat_counters: Option<RegionQueryStatCounters>,
/// 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<AtomicU64>,
/// The total scanned bytes.
pub query_scanned_bytes: Arc<AtomicU64>,
}
/// 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<u64>) {
self.query_load_region_id = region_id;
}
pub fn set_query_stat_counters(&mut self, counters: Option<RegionQueryStatCounters>) {
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::<datafusion::error::Result<DfRecordBatch>>(),
),
);
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!({

2
src/meta-srv/src/gc/handler.rs
View File

@@ -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,
}

2
src/meta-srv/src/gc/mock.rs
View File

@@ -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,
}
}

2
src/meta-srv/src/handler/collect_leader_region_handler.rs
View File

@@ -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,
}
}

2
src/meta-srv/src/handler/failure_handler.rs
View File

@@ -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 {

2
src/meta-srv/src/handler/persist_stats_handler.rs
View File

@@ -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,
}

2
src/meta-srv/src/handler/region_lease_handler.rs
View File

@@ -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,
}
}

6
src/meta-srv/src/selector/weight_compute.rs
View File

@@ -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()
}

68
src/metric-engine/src/utils.rs
View File

@@ -48,24 +48,9 @@ pub fn get_region_statistic(mito: &MitoEngine, region_id: RegionId) -> Option<Re
let data_stat = mito.region_statistic(data_region_id);
match (&metadata_stat, &data_stat) {
(Some(metadata_stat), Some(data_stat)) => 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<Re
}
}
fn merge_region_statistic(
metadata_stat: &RegionStatistic,
data_stat: &RegionStatistic,
) -> 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);
}
}

1
src/mito2/src/engine.rs
View File

@@ -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())

4
src/mito2/src/engine/basic_test.rs
View File

@@ -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<LogStoreFactory>, 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();

25
src/mito2/src/read/scan_region.rs
View File

@@ -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<RegionQueryStatCounters>,
#[cfg(feature = "enterprise")]
extension_range_provider: Option<BoxedExtensionRangeProvider>,
}
@@ -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<SequenceNumber>,
/// Whether this scan is for compaction.
pub(crate) compaction: bool,
/// Counters that should receive query-load metrics.
pub(crate) query_stat_counters: Option<RegionQueryStatCounters>,
#[cfg(feature = "enterprise")]
extension_ranges: Vec<BoxedExtensionRange>,
}
@@ -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<RegionQueryStatCounters>,
) -> 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 {

3
src/mito2/src/read/seq_scan.rs
View File

@@ -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()];

3
src/mito2/src/read/series_scan.rs
View File

@@ -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()];

3
src/mito2/src/read/unordered_scan.rs
View File

@@ -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()];

49
src/mito2/src/region.rs
View File

@@ -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<AtomicU64>,
/// 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<AtomicU64>,
/// The total query CPU time of the region in nanoseconds.
pub(crate) query_cpu_time: Arc<AtomicU64>,
/// The total scanned bytes of the region.
pub(crate) query_scanned_bytes: Arc<AtomicU64>,
}
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<MitoRegion>;
#[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<Option<P
#[cfg(test)]
mod tests {
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use common_datasource::compression::CompressionType;
use common_test_util::temp_dir::create_temp_dir;
@@ -1766,7 +1801,7 @@ mod tests {
};
use crate::manifest::manager::{RegionManifestManager, RegionManifestOptions};
use crate::region::{
ManifestContext, ManifestStats, MitoRegion, RegionLeaderState, RegionRoleState,
ManifestContext, ManifestStats, MitoRegion, RegionLeaderState, RegionRoleState, RegionStats,
};
use crate::sst::FormatType;
use crate::sst::index::intermediate::IntermediateManager;
@@ -1836,7 +1871,7 @@ mod tests {
last_schedule_compaction_millis: Default::default(),
time_provider: Arc::new(StdTimeProvider),
topic_latest_entry_id: Default::default(),
written_bytes: Arc::new(AtomicU64::new(0)),
region_stats: RegionStats::new(),
stats: ManifestStats::default(),
}
}
@@ -2195,7 +2230,7 @@ mod tests {
last_schedule_compaction_millis: Default::default(),
time_provider: Arc::new(StdTimeProvider),
topic_latest_entry_id: Default::default(),
written_bytes: Arc::new(AtomicU64::new(0)),
region_stats: RegionStats::new(),
stats: ManifestStats::default(),
};

5
src/mito2/src/region/opener.rs
View File

@@ -62,6 +62,7 @@ use crate::region::options::RegionOptions;
use crate::region::version::{VersionBuilder, VersionControl, VersionControlRef};
use crate::region::{
ManifestContext, ManifestStats, MitoRegion, MitoRegionRef, RegionLeaderState, RegionRoleState,
RegionStats,
};
use crate::region_write_ctx::RegionWriteCtx;
use crate::request::OptionOutputTx;
@@ -397,7 +398,7 @@ impl RegionOpener {
last_schedule_compaction_millis: AtomicI64::new(now),
time_provider: self.time_provider.clone(),
topic_latest_entry_id: AtomicU64::new(flushed_entry_id),
written_bytes: Arc::new(AtomicU64::new(0)),
region_stats: RegionStats::new(),
stats: self.stats,
}))
}
@@ -642,7 +643,7 @@ impl RegionOpener {
last_schedule_compaction_millis: AtomicI64::new(now),
time_provider: self.time_provider.clone(),
topic_latest_entry_id: AtomicU64::new(topic_latest_entry_id),
written_bytes: Arc::new(AtomicU64::new(0)),
region_stats: RegionStats::new(),
stats: self.stats.clone(),
};

4
src/mito2/src/worker/handle_write.rs
View File

@@ -328,7 +328,7 @@ impl<S> RegionWorkerLoop<S> {
region.region_id,
&region.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<S> RegionWorkerLoop<S> {
region.region_id,
&region.version_control,
region.provider.clone(),
Some(region.written_bytes.clone()),
Some(region.region_stats.written_bytes.clone()),
);
e.insert(region_ctx);

127
src/query/src/datafusion.rs
View File

@@ -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<dyn ExecutionPlan>) -> Option<u64> {
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<dyn ExecutionPlan>) -> Option<RegionQueryStatCounters> {
let mut counters: Option<RegionQueryStatCounters> = 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::<RegionScanExec>()
&& 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<QueryEngineState>,
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<RegionId>,
) -> Arc<dyn ExecutionPlan> {
build_region_scan(query_load_region_id, None)
}
fn build_query_stat_counter_region_scan(
counters: RegionQueryStatCounters,
) -> Arc<dyn ExecutionPlan> {
build_region_scan(None, Some(counters))
}
fn build_region_scan(
query_load_region_id: Option<RegionId>,
query_stat_counters: Option<RegionQueryStatCounters>,
) -> Arc<dyn ExecutionPlan> {
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<dyn PhysicalExpr>,
Arc::new(Column::new("ts", 0)) as Arc<dyn PhysicalExpr>,
)];
let plan: Arc<dyn ExecutionPlan> = 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<dyn PhysicalExpr>,
Arc::new(Column::new("ts", 0)) as Arc<dyn PhysicalExpr>,
)];
let plan: Arc<dyn ExecutionPlan> = 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 {

26
src/query/src/dist_plan/merge_scan.rs
View File

@@ -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]

2
src/standalone/src/information_extension.rs
View File

@@ -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::<Vec<_>>();

54
src/store-api/src/region_engine.rs
View File

@@ -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<RegionId>,
/// Counters that should receive query-load metrics for this scanner.
query_stat_counters: Option<RegionQueryStatCounters>,
}
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<RegionQueryStatCounters> {
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 {

9
src/table/src/table/scan.rs
View File

@@ -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<RegionQueryStatCounters> {
self.scanner
.lock()
.unwrap()
.properties()
.query_stat_counters()
}
}
impl ExecutionPlan for RegionScanExec {

13
tests/cases/standalone/common/information_schema/region_statistics.result
View File

@@ -25,15 +25,16 @@ Affected Rows: 3
-- SQLNESS REPLACE (\s+\d+\s+) <NUM>
-- 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) |
+-------------------------------------------------------+--------------------------------------------------------------------+---------------------------------------------------------+----------------------------------------------------+------------------------------------------------------+
|<NUM>|<NUM>|<NUM>|<NUM>|<NUM>|
+-------------------------------------------------------+--------------------------------------------------------------------+---------------------------------------------------------+----------------------------------------------------+------------------------------------------------------+
+-------------------------------------------------------+--------------------------------------------------------------------+-----------------------------------------------------------------+---------------------------------------------------------------+---------------------------------------------------------+----------------------------------------------------+------------------------------------------------------+
| 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) |
+-------------------------------------------------------+--------------------------------------------------------------------+-----------------------------------------------------------------+---------------------------------------------------------------+---------------------------------------------------------+----------------------------------------------------+------------------------------------------------------+
|<NUM>|<NUM>|<NUM>|<NUM>|<NUM>|<NUM>|<NUM>|
+-------------------------------------------------------+--------------------------------------------------------------------+-----------------------------------------------------------------+---------------------------------------------------------------+---------------------------------------------------------+----------------------------------------------------+------------------------------------------------------+
-- SQLNESS REPLACE (\s+\d+\s+) <NUM>
SELECT data_length, index_length, avg_row_length, table_rows FROM INFORMATION_SCHEMA.TABLES WHERE table_name = 'test';

3
tests/cases/standalone/common/information_schema/region_statistics.sql
View File

@@ -20,7 +20,8 @@ INSERT INTO test VALUES
-- SQLNESS REPLACE (\s+\d+\s+) <NUM>
-- 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');

18
tests/cases/standalone/common/system/information_schema.result
View File

@@ -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 | | |