common_meta/key/flow/
flow_info.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::{BTreeMap, HashMap};
16use std::sync::Arc;
17
18use chrono::{DateTime, Utc};
19use futures::TryStreamExt;
20use futures::stream::BoxStream;
21use lazy_static::lazy_static;
22use regex::Regex;
23use serde::{Deserialize, Serialize};
24use snafu::OptionExt;
25use table::metadata::TableId;
26use table::table_name::TableName;
27
28use crate::FlownodeId;
29use crate::error::{self, Result};
30use crate::key::flow::FlowScoped;
31use crate::key::txn_helper::TxnOpGetResponseSet;
32use crate::key::{
33    BytesAdapter, DeserializedValueWithBytes, FlowId, FlowPartitionId, MetadataKey, MetadataValue,
34};
35use crate::kv_backend::KvBackendRef;
36use crate::kv_backend::txn::{Compare, CompareOp, Txn, TxnOp};
37use crate::range_stream::{DEFAULT_PAGE_SIZE, PaginationStream};
38use crate::rpc::KeyValue;
39use crate::rpc::store::RangeRequest;
40
41pub const FLOW_INFO_KEY_PREFIX: &str = "info";
42
43/// The lifecycle status of a flow stored in metadata.
44#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
45pub enum FlowStatus {
46    /// The flow metadata exists, but at least one source table did not exist at create time.
47    PendingSources,
48    /// The flow has resolved source tables and can be scheduled on flownodes.
49    #[default]
50    Active,
51}
52
53lazy_static! {
54    static ref FLOW_INFO_KEY_PATTERN: Regex =
55        Regex::new(&format!("^{FLOW_INFO_KEY_PREFIX}/([0-9]+)$")).unwrap();
56}
57
58/// The key stores the metadata of the flow.
59///
60/// The layout: `__flow/info/{flow_id}`.
61pub struct FlowInfoKey(FlowScoped<FlowInfoKeyInner>);
62
63pub type FlowInfoDecodeResult = Result<Option<DeserializedValueWithBytes<FlowInfoValue>>>;
64
65impl<'a> MetadataKey<'a, FlowInfoKey> for FlowInfoKey {
66    fn to_bytes(&self) -> Vec<u8> {
67        self.0.to_bytes()
68    }
69
70    fn from_bytes(bytes: &'a [u8]) -> Result<FlowInfoKey> {
71        Ok(FlowInfoKey(FlowScoped::<FlowInfoKeyInner>::from_bytes(
72            bytes,
73        )?))
74    }
75}
76
77impl FlowInfoKey {
78    /// Returns the [FlowInfoKey].
79    pub fn new(flow_id: FlowId) -> FlowInfoKey {
80        let inner = FlowInfoKeyInner::new(flow_id);
81        FlowInfoKey(FlowScoped::new(inner))
82    }
83
84    /// Returns the [FlowId].
85    pub fn flow_id(&self) -> FlowId {
86        self.0.flow_id
87    }
88}
89
90/// The key of flow metadata.
91#[derive(Debug, Clone, Copy, PartialEq)]
92struct FlowInfoKeyInner {
93    flow_id: FlowId,
94}
95
96impl FlowInfoKeyInner {
97    /// Returns a [FlowInfoKey] with the specified `flow_id`.
98    pub fn new(flow_id: FlowId) -> FlowInfoKeyInner {
99        FlowInfoKeyInner { flow_id }
100    }
101}
102
103impl<'a> MetadataKey<'a, FlowInfoKeyInner> for FlowInfoKeyInner {
104    fn to_bytes(&self) -> Vec<u8> {
105        format!("{FLOW_INFO_KEY_PREFIX}/{}", self.flow_id).into_bytes()
106    }
107
108    fn from_bytes(bytes: &'a [u8]) -> Result<FlowInfoKeyInner> {
109        let key = std::str::from_utf8(bytes).map_err(|e| {
110            error::InvalidMetadataSnafu {
111                err_msg: format!(
112                    "FlowInfoKeyInner '{}' is not a valid UTF8 string: {e}",
113                    String::from_utf8_lossy(bytes)
114                ),
115            }
116            .build()
117        })?;
118        let captures =
119            FLOW_INFO_KEY_PATTERN
120                .captures(key)
121                .context(error::InvalidMetadataSnafu {
122                    err_msg: format!("Invalid FlowInfoKeyInner '{key}'"),
123                })?;
124        // Safety: pass the regex check above
125        let flow_id = captures[1].parse::<FlowId>().unwrap();
126        Ok(FlowInfoKeyInner { flow_id })
127    }
128}
129
130/// Internal typed schedule configuration for `EVAL INTERVAL` flows.
131///
132/// This struct is the canonical schedule state for `EVAL INTERVAL` flows and
133/// is stored alongside `FlowInfoValue` (not inside `options`).
134#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
135pub struct FlowScheduleConfig {
136    /// Anchor timestamp in seconds since Unix epoch (default: 0 = epoch).
137    pub anchor_secs: i64,
138    /// Start timestamp in seconds since Unix epoch.
139    pub start_secs: i64,
140    /// Policy for handling missed ticks.
141    #[serde(default)]
142    pub missed_tick_policy: FlowMissedTickPolicy,
143    /// Maximum number of catch-up runs when using bounded catch-up.
144    ///
145    /// A catch-up run is an evaluation for a scheduled timestamp that is
146    /// already due but was missed because the flownode was stopped or busy.
147    #[serde(default = "FlowScheduleConfig::default_catchup_max_runs")]
148    pub catchup_max_runs: u32,
149    /// Maximum age (in seconds) of a due scheduled time to still include in catch-up.
150    #[serde(default = "FlowScheduleConfig::default_catchup_max_lag_secs")]
151    pub catchup_max_lag_secs: i64,
152}
153
154impl FlowScheduleConfig {
155    pub const DEFAULT_ANCHOR_SECS: i64 = 0;
156    pub const DEFAULT_CATCHUP_MAX_RUNS: u32 = 3;
157    pub const DEFAULT_CATCHUP_MAX_LAG_SECS: i64 = 300;
158
159    pub fn default_catchup_max_runs() -> u32 {
160        Self::DEFAULT_CATCHUP_MAX_RUNS
161    }
162
163    pub fn default_catchup_max_lag_secs() -> i64 {
164        Self::DEFAULT_CATCHUP_MAX_LAG_SECS
165    }
166
167    pub fn catchup_max_lag_secs_for_interval(eval_interval_secs: i64) -> i64 {
168        std::cmp::max(
169            Self::DEFAULT_CATCHUP_MAX_LAG_SECS,
170            3_i64.saturating_mul(eval_interval_secs),
171        )
172    }
173
174    pub fn default_with_start(start_secs: i64, eval_interval_secs: i64) -> Self {
175        Self {
176            anchor_secs: Self::DEFAULT_ANCHOR_SECS,
177            start_secs,
178            missed_tick_policy: FlowMissedTickPolicy::BoundedCatchUp,
179            catchup_max_runs: Self::DEFAULT_CATCHUP_MAX_RUNS,
180            catchup_max_lag_secs: Self::catchup_max_lag_secs_for_interval(eval_interval_secs),
181        }
182    }
183}
184
185impl Default for FlowScheduleConfig {
186    fn default() -> Self {
187        Self {
188            anchor_secs: Self::DEFAULT_ANCHOR_SECS,
189            start_secs: 0,
190            missed_tick_policy: FlowMissedTickPolicy::default(),
191            catchup_max_runs: Self::default_catchup_max_runs(),
192            catchup_max_lag_secs: Self::default_catchup_max_lag_secs(),
193        }
194    }
195}
196
197/// Policy for handling flow evaluation scheduled times that were missed.
198#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
199pub enum FlowMissedTickPolicy {
200    /// Keep the most recent `catchup_max_runs` due scheduled times within
201    /// `catchup_max_lag_secs`, drop older ones.
202    #[default]
203    #[serde(rename = "bounded_catch_up")]
204    BoundedCatchUp,
205    /// Skip all missed scheduled times; only execute the single most recent one.
206    #[serde(rename = "skip")]
207    Skip,
208}
209
210// The metadata of the flow.
211#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
212pub struct FlowInfoValue {
213    /// The source tables used by the flow.
214    #[serde(default)]
215    pub source_table_ids: Vec<TableId>,
216    #[serde(default)]
217    pub all_source_table_names: Vec<TableName>,
218    #[serde(default)]
219    pub unresolved_source_table_names: Vec<TableName>,
220    /// The sink table used by the flow.
221    pub sink_table_name: TableName,
222    /// Which flow nodes this flow is running on.
223    pub flownode_ids: BTreeMap<FlowPartitionId, FlownodeId>,
224    /// The catalog name.
225    pub catalog_name: String,
226    /// The query context used when create flow.
227    /// Although flow doesn't belong to any schema, this query_context is needed to remember
228    /// the query context when `create_flow` is executed
229    /// for recovering flow using the same sql&query_context after db restart.
230    /// if none, should use default query context
231    #[serde(default)]
232    pub query_context: Option<crate::rpc::ddl::QueryContext>,
233    /// The flow name.
234    pub flow_name: String,
235    /// The raw sql.
236    pub raw_sql: String,
237    /// The expr of expire.
238    /// Duration in seconds as `i64`.
239    pub expire_after: Option<i64>,
240    /// The eval interval.
241    /// Duration in seconds as `i64`.
242    /// If `None`, will automatically decide when to evaluate the flow.
243    /// If `Some`, it will be evaluated every `eval_interval` seconds.
244    #[serde(default)]
245    pub eval_interval_secs: Option<i64>,
246    /// The comment.
247    pub comment: String,
248    /// The options.
249    pub options: HashMap<String, String>,
250    #[serde(default)]
251    pub status: FlowStatus,
252    /// The created time
253    #[serde(default)]
254    pub created_time: DateTime<Utc>,
255    /// The updated time.
256    #[serde(default)]
257    pub updated_time: DateTime<Utc>,
258    /// Typed schedule configuration for `EVAL INTERVAL` flows.
259    /// When `eval_interval_secs` is set, this field carries the resolved
260    /// schedule parameters (anchor, start, missed-tick policy, catch-up
261    /// limits). Absent for flows without `EVAL INTERVAL`.
262    #[serde(default)]
263    pub eval_schedule: Option<FlowScheduleConfig>,
264}
265
266impl FlowInfoValue {
267    pub fn is_pending(&self) -> bool {
268        self.status == FlowStatus::PendingSources
269    }
270
271    pub fn is_active(&self) -> bool {
272        self.status == FlowStatus::Active
273    }
274
275    /// Returns the `flownode_id`.
276    pub fn flownode_ids(&self) -> &BTreeMap<FlowPartitionId, FlownodeId> {
277        &self.flownode_ids
278    }
279
280    /// Insert a new flownode id for a partition.
281    pub fn insert_flownode_id(
282        &mut self,
283        partition: FlowPartitionId,
284        node: FlownodeId,
285    ) -> Option<FlownodeId> {
286        self.flownode_ids.insert(partition, node)
287    }
288
289    /// Returns the `source_table`.
290    pub fn source_table_ids(&self) -> &[TableId] {
291        &self.source_table_ids
292    }
293
294    pub fn all_source_table_names(&self) -> &[TableName] {
295        &self.all_source_table_names
296    }
297
298    pub fn unresolved_source_table_names(&self) -> &[TableName] {
299        &self.unresolved_source_table_names
300    }
301
302    pub fn catalog_name(&self) -> &String {
303        &self.catalog_name
304    }
305
306    pub fn query_context(&self) -> &Option<crate::rpc::ddl::QueryContext> {
307        &self.query_context
308    }
309
310    pub fn flow_name(&self) -> &String {
311        &self.flow_name
312    }
313
314    pub fn sink_table_name(&self) -> &TableName {
315        &self.sink_table_name
316    }
317
318    pub fn raw_sql(&self) -> &String {
319        &self.raw_sql
320    }
321
322    pub fn expire_after(&self) -> Option<i64> {
323        self.expire_after
324    }
325
326    pub fn eval_interval(&self) -> Option<i64> {
327        self.eval_interval_secs
328    }
329
330    pub fn comment(&self) -> &String {
331        &self.comment
332    }
333
334    pub fn options(&self) -> &HashMap<String, String> {
335        &self.options
336    }
337
338    pub fn status(&self) -> &FlowStatus {
339        &self.status
340    }
341
342    pub fn created_time(&self) -> &DateTime<Utc> {
343        &self.created_time
344    }
345
346    pub fn updated_time(&self) -> &DateTime<Utc> {
347        &self.updated_time
348    }
349}
350
351pub type FlowInfoManagerRef = Arc<FlowInfoManager>;
352
353/// The manager of [FlowInfoKey].
354pub struct FlowInfoManager {
355    kv_backend: KvBackendRef,
356}
357
358pub fn flow_info_decoder(kv: KeyValue) -> Result<(FlowInfoKey, FlowInfoValue)> {
359    let key = FlowInfoKey::from_bytes(&kv.key)?;
360    let value = FlowInfoValue::try_from_raw_value(&kv.value)?;
361    Ok((key, value))
362}
363
364impl FlowInfoManager {
365    /// Returns a new [FlowInfoManager].
366    pub fn new(kv_backend: KvBackendRef) -> Self {
367        Self { kv_backend }
368    }
369
370    /// Returns the [FlowInfoValue] of specified `flow_id`.
371    pub async fn get(&self, flow_id: FlowId) -> Result<Option<FlowInfoValue>> {
372        let key = FlowInfoKey::new(flow_id).to_bytes();
373        self.kv_backend
374            .get(&key)
375            .await?
376            .map(|x| FlowInfoValue::try_from_raw_value(&x.value))
377            .transpose()
378    }
379
380    /// Returns the [FlowInfoValue] with original bytes of specified `flow_id`.
381    pub async fn get_raw(
382        &self,
383        flow_id: FlowId,
384    ) -> Result<Option<DeserializedValueWithBytes<FlowInfoValue>>> {
385        let key = FlowInfoKey::new(flow_id).to_bytes();
386        self.kv_backend
387            .get(&key)
388            .await?
389            .map(|x| DeserializedValueWithBytes::from_inner_slice(&x.value))
390            .transpose()
391    }
392
393    pub fn flow_infos(&self) -> BoxStream<'static, Result<(FlowId, FlowInfoValue)>> {
394        let start_key = FlowScoped::new(BytesAdapter::from(
395            format!("{FLOW_INFO_KEY_PREFIX}/").into_bytes(),
396        ))
397        .to_bytes();
398        let req = RangeRequest::new().with_prefix(start_key);
399        let stream = PaginationStream::new(
400            self.kv_backend.clone(),
401            req,
402            DEFAULT_PAGE_SIZE,
403            flow_info_decoder,
404        )
405        .into_stream();
406
407        Box::pin(stream.map_ok(|(key, value)| (key.flow_id(), value)))
408    }
409
410    /// Builds a create flow transaction.
411    /// It is expected that the `__flow/info/{flow_id}` wasn't occupied.
412    /// Otherwise, the transaction will retrieve existing value.
413    pub(crate) fn build_create_txn(
414        &self,
415        flow_id: FlowId,
416        flow_value: &FlowInfoValue,
417    ) -> Result<(
418        Txn,
419        impl FnOnce(&mut TxnOpGetResponseSet) -> FlowInfoDecodeResult,
420    )> {
421        let key = FlowInfoKey::new(flow_id).to_bytes();
422        let txn = Txn::put_if_not_exists(key.clone(), flow_value.try_as_raw_value()?);
423
424        Ok((
425            txn,
426            TxnOpGetResponseSet::decode_with(TxnOpGetResponseSet::filter(key)),
427        ))
428    }
429
430    /// Builds a update flow transaction.
431    /// It is expected that the `__flow/info/{flow_id}` IS ALREADY occupied and equal to `prev_flow_value`,
432    /// but the new value can be the same, so to allow replace operation to happen even when the value is the same.
433    /// Otherwise, the transaction will retrieve existing value and fail.
434    pub(crate) fn build_update_txn(
435        &self,
436        flow_id: FlowId,
437        current_flow_value: &DeserializedValueWithBytes<FlowInfoValue>,
438        new_flow_value: &FlowInfoValue,
439    ) -> Result<(
440        Txn,
441        impl FnOnce(&mut TxnOpGetResponseSet) -> FlowInfoDecodeResult,
442    )> {
443        let key = FlowInfoKey::new(flow_id).to_bytes();
444        let raw_value = new_flow_value.try_as_raw_value()?;
445        let prev_value = current_flow_value.get_raw_bytes();
446        let txn = Txn::new()
447            .when(vec![Compare::new(
448                key.clone(),
449                CompareOp::Equal,
450                Some(prev_value),
451            )])
452            .and_then(vec![TxnOp::Put(key.clone(), raw_value)])
453            .or_else(vec![TxnOp::Get(key.clone())]);
454
455        Ok((
456            txn,
457            TxnOpGetResponseSet::decode_with(TxnOpGetResponseSet::filter(key)),
458        ))
459    }
460}
461
462#[cfg(test)]
463mod tests {
464    use super::*;
465
466    #[test]
467    fn test_key_serialization() {
468        let flow_info = FlowInfoKey::new(2);
469        assert_eq!(b"__flow/info/2".to_vec(), flow_info.to_bytes());
470    }
471
472    #[test]
473    fn test_key_deserialization() {
474        let bytes = b"__flow/info/2".to_vec();
475        let key = FlowInfoKey::from_bytes(&bytes).unwrap();
476        assert_eq!(key.flow_id(), 2);
477    }
478}
common_meta/key/flow/flow_info.rs

common_meta/key/flow/
flow_info.rs
