rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-13 23:42:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(flow): add flow statistics and SHOW FLOW STATUS (release slice)

Browse Source 
Signed-off-by: discord9 <discord9@163.com>
(cherry picked from commit 523c60c53bdc21fd1b1fa43b857e78de710f0d58)
This commit is contained in:
discord9
2026-08-20 14:44:57 +08:00
parent cf69ee5df9
commit 0b6702b7cc
27 changed files with 700 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/catalog/src/system_schema/information_schema.rs
+11
View File
@@ -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(),
src/catalog/src/system_schema/information_schema/flow_statistics.rs
+285
View File
@@ -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<dyn CatalogManager>,
flow_metadata_manager: Arc<FlowMetadataManager>,
}
impl InformationSchemaFlowStatistics {
pub(super) fn new(
catalog_name: String,
catalog_manager: Weak<dyn CatalogManager>,
flow_metadata_manager: Arc<FlowMetadataManager>,
) -> 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<SendableRecordBatchStream> {
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<dyn CatalogManager>,
flow_metadata_manager: Arc<FlowMetadataManager>,
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<dyn CatalogManager>,
flow_metadata_manager: &Arc<FlowMetadataManager>,
) -> 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<ScanRequest>) -> Result<RecordBatch> {
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<FlowStat>,
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<RecordBatch> {
let columns: Vec<VectorRef> = 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<TaskContext>) -> DfSendableRecordBatchStream {
let schema: Arc<arrow_schema::Schema> = 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)
}),
))
}
}
src/catalog/src/system_schema/information_schema/table_names.rs
+1
View File
@@ -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";
src/cli/src/metadata/control/put/key.rs
+1 -3
View File
@@ -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();
}
src/common/catalog/src/consts.rs
+2
View File
@@ -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 -----
src/flow/src/adapter/flownode_impl.rs
+4
View File
@@ -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,
}
}
src/flow/src/adapter/stat.rs
+10 -16
View File
@@ -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,
}
}
}
src/flow/src/adapter/worker.rs
+35 -41
View File
@@ -202,30 +202,24 @@ impl WorkerHandle {
}
}
pub async fn get_state_size(&self) -> Result<BTreeMap<FlowId, usize>, Error> {
pub async fn get_full_flow_stat(
&self,
) -> Result<
(
BTreeMap<FlowId, usize>,
BTreeMap<FlowId, i64>,
BTreeMap<FlowId, i64>,
),
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<BTreeMap<FlowId, i64>, 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<FlowId, usize>,
},
QueryLastExecTimeMap {
/// each flow tasks' last execution time
result: BTreeMap<FlowId, i64>,
QueryFullFlowStat {
state_size: BTreeMap<FlowId, usize>,
last_exec_time_map: BTreeMap<FlowId, i64>,
start_time_map: BTreeMap<FlowId, i64>,
},
}
@@ -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();
src/flow/src/batching_mode/engine.rs
+16 -6
View File
@@ -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,
}
}
}
src/flow/src/batching_mode/state.rs
+20 -2
View File
@@ -46,6 +46,8 @@ pub struct TaskState {
last_query_duration: Duration,
/// Last successful execution time in unix timestamp milliseconds.
last_exec_time_millis: Option<i64>,
/// First execution time in unix timestamp milliseconds, set once.
start_time_millis: Option<i64>,
/// 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<i64> {
self.start_time_millis
}
pub fn checkpoint_mode(&self) -> CheckpointMode {
self.checkpoint_mode
}
src/flow/src/batching_mode/task.rs
+8
View File
@@ -290,6 +290,10 @@ impl BatchingTask {
self.state.read().unwrap().last_execution_time_millis()
}
pub fn start_time_millis(&self) -> Option<i64> {
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<String, String> {
@@ -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,
src/flow/src/compute/state.rs
+12
View File
@@ -47,6 +47,8 @@ pub struct DataflowState {
expire_after: Option<Timestamp>,
/// the last time each subgraph executed
last_exec_time: Option<Timestamp>,
/// the time the flow first executed, in unix timestamp milliseconds
start_time: Option<Timestamp>,
}
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<Timestamp> {
self.last_exec_time
}
/// Returns the time the flow first executed, in unix timestamp milliseconds.
pub fn start_time(&self) -> Option<Timestamp> {
self.start_time
}
}
#[derive(Debug, Clone)]
src/frontend/src/instance.rs
+5
View File
@@ -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
src/meta-srv/src/handler/flow_state_handler.rs
+3
View File
@@ -55,6 +55,9 @@ impl HeartbeatHandler for FlowStateHandler {
.iter()
.map(|(k, v)| (*k, *v))
.collect();
// The beta2 heartbeat protobuf has only the two historical fields.
// Distributed start_time/uptime_seconds are intentionally unsupported and
// therefore remain NULL; start_time is retained only in local FlowStat.
let value: FlowStateValue = FlowStateValue::new(state_size, last_exec_time_map);
self.flow_state_manager
.put(value)
src/operator/src/statement.rs
+1
View File
@@ -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,
src/operator/src/statement/show.rs
+14 -2
View File
@@ -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<Output> {
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(
src/query/src/sql.rs
+44 -4
View File
@@ -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<Output> {
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,
src/servers/src/postgres/handler.rs
+47
View File
@@ -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(_),
src/sql/src/parsers/show_parser.rs
+30 -2
View File
@@ -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<Statement> {
let kind = self.parse_show_kind()?;
Ok(Statement::ShowFlowStatus(ShowFlowStatus { kind }))
}
fn parse_show_processlist(&mut self, full: bool) -> Result<Statement> {
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";
src/sql/src/statements/show.rs
+15
View File
@@ -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 {
src/sql/src/statements/statement.rs
+6 -2
View File
@@ -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),
src/sql/src/util.rs
+1
View File
@@ -335,6 +335,7 @@ fn extract_tables_from_statement(stmt: &Statement, names: &mut HashSet<ObjectNam
| Statement::ShowCreateDatabase(_)
| Statement::ShowCreateFlow(_)
| Statement::ShowFlows(_)
| Statement::ShowFlowStatus(_)
| Statement::ShowStatus(_)
| Statement::ShowSearchPath(_)
| Statement::ShowViews(_)
tests/cases/standalone/common/flow/flow_status.result
+76
View File
@@ -0,0 +1,76 @@
CREATE TABLE numbers_input_flow_status (
"number" INT,
ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY(number),
TIME INDEX(ts)
);
Affected Rows: 0
CREATE FLOW test_flow_status SINK TO out_flow_status AS
SELECT
sum(number),
date_bin(INTERVAL '1 second', ts, '2021-07-01 00:00:00') as time_window
FROM
numbers_input_flow_status
GROUP BY
time_window;
Affected Rows: 0
INSERT INTO
numbers_input_flow_status
VALUES
(20, "2021-07-01 00:00:00.200"),
(22, "2021-07-01 00:00:00.600");
Affected Rows: 2
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
ADMIN FLUSH_FLOW('test_flow_status');
+--------------------------------------+
| ADMIN FLUSH_FLOW('test_flow_status') |
+--------------------------------------+
| FLOW_FLUSHED |
+--------------------------------------+
-- flow_name is deterministic; the rest of the columns are runtime dependent.
SELECT flow_name FROM information_schema.flow_statistics WHERE flow_name = 'test_flow_status';
+------------------+
| flow_name |
+------------------+
| test_flow_status |
+------------------+
-- SHOW FLOW STATUS is exercised below via the empty-result form (six-column header).
-- Full row values differ between standalone (populated) and distributed (heartbeat
-- has not yet propagated start_time), so assert only the stable columns here.
-- SQLNESS REPLACE (\|\s+[0-9]+\s+\|) | ID |
SELECT flow_id, flow_name FROM information_schema.flow_statistics WHERE flow_name = 'test_flow_status';
+---------+------------------+
| flow_id | flow_name |
+---------+------------------+
| ID | test_flow_status |
+---------+------------------+
-- the like filter matches against flow_name; no matching flow returns empty.
SHOW FLOW STATUS LIKE 'no_such_flow';
++
++
DROP FLOW test_flow_status;
Affected Rows: 0
DROP TABLE numbers_input_flow_status;
Affected Rows: 0
DROP TABLE out_flow_status;
Affected Rows: 0
tests/cases/standalone/common/flow/flow_status.sql
+42
View File
@@ -0,0 +1,42 @@
CREATE TABLE numbers_input_flow_status (
"number" INT,
ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY(number),
TIME INDEX(ts)
);
CREATE FLOW test_flow_status SINK TO out_flow_status AS
SELECT
sum(number),
date_bin(INTERVAL '1 second', ts, '2021-07-01 00:00:00') as time_window
FROM
numbers_input_flow_status
GROUP BY
time_window;
INSERT INTO
numbers_input_flow_status
VALUES
(20, "2021-07-01 00:00:00.200"),
(22, "2021-07-01 00:00:00.600");
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
ADMIN FLUSH_FLOW('test_flow_status');
-- flow_name is deterministic; the rest of the columns are runtime dependent.
SELECT flow_name FROM information_schema.flow_statistics WHERE flow_name = 'test_flow_status';
-- SHOW FLOW STATUS is exercised below via the empty-result form (six-column header).
-- Full row values differ between standalone (populated) and distributed (heartbeat
-- has not yet propagated start_time), so assert only the stable columns here.
-- SQLNESS REPLACE (\|\s+[0-9]+\s+\|) | ID |
SELECT flow_id, flow_name FROM information_schema.flow_statistics WHERE flow_name = 'test_flow_status';
-- the like filter matches against flow_name; no matching flow returns empty.
SHOW FLOW STATUS LIKE 'no_such_flow';
DROP FLOW test_flow_status;
DROP TABLE numbers_input_flow_status;
DROP TABLE out_flow_status;
tests/cases/standalone/common/show/show_databases_tables.result
+3
View File
@@ -39,6 +39,7 @@ SHOW TABLES;
| engines |
| events |
| files |
| flow_statistics |
| flows |
| global_status |
| key_column_usage |
@@ -92,6 +93,7 @@ SHOW FULL TABLES;
| engines | LOCAL TEMPORARY |
| events | LOCAL TEMPORARY |
| files | LOCAL TEMPORARY |
| flow_statistics | LOCAL TEMPORARY |
| flows | LOCAL TEMPORARY |
| global_status | LOCAL TEMPORARY |
| key_column_usage | LOCAL TEMPORARY |
@@ -139,6 +141,7 @@ SHOW TABLE STATUS;
|engines||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
|events||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
|files||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
|flow_statistics||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
|flows||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
|global_status||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
|key_column_usage||11|Fixed|0|0|0|0|0|0|0|DATETIME|DATETIME||utf8_bin|0|||
tests/cases/standalone/common/system/information_schema.result
+7
View File
@@ -26,6 +26,7 @@ order by table_schema, table_name;
|greptime|information_schema|engines|LOCALTEMPORARY|5|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
|greptime|information_schema|events|LOCALTEMPORARY|13|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
|greptime|information_schema|files|LOCALTEMPORARY|14|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
|greptime|information_schema|flow_statistics|LOCALTEMPORARY|45|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
|greptime|information_schema|flows|LOCALTEMPORARY|33|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
|greptime|information_schema|global_status|LOCALTEMPORARY|25|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
|greptime|information_schema|key_column_usage|LOCALTEMPORARY|16|0|0|0|0|0||11|Fixed|0|0|0|DATETIME|DATETIME||utf8_bin|0|||Y|
@@ -198,6 +199,12 @@ select * from information_schema.columns order by table_schema, table_name, colu
| greptime | information_schema | files | update_count | 13 | | | 19 | 0 | | | | | | select,insert | | Int64 | bigint | FIELD | | NO | bigint | | |
| greptime | information_schema | files | update_time | 34 | | | | | 6 | | | | | select,insert | | TimestampMicrosecond | timestamp(6) | FIELD | | NO | timestamp(6) | | |
| greptime | information_schema | files | version | 25 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | |
| greptime | information_schema | flow_statistics | flow_id | 1 | | | 10 | 0 | | | | | | select,insert | | UInt32 | int unsigned | FIELD | | NO | int unsigned | | |
| greptime | information_schema | flow_statistics | flow_name | 2 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | NO | string | | |
| greptime | information_schema | flow_statistics | last_execution_time | 4 | | | | | 3 | | | | | select,insert | | TimestampMillisecond | timestamp(3) | FIELD | | YES | timestamp(3) | | |
| greptime | information_schema | flow_statistics | start_time | 3 | | | | | 3 | | | | | select,insert | | TimestampMillisecond | timestamp(3) | FIELD | | YES | timestamp(3) | | |
| greptime | information_schema | flow_statistics | state_size | 6 | | | 20 | 0 | | | | | | select,insert | | UInt64 | bigint unsigned | FIELD | | YES | bigint unsigned | | |
| greptime | information_schema | flow_statistics | uptime_seconds | 5 | | | 19 | 0 | | | | | | select,insert | | Int64 | bigint | FIELD | | YES | bigint | | |
| greptime | information_schema | flows | comment | 6 | 2147483647 | 2147483647 | | | | utf8 | utf8_bin | | | select,insert | | String | string | FIELD | | YES | string | | |
| greptime | information_schema | flows | created_time | 12 | | | | | 3 | | | | | select,insert | | TimestampMillisecond | timestamp(3) | FIELD | | NO | timestamp(3) | | |
| greptime | information_schema | flows | expire_after | 7 | | | 19 | 0 | | | | | | select,insert | | Int64 | bigint | FIELD | | YES | bigint | | |
tests/cases/standalone/common/view/create.result
+1
View File
@@ -105,6 +105,7 @@ SELECT * FROM INFORMATION_SCHEMA.TABLES ORDER BY TABLE_NAME, TABLE_TYPE;
|greptime|information_schema|engines|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
|greptime|information_schema|events|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
|greptime|information_schema|files|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
|greptime|information_schema|flow_statistics|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
|greptime|information_schema|flows|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
|greptime|information_schema|global_status|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|
|greptime|information_schema|key_column_usage|LOCALTEMPORARY|ID|ID|ID|ID|ID|ID||ID|Fixed|ID|ID|ID|DATETIME|DATETIME||utf8_bin|ID|||Y|