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::with_anchor(Self::DEFAULT_ANCHOR_SECS, start_secs, eval_interval_secs)
176    }
177
178    /// Builds a schedule anchored at `anchor_secs` (the `EVAL OFFSET` phase,
179    /// in Unix epoch seconds) with `start_secs` as the first scheduled time.
180    /// Callers must ensure `0 <= anchor_secs < eval_interval_secs` and that
181    /// `start_secs` lies on an `anchor_secs + k * eval_interval_secs` boundary.
182    pub fn with_anchor(anchor_secs: i64, start_secs: i64, eval_interval_secs: i64) -> Self {
183        Self {
184            anchor_secs,
185            start_secs,
186            missed_tick_policy: FlowMissedTickPolicy::BoundedCatchUp,
187            catchup_max_runs: Self::DEFAULT_CATCHUP_MAX_RUNS,
188            catchup_max_lag_secs: Self::catchup_max_lag_secs_for_interval(eval_interval_secs),
189        }
190    }
191}
192
193impl Default for FlowScheduleConfig {
194    fn default() -> Self {
195        Self {
196            anchor_secs: Self::DEFAULT_ANCHOR_SECS,
197            start_secs: 0,
198            missed_tick_policy: FlowMissedTickPolicy::default(),
199            catchup_max_runs: Self::default_catchup_max_runs(),
200            catchup_max_lag_secs: Self::default_catchup_max_lag_secs(),
201        }
202    }
203}
204
205/// Policy for handling flow evaluation scheduled times that were missed.
206#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
207pub enum FlowMissedTickPolicy {
208    /// Keep the most recent `catchup_max_runs` due scheduled times within
209    /// `catchup_max_lag_secs`, drop older ones.
210    #[default]
211    #[serde(rename = "bounded_catch_up")]
212    BoundedCatchUp,
213    /// Skip all missed scheduled times; only execute the single most recent one.
214    #[serde(rename = "skip")]
215    Skip,
216}
217
218// The metadata of the flow.
219#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
220pub struct FlowInfoValue {
221    /// The source tables used by the flow.
222    #[serde(default)]
223    pub source_table_ids: Vec<TableId>,
224    #[serde(default)]
225    pub all_source_table_names: Vec<TableName>,
226    #[serde(default)]
227    pub unresolved_source_table_names: Vec<TableName>,
228    /// The sink table used by the flow.
229    pub sink_table_name: TableName,
230    /// Which flow nodes this flow is running on.
231    pub flownode_ids: BTreeMap<FlowPartitionId, FlownodeId>,
232    /// The catalog name.
233    pub catalog_name: String,
234    /// The query context used when create flow.
235    /// Although flow doesn't belong to any schema, this query_context is needed to remember
236    /// the query context when `create_flow` is executed
237    /// for recovering flow using the same sql&query_context after db restart.
238    /// if none, should use default query context
239    #[serde(default)]
240    pub query_context: Option<crate::rpc::ddl::QueryContext>,
241    /// The flow name.
242    pub flow_name: String,
243    /// The raw sql.
244    pub raw_sql: String,
245    /// The expr of expire.
246    /// Duration in seconds as `i64`.
247    pub expire_after: Option<i64>,
248    /// The eval interval.
249    /// Duration in seconds as `i64`.
250    /// If `None`, will automatically decide when to evaluate the flow.
251    /// If `Some`, it will be evaluated every `eval_interval` seconds.
252    #[serde(default)]
253    pub eval_interval_secs: Option<i64>,
254    /// The comment.
255    pub comment: String,
256    /// The options.
257    pub options: HashMap<String, String>,
258    #[serde(default)]
259    pub status: FlowStatus,
260    /// The created time
261    #[serde(default)]
262    pub created_time: DateTime<Utc>,
263    /// The updated time.
264    #[serde(default)]
265    pub updated_time: DateTime<Utc>,
266    /// Typed schedule configuration for `EVAL INTERVAL` flows.
267    /// When `eval_interval_secs` is set, this field carries the resolved
268    /// schedule parameters (anchor, start, missed-tick policy, catch-up
269    /// limits). Absent for flows without `EVAL INTERVAL`.
270    #[serde(default)]
271    pub eval_schedule: Option<FlowScheduleConfig>,
272}
273
274impl FlowInfoValue {
275    pub fn is_pending(&self) -> bool {
276        self.status == FlowStatus::PendingSources
277    }
278
279    pub fn is_active(&self) -> bool {
280        self.status == FlowStatus::Active
281    }
282
283    /// Returns the `flownode_id`.
284    pub fn flownode_ids(&self) -> &BTreeMap<FlowPartitionId, FlownodeId> {
285        &self.flownode_ids
286    }
287
288    /// Insert a new flownode id for a partition.
289    pub fn insert_flownode_id(
290        &mut self,
291        partition: FlowPartitionId,
292        node: FlownodeId,
293    ) -> Option<FlownodeId> {
294        self.flownode_ids.insert(partition, node)
295    }
296
297    /// Returns the `source_table`.
298    pub fn source_table_ids(&self) -> &[TableId] {
299        &self.source_table_ids
300    }
301
302    pub fn all_source_table_names(&self) -> &[TableName] {
303        &self.all_source_table_names
304    }
305
306    pub fn unresolved_source_table_names(&self) -> &[TableName] {
307        &self.unresolved_source_table_names
308    }
309
310    pub fn catalog_name(&self) -> &String {
311        &self.catalog_name
312    }
313
314    pub fn query_context(&self) -> &Option<crate::rpc::ddl::QueryContext> {
315        &self.query_context
316    }
317
318    pub fn flow_name(&self) -> &String {
319        &self.flow_name
320    }
321
322    pub fn sink_table_name(&self) -> &TableName {
323        &self.sink_table_name
324    }
325
326    pub fn raw_sql(&self) -> &String {
327        &self.raw_sql
328    }
329
330    pub fn expire_after(&self) -> Option<i64> {
331        self.expire_after
332    }
333
334    pub fn eval_interval(&self) -> Option<i64> {
335        self.eval_interval_secs
336    }
337
338    pub fn comment(&self) -> &String {
339        &self.comment
340    }
341
342    pub fn options(&self) -> &HashMap<String, String> {
343        &self.options
344    }
345
346    pub fn status(&self) -> &FlowStatus {
347        &self.status
348    }
349
350    pub fn created_time(&self) -> &DateTime<Utc> {
351        &self.created_time
352    }
353
354    pub fn updated_time(&self) -> &DateTime<Utc> {
355        &self.updated_time
356    }
357}
358
359pub type FlowInfoManagerRef = Arc<FlowInfoManager>;
360
361/// The manager of [FlowInfoKey].
362pub struct FlowInfoManager {
363    kv_backend: KvBackendRef,
364}
365
366pub fn flow_info_decoder(kv: KeyValue) -> Result<(FlowInfoKey, FlowInfoValue)> {
367    let key = FlowInfoKey::from_bytes(&kv.key)?;
368    let value = FlowInfoValue::try_from_raw_value(&kv.value)?;
369    Ok((key, value))
370}
371
372impl FlowInfoManager {
373    /// Returns a new [FlowInfoManager].
374    pub fn new(kv_backend: KvBackendRef) -> Self {
375        Self { kv_backend }
376    }
377
378    /// Returns the [FlowInfoValue] of specified `flow_id`.
379    pub async fn get(&self, flow_id: FlowId) -> Result<Option<FlowInfoValue>> {
380        let key = FlowInfoKey::new(flow_id).to_bytes();
381        self.kv_backend
382            .get(&key)
383            .await?
384            .map(|x| FlowInfoValue::try_from_raw_value(&x.value))
385            .transpose()
386    }
387
388    /// Returns the [FlowInfoValue] with original bytes of specified `flow_id`.
389    pub async fn get_raw(
390        &self,
391        flow_id: FlowId,
392    ) -> Result<Option<DeserializedValueWithBytes<FlowInfoValue>>> {
393        let key = FlowInfoKey::new(flow_id).to_bytes();
394        self.kv_backend
395            .get(&key)
396            .await?
397            .map(|x| DeserializedValueWithBytes::from_inner_slice(&x.value))
398            .transpose()
399    }
400
401    pub fn flow_infos(&self) -> BoxStream<'static, Result<(FlowId, FlowInfoValue)>> {
402        let start_key = FlowScoped::new(BytesAdapter::from(
403            format!("{FLOW_INFO_KEY_PREFIX}/").into_bytes(),
404        ))
405        .to_bytes();
406        let req = RangeRequest::new().with_prefix(start_key);
407        let stream = PaginationStream::new(
408            self.kv_backend.clone(),
409            req,
410            DEFAULT_PAGE_SIZE,
411            flow_info_decoder,
412        )
413        .into_stream();
414
415        Box::pin(stream.map_ok(|(key, value)| (key.flow_id(), value)))
416    }
417
418    /// Builds a create flow transaction.
419    /// It is expected that the `__flow/info/{flow_id}` wasn't occupied.
420    /// Otherwise, the transaction will retrieve existing value.
421    pub(crate) fn build_create_txn(
422        &self,
423        flow_id: FlowId,
424        flow_value: &FlowInfoValue,
425    ) -> Result<(
426        Txn,
427        impl FnOnce(&mut TxnOpGetResponseSet) -> FlowInfoDecodeResult,
428    )> {
429        let key = FlowInfoKey::new(flow_id).to_bytes();
430        let txn = Txn::put_if_not_exists(key.clone(), flow_value.try_as_raw_value()?);
431
432        Ok((
433            txn,
434            TxnOpGetResponseSet::decode_with(TxnOpGetResponseSet::filter(key)),
435        ))
436    }
437
438    /// Builds a update flow transaction.
439    /// It is expected that the `__flow/info/{flow_id}` IS ALREADY occupied and equal to `prev_flow_value`,
440    /// but the new value can be the same, so to allow replace operation to happen even when the value is the same.
441    /// Otherwise, the transaction will retrieve existing value and fail.
442    pub(crate) fn build_update_txn(
443        &self,
444        flow_id: FlowId,
445        current_flow_value: &DeserializedValueWithBytes<FlowInfoValue>,
446        new_flow_value: &FlowInfoValue,
447    ) -> Result<(
448        Txn,
449        impl FnOnce(&mut TxnOpGetResponseSet) -> FlowInfoDecodeResult,
450    )> {
451        let key = FlowInfoKey::new(flow_id).to_bytes();
452        let raw_value = new_flow_value.try_as_raw_value()?;
453        let prev_value = current_flow_value.get_raw_bytes();
454        let txn = Txn::new()
455            .when(vec![Compare::new(
456                key.clone(),
457                CompareOp::Equal,
458                Some(prev_value),
459            )])
460            .and_then(vec![TxnOp::Put(key.clone(), raw_value)])
461            .or_else(vec![TxnOp::Get(key.clone())]);
462
463        Ok((
464            txn,
465            TxnOpGetResponseSet::decode_with(TxnOpGetResponseSet::filter(key)),
466        ))
467    }
468}
469
470#[cfg(test)]
471mod tests {
472    use super::*;
473
474    #[test]
475    fn test_key_serialization() {
476        let flow_info = FlowInfoKey::new(2);
477        assert_eq!(b"__flow/info/2".to_vec(), flow_info.to_bytes());
478    }
479
480    #[test]
481    fn test_key_deserialization() {
482        let bytes = b"__flow/info/2".to_vec();
483        let key = FlowInfoKey::from_bytes(&bytes).unwrap();
484        assert_eq!(key.flow_id(), 2);
485    }
486}
common_meta/key/flow/flow_info.rs

common_meta/key/flow/
flow_info.rs
