diff --git a/src/catalog/src/system_schema/information_schema/flows.rs b/src/catalog/src/system_schema/information_schema/flows.rs index 8002b17888..74cde4974b 100644 --- a/src/catalog/src/system_schema/information_schema/flows.rs +++ b/src/catalog/src/system_schema/information_schema/flows.rs @@ -16,7 +16,7 @@ use std::sync::{Arc, Weak}; use common_catalog::consts::INFORMATION_SCHEMA_FLOW_TABLE_ID; use common_error::ext::BoxedError; -use common_meta::ddl::create_flow::FlowType; +use common_meta::ddl::create_flow::{FlowType, INTERNAL_INCREMENTAL_MODE_KEY}; use common_meta::key::FlowId; use common_meta::key::flow::FlowMetadataManager; use common_meta::key::flow::flow_info::FlowInfoValue; @@ -171,7 +171,7 @@ impl InformationSchemaFlows { comment, flow_options: sql::statements::OptionMap::from_filtered_string_map( flow_info.options(), - &[FlowType::FLOW_TYPE_KEY], + &[FlowType::FLOW_TYPE_KEY, INTERNAL_INCREMENTAL_MODE_KEY], ), query, }; @@ -464,3 +464,57 @@ impl DfPartitionStream for InformationSchemaFlows { )) } } + +#[cfg(test)] +mod tests { + use std::collections::{BTreeMap, HashMap}; + + 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 table::table_name::TableName; + + use super::InformationSchemaFlows; + + fn test_flow_info() -> FlowInfoValue { + 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: 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, + } + } + + #[test] + fn test_generate_show_create_flow_hides_internal_options() { + let flow_info = test_flow_info(); + let sql = InformationSchemaFlows::generate_show_create_flow(&flow_info).unwrap(); + + // 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)); + } +} diff --git a/src/common/meta/src/ddl/create_flow.rs b/src/common/meta/src/ddl/create_flow.rs index cec772f7b0..e0cb2e5df9 100644 --- a/src/common/meta/src/ddl/create_flow.rs +++ b/src/common/meta/src/ddl/create_flow.rs @@ -473,6 +473,10 @@ pub const DEFER_ON_MISSING_SOURCE_KEY: &str = "defer_on_missing_source"; /// field in the flow create request. pub const INTERNAL_EVAL_SCHEDULE_KEY: &str = "__greptime_internal_eval_schedule"; +/// Reserved internal per-flow runtime intent (`memtable_only` | `sequence_range`), +/// injected only by internal producers and persisted with flow metadata for recovery. +pub const INTERNAL_INCREMENTAL_MODE_KEY: &str = "__greptime_internal_incremental_mode"; + const FLOW_SCHEDULED_TIME_MILLIS_EXTENSION_KEY: &str = "flow.scheduled_time_millis"; fn without_scheduled_time_extension(mut query_context: QueryContext) -> QueryContext { @@ -519,6 +523,7 @@ pub fn validate_flow_options(flow_task: &CreateFlowTask) -> Result<()> { match key.as_str() { DEFER_ON_MISSING_SOURCE_KEY | FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY + | INTERNAL_INCREMENTAL_MODE_KEY | FlowType::FLOW_TYPE_KEY => {} unknown => { return UnexpectedSnafu { @@ -772,9 +777,6 @@ impl From<&CreateFlowData> for (FlowInfoValue, Vec<(FlowPartitionId, FlowRouteVa let sql = value.task.sql.clone(); let eval_schedule = value.task.eval_schedule.clone(); - // Start with a clean options map. The transient schedule payload is - // only for the meta→flownode CreateRequest boundary and must not be - // persisted in FlowInfoValue.options. let mut options: HashMap = value .task .flow_options diff --git a/src/common/meta/src/ddl/tests/create_flow.rs b/src/common/meta/src/ddl/tests/create_flow.rs index b526856bda..deadbe00ab 100644 --- a/src/common/meta/src/ddl/tests/create_flow.rs +++ b/src/common/meta/src/ddl/tests/create_flow.rs @@ -25,9 +25,9 @@ use table::table_name::TableName; use crate::ddl::DdlContext; use crate::ddl::create_flow::{ CreateFlowData, CreateFlowProcedure, CreateFlowState, DEFER_ON_MISSING_SOURCE_KEY, - FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY, FlowType, defer_on_missing_source, - effective_eval_schedule_from_flow_info, resolve_schedule_defaults_into_task, - validate_flow_options, + FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY, FlowType, INTERNAL_INCREMENTAL_MODE_KEY, + defer_on_missing_source, effective_eval_schedule_from_flow_info, + resolve_schedule_defaults_into_task, validate_flow_options, }; use crate::ddl::test_util::create_table::test_create_table_task; use crate::ddl::test_util::flownode_handler::NaiveFlownodeHandler; @@ -325,6 +325,24 @@ fn test_validate_flow_options_allows_incremental_read_option() { validate_flow_options(&task).unwrap(); } +#[test] +fn test_validate_flow_options_accepts_internal_incremental_mode_key() { + // Meta accepts the reserved key because an internal producer may inject + // it after user option validation; value validation is the flow crate's job. + let mut task = test_create_flow_task( + "my_flow", + vec![], + TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_sink_table"), + false, + ); + task.flow_options.insert( + INTERNAL_INCREMENTAL_MODE_KEY.to_string(), + "sequence_range".to_string(), + ); + + validate_flow_options(&task).unwrap(); +} + #[test] fn test_validate_flow_options_rejects_schedule_and_internal_keys_as_unknown() { for key in [ @@ -346,11 +364,16 @@ fn test_validate_flow_options_rejects_schedule_and_internal_keys_as_unknown() { .insert(key.to_string(), "value".to_string()); let err = validate_flow_options(&task).unwrap_err(); + let msg = err.to_string(); assert!( - err.to_string() - .contains(&format!("Unknown flow option '{key}'")), + msg.contains(&format!("Unknown flow option '{key}'")), "unexpected error for {key}: {err}" ); + let supported = msg + .split("supported user options: ") + .nth(1) + .unwrap_or_default(); + assert!(!supported.contains(INTERNAL_INCREMENTAL_MODE_KEY), "{err}"); } } diff --git a/src/flow/src/adapter/flownode_impl.rs b/src/flow/src/adapter/flownode_impl.rs index 7e83d5131d..859aea2927 100644 --- a/src/flow/src/adapter/flownode_impl.rs +++ b/src/flow/src/adapter/flownode_impl.rs @@ -26,7 +26,8 @@ use catalog::CatalogManager; use common_base::Plugins; use common_error::ext::BoxedError; use common_meta::ddl::create_flow::{ - FlowType, INTERNAL_EVAL_SCHEDULE_KEY, effective_eval_schedule_from_flow_info, + FlowType, INTERNAL_EVAL_SCHEDULE_KEY, INTERNAL_INCREMENTAL_MODE_KEY, + effective_eval_schedule_from_flow_info, }; use common_meta::error::Result as MetaResult; use common_meta::key::flow::FlowMetadataManager; @@ -663,6 +664,24 @@ impl SrcTableToFlow { } } +/// The streaming engine ignores `flow_options`, so injected incremental intent +/// must surface as an internal error rather than be silently dropped. +fn validate_flow_options_for_engine( + flow_type: FlowType, + flow_options: &HashMap, +) -> Result<(), Error> { + if flow_type == FlowType::Streaming && flow_options.contains_key(INTERNAL_INCREMENTAL_MODE_KEY) + { + return InternalSnafu { + reason: format!( + "internal flow option '{INTERNAL_INCREMENTAL_MODE_KEY}' is only valid for batching flows" + ), + } + .fail(); + } + Ok(()) +} + impl FlowEngine for FlowDualEngine { async fn create_flow(&self, args: CreateFlowArgs) -> Result, Error> { let flow_type = args @@ -682,6 +701,8 @@ impl FlowEngine for FlowDualEngine { } }; + validate_flow_options_for_engine(flow_type, &args.flow_options)?; + let flow_id = args.flow_id; let src_table_ids = args.source_table_ids.clone(); @@ -1164,9 +1185,11 @@ impl StreamingEngine { mod tests { use std::collections::HashMap; - use common_meta::ddl::create_flow::INTERNAL_EVAL_SCHEDULE_KEY; + use common_meta::ddl::create_flow::{ + FlowType, INTERNAL_EVAL_SCHEDULE_KEY, INTERNAL_INCREMENTAL_MODE_KEY, + }; - use super::decode_internal_eval_schedule; + use super::{decode_internal_eval_schedule, validate_flow_options_for_engine}; use crate::error::Error; #[test] @@ -1185,4 +1208,26 @@ mod tests { )); assert!(!flow_options.contains_key(INTERNAL_EVAL_SCHEDULE_KEY)); } + + #[test] + fn test_internal_incremental_mode_guard_rejects_streaming_accepts_batching() { + let mut stream_options = HashMap::new(); + stream_options.insert( + INTERNAL_INCREMENTAL_MODE_KEY.to_string(), + "sequence_range".to_string(), + ); + let err = + validate_flow_options_for_engine(FlowType::Streaming, &stream_options).unwrap_err(); + assert!(matches!( + err, + Error::Internal { reason, .. } if reason.contains(INTERNAL_INCREMENTAL_MODE_KEY) + )); + + let mut batch_options = HashMap::new(); + batch_options.insert( + INTERNAL_INCREMENTAL_MODE_KEY.to_string(), + "sequence_range".to_string(), + ); + validate_flow_options_for_engine(FlowType::Batching, &batch_options).unwrap(); + } } diff --git a/src/flow/src/batching_mode.rs b/src/flow/src/batching_mode.rs index bb946e7574..e0ae41ed4b 100644 --- a/src/flow/src/batching_mode.rs +++ b/src/flow/src/batching_mode.rs @@ -30,6 +30,18 @@ mod task; mod time_window; pub(crate) mod utils; +/// Incremental read mode for a batching flow, selected only through the +/// reserved internal flow option +/// [`common_meta::ddl::create_flow::INTERNAL_INCREMENTAL_MODE_KEY`]. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum IncrementalMode { + /// Read only memtables newer than the checkpoint, skipping SSTs entirely. + #[default] + MemtableOnly, + /// Exact row-level sequence delta `(C, H]` across memtables and all SSTs. + SequenceRange, +} + #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct BatchingModeOptions { /// The default batching engine query timeout is 10 minutes @@ -59,6 +71,11 @@ pub struct BatchingModeOptions { /// /// When disabled, batching flows always execute full-snapshot queries. pub experimental_enable_incremental_read: bool, + /// Internal incremental read mode for batching flows, injected only + /// through the reserved internal flow option + /// [`common_meta::ddl::create_flow::INTERNAL_INCREMENTAL_MODE_KEY`]. + #[serde(skip)] + pub incremental_mode: IncrementalMode, /// Read preference of the Frontend client. pub read_preference: ReadPreference, /// TLS option for client connections to frontends. @@ -77,8 +94,27 @@ impl Default for BatchingModeOptions { experimental_max_filter_num_per_query: 20, experimental_time_window_merge_threshold: 3, experimental_enable_incremental_read: false, + incremental_mode: IncrementalMode::default(), read_preference: Default::default(), frontend_tls: None, } } } + +#[cfg(test)] +mod tests { + use super::{BatchingModeOptions, IncrementalMode}; + + #[test] + fn test_incremental_mode_not_exposed_by_options_serialization() { + // `#[serde(skip)]`: the runtime-only field never appears in + // serialized options, and injected input is ignored. + let serialized = serde_json::to_string(&BatchingModeOptions::default()).unwrap(); + assert!(!serialized.contains("incremental_mode")); + + let mut json = serde_json::to_value(BatchingModeOptions::default()).unwrap(); + json["incremental_mode"] = serde_json::json!("sequence_range"); + let opts: BatchingModeOptions = serde_json::from_value(json).unwrap(); + assert_eq!(opts.incremental_mode, IncrementalMode::default()); + } +} diff --git a/src/flow/src/batching_mode/engine.rs b/src/flow/src/batching_mode/engine.rs index d05eec9251..4daf911271 100644 --- a/src/flow/src/batching_mode/engine.rs +++ b/src/flow/src/batching_mode/engine.rs @@ -21,7 +21,9 @@ use std::time::Duration; use api::v1::flow::DirtyWindowRequests; use catalog::CatalogManagerRef; use common_error::ext::BoxedError; -use common_meta::ddl::create_flow::{FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY, FlowType}; +use common_meta::ddl::create_flow::{ + FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY, FlowType, INTERNAL_INCREMENTAL_MODE_KEY, +}; use common_meta::key::TableMetadataManagerRef; use common_meta::key::flow::FlowMetadataManagerRef; use common_meta::key::flow::flow_state::FlowStat; @@ -43,17 +45,18 @@ use store_api::storage::{RegionId, TableId}; use table::table_reference::TableReference; use tokio::sync::{RwLock, oneshot}; -use crate::batching_mode::BatchingModeOptions; use crate::batching_mode::eval_schedule::EvalSchedule; use crate::batching_mode::frontend_client::FrontendClient; use crate::batching_mode::state::DirtyTimeWindows; use crate::batching_mode::task::{BatchingTask, TaskArgs}; use crate::batching_mode::time_window::{TimeWindowExpr, find_time_window_expr}; use crate::batching_mode::utils::sql_to_df_plan; +use crate::batching_mode::{BatchingModeOptions, IncrementalMode}; use crate::engine::{FlowEngine, FlowStatProvider}; use crate::error::{ CreateFlowSnafu, DatafusionSnafu, ExternalSnafu, FlowAlreadyExistSnafu, FlowNotFoundSnafu, - InvalidQuerySnafu, JoinTaskSnafu, TableNotFoundMetaSnafu, UnexpectedSnafu, UnsupportedSnafu, + InternalSnafu, InvalidQuerySnafu, JoinTaskSnafu, TableNotFoundMetaSnafu, UnexpectedSnafu, + UnsupportedSnafu, }; use crate::metrics::METRIC_FLOW_BATCHING_ENGINE_BULK_MARK_TIME_WINDOW; use crate::{CreateFlowArgs, Error, FlowId, TableName}; @@ -482,6 +485,24 @@ impl BatchingEngine { })?; } + if let Some(incremental_mode) = flow_options.get(INTERNAL_INCREMENTAL_MODE_KEY) { + let lowered = incremental_mode.trim().to_ascii_lowercase(); + batch_opts.incremental_mode = match lowered.as_str() { + "memtable_only" => IncrementalMode::MemtableOnly, + "sequence_range" => IncrementalMode::SequenceRange, + _ => { + // A bad value in the reserved internal key is a broken + // internal contract, not a user input error. + return Err(InternalSnafu { + reason: format!( + "Invalid internal flow option {INTERNAL_INCREMENTAL_MODE_KEY}: {incremental_mode}" + ), + } + .build()); + } + }; + } + Ok(Arc::new(batch_opts)) } @@ -1099,6 +1120,33 @@ mod tests { assert!(enabled_opts.experimental_enable_incremental_read); } + #[tokio::test] + async fn test_flow_option_parses_incremental_mode() { + let engine = new_test_engine().await; + + let default_opts = engine.batch_opts_for_flow_options(&HashMap::new()).unwrap(); + assert_eq!(IncrementalMode::MemtableOnly, default_opts.incremental_mode); + + let sequence_range_opts = engine + .batch_opts_for_flow_options(&HashMap::from([( + INTERNAL_INCREMENTAL_MODE_KEY.to_string(), + " sequence_range ".to_string(), + )])) + .unwrap(); + assert_eq!( + IncrementalMode::SequenceRange, + sequence_range_opts.incremental_mode + ); + + let err = engine + .batch_opts_for_flow_options(&HashMap::from([( + INTERNAL_INCREMENTAL_MODE_KEY.to_string(), + "bogus".to_string(), + )])) + .unwrap_err(); + assert!(matches!(err, Error::Internal { .. }), "{err}"); + } + #[test] fn test_table_options_enable_append_mode() { assert!(!BatchingEngine::table_options_enable_append_mode( diff --git a/src/flow/src/batching_mode/task/inc.rs b/src/flow/src/batching_mode/task/inc.rs index fbea3abd25..86ac37c412 100644 --- a/src/flow/src/batching_mode/task/inc.rs +++ b/src/flow/src/batching_mode/task/inc.rs @@ -20,12 +20,13 @@ use common_telemetry::tracing::warn; use datafusion_expr::{DmlStatement, LogicalPlan}; use query::options::{ FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE, FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY, - FLOW_SINK_TABLE_ID, + FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE, FLOW_SINK_TABLE_ID, }; use snafu::ResultExt; use table::metadata::TableId; use crate::Error; +use crate::batching_mode::IncrementalMode; use crate::batching_mode::state::CheckpointMode; use crate::batching_mode::table_creator::QueryType; use crate::batching_mode::task::BatchingTask; @@ -221,10 +222,11 @@ impl BatchingTask { if let Some(checkpoints_json) = incremental_checkpoints_json { let sink_table_id = self.sink_table_id().await?; extensions.push((FLOW_SINK_TABLE_ID, sink_table_id.to_string())); - extensions.push(( - FLOW_INCREMENTAL_MODE, - FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY.to_string(), - )); + let incremental_mode = match self.config.batch_opts.incremental_mode { + IncrementalMode::SequenceRange => FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE, + IncrementalMode::MemtableOnly => FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY, + }; + extensions.push((FLOW_INCREMENTAL_MODE, incremental_mode.to_string())); extensions.push((FLOW_INCREMENTAL_AFTER_SEQS, checkpoints_json)); } diff --git a/src/flow/src/batching_mode/task/test.rs b/src/flow/src/batching_mode/task/test.rs index 28468aa595..dcc9cf9a01 100644 --- a/src/flow/src/batching_mode/task/test.rs +++ b/src/flow/src/batching_mode/task/test.rs @@ -31,14 +31,15 @@ use datatypes::vectors::{ }; use pretty_assertions::assert_eq; use query::options::{ - FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY, FLOW_SCHEDULED_TIME_MILLIS, - QueryOptions, + FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE, FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY, + FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE, FLOW_SCHEDULED_TIME_MILLIS, QueryOptions, }; use session::context::QueryContext; use snafu::ResultExt; use table::test_util::MemTable; use super::*; +use crate::batching_mode::IncrementalMode; use crate::batching_mode::checkpoint::{ CHECKPOINT_DECISION_ADVANCE, CHECKPOINT_DECISION_FALLBACK, CHECKPOINT_REASON_NONE, FlowCheckpointDecision, FlowQueryFallbackReason, @@ -1729,6 +1730,39 @@ async fn test_build_flow_query_extensions_switches_with_checkpoint_mode() { ); } +// `sequence_range` must flow into the emitted extensions only with a +// checkpoint map present, requesting exact filtering across memtables and all +// SSTs; the default mode keeps emitting `memtable_only`. +#[tokio::test] +async fn test_build_flow_query_extensions_sequence_range_mode() { + let (task, _) = new_test_task_engine_and_plan_with_query_and_opts( + "SELECT number, ts FROM numbers_with_ts", + "numbers_with_ts", + Arc::new(BatchingModeOptions { + experimental_enable_incremental_read: true, + incremental_mode: IncrementalMode::SequenceRange, + ..Default::default() + }), + ) + .await + .into_task_and_plan(); + + task.state + .write() + .unwrap() + .advance_checkpoints(HashMap::from([(1_u64, 10_u64)])); + + let extensions = task.build_flow_query_extensions(true, true).await.unwrap(); + assert!(extensions.contains(&( + FLOW_INCREMENTAL_MODE, + FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE.to_string() + ))); + assert!(extensions.contains(&( + FLOW_INCREMENTAL_AFTER_SEQS, + serde_json::json!({"1": 10}).to_string(), + ))); +} + #[tokio::test] async fn test_full_snapshot_scoped_plan_marks_checkpoint_advance_safe_only_after_backlog_drained() { let TestTaskParts { diff --git a/src/operator/src/statement/ddl.rs b/src/operator/src/statement/ddl.rs index 98cb483d94..08913cdc82 100644 --- a/src/operator/src/statement/ddl.rs +++ b/src/operator/src/statement/ddl.rs @@ -42,6 +42,7 @@ use common_meta::cache_invalidator::CacheInvalidatorRef; use common_meta::cache_invalidator::Context; use common_meta::ddl::create_flow::{ DEFER_ON_MISSING_SOURCE_KEY, FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY, FlowType, + INTERNAL_INCREMENTAL_MODE_KEY, }; #[cfg(feature = "enterprise")] use common_meta::ddl_manager::FlowExtensionRef; @@ -128,7 +129,12 @@ struct DdlSubmitOptions { timeout: Duration, } -const ALLOWED_FLOW_OPTIONS: &[&str] = &[ +/// User-facing flow options accepted from SQL `WITH (...)`. The reserved +/// internal option [`INTERNAL_INCREMENTAL_MODE_KEY`] is not a user option: +/// it is rejected here and can only be injected into the underlying +/// `CreateFlowTask` by an internal producer (e.g. FlowExtension) after user +/// option validation. +const ALLOWED_FLOW_OPTIONS: [&str; 2] = [ DEFER_ON_MISSING_SOURCE_KEY, FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY, ]; @@ -219,7 +225,7 @@ fn validate_and_normalize_flow_options( let mut extension_options = HashMap::new(); for (key, value) in options { - if key == FlowType::FLOW_TYPE_KEY { + if key == FlowType::FLOW_TYPE_KEY || key == INTERNAL_INCREMENTAL_MODE_KEY { return InvalidSqlSnafu { err_msg: format!("flow option '{key}' is reserved for internal use"), } @@ -3398,6 +3404,22 @@ mod test { ); } + #[test] + fn test_validate_and_normalize_flow_options_rejects_internal_incremental_mode_key() { + let err = validate_and_normalize_flow_options( + HashMap::from([( + INTERNAL_INCREMENTAL_MODE_KEY.to_string(), + "sequence_range".to_string(), + )]), + None, + ) + .unwrap_err(); + + assert!(err.to_string().contains(&format!( + "flow option '{INTERNAL_INCREMENTAL_MODE_KEY}' is reserved for internal use" + ))); + } + #[test] fn test_validate_and_normalize_flow_options_invalid_bool() { let err = validate_and_normalize_flow_options( diff --git a/src/query/src/dummy_catalog.rs b/src/query/src/dummy_catalog.rs index e38813b3e6..44cd112da1 100644 --- a/src/query/src/dummy_catalog.rs +++ b/src/query/src/dummy_catalog.rs @@ -371,6 +371,10 @@ struct FlowScanDecision { memtable_max_sequence: Option, /// 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 Result Result { 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([( diff --git a/src/query/src/sql.rs b/src/query/src/sql.rs index b2d979c666..7caec64c34 100644 --- a/src/query/src/sql.rs +++ b/src/query/src/sql.rs @@ -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)); + } }