diff --git a/src/catalog/src/system_schema/information_schema.rs b/src/catalog/src/system_schema/information_schema.rs index a51810e048..d0539a062c 100644 --- a/src/catalog/src/system_schema/information_schema.rs +++ b/src/catalog/src/system_schema/information_schema.rs @@ -14,6 +14,7 @@ mod cluster_info; pub mod columns; +pub mod flow_statistics; pub mod flows; mod information_memory_table; pub mod key_column_usage; @@ -73,6 +74,7 @@ use crate::CatalogManager; use crate::error::{Error, Result}; use crate::process_manager::ProcessManagerRef; use crate::system_schema::information_schema::cluster_info::InformationSchemaClusterInfo; +use crate::system_schema::information_schema::flow_statistics::InformationSchemaFlowStatistics; use crate::system_schema::information_schema::flows::InformationSchemaFlows; use crate::system_schema::information_schema::information_memory_table::get_schema_columns; use crate::system_schema::information_schema::key_column_usage::InformationSchemaKeyColumnUsage; @@ -271,6 +273,11 @@ impl SystemSchemaProviderInner for InformationSchemaProvider { self.catalog_manager.clone(), self.flow_metadata_manager.clone(), )) as _), + FLOW_STATISTICS => Some(Arc::new(InformationSchemaFlowStatistics::new( + self.catalog_name.clone(), + self.catalog_manager.clone(), + self.flow_metadata_manager.clone(), + )) as _), PROCEDURE_INFO => Some( Arc::new(procedure_info::InformationSchemaProcedureInfo::new( self.catalog_manager.clone(), @@ -406,6 +413,10 @@ impl InformationSchemaProvider { self.build_table(STATISTICS).unwrap(), ); tables.insert(FLOWS.to_string(), self.build_table(FLOWS).unwrap()); + tables.insert( + FLOW_STATISTICS.to_string(), + self.build_table(FLOW_STATISTICS).unwrap(), + ); #[cfg(feature = "enterprise")] tables.insert( RECYCLE_BIN.to_string(), diff --git a/src/catalog/src/system_schema/information_schema/flow_statistics.rs b/src/catalog/src/system_schema/information_schema/flow_statistics.rs new file mode 100644 index 0000000000..6163461eda --- /dev/null +++ b/src/catalog/src/system_schema/information_schema/flow_statistics.rs @@ -0,0 +1,285 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::sync::{Arc, Weak}; + +use common_catalog::consts::INFORMATION_SCHEMA_FLOW_STATISTICS_TABLE_ID; +use common_error::ext::BoxedError; +use common_meta::key::FlowId; +use common_meta::key::flow::FlowMetadataManager; +use common_meta::key::flow::flow_state::FlowStat; +use common_recordbatch::adapter::RecordBatchStreamAdapter; +use common_recordbatch::{DfSendableRecordBatchStream, RecordBatch, SendableRecordBatchStream}; +use common_time::util::current_time_millis; +use datafusion::execution::TaskContext; +use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter; +use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream; +use datatypes::prelude::ConcreteDataType as CDT; +use datatypes::scalars::ScalarVectorBuilder; +use datatypes::schema::{ColumnSchema, Schema, SchemaRef}; +use datatypes::timestamp::TimestampMillisecond; +use datatypes::value::Value; +use datatypes::vectors::{ + Int64VectorBuilder, StringVectorBuilder, TimestampMillisecondVectorBuilder, + UInt32VectorBuilder, UInt64VectorBuilder, VectorRef, +}; +use futures::TryStreamExt; +use snafu::ResultExt; +use store_api::storage::{ScanRequest, TableId}; + +use crate::CatalogManager; +use crate::error::{CreateRecordBatchSnafu, InternalSnafu, ListFlowsSnafu, Result}; +use crate::information_schema::{FLOW_STATISTICS, Predicates}; +use crate::system_schema::information_schema::InformationTable; +use crate::system_schema::utils; + +const INIT_CAPACITY: usize = 42; + +// rows of information_schema.flow_statistics +pub const FLOW_ID: &str = "flow_id"; +pub const FLOW_NAME: &str = "flow_name"; +pub const START_TIME: &str = "start_time"; +pub const LAST_EXECUTION_TIME: &str = "last_execution_time"; +pub const UPTIME_SECONDS: &str = "uptime_seconds"; +pub const STATE_SIZE: &str = "state_size"; + +/// The `information_schema.flow_statistics` provides runtime statistics about flows. +#[derive(Debug)] +pub(super) struct InformationSchemaFlowStatistics { + schema: SchemaRef, + catalog_name: String, + catalog_manager: Weak, + flow_metadata_manager: Arc, +} + +impl InformationSchemaFlowStatistics { + pub(super) fn new( + catalog_name: String, + catalog_manager: Weak, + flow_metadata_manager: Arc, + ) -> Self { + Self { + schema: Self::schema(), + catalog_name, + catalog_manager, + flow_metadata_manager, + } + } + + pub(crate) fn schema() -> SchemaRef { + Arc::new(Schema::new( + vec![ + (FLOW_ID, CDT::uint32_datatype(), false), + (FLOW_NAME, CDT::string_datatype(), false), + (START_TIME, CDT::timestamp_millisecond_datatype(), true), + ( + LAST_EXECUTION_TIME, + CDT::timestamp_millisecond_datatype(), + true, + ), + (UPTIME_SECONDS, CDT::int64_datatype(), true), + (STATE_SIZE, CDT::uint64_datatype(), true), + ] + .into_iter() + .map(|(name, ty, nullable)| ColumnSchema::new(name, ty, nullable)) + .collect(), + )) + } + + fn builder(&self) -> InformationSchemaFlowStatisticsBuilder { + InformationSchemaFlowStatisticsBuilder::new( + self.schema.clone(), + self.catalog_name.clone(), + self.catalog_manager.clone(), + &self.flow_metadata_manager, + ) + } +} + +impl InformationTable for InformationSchemaFlowStatistics { + fn table_id(&self) -> TableId { + INFORMATION_SCHEMA_FLOW_STATISTICS_TABLE_ID + } + + fn table_name(&self) -> &'static str { + FLOW_STATISTICS + } + + fn schema(&self) -> SchemaRef { + self.schema.clone() + } + + fn to_stream(&self, request: ScanRequest) -> Result { + let schema = self.schema.arrow_schema().clone(); + let mut builder = self.builder(); + let stream = Box::pin(DfRecordBatchStreamAdapter::new( + schema, + futures::stream::once(async move { + builder + .make_flow_statistics(Some(request)) + .await + .map(|x| x.into_df_record_batch()) + .map_err(|err| datafusion::error::DataFusionError::External(Box::new(err))) + }), + )); + Ok(Box::pin( + RecordBatchStreamAdapter::try_new(stream) + .map_err(BoxedError::new) + .context(InternalSnafu)?, + )) + } +} + +/// Builds the `information_schema.flow_statistics` table row by row. +struct InformationSchemaFlowStatisticsBuilder { + schema: SchemaRef, + catalog_name: String, + catalog_manager: Weak, + flow_metadata_manager: Arc, + + flow_ids: UInt32VectorBuilder, + flow_names: StringVectorBuilder, + start_times: TimestampMillisecondVectorBuilder, + last_execution_times: TimestampMillisecondVectorBuilder, + uptime_seconds: Int64VectorBuilder, + state_sizes: UInt64VectorBuilder, +} + +impl InformationSchemaFlowStatisticsBuilder { + fn new( + schema: SchemaRef, + catalog_name: String, + catalog_manager: Weak, + flow_metadata_manager: &Arc, + ) -> Self { + Self { + schema, + catalog_name, + catalog_manager, + flow_metadata_manager: flow_metadata_manager.clone(), + + flow_ids: UInt32VectorBuilder::with_capacity(INIT_CAPACITY), + flow_names: StringVectorBuilder::with_capacity(INIT_CAPACITY), + start_times: TimestampMillisecondVectorBuilder::with_capacity(INIT_CAPACITY), + last_execution_times: TimestampMillisecondVectorBuilder::with_capacity(INIT_CAPACITY), + uptime_seconds: Int64VectorBuilder::with_capacity(INIT_CAPACITY), + state_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY), + } + } + + /// Construct the `information_schema.flow_statistics` virtual table. + async fn make_flow_statistics(&mut self, request: Option) -> Result { + let catalog_name = self.catalog_name.clone(); + let predicates = Predicates::from_scan_request(&request); + + let flow_info_manager = self.flow_metadata_manager.clone(); + + let mut stream = flow_info_manager + .flow_name_manager() + .flow_names(&catalog_name) + .await; + + let flow_stat = { + let information_extension = utils::information_extension(&self.catalog_manager)?; + information_extension.flow_stats().await? + }; + + let now = current_time_millis(); + + while let Some((flow_name, flow_id)) = stream + .try_next() + .await + .map_err(BoxedError::new) + .context(ListFlowsSnafu { + catalog: &catalog_name, + })? + { + self.add_flow_statistic(&predicates, flow_id.flow_id(), &flow_name, &flow_stat, now); + } + + self.finish() + } + + fn add_flow_statistic( + &mut self, + predicates: &Predicates, + flow_id: FlowId, + flow_name: &str, + flow_stat: &Option, + now: i64, + ) { + let row = [ + (FLOW_ID, &Value::from(flow_id)), + (FLOW_NAME, &Value::from(flow_name.to_string())), + ]; + if !predicates.eval(&row) { + return; + } + + let start_time = flow_stat + .as_ref() + .and_then(|stat| stat.start_time_map.get(&flow_id).copied()); + + self.flow_ids.push(Some(flow_id)); + self.flow_names.push(Some(flow_name)); + self.start_times + .push(start_time.map(TimestampMillisecond::new)); + self.last_execution_times + .push(flow_stat.as_ref().and_then(|stat| { + stat.last_exec_time_map + .get(&flow_id) + .map(|v| TimestampMillisecond::new(*v)) + })); + self.uptime_seconds + .push(start_time.map(|start| ((now - start) / 1000).max(0))); + self.state_sizes.push( + flow_stat + .as_ref() + .and_then(|stat| stat.state_size.get(&flow_id).map(|v| *v as u64)), + ); + } + + fn finish(&mut self) -> Result { + let columns: Vec = vec![ + Arc::new(self.flow_ids.finish()), + Arc::new(self.flow_names.finish()), + Arc::new(self.start_times.finish()), + Arc::new(self.last_execution_times.finish()), + Arc::new(self.uptime_seconds.finish()), + Arc::new(self.state_sizes.finish()), + ]; + RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu) + } +} + +impl DfPartitionStream for InformationSchemaFlowStatistics { + fn schema(&self) -> &arrow_schema::SchemaRef { + self.schema.arrow_schema() + } + + fn execute(&self, _: Arc) -> DfSendableRecordBatchStream { + let schema: Arc = self.schema.arrow_schema().clone(); + let mut builder = self.builder(); + Box::pin(DfRecordBatchStreamAdapter::new( + schema, + futures::stream::once(async move { + builder + .make_flow_statistics(None) + .await + .map(|x| x.into_df_record_batch()) + .map_err(Into::into) + }), + )) + } +} diff --git a/src/catalog/src/system_schema/information_schema/table_names.rs b/src/catalog/src/system_schema/information_schema/table_names.rs index 5fa1aa517c..f161fecb54 100644 --- a/src/catalog/src/system_schema/information_schema/table_names.rs +++ b/src/catalog/src/system_schema/information_schema/table_names.rs @@ -44,6 +44,7 @@ pub const TABLE_CONSTRAINTS: &str = "table_constraints"; pub const CLUSTER_INFO: &str = "cluster_info"; pub const VIEWS: &str = "views"; pub const FLOWS: &str = "flows"; +pub const FLOW_STATISTICS: &str = "flow_statistics"; pub const PROCEDURE_INFO: &str = "procedure_info"; pub const REGION_INFO: &str = "region_info"; pub const REGION_STATISTICS: &str = "region_statistics"; diff --git a/src/cli/src/metadata/control/put/key.rs b/src/cli/src/metadata/control/put/key.rs index 7becfd72dc..49d4b61e2f 100644 --- a/src/cli/src/metadata/control/put/key.rs +++ b/src/cli/src/metadata/control/put/key.rs @@ -345,9 +345,7 @@ mod tests { #[test] fn test_validate_exact_flow_state_key() { - let value = FlowStateValue::new(BTreeMap::new(), BTreeMap::new()) - .try_as_raw_value() - .unwrap(); + let value = FlowStateValue::default().try_as_raw_value().unwrap(); validate_metadata_value(&flow_state_full_key(), &value).unwrap(); } diff --git a/src/common/catalog/src/consts.rs b/src/common/catalog/src/consts.rs index adbd3a8d47..3149f6ccd4 100644 --- a/src/common/catalog/src/consts.rs +++ b/src/common/catalog/src/consts.rs @@ -120,6 +120,8 @@ pub const INFORMATION_SCHEMA_TABLE_SEMANTICS_TABLE_ID: u32 = 42; pub const INFORMATION_SCHEMA_STATISTICS_TABLE_ID: u32 = 43; /// id for information_schema.recycle_bin pub const INFORMATION_SCHEMA_RECYCLE_BIN_TABLE_ID: u32 = 44; +/// id for information_schema.flow_statistics +pub const INFORMATION_SCHEMA_FLOW_STATISTICS_TABLE_ID: u32 = 45; // ----- End of information_schema tables ----- diff --git a/src/common/meta/src/key/flow/flow_state.rs b/src/common/meta/src/key/flow/flow_state.rs index 1b161929fe..77ccd6206a 100644 --- a/src/common/meta/src/key/flow/flow_state.rs +++ b/src/common/meta/src/key/flow/flow_state.rs @@ -93,22 +93,28 @@ impl<'a> MetadataKey<'a, FlowStateKey> for FlowStateKey { } /// The value of flow state size -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] pub struct FlowStateValue { /// For each key, the bytes of the state in memory pub state_size: BTreeMap, /// For each key, the last execution time of flow in unix timestamp milliseconds. pub last_exec_time_map: BTreeMap, + /// For each flow, the time the flow first executed, in unix timestamp milliseconds. + /// TODO(#7987-followup): not yet propagated via the heartbeat wire format in distributed mode. + #[serde(default)] + pub start_time_map: BTreeMap, } impl FlowStateValue { pub fn new( state_size: BTreeMap, last_exec_time_map: BTreeMap, + start_time_map: BTreeMap, ) -> Self { Self { state_size, last_exec_time_map, + start_time_map, } } } @@ -147,12 +153,15 @@ impl FlowStateManager { } /// Flow's state report, send regularly through heartbeat message -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Default)] pub struct FlowStat { /// For each key, the bytes of the state in memory pub state_size: BTreeMap, /// For each key, the last execution time of flow in unix timestamp milliseconds. pub last_exec_time_map: BTreeMap, + /// For each flow, the time the flow first executed, in unix timestamp milliseconds. + /// TODO(#7987-followup): not yet propagated via the heartbeat wire format in distributed mode. + pub start_time_map: BTreeMap, } impl From for FlowStat { @@ -160,6 +169,7 @@ impl From for FlowStat { Self { state_size: value.state_size, last_exec_time_map: value.last_exec_time_map, + start_time_map: value.start_time_map, } } } @@ -169,6 +179,59 @@ impl From for FlowStateValue { Self { state_size: value.state_size, last_exec_time_map: value.last_exec_time_map, + start_time_map: value.start_time_map, } } } + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use crate::key::FlowId; + use crate::key::flow::flow_state::FlowStateValue; + + #[test] + fn test_deserialize_legacy_flow_state_value() { + // Legacy format: only state_size and last_exec_time_map are present, + // without the start_time_map field added in PR #8392. + let legacy_json = + r#"{"state_size":{"1":1024,"2":2048},"last_exec_time_map":{"1":1700000000000}}"#; + let value: FlowStateValue = serde_json::from_str(legacy_json).unwrap(); + + let mut expected_state_size = BTreeMap::new(); + expected_state_size.insert(FlowId::from(1u32), 1024usize); + expected_state_size.insert(FlowId::from(2u32), 2048usize); + assert_eq!(value.state_size, expected_state_size); + + let mut expected_last_exec_time_map = BTreeMap::new(); + expected_last_exec_time_map.insert(FlowId::from(1u32), 1700000000000i64); + assert_eq!(value.last_exec_time_map, expected_last_exec_time_map); + + // serde(default) kicks in: old persisted data must not break, + // and the new field defaults to empty. + assert!(value.start_time_map.is_empty()); + } + + #[test] + fn test_flow_state_value_roundtrip_includes_start_time_map() { + let mut state_size = BTreeMap::new(); + state_size.insert(FlowId::from(1u32), 1024usize); + let mut last_exec_time_map = BTreeMap::new(); + last_exec_time_map.insert(FlowId::from(1u32), 1700000000000i64); + let mut start_time_map = BTreeMap::new(); + start_time_map.insert(FlowId::from(1u32), 1700000000000i64); + + let value = FlowStateValue { + state_size, + last_exec_time_map, + start_time_map, + }; + + let json = serde_json::to_string(&value).unwrap(); + assert!(json.contains("start_time_map")); + + let decoded: FlowStateValue = serde_json::from_str(&json).unwrap(); + assert_eq!(decoded, value); + } +} diff --git a/src/flow/src/adapter/flownode_impl.rs b/src/flow/src/adapter/flownode_impl.rs index 9e3399c182..7e83d5131d 100644 --- a/src/flow/src/adapter/flownode_impl.rs +++ b/src/flow/src/adapter/flownode_impl.rs @@ -174,9 +174,13 @@ impl FlowDualEngine { let mut last_exec_time_map = streaming.last_exec_time_map; last_exec_time_map.extend(batching.last_exec_time_map); + let mut start_time_map = streaming.start_time_map; + start_time_map.extend(batching.start_time_map); + FlowStat { state_size, last_exec_time_map, + start_time_map, } } diff --git a/src/flow/src/adapter/stat.rs b/src/flow/src/adapter/stat.rs index 68f8160c87..9521a24c5c 100644 --- a/src/flow/src/adapter/stat.rs +++ b/src/flow/src/adapter/stat.rs @@ -21,33 +21,27 @@ use crate::engine::FlowStatProvider; impl FlowStatProvider for StreamingEngine { async fn flow_stat(&self) -> FlowStat { - let mut full_report = BTreeMap::new(); + let mut state_size_map = BTreeMap::new(); let mut last_exec_time_map = BTreeMap::new(); + let mut start_time_map = BTreeMap::new(); for worker in self.worker_handles.iter() { - match worker.get_state_size().await { - Ok(state_size) => { - full_report.extend(state_size.into_iter().map(|(k, v)| (k as u32, v))); + match worker.get_full_flow_stat().await { + Ok((sizes, exec_times, start_times)) => { + state_size_map.extend(sizes.into_iter().map(|(k, v)| (k as u32, v))); + last_exec_time_map.extend(exec_times.into_iter().map(|(k, v)| (k as u32, v))); + start_time_map.extend(start_times.into_iter().map(|(k, v)| (k as u32, v))); } Err(err) => { - common_telemetry::error!(err; "Get flow stat size error"); - } - } - - match worker.get_last_exec_time_map().await { - Ok(last_exec_time) => { - last_exec_time_map - .extend(last_exec_time.into_iter().map(|(k, v)| (k as u32, v))); - } - Err(err) => { - common_telemetry::error!(err; "Get last exec time error"); + common_telemetry::error!(err; "Get full flow stat error"); } } } FlowStat { - state_size: full_report, + state_size: state_size_map, last_exec_time_map, + start_time_map, } } } diff --git a/src/flow/src/adapter/worker.rs b/src/flow/src/adapter/worker.rs index 32cc4eb7f5..71fbacb18f 100644 --- a/src/flow/src/adapter/worker.rs +++ b/src/flow/src/adapter/worker.rs @@ -202,30 +202,24 @@ impl WorkerHandle { } } - pub async fn get_state_size(&self) -> Result, Error> { + pub async fn get_full_flow_stat( + &self, + ) -> Result< + ( + BTreeMap, + BTreeMap, + BTreeMap, + ), + Error, + > { let ret = self .itc_client - .call_with_resp(Request::QueryStateSize) + .call_with_resp(Request::QueryFullFlowStat) .await?; - ret.into_query_state_size().map_err(|ret| { + ret.into_query_full_flow_stat().map_err(|ret| { InternalSnafu { reason: format!( - "Flow Node/Worker itc failed, expect Response::QueryStateSize, found {ret:?}" - ), - } - .build() - }) - } - - pub async fn get_last_exec_time_map(&self) -> Result, Error> { - let ret = self - .itc_client - .call_with_resp(Request::QueryLastExecTimeMap) - .await?; - ret.into_query_last_exec_time_map().map_err(|ret| { - InternalSnafu { - reason: format!( - "Flow Node/Worker get_last_exec_time_map failed, expect Response::QueryLastExecTimeMap, found {ret:?}" + "Flow Node/Worker get_full_flow_stat failed, expected Response::QueryFullFlowStat, found {ret:?}" ), } .build() @@ -408,21 +402,24 @@ impl<'s> Worker<'s> { Some(Response::ContainTask { result: ret }) } Request::Shutdown => return Err(()), - Request::QueryStateSize => { - let mut ret = BTreeMap::new(); + Request::QueryFullFlowStat => { + let mut state_size = BTreeMap::new(); + let mut last_exec_time_map = BTreeMap::new(); + let mut start_time_map = BTreeMap::new(); for (flow_id, task_state) in self.task_states.iter() { - ret.insert(*flow_id, task_state.state.get_state_size()); - } - Some(Response::QueryStateSize { result: ret }) - } - Request::QueryLastExecTimeMap => { - let mut ret = BTreeMap::new(); - for (flow_id, task_state) in self.task_states.iter() { - if let Some(last_exec_time) = task_state.state.last_exec_time() { - ret.insert(*flow_id, last_exec_time); + state_size.insert(*flow_id, task_state.state.get_state_size()); + if let Some(t) = task_state.state.last_exec_time() { + last_exec_time_map.insert(*flow_id, t); + } + if let Some(t) = task_state.state.start_time() { + start_time_map.insert(*flow_id, t); } } - Some(Response::QueryLastExecTimeMap { result: ret }) + Some(Response::QueryFullFlowStat { + state_size, + last_exec_time_map, + start_time_map, + }) } }; Ok(ret) @@ -455,8 +452,7 @@ pub enum Request { flow_id: FlowId, }, Shutdown, - QueryStateSize, - QueryLastExecTimeMap, + QueryFullFlowStat, } #[derive(Debug, EnumAsInner)] @@ -472,13 +468,10 @@ enum Response { result: bool, }, RunAvail, - QueryStateSize { - /// each flow tasks' state size - result: BTreeMap, - }, - QueryLastExecTimeMap { - /// each flow tasks' last execution time - result: BTreeMap, + QueryFullFlowStat { + state_size: BTreeMap, + last_exec_time_map: BTreeMap, + start_time_map: BTreeMap, }, } @@ -604,7 +597,8 @@ mod test { ); tx.send(Batch::empty()).unwrap(); handle.run_available(0, true).await.unwrap(); - assert_eq!(handle.get_state_size().await.unwrap().len(), 1); + let (state_size, _, _) = handle.get_full_flow_stat().await.unwrap(); + assert_eq!(state_size.len(), 1); assert_eq!(sink_rx.recv().await.unwrap(), Batch::empty()); drop(handle); worker_thread_handle.join().unwrap(); diff --git a/src/flow/src/batching_mode/engine.rs b/src/flow/src/batching_mode/engine.rs index 3f65761b55..d05eec9251 100644 --- a/src/flow/src/batching_mode/engine.rs +++ b/src/flow/src/batching_mode/engine.rs @@ -414,14 +414,24 @@ impl BatchingEngine { impl FlowStatProvider for BatchingEngine { async fn flow_stat(&self) -> FlowStat { + let runtime = self.runtime.read().await; + let mut last_exec_time_map = BTreeMap::new(); + let mut start_time_map = BTreeMap::new(); + + for (flow_id, task) in runtime.tasks.iter() { + let id = *flow_id as u32; + if let Some(ts) = task.last_execution_time_millis() { + last_exec_time_map.insert(id, ts); + } + if let Some(ts) = task.start_time_millis() { + start_time_map.insert(id, ts); + } + } + FlowStat { state_size: BTreeMap::new(), - last_exec_time_map: self - .get_last_exec_time_map() - .await - .into_iter() - .map(|(flow_id, timestamp)| (flow_id as u32, timestamp)) - .collect(), + last_exec_time_map, + start_time_map, } } } diff --git a/src/flow/src/batching_mode/state.rs b/src/flow/src/batching_mode/state.rs index 225d76a864..59c4f0fe10 100644 --- a/src/flow/src/batching_mode/state.rs +++ b/src/flow/src/batching_mode/state.rs @@ -46,6 +46,8 @@ pub struct TaskState { last_query_duration: Duration, /// Last successful execution time in unix timestamp milliseconds. last_exec_time_millis: Option, + /// First execution time in unix timestamp milliseconds, set once. + start_time_millis: Option, /// Dirty Time windows need to be updated /// mapping of `start -> end` and non-overlapping pub(crate) dirty_time_windows: DirtyTimeWindows, @@ -79,6 +81,7 @@ impl TaskState { last_update_time: Instant::now(), last_query_duration: Duration::from_secs(0), last_exec_time_millis: None, + start_time_millis: None, dirty_time_windows, checkpoint_mode: CheckpointMode::FullSnapshot, pending_fenced_repair: None, @@ -90,8 +93,18 @@ impl TaskState { } } - /// called after last query is done - /// `is_succ` indicate whether the last query is successful + /// Record the first-execution start time. Call this once, just before + /// the first frontend query is dispatched, not after it completes. + pub fn record_start_time_if_first(&mut self) { + if self.start_time_millis.is_none() { + // start_time is recorded just before the first frontend query is dispatched + // (pre-execution), so it may be marginally earlier than the streaming engine's + // start_time which is set post-execution. Both are valid approximations of + // "when this flow first ran". + self.start_time_millis = Some(common_time::util::current_time_millis()); + } + } + pub fn after_query_exec(&mut self, elapsed: Duration, is_succ: bool) { self.exec_state = ExecState::Idle; self.last_query_duration = elapsed; @@ -105,6 +118,11 @@ impl TaskState { self.last_exec_time_millis } + /// First execution time in unix timestamp milliseconds, set once. + pub fn start_time_millis(&self) -> Option { + self.start_time_millis + } + pub fn checkpoint_mode(&self) -> CheckpointMode { self.checkpoint_mode } diff --git a/src/flow/src/batching_mode/task.rs b/src/flow/src/batching_mode/task.rs index 8428198de3..196c9ea42d 100644 --- a/src/flow/src/batching_mode/task.rs +++ b/src/flow/src/batching_mode/task.rs @@ -290,6 +290,10 @@ impl BatchingTask { self.state.read().unwrap().last_execution_time_millis() } + pub fn start_time_millis(&self) -> Option { + self.state.read().unwrap().start_time_millis() + } + /// Collect flow-related extensions from the task's query context that should be /// forwarded to the frontend (e.g. scheduled time). fn frontend_extensions(&self) -> HashMap { @@ -717,6 +721,10 @@ impl BatchingTask { }; let snapshot_seqs = coverage.snapshot_seqs(); + { + let mut state = self.state.write().unwrap(); + state.record_start_time_if_first(); + } frontend_client .query_with_terminal_metrics( catalog, diff --git a/src/flow/src/compute/state.rs b/src/flow/src/compute/state.rs index 2ccd366194..b71633f517 100644 --- a/src/flow/src/compute/state.rs +++ b/src/flow/src/compute/state.rs @@ -47,6 +47,8 @@ pub struct DataflowState { expire_after: Option, /// the last time each subgraph executed last_exec_time: Option, + /// the time the flow first executed, in unix timestamp milliseconds + start_time: Option, } impl DataflowState { @@ -120,11 +122,21 @@ impl DataflowState { pub fn set_last_exec_time(&mut self, time: Timestamp) { self.last_exec_time = Some(time); + if self.start_time.is_none() { + // start_time is recorded at the completion of the first execution + // (post-execution), consistent with how last_exec_time is recorded. + self.start_time = Some(time); + } } pub fn last_exec_time(&self) -> Option { self.last_exec_time } + + /// Returns the time the flow first executed, in unix timestamp milliseconds. + pub fn start_time(&self) -> Option { + self.start_time + } } #[derive(Debug, Clone)] diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs index 7a4b2f9a41..706e651462 100644 --- a/src/frontend/src/instance.rs +++ b/src/frontend/src/instance.rs @@ -1559,6 +1559,11 @@ pub fn check_permission( Statement::ShowFlows(stmt) => { validate_db_permission!(stmt, query_ctx); } + Statement::ShowFlowStatus(_stmt) => { + // Flow statistics are organized based on the catalog dimension and + // filtered by the current catalog, so there is no need to check the + // permission of the database(schema). + } #[cfg(feature = "enterprise")] Statement::ShowTriggers(_stmt) => { // The trigger is organized based on the catalog dimension, so there diff --git a/src/meta-srv/src/handler/flow_state_handler.rs b/src/meta-srv/src/handler/flow_state_handler.rs index 9683fd760a..a13c18a9e0 100644 --- a/src/meta-srv/src/handler/flow_state_handler.rs +++ b/src/meta-srv/src/handler/flow_state_handler.rs @@ -55,7 +55,11 @@ impl HeartbeatHandler for FlowStateHandler { .iter() .map(|(k, v)| (*k, *v)) .collect(); - let value: FlowStateValue = FlowStateValue::new(state_size, last_exec_time_map); + // TODO(#7987-followup): start_time_map is not yet propagated through the heartbeat + // wire format (`api::v1::meta::FlowStat`); it will always be empty in distributed + // mode until a follow-up PR adds heartbeat propagation. + let value: FlowStateValue = + FlowStateValue::new(state_size, last_exec_time_map, Default::default()); self.flow_state_manager .put(value) .await diff --git a/src/operator/src/statement.rs b/src/operator/src/statement.rs index 92e0c04a92..320e7c2e64 100644 --- a/src/operator/src/statement.rs +++ b/src/operator/src/statement.rs @@ -259,6 +259,7 @@ impl StatementExecutor { Statement::ShowViews(stmt) => self.show_views(stmt, query_ctx).await, Statement::ShowFlows(stmt) => self.show_flows(stmt, query_ctx).await, + Statement::ShowFlowStatus(stmt) => self.show_flow_status(stmt, query_ctx).await, #[cfg(feature = "enterprise")] Statement::ShowTriggers(stmt) => self.show_triggers(stmt, query_ctx).await, diff --git a/src/operator/src/statement/show.rs b/src/operator/src/statement/show.rs index 33cdb0f367..4a3fa5ae9b 100644 --- a/src/operator/src/statement/show.rs +++ b/src/operator/src/statement/show.rs @@ -26,8 +26,9 @@ use sql::ast::ObjectNamePartExt; use sql::statements::OptionMap; use sql::statements::create::Partitions; use sql::statements::show::{ - ShowColumns, ShowCreateFlow, ShowCreateView, ShowDatabases, ShowFlows, ShowIndex, ShowKind, - ShowProcessList, ShowRegion, ShowTableStatus, ShowTables, ShowVariables, ShowViews, + ShowColumns, ShowCreateFlow, ShowCreateView, ShowDatabases, ShowFlowStatus, ShowFlows, + ShowIndex, ShowKind, ShowProcessList, ShowRegion, ShowTableStatus, ShowTables, ShowVariables, + ShowViews, }; use table::TableRef; use table::metadata::{TableInfo, TableType}; @@ -250,6 +251,17 @@ impl StatementExecutor { .context(ExecuteStatementSnafu) } + #[tracing::instrument(skip_all)] + pub(super) async fn show_flow_status( + &self, + stmt: ShowFlowStatus, + query_ctx: QueryContextRef, + ) -> Result { + query::sql::show_flow_status(stmt, &self.query_engine, &self.catalog_manager, query_ctx) + .await + .context(ExecuteStatementSnafu) + } + #[cfg(feature = "enterprise")] #[tracing::instrument(skip_all)] pub(super) async fn show_triggers( diff --git a/src/query/src/sql.rs b/src/query/src/sql.rs index 90c3b4ccd4..fc755deb8a 100644 --- a/src/query/src/sql.rs +++ b/src/query/src/sql.rs @@ -20,8 +20,9 @@ use std::sync::Arc; use catalog::CatalogManagerRef; use catalog::information_schema::{ - CHARACTER_SETS, COLLATIONS, COLUMNS, FLOWS, REGION_PEERS, SCHEMATA, STATISTICS, TABLES, VIEWS, - columns, flows, process_list, region_peers, schemata, statistics, tables, + CHARACTER_SETS, COLLATIONS, COLUMNS, FLOW_STATISTICS, FLOWS, REGION_PEERS, SCHEMATA, + STATISTICS, TABLES, VIEWS, columns, flow_statistics, flows, process_list, region_peers, + schemata, statistics, tables, }; use common_catalog::consts::{ INFORMATION_SCHEMA_NAME, SEMANTIC_TYPE_FIELD, SEMANTIC_TYPE_PRIMARY_KEY, @@ -57,8 +58,8 @@ use sql::parser::ParserContext; use sql::statements::OptionMap; use sql::statements::create::{CreateDatabase, CreateFlow, CreateView, Partitions, SqlOrTql}; use sql::statements::show::{ - ShowColumns, ShowDatabases, ShowFlows, ShowIndex, ShowKind, ShowProcessList, ShowRegion, - ShowTableStatus, ShowTables, ShowVariables, ShowViews, + ShowColumns, ShowDatabases, ShowFlowStatus, ShowFlows, ShowIndex, ShowKind, ShowProcessList, + ShowRegion, ShowTableStatus, ShowTables, ShowVariables, ShowViews, }; use sql::statements::statement::Statement; use sqlparser::ast::ObjectName; @@ -971,6 +972,45 @@ pub async fn show_flows( .await } +/// Execute [`ShowFlowStatus`] statement and return the [`Output`] if success. +pub async fn show_flow_status( + stmt: ShowFlowStatus, + query_engine: &QueryEngineRef, + catalog_manager: &CatalogManagerRef, + query_ctx: QueryContextRef, +) -> Result { + let projects = vec![ + (flow_statistics::FLOW_ID, flow_statistics::FLOW_ID), + (flow_statistics::FLOW_NAME, flow_statistics::FLOW_NAME), + (flow_statistics::START_TIME, flow_statistics::START_TIME), + ( + flow_statistics::LAST_EXECUTION_TIME, + flow_statistics::LAST_EXECUTION_TIME, + ), + ( + flow_statistics::UPTIME_SECONDS, + flow_statistics::UPTIME_SECONDS, + ), + (flow_statistics::STATE_SIZE, flow_statistics::STATE_SIZE), + ]; + let like_field = Some(flow_statistics::FLOW_NAME); + let sort = vec![col(flow_statistics::FLOW_NAME).sort(true, true)]; + + query_from_information_schema_table( + query_engine, + catalog_manager, + query_ctx, + FLOW_STATISTICS, + vec![], + projects, + vec![], + like_field, + sort, + stmt.kind, + ) + .await +} + #[cfg(feature = "enterprise")] pub async fn show_triggers( stmt: sql::statements::show::trigger::ShowTriggers, diff --git a/src/servers/src/postgres/handler.rs b/src/servers/src/postgres/handler.rs index e372a54237..484bb6a1f1 100644 --- a/src/servers/src/postgres/handler.rs +++ b/src/servers/src/postgres/handler.rs @@ -600,6 +600,53 @@ fn describe_fields( format.format_for(1), ), ]), + // SHOW FLOW STATUS returns six columns; return their descriptions so + // prepared/extended-protocol clients receive the correct row description. + SqlPlan::Statement(Statement::ShowFlowStatus(_), _) => Ok(vec![ + FieldInfo::new( + "flow_id".to_string(), + None, + None, + Type::INT8, // matches type_gt_to_pg(UInt32) — do not use INT4 + format.format_for(0), + ), + FieldInfo::new( + "flow_name".to_string(), + None, + None, + Type::TEXT, + format.format_for(1), + ), + FieldInfo::new( + "start_time".to_string(), + None, + None, + Type::TIMESTAMP, + format.format_for(2), + ), + FieldInfo::new( + "last_execution_time".to_string(), + None, + None, + Type::TIMESTAMP, + format.format_for(3), + ), + FieldInfo::new( + "uptime_seconds".to_string(), + None, + None, + Type::INT8, + format.format_for(4), + ), + FieldInfo::new( + "state_size".to_string(), + None, + None, + Type::NUMERIC, + format.format_for(5), + ), + ]), + // single column show statements SqlPlan::Statement( Statement::ShowTables(_) | Statement::ShowFlows(_) | Statement::ShowViews(_), diff --git a/src/sql/src/parsers/show_parser.rs b/src/sql/src/parsers/show_parser.rs index d6fc35c675..075c17dabe 100644 --- a/src/sql/src/parsers/show_parser.rs +++ b/src/sql/src/parsers/show_parser.rs @@ -26,8 +26,8 @@ use crate::error::{ use crate::parser::ParserContext; use crate::statements::show::{ ShowColumns, ShowCreateDatabase, ShowCreateFlow, ShowCreateTable, ShowCreateTableVariant, - ShowCreateView, ShowDatabases, ShowFlows, ShowIndex, ShowKind, ShowProcessList, ShowRegion, - ShowSearchPath, ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews, + ShowCreateView, ShowDatabases, ShowFlowStatus, ShowFlows, ShowIndex, ShowKind, ShowProcessList, + ShowRegion, ShowSearchPath, ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews, }; use crate::statements::statement::Statement; @@ -57,6 +57,12 @@ impl ParserContext<'_> { self.parse_show_views() } else if self.consume_token("FLOWS") { self.parse_show_flows() + } else if self.consume_token("FLOW") { + if self.consume_token("STATUS") { + self.parse_show_flow_status() + } else { + self.unsupported(self.peek_token_as_string()) + } } else if self.matches_keyword(Keyword::CHARSET) { self.parser.next_token(); Ok(Statement::ShowCharset(self.parse_show_kind()?)) @@ -587,6 +593,12 @@ impl ParserContext<'_> { Ok(Statement::ShowFlows(ShowFlows { kind, database })) } + fn parse_show_flow_status(&mut self) -> Result { + let kind = self.parse_show_kind()?; + + Ok(Statement::ShowFlowStatus(ShowFlowStatus { kind })) + } + fn parse_show_processlist(&mut self, full: bool) -> Result { match self.parser.next_token().token { Token::EOF | Token::SemiColon => { @@ -1250,6 +1262,22 @@ mod tests { assert_eq!(sql, stmts[0].to_string()); } + #[test] + pub fn test_show_flow_status() { + let sql = "SHOW FLOW STATUS"; + let result = + ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default()); + let stmts = result.unwrap(); + assert_eq!(1, stmts.len()); + assert_eq!( + stmts[0], + Statement::ShowFlowStatus(ShowFlowStatus { + kind: ShowKind::All, + }) + ); + assert_eq!(sql, stmts[0].to_string()); + } + #[test] pub fn test_show_processlist() { let sql = "SHOW PROCESSLIST"; diff --git a/src/sql/src/statements/show.rs b/src/sql/src/statements/show.rs index 77880e4a50..67624192b8 100644 --- a/src/sql/src/statements/show.rs +++ b/src/sql/src/statements/show.rs @@ -256,6 +256,21 @@ impl Display for ShowFlows { } } +/// SQL structure for `SHOW FLOW STATUS`. +#[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut, Serialize)] +pub struct ShowFlowStatus { + pub kind: ShowKind, +} + +impl Display for ShowFlowStatus { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "SHOW FLOW STATUS")?; + format_kind!(self, f); + + Ok(()) + } +} + /// SQL structure for `SHOW CREATE VIEW`. #[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut, Serialize)] pub struct ShowCreateView { diff --git a/src/sql/src/statements/statement.rs b/src/sql/src/statements/statement.rs index 0f652f0179..096acb654d 100644 --- a/src/sql/src/statements/statement.rs +++ b/src/sql/src/statements/statement.rs @@ -40,8 +40,8 @@ use crate::statements::query::Query; use crate::statements::set_variables::SetVariables; use crate::statements::show::{ ShowColumns, ShowCreateDatabase, ShowCreateFlow, ShowCreateTable, ShowCreateView, - ShowDatabases, ShowFlows, ShowIndex, ShowKind, ShowProcessList, ShowRegion, ShowSearchPath, - ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews, + ShowDatabases, ShowFlowStatus, ShowFlows, ShowIndex, ShowKind, ShowProcessList, ShowRegion, + ShowSearchPath, ShowStatus, ShowTableStatus, ShowTables, ShowVariables, ShowViews, }; use crate::statements::tql::Tql; use crate::statements::truncate::TruncateTable; @@ -118,6 +118,8 @@ pub enum Statement { ShowCreateTrigger(crate::statements::show::trigger::ShowCreateTrigger), /// SHOW FLOWS ShowFlows(ShowFlows), + /// SHOW FLOW STATUS + ShowFlowStatus(ShowFlowStatus), // SHOW TRIGGERS #[cfg(feature = "enterprise")] ShowTriggers(crate::statements::show::trigger::ShowTriggers), @@ -178,6 +180,7 @@ impl Statement { | Statement::ShowCreateTable(_) | Statement::ShowCreateFlow(_) | Statement::ShowFlows(_) + | Statement::ShowFlowStatus(_) | Statement::ShowCreateView(_) | Statement::ShowStatus(_) | Statement::ShowSearchPath(_) @@ -268,6 +271,7 @@ impl Display for Statement { #[cfg(feature = "enterprise")] Statement::ShowCreateTrigger(s) => s.fmt(f), Statement::ShowFlows(s) => s.fmt(f), + Statement::ShowFlowStatus(s) => s.fmt(f), #[cfg(feature = "enterprise")] Statement::ShowTriggers(s) => s.fmt(f), Statement::ShowCreateDatabase(s) => s.fmt(f), diff --git a/src/sql/src/util.rs b/src/sql/src/util.rs index 03071cfaa2..c0c237cb0c 100644 --- a/src/sql/src/util.rs +++ b/src/sql/src/util.rs @@ -335,6 +335,7 @@ fn extract_tables_from_statement(stmt: &Statement, names: &mut HashSet