feat(flow): activate sequence_range incremental mode

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
2026-10-27 14:16:18 +00:00 · 2026-08-14 03:22:18 +08:00
parent a29a2a5ede
commit 23fe843cc5
12 changed files with 528 additions and 37 deletions
@@ -371,6 +371,10 @@ struct FlowScanDecision {
    memtable_max_sequence: Option<u64>,
    /// Whether to skip SST files for memtable-only incremental source scans.
    skip_sst_files: bool,
+    /// Explicit intent to read an exact row-level sequence delta `(min, max]`
+    /// across memtables and all SST files (`flow.incremental_mode=sequence_range`).
+    /// Historical `memtable_only` reads never set this.
+    exact_sequence_range: bool,
 }

 impl FlowScanDecision {
@@ -381,6 +385,7 @@ impl FlowScanDecision {
            memtable_min_sequence: None,
            memtable_max_sequence: None,
            skip_sst_files: false,
+            exact_sequence_range: false,
        }
    }
 }
@@ -395,6 +400,7 @@ fn decide_flow_scan(query_ctx: &QueryContext, region_id: RegionId) -> Result<Flo
            memtable_min_sequence: None,
            memtable_max_sequence: query_ctx.get_snapshot(region_id.as_u64()),
            skip_sst_files: false,
+            exact_sequence_range: false,
        });
    };

@@ -426,9 +432,14 @@ fn decide_flow_scan(query_ctx: &QueryContext, region_id: RegionId) -> Result<Flo
    // into SSTs, instead of silently bypassing the stale-fence check. If a future
    // incremental delta also carries an upper bound, the lower-bound stale check
    // still proves whether memtable-only is safe.
-    let skip_sst_files = apply_incremental
-        && memtable_min_sequence.is_some()
-        && flow_extensions.incremental_mode == Some(FlowIncrementalMode::MemtableOnly);
+    let is_memtable_only =
+        flow_extensions.incremental_mode == Some(FlowIncrementalMode::MemtableOnly);
+    // `sequence_range` is the exact row-level delta: it always includes SSTs and
+    // explicitly requests row-level filtering across memtables and every SST.
+    let is_sequence_range =
+        flow_extensions.incremental_mode == Some(FlowIncrementalMode::SequenceRange);
+
+    let skip_sst_files = apply_incremental && memtable_min_sequence.is_some() && is_memtable_only;

    Ok(FlowScanDecision {
        is_sink_scan: false,
@@ -437,6 +448,9 @@ fn decide_flow_scan(query_ctx: &QueryContext, region_id: RegionId) -> Result<Flo
        memtable_min_sequence,
        memtable_max_sequence,
        skip_sst_files,
+        exact_sequence_range: apply_incremental
+            && memtable_min_sequence.is_some()
+            && is_sequence_range,
    })
 }

@@ -456,6 +470,7 @@ fn build_scan_request(
        snapshot_on_scan: decision.snapshot_on_scan,
        memtable_min_sequence: decision.memtable_min_sequence,
        memtable_max_sequence: decision.memtable_max_sequence,
+        exact_sequence_range: decision.exact_sequence_range,
        ..Default::default()
    }
 }
@@ -791,6 +806,62 @@ mod tests {
        assert_eq!(request.memtable_max_sequence, Some(42));
        assert!(!request.snapshot_on_scan);
        assert!(request.skip_sst_files);
+        // Historical memtable-only never requests the exact capability.
+        assert!(!request.exact_sequence_range);
+    }
+
+    #[test]
+    fn test_scan_request_from_query_context_sequence_range_mode_includes_ssts() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default()
+            .extensions(HashMap::from([
+                (
+                    FLOW_INCREMENTAL_MODE.to_string(),
+                    "sequence_range".to_string(),
+                ),
+                (
+                    FLOW_INCREMENTAL_AFTER_SEQS.to_string(),
+                    format!(r#"{{"{}":55}}"#, region_id.as_u64()),
+                ),
+            ]))
+            .snapshot_seqs(Arc::new(RwLock::new(HashMap::from([(
+                region_id.as_u64(),
+                77_u64,
+            )]))))
+            .build();
+
+        let request = scan_request_from_query_context(region_id, &query_ctx).unwrap();
+
+        assert_eq!(request.memtable_min_sequence, Some(55));
+        assert_eq!(request.memtable_max_sequence, Some(77));
+        assert!(!request.skip_sst_files);
+        assert!(request.exact_sequence_range);
+        assert!(!request.snapshot_on_scan);
+    }
+
+    #[test]
+    fn test_scan_request_from_query_context_sequence_range_binds_snapshot_on_open() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default()
+            .extensions(HashMap::from([
+                (
+                    FLOW_INCREMENTAL_MODE.to_string(),
+                    "sequence_range".to_string(),
+                ),
+                (
+                    FLOW_INCREMENTAL_AFTER_SEQS.to_string(),
+                    format!(r#"{{"{}":55}}"#, region_id.as_u64()),
+                ),
+            ]))
+            .build();
+
+        let request = scan_request_from_query_context(region_id, &query_ctx).unwrap();
+
+        assert_eq!(request.memtable_min_sequence, Some(55));
+        assert_eq!(request.memtable_max_sequence, None);
+        assert!(request.snapshot_on_scan);
+        assert!(!request.skip_sst_files);
+        assert!(request.exact_sequence_range);
    }

    #[test]
@@ -37,6 +37,7 @@ pub const QUERY_ENABLE_REMOTE_DYNAMIC_FILTER_PUSHDOWN: &str =
    "query.enable_remote_dynamic_filter_pushdown";

 pub const FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY: &str = "memtable_only";
+pub const FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE: &str = "sequence_range";

 /// Query engine config
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -69,7 +70,15 @@ impl Default for QueryOptions {

 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum FlowIncrementalMode {
+    /// Historical incremental delta: read only memtables newer than the
+    /// checkpoint, skipping SSTs entirely. Fences are never relaxed by region
+    /// options; a checkpoint behind the flushed frontier stays stale.
    MemtableOnly,
+    /// Exact row-level sequence delta `(C, H]` across memtables and all SST
+    /// files. Requires the source region to preserve per-row sequences; when the
+    /// exact capability is unavailable the engine returns a structured
+    /// stale/unsupported error instead of silently approximating.
+    SequenceRange,
 }

 #[derive(Debug, Clone, PartialEq, Eq, Default)]
@@ -106,6 +115,9 @@ impl FlowQueryExtensions {
                v if v.eq_ignore_ascii_case(FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY) => {
                    Ok(FlowIncrementalMode::MemtableOnly)
                }
+                v if v.eq_ignore_ascii_case(FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE) => {
+                    Ok(FlowIncrementalMode::SequenceRange)
+                }
                _ => Err(invalid_query_context_extension(format!(
                    "Invalid value for {}: {}",
                    FLOW_INCREMENTAL_MODE, value
@@ -136,13 +148,25 @@ impl FlowQueryExtensions {
            })
            .transpose()?;

-        if matches!(incremental_mode, Some(FlowIncrementalMode::MemtableOnly)) {
+        if matches!(
+            incremental_mode,
+            Some(FlowIncrementalMode::MemtableOnly | FlowIncrementalMode::SequenceRange)
+        ) {
            let after_seqs = incremental_after_seqs.as_ref().ok_or_else(|| {
                invalid_query_context_extension(format!(
                    "{} is required when {}={}.",
                    FLOW_INCREMENTAL_AFTER_SEQS,
                    FLOW_INCREMENTAL_MODE,
-                    FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY
+                    incremental_mode
+                        .map(|m| match m {
+                            FlowIncrementalMode::MemtableOnly => {
+                                FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY
+                            }
+                            FlowIncrementalMode::SequenceRange => {
+                                FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE
+                            }
+                        })
+                        .unwrap_or_default()
                ))
            })?;
            if after_seqs.is_empty() {
@@ -150,7 +174,16 @@ impl FlowQueryExtensions {
                    "{} must not be empty when {}={}.",
                    FLOW_INCREMENTAL_AFTER_SEQS,
                    FLOW_INCREMENTAL_MODE,
-                    FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY
+                    incremental_mode
+                        .map(|m| match m {
+                            FlowIncrementalMode::MemtableOnly => {
+                                FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY
+                            }
+                            FlowIncrementalMode::SequenceRange => {
+                                FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE
+                            }
+                        })
+                        .unwrap_or_default()
                )));
            }
        }
@@ -170,18 +203,18 @@ impl FlowQueryExtensions {

        if matches!(
            self.incremental_mode,
-            Some(FlowIncrementalMode::MemtableOnly)
+            Some(FlowIncrementalMode::MemtableOnly | FlowIncrementalMode::SequenceRange)
        ) {
            let after_seqs = self.incremental_after_seqs.as_ref().ok_or_else(|| {
                invalid_query_context_extension(format!(
-                    "{} is required when {}=memtable_only.",
+                    "{} is required when {}=memtable_only or sequence_range.",
                    FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE
                ))
            })?;

            if !after_seqs.contains_key(&source_region_id.as_u64()) {
                return Err(invalid_query_context_extension(format!(
-                    "Missing region {} in {} when {}=memtable_only.",
+                    "Missing region {} in {} when {}=memtable_only or sequence_range.",
                    source_region_id, FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE
                )));
            }
@@ -449,6 +482,67 @@ mod flow_extension_tests {
        assert_eq!(parsed.sink_table_id, Some(1024));
    }

+    #[test]
+    fn test_parse_flow_extensions_sequence_range_success() {
+        let exts = HashMap::from([
+            (
+                FLOW_INCREMENTAL_MODE.to_string(),
+                FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE.to_string(),
+            ),
+            (
+                FLOW_INCREMENTAL_AFTER_SEQS.to_string(),
+                r#"{"1":10,"2":20}"#.to_string(),
+            ),
+            (FLOW_RETURN_REGION_SEQ.to_string(), "true".to_string()),
+        ]);
+
+        let parsed = FlowQueryExtensions::parse_flow_extensions(&exts)
+            .unwrap()
+            .unwrap();
+        assert_eq!(
+            parsed.incremental_mode,
+            Some(FlowIncrementalMode::SequenceRange)
+        );
+        assert_eq!(
+            parsed.incremental_after_seqs.unwrap(),
+            HashMap::from([(1, 10), (2, 20)])
+        );
+        assert!(parsed.return_region_seq);
+    }
+
+    #[test]
+    fn test_parse_flow_extensions_sequence_range_requires_after_seqs() {
+        let exts = HashMap::from([(
+            FLOW_INCREMENTAL_MODE.to_string(),
+            FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE.to_string(),
+        )]);
+
+        let err = FlowQueryExtensions::parse_flow_extensions(&exts).unwrap_err();
+        assert!(format!("{err}").contains(FLOW_INCREMENTAL_AFTER_SEQS));
+    }
+
+    #[test]
+    fn test_validate_for_scan_sequence_range_missing_source_region() {
+        let source_region_id = RegionId::new(100, 2);
+        let existing_region_id = RegionId::new(100, 1);
+        let exts = HashMap::from([
+            (
+                FLOW_INCREMENTAL_MODE.to_string(),
+                FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE.to_string(),
+            ),
+            (
+                FLOW_INCREMENTAL_AFTER_SEQS.to_string(),
+                format!(r#"{{"{}":10}}"#, existing_region_id.as_u64()),
+            ),
+        ]);
+
+        let parsed = FlowQueryExtensions::parse_flow_extensions(&exts)
+            .unwrap()
+            .unwrap();
+        let err = parsed.validate_for_scan(source_region_id).unwrap_err();
+        assert!(format!("{err}").contains("Missing region"));
+    }
+
    #[test]
    fn test_parse_flow_extensions_mode_requires_after_seqs() {
        let exts = HashMap::from([(
@@ -33,7 +33,7 @@ use common_datasource::file_format::{FileFormat, Format, infer_schemas};
 use common_datasource::lister::{Lister, Source};
 use common_datasource::object_store::{LocalFileAccess, build_backend_with_path};
 use common_meta::SchemaOptions;
-use common_meta::ddl::create_flow::FlowType;
+use common_meta::ddl::create_flow::{FlowType, INTERNAL_INCREMENTAL_MODE_KEY};
 use common_meta::key::flow::flow_info::FlowInfoValue;
 use common_query::Output;
 use common_query::prelude::greptime_timestamp;
@@ -1079,7 +1079,7 @@ pub fn show_create_flow(
        comment,
        flow_options: OptionMap::from_filtered_string_map(
            flow_val.options(),
-            &[FlowType::FLOW_TYPE_KEY],
+            &[FlowType::FLOW_TYPE_KEY, INTERNAL_INCREMENTAL_MODE_KEY],
        ),
        query,
    };
@@ -1384,8 +1384,14 @@ pub async fn show_processlist(

 #[cfg(test)]
 mod test {
+    use std::collections::BTreeMap;
    use std::sync::Arc;

+    use chrono::Utc;
+    use common_meta::ddl::create_flow::{
+        DEFER_ON_MISSING_SOURCE_KEY, FlowType, INTERNAL_INCREMENTAL_MODE_KEY,
+    };
+    use common_meta::key::flow::flow_info::{FlowInfoValue, FlowStatus};
    use common_query::{Output, OutputData};
    use common_recordbatch::{RecordBatch, RecordBatches};
    use common_time::Timezone;
@@ -1399,9 +1405,10 @@ mod test {
    use sql::ast::{Ident, ObjectName};
    use sql::statements::show::ShowVariables;
    use table::TableRef;
+    use table::table_name::TableName;
    use table::test_util::MemTable;

-    use super::{describe_column_type_name, show_variable};
+    use super::{describe_column_type_name, show_create_flow, show_variable};
    use crate::error;
    use crate::error::Result;
    use crate::sql::{
@@ -1530,4 +1537,57 @@ mod test {
            Err(e) => Err(e),
        }
    }
+
+    #[test]
+    fn test_show_create_flow_hides_internal_options() {
+        let flow_info = FlowInfoValue {
+            source_table_ids: vec![],
+            all_source_table_names: vec![],
+            unresolved_source_table_names: vec![],
+            sink_table_name: TableName::new("greptime", "public", "my_sink"),
+            flownode_ids: BTreeMap::new(),
+            catalog_name: "greptime".to_string(),
+            query_context: None,
+            flow_name: "my_flow".to_string(),
+            raw_sql: "SELECT number FROM numbers".to_string(),
+            expire_after: None,
+            eval_interval_secs: None,
+            comment: String::new(),
+            options: std::collections::HashMap::from([
+                (DEFER_ON_MISSING_SOURCE_KEY.to_string(), "true".to_string()),
+                (FlowType::FLOW_TYPE_KEY.to_string(), "batching".to_string()),
+                (
+                    INTERNAL_INCREMENTAL_MODE_KEY.to_string(),
+                    "sequence_range".to_string(),
+                ),
+            ]),
+            status: FlowStatus::Active,
+            created_time: Utc::now(),
+            updated_time: Utc::now(),
+            eval_schedule: None,
+        };
+        let ctx = Arc::new(QueryContextBuilder::default().build());
+        let output = show_create_flow(
+            ObjectName::from(vec![Ident::new("my_flow")]),
+            flow_info,
+            ctx,
+        )
+        .unwrap();
+        let sql = match output.data {
+            OutputData::RecordBatches(record) => record
+                .take()
+                .first()
+                .unwrap()
+                .iter_column_as_string(1)
+                .next()
+                .unwrap()
+                .unwrap(),
+            _ => unreachable!(),
+        };
+
+        // The user option survives and the reserved internal keys are hidden.
+        assert!(sql.contains("defer_on_missing_source = 'true'"));
+        assert!(!sql.contains(FlowType::FLOW_TYPE_KEY));
+        assert!(!sql.contains(INTERNAL_INCREMENTAL_MODE_KEY));
+    }
 }