feat(flow): inc query scan bind seq (#7879)

* feat: add sequence scan boundaries and stale detection

Signed-off-by: discord9 <discord9@163.com>

* fix: only scan bind once

Signed-off-by: discord9 <discord9@163.com>

* refactor: flow snapshot decision

Signed-off-by: discord9 <discord9@163.com>

* refactor: one place modify

Signed-off-by: discord9 <discord9@163.com>

* chore: per review

Signed-off-by: discord9 <discord9@163.com>

* refactor: pre review

Signed-off-by: discord9 <discord9@163.com>

* per review

Signed-off-by: discord9 <discord9@163.com>

---------

Signed-off-by: discord9 <discord9@163.com>

src/query/src/dummy_catalog.rs

@@ -43,6 +43,7 @@ use table::metadata::{TableId, TableInfoRef};
 use table::table::scan::RegionScanExec;
 
 use crate::error::{GetRegionMetadataSnafu, Result};
+use crate::options::FlowQueryExtensions;
 
 /// Resolve to the given region (specified by [RegionId]) unconditionally.
 #[derive(Clone, Debug)]
@@ -182,11 +183,26 @@ impl TableProvider for DummyTableProvider {
         request.filters = filters.to_vec();
         request.limit = limit;
 
+        if let Some(query_ctx) = &self.query_ctx {
+            let is_sink_scan = is_sink_scan(query_ctx, self.region_id)
+                .map_err(|e| DataFusionError::External(Box::new(e)))?;
+            apply_cached_snapshot_to_request(query_ctx, self.region_id, is_sink_scan, &mut request);
+        }
+
         let scanner = self
             .engine
             .handle_query(self.region_id, request.clone())
             .await
             .map_err(|e| DataFusionError::External(Box::new(e)))?;
+
+        if request.snapshot_on_scan
+            && let Some(query_ctx) = &self.query_ctx
+            && let Some(snapshot_sequence) = scanner.snapshot_sequence()
+        {
+            bind_snapshot_bound_region_seq(query_ctx, self.region_id, snapshot_sequence)
+                .map_err(|e| DataFusionError::External(Box::new(e)))?;
+        }
+
         let query_memory_tracker = self.engine.query_memory_tracker();
         let mut scan_exec = RegionScanExec::new(scanner, request, query_memory_tracker)?;
         if let Some(query_ctx) = &self.query_ctx {
@@ -295,14 +311,11 @@ impl DummyTableProviderFactory {
                     region_id,
                 })?;
 
-        let scan_request = query_ctx
-            .as_ref()
-            .map(|ctx| ScanRequest {
-                memtable_max_sequence: ctx.get_snapshot(region_id.as_u64()),
-                sst_min_sequence: ctx.sst_min_sequence(region_id.as_u64()),
-                ..Default::default()
-            })
-            .unwrap_or_default();
+        let scan_request = if let Some(ctx) = query_ctx.as_ref() {
+            scan_request_from_query_context(region_id, ctx)?
+        } else {
+            ScanRequest::default()
+        };
 
         Ok(DummyTableProvider {
             region_id,
@@ -314,6 +327,152 @@ impl DummyTableProviderFactory {
     }
 }
 
+fn scan_request_from_query_context(
+    region_id: RegionId,
+    query_ctx: &QueryContext,
+) -> Result<ScanRequest> {
+    let decision = decide_flow_scan(query_ctx, region_id)?;
+    Ok(build_scan_request(query_ctx, region_id, &decision))
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct FlowScanDecision {
+    /// Whether this region is the flow sink-table scan.
+    /// Sink scans intentionally bypass incremental and snapshot-binding semantics.
+    is_sink_scan: bool,
+    /// Whether this scan should bind a memtable upper bound when opening the scan.
+    /// This is only the initial intent; if a cached bound already exists in `query_ctx`,
+    /// we reuse that cached bound instead and clear this flag.
+    snapshot_on_scan: bool,
+    /// Optional lower exclusive memtable sequence bound for incremental reads.
+    /// When set, only rows with sequence strictly greater than this bound are read from memtables.
+    memtable_min_sequence: Option<u64>,
+    /// Optional cached per-region snapshot already bound in `query_ctx`.
+    /// When present, this becomes the effective memtable upper bound and suppresses
+    /// binding a new snapshot on scan open.
+    memtable_max_sequence: Option<u64>,
+}
+
+impl FlowScanDecision {
+    fn plain_scan() -> Self {
+        Self {
+            is_sink_scan: true,
+            snapshot_on_scan: false,
+            memtable_min_sequence: None,
+            memtable_max_sequence: None,
+        }
+    }
+}
+
+fn decide_flow_scan(query_ctx: &QueryContext, region_id: RegionId) -> Result<FlowScanDecision> {
+    let Some(flow_extensions) =
+        FlowQueryExtensions::parse_flow_extensions(&query_ctx.extensions())?
+    else {
+        return Ok(FlowScanDecision {
+            is_sink_scan: false,
+            snapshot_on_scan: false,
+            memtable_min_sequence: None,
+            memtable_max_sequence: query_ctx.get_snapshot(region_id.as_u64()),
+        });
+    };
+
+    // Sink-table scans intentionally bypass all flow scan semantics. They should
+    // behave like plain reads and must not participate in incremental lower bounds
+    // or per-region snapshot binding/reuse.
+    if flow_extensions.sink_table_id == Some(region_id.table_id()) {
+        return Ok(FlowScanDecision::plain_scan());
+    }
+
+    let apply_incremental = flow_extensions.validate_for_scan(region_id)?;
+
+    let memtable_min_sequence = if apply_incremental {
+        flow_extensions
+            .incremental_after_seqs
+            .as_ref()
+            .and_then(|seqs| seqs.get(&region_id.as_u64()))
+            .copied()
+    } else {
+        None
+    };
+
+    let memtable_max_sequence = query_ctx.get_snapshot(region_id.as_u64());
+
+    Ok(FlowScanDecision {
+        is_sink_scan: false,
+        snapshot_on_scan: memtable_max_sequence.is_none()
+            && flow_extensions.should_collect_region_watermark(),
+        memtable_min_sequence,
+        memtable_max_sequence,
+    })
+}
+
+fn build_scan_request(
+    query_ctx: &QueryContext,
+    region_id: RegionId,
+    decision: &FlowScanDecision,
+) -> ScanRequest {
+    // Build the initial scan request from the final decision known at provider creation
+    // time. A later scan may still refresh `memtable_max_sequence` if another source scan
+    // has bound a snapshot into `query_ctx` after this provider was created.
+    ScanRequest {
+        sst_min_sequence: (!decision.is_sink_scan)
+            .then(|| query_ctx.sst_min_sequence(region_id.as_u64()))
+            .flatten(),
+        snapshot_on_scan: decision.snapshot_on_scan,
+        memtable_min_sequence: decision.memtable_min_sequence,
+        memtable_max_sequence: decision.memtable_max_sequence,
+        ..Default::default()
+    }
+}
+
+fn is_sink_scan(query_ctx: &QueryContext, region_id: RegionId) -> Result<bool> {
+    Ok(
+        FlowQueryExtensions::parse_flow_extensions(&query_ctx.extensions())?
+            .is_some_and(|exts| exts.sink_table_id == Some(region_id.table_id())),
+    )
+}
+
+fn apply_cached_snapshot_to_request(
+    query_ctx: &QueryContext,
+    region_id: RegionId,
+    is_sink_scan: bool,
+    scan_request: &mut ScanRequest,
+) {
+    if is_sink_scan {
+        return;
+    }
+
+    if let Some(snapshot_sequence) = query_ctx.get_snapshot(region_id.as_u64()) {
+        // Reuse the previously bound per-region snapshot instead of rebinding a new
+        // upper bound on scan open. This refresh is still needed at scan time because
+        // the provider's cached request may have been built before another source scan
+        // bound the shared query-level snapshot into `query_ctx`.
+        scan_request.memtable_max_sequence = Some(snapshot_sequence);
+        scan_request.snapshot_on_scan = false;
+    }
+}
+
+fn bind_snapshot_bound_region_seq(
+    query_ctx: &QueryContext,
+    region_id: RegionId,
+    snapshot_sequence: u64,
+) -> Result<u64> {
+    if let Some(existing) = query_ctx.get_snapshot(region_id.as_u64()) {
+        if existing != snapshot_sequence {
+            return crate::error::ConflictingSnapshotSequenceSnafu {
+                region_id,
+                existing,
+                new: snapshot_sequence,
+            }
+            .fail();
+        }
+        Ok(existing)
+    } else {
+        query_ctx.set_snapshot(region_id.as_u64(), snapshot_sequence);
+        Ok(snapshot_sequence)
+    }
+}
+
 #[async_trait]
 impl TableProviderFactory for DummyTableProviderFactory {
     async fn create(
@@ -443,3 +602,277 @@ impl CatalogManager for DummyCatalogManager {
         Box::pin(futures::stream::empty())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+    use std::sync::{Arc, RwLock};
+
+    use common_error::ext::ErrorExt;
+    use common_error::status_code::StatusCode;
+    use session::context::QueryContextBuilder;
+
+    use super::*;
+    use crate::error::Error;
+    use crate::options::{FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE, FLOW_SINK_TABLE_ID};
+
+    fn test_region_id() -> RegionId {
+        RegionId::new(1024, 1)
+    }
+
+    #[test]
+    fn test_scan_request_from_query_context_uses_snapshot_bound_intent() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default()
+            .extensions(HashMap::from([(
+                "flow.return_region_seq".to_string(),
+                "true".to_string(),
+            )]))
+            .snapshot_seqs(Arc::new(RwLock::new(HashMap::from([(
+                region_id.as_u64(),
+                42_u64,
+            )]))))
+            .sst_min_sequences(Arc::new(RwLock::new(HashMap::from([(
+                region_id.as_u64(),
+                7_u64,
+            )]))))
+            .build();
+
+        let request = scan_request_from_query_context(region_id, &query_ctx).unwrap();
+
+        assert!(!request.snapshot_on_scan);
+        assert_eq!(request.memtable_max_sequence, Some(42));
+        assert_eq!(request.sst_min_sequence, Some(7));
+    }
+
+    #[test]
+    fn test_scan_request_from_incremental_context_uses_snapshot_bound_intent() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default()
+            .extensions(HashMap::from([(
+                "flow.incremental_after_seqs".to_string(),
+                format!(r#"{{"{}":10}}"#, region_id.as_u64()),
+            )]))
+            .build();
+
+        let request = scan_request_from_query_context(region_id, &query_ctx).unwrap();
+
+        assert!(request.snapshot_on_scan);
+        assert_eq!(request.memtable_min_sequence, Some(10));
+        assert_eq!(request.memtable_max_sequence, None);
+    }
+
+    #[test]
+    fn test_scan_request_from_query_context_keeps_snapshot_fields() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default()
+            .snapshot_seqs(Arc::new(RwLock::new(HashMap::from([(
+                region_id.as_u64(),
+                100,
+            )]))))
+            .sst_min_sequences(Arc::new(RwLock::new(HashMap::from([(
+                region_id.as_u64(),
+                90,
+            )]))))
+            .build();
+
+        let request = scan_request_from_query_context(region_id, &query_ctx).unwrap();
+        assert_eq!(request.memtable_max_sequence, Some(100));
+        assert_eq!(request.sst_min_sequence, Some(90));
+        assert_eq!(request.memtable_min_sequence, None);
+        assert!(!request.snapshot_on_scan);
+    }
+
+    #[test]
+    fn test_scan_request_from_query_context_reuses_existing_snapshot_for_incremental_scan() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default()
+            .extensions(HashMap::from([(
+                FLOW_INCREMENTAL_AFTER_SEQS.to_string(),
+                format!(r#"{{"{}":10}}"#, region_id.as_u64()),
+            )]))
+            .snapshot_seqs(Arc::new(RwLock::new(HashMap::from([(
+                region_id.as_u64(),
+                42_u64,
+            )]))))
+            .build();
+
+        let request = scan_request_from_query_context(region_id, &query_ctx).unwrap();
+
+        assert_eq!(request.memtable_min_sequence, Some(10));
+        assert_eq!(request.memtable_max_sequence, Some(42));
+        assert!(!request.snapshot_on_scan);
+    }
+
+    #[test]
+    fn test_apply_cached_snapshot_to_request_updates_cached_scan_request() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default()
+            .snapshot_seqs(Arc::new(RwLock::new(HashMap::from([(
+                region_id.as_u64(),
+                88_u64,
+            )]))))
+            .build();
+        let mut request = ScanRequest {
+            snapshot_on_scan: true,
+            ..Default::default()
+        };
+
+        apply_cached_snapshot_to_request(&query_ctx, region_id, false, &mut request);
+
+        assert_eq!(request.memtable_max_sequence, Some(88));
+        assert!(!request.snapshot_on_scan);
+    }
+
+    #[test]
+    fn test_apply_cached_snapshot_to_request_skips_sink_scan() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default()
+            .snapshot_seqs(Arc::new(RwLock::new(HashMap::from([(
+                region_id.as_u64(),
+                88_u64,
+            )]))))
+            .build();
+        let mut request = ScanRequest {
+            snapshot_on_scan: true,
+            ..Default::default()
+        };
+
+        apply_cached_snapshot_to_request(&query_ctx, region_id, true, &mut request);
+
+        assert_eq!(request.memtable_max_sequence, None);
+        assert!(request.snapshot_on_scan);
+    }
+
+    #[test]
+    fn test_bind_snapshot_bound_region_seq_reuses_existing_snapshot() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default()
+            .snapshot_seqs(Arc::new(RwLock::new(HashMap::from([(
+                region_id.as_u64(),
+                42_u64,
+            )]))))
+            .build();
+
+        let err = bind_snapshot_bound_region_seq(&query_ctx, region_id, 99).unwrap_err();
+
+        assert!(matches!(err, Error::ConflictingSnapshotSequence { .. }));
+        assert_eq!(query_ctx.get_snapshot(region_id.as_u64()), Some(42));
+    }
+
+    #[test]
+    fn test_bind_snapshot_bound_region_seq_sets_snapshot_once() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default().build();
+
+        let seq = bind_snapshot_bound_region_seq(&query_ctx, region_id, 99).unwrap();
+
+        assert_eq!(seq, 99);
+        assert_eq!(query_ctx.get_snapshot(region_id.as_u64()), Some(99));
+    }
+
+    #[test]
+    fn test_scan_request_from_query_context_applies_incremental_after_seq_for_source_scan() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default()
+            .extensions(HashMap::from([
+                (
+                    FLOW_INCREMENTAL_MODE.to_string(),
+                    "memtable_only".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));
+    }
+
+    #[test]
+    fn test_scan_request_from_query_context_does_not_apply_incremental_for_sink_table() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default()
+            .extensions(HashMap::from([
+                (
+                    FLOW_INCREMENTAL_MODE.to_string(),
+                    "memtable_only".to_string(),
+                ),
+                (
+                    FLOW_INCREMENTAL_AFTER_SEQS.to_string(),
+                    format!(r#"{{"{}":55}}"#, region_id.as_u64()),
+                ),
+                (
+                    FLOW_SINK_TABLE_ID.to_string(),
+                    region_id.table_id().to_string(),
+                ),
+            ]))
+            .snapshot_seqs(Arc::new(RwLock::new(HashMap::from([(
+                region_id.as_u64(),
+                88_u64,
+            )]))))
+            .sst_min_sequences(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, None);
+        assert_eq!(request.memtable_max_sequence, None);
+        assert_eq!(request.sst_min_sequence, None);
+        assert!(!request.snapshot_on_scan);
+    }
+
+    #[test]
+    fn test_scan_request_from_query_context_rejects_missing_memtable_only_region() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default()
+            .extensions(HashMap::from([
+                (
+                    FLOW_INCREMENTAL_MODE.to_string(),
+                    "memtable_only".to_string(),
+                ),
+                (
+                    FLOW_INCREMENTAL_AFTER_SEQS.to_string(),
+                    r#"{"9":55}"#.to_string(),
+                ),
+            ]))
+            .build();
+
+        let err = scan_request_from_query_context(region_id, &query_ctx).unwrap_err();
+        assert!(matches!(err, Error::InvalidQueryContextExtension { .. }));
+    }
+
+    #[test]
+    fn test_scan_request_from_query_context_rejects_invalid_incremental_json() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default()
+            .extensions(HashMap::from([(
+                FLOW_INCREMENTAL_AFTER_SEQS.to_string(),
+                "not-json".to_string(),
+            )]))
+            .build();
+
+        let err = scan_request_from_query_context(region_id, &query_ctx).unwrap_err();
+        assert!(matches!(err, Error::InvalidQueryContextExtension { .. }));
+        assert_eq!(err.status_code(), StatusCode::InvalidArguments);
+    }
+
+    #[test]
+    fn test_scan_request_from_query_context_rejects_invalid_sink_table_id() {
+        let region_id = test_region_id();
+        let query_ctx = QueryContextBuilder::default()
+            .extensions(HashMap::from([(
+                FLOW_SINK_TABLE_ID.to_string(),
+                "abc".to_string(),
+            )]))
+            .build();
+
+        let err = scan_request_from_query_context(region_id, &query_ctx).unwrap_err();
+        assert!(matches!(err, Error::InvalidQueryContextExtension { .. }));
+        assert_eq!(err.status_code(), StatusCode::InvalidArguments);
+    }
+}

src/query/src/error.rs

@@ -375,6 +375,20 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display(
+        "Conflicting snapshot sequence observed for region {} in a single query (existing={}, new={})",
+        region_id,
+        existing,
+        new
+    ))]
+    ConflictingSnapshotSequence {
+        region_id: RegionId,
+        existing: u64,
+        new: u64,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(transparent)]
     Datatypes {
         source: datatypes::error::Error,
@@ -407,7 +421,8 @@ impl ErrorExt for Error {
             | CteColumnSchemaMismatch { .. }
             | ConvertValue { .. }
             | TryIntoDuration { .. }
-            | InvalidQueryContextExtension { .. } => StatusCode::InvalidArguments,
+            | InvalidQueryContextExtension { .. }
+            | ConflictingSnapshotSequence { .. } => StatusCode::InvalidArguments,
 
             BuildBackend { .. } | ListObjects { .. } => StatusCode::StorageUnavailable,
 

src/query/src/options.rs

@@ -71,7 +71,21 @@ pub struct FlowQueryExtensions {
 }
 
 impl FlowQueryExtensions {
-    pub fn from_extensions(extensions: &HashMap<String, String>) -> Result<Self> {
+    /// Parses flow-specific query extensions when any flow key is present.
+    ///
+    /// Returns `Ok(None)` for ordinary queries with no flow-related extensions,
+    /// `Ok(Some(_))` when flow context is present and valid, and `Err(_)` when a
+    /// flow-related extension is present but malformed or incomplete.
+    pub fn parse_flow_extensions(extensions: &HashMap<String, String>) -> Result<Option<Self>> {
+        let has_flow_context = extensions.contains_key(FLOW_INCREMENTAL_AFTER_SEQS)
+            || extensions.contains_key(FLOW_INCREMENTAL_MODE)
+            || extensions.contains_key(FLOW_RETURN_REGION_SEQ)
+            || extensions.contains_key(FLOW_SINK_TABLE_ID);
+
+        if !has_flow_context {
+            return Ok(None);
+        }
+
         let incremental_mode = extensions
             .get(FLOW_INCREMENTAL_MODE)
             .map(|value| match value.as_str() {
@@ -127,12 +141,12 @@ impl FlowQueryExtensions {
             }
         }
 
-        Ok(Self {
+        Ok(Some(Self {
             incremental_after_seqs,
             incremental_mode,
             return_region_seq,
             sink_table_id,
-        })
+        }))
     }
 
     pub fn validate_for_scan(&self, source_region_id: RegionId) -> Result<bool> {
@@ -230,14 +244,11 @@ mod flow_extension_tests {
     use super::*;
 
     #[test]
-    fn test_parse_flow_extensions_default() {
+    fn test_parse_flow_extensions_returns_none_for_non_flow_query() {
         let exts = HashMap::new();
-        let parsed = FlowQueryExtensions::from_extensions(&exts).unwrap();
+        let parsed = FlowQueryExtensions::parse_flow_extensions(&exts).unwrap();
 
-        assert_eq!(parsed.incremental_mode, None);
-        assert_eq!(parsed.incremental_after_seqs, None);
-        assert!(!parsed.return_region_seq);
-        assert_eq!(parsed.sink_table_id, None);
+        assert_eq!(parsed, None);
     }
 
     #[test]
@@ -255,7 +266,9 @@ mod flow_extension_tests {
             (FLOW_SINK_TABLE_ID.to_string(), "1024".to_string()),
         ]);
 
-        let parsed = FlowQueryExtensions::from_extensions(&exts).unwrap();
+        let parsed = FlowQueryExtensions::parse_flow_extensions(&exts)
+            .unwrap()
+            .unwrap();
         assert_eq!(
             parsed.incremental_mode,
             Some(FlowIncrementalMode::MemtableOnly)
@@ -275,7 +288,7 @@ mod flow_extension_tests {
             FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY.to_string(),
         )]);
 
-        let err = FlowQueryExtensions::from_extensions(&exts).unwrap_err();
+        let err = FlowQueryExtensions::parse_flow_extensions(&exts).unwrap_err();
         assert!(format!("{err}").contains(FLOW_INCREMENTAL_AFTER_SEQS));
     }
 
@@ -283,7 +296,7 @@ mod flow_extension_tests {
     fn test_parse_flow_extensions_invalid_mode() {
         let exts = HashMap::from([(FLOW_INCREMENTAL_MODE.to_string(), "foo".to_string())]);
 
-        let err = FlowQueryExtensions::from_extensions(&exts).unwrap_err();
+        let err = FlowQueryExtensions::parse_flow_extensions(&exts).unwrap_err();
         assert!(format!("{err}").contains(FLOW_INCREMENTAL_MODE));
     }
 
@@ -300,7 +313,7 @@ mod flow_extension_tests {
             ),
         ]);
 
-        let err = FlowQueryExtensions::from_extensions(&exts).unwrap_err();
+        let err = FlowQueryExtensions::parse_flow_extensions(&exts).unwrap_err();
         assert!(format!("{err}").contains(FLOW_INCREMENTAL_AFTER_SEQS));
     }
 
@@ -311,7 +324,9 @@ mod flow_extension_tests {
             r#"{"1":"10","2":"20"}"#.to_string(),
         )]);
 
-        let parsed = FlowQueryExtensions::from_extensions(&exts).unwrap();
+        let parsed = FlowQueryExtensions::parse_flow_extensions(&exts)
+            .unwrap()
+            .unwrap();
         assert_eq!(
             parsed.incremental_after_seqs.unwrap(),
             HashMap::from([(1, 10), (2, 20)])
@@ -325,7 +340,7 @@ mod flow_extension_tests {
             r#"{"1":true}"#.to_string(),
         )]);
 
-        let err = FlowQueryExtensions::from_extensions(&exts).unwrap_err();
+        let err = FlowQueryExtensions::parse_flow_extensions(&exts).unwrap_err();
         assert!(format!("{err}").contains(FLOW_INCREMENTAL_AFTER_SEQS));
     }
 
@@ -333,7 +348,7 @@ mod flow_extension_tests {
     fn test_parse_flow_extensions_invalid_sink_table_id() {
         let exts = HashMap::from([(FLOW_SINK_TABLE_ID.to_string(), "x".to_string())]);
 
-        let err = FlowQueryExtensions::from_extensions(&exts).unwrap_err();
+        let err = FlowQueryExtensions::parse_flow_extensions(&exts).unwrap_err();
         assert!(format!("{err}").contains(FLOW_SINK_TABLE_ID));
     }
 
@@ -352,7 +367,9 @@ mod flow_extension_tests {
             ),
         ]);
 
-        let parsed = FlowQueryExtensions::from_extensions(&exts).unwrap();
+        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"));
     }
@@ -372,7 +389,9 @@ mod flow_extension_tests {
             (FLOW_SINK_TABLE_ID.to_string(), "1024".to_string()),
         ]);
 
-        let parsed = FlowQueryExtensions::from_extensions(&exts).unwrap();
+        let parsed = FlowQueryExtensions::parse_flow_extensions(&exts)
+            .unwrap()
+            .unwrap();
         let apply_incremental = parsed.validate_for_scan(source_region_id).unwrap();
         assert!(!apply_incremental);
     }
@@ -400,4 +419,43 @@ mod flow_extension_tests {
         };
         assert!(parsed.should_collect_region_watermark());
     }
+
+    #[test]
+    fn test_parse_flow_extensions_return_region_seq_only_returns_some() {
+        let exts = HashMap::from([(FLOW_RETURN_REGION_SEQ.to_string(), "true".to_string())]);
+
+        let parsed = FlowQueryExtensions::parse_flow_extensions(&exts)
+            .unwrap()
+            .unwrap();
+
+        assert!(parsed.return_region_seq);
+    }
+
+    #[test]
+    fn test_parse_flow_extensions_sink_table_only_returns_some() {
+        let exts = HashMap::from([(FLOW_SINK_TABLE_ID.to_string(), "1024".to_string())]);
+
+        let parsed = FlowQueryExtensions::parse_flow_extensions(&exts)
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(parsed.sink_table_id, Some(1024));
+    }
+
+    #[test]
+    fn test_parse_flow_extensions_incremental_after_seqs_only_returns_some() {
+        let exts = HashMap::from([(
+            FLOW_INCREMENTAL_AFTER_SEQS.to_string(),
+            r#"{"1":10}"#.to_string(),
+        )]);
+
+        let parsed = FlowQueryExtensions::parse_flow_extensions(&exts)
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(
+            parsed.incremental_after_seqs,
+            Some(HashMap::from([(1, 10)]))
+        );
+    }
 }