diff --git a/src/common/meta/src/ddl/create_flow.rs b/src/common/meta/src/ddl/create_flow.rs index 2679e61e742..e36c89fbb34 100644 --- a/src/common/meta/src/ddl/create_flow.rs +++ b/src/common/meta/src/ddl/create_flow.rs @@ -568,6 +568,21 @@ pub fn validate_flow_options(flow_task: &CreateFlowTask) -> Result<()> { } } + if let Some(value) = flow_task + .flow_options + .get(FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY) + && value != FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_SEQUENCE_RANGE + { + value.parse::().map_err(|_| { + UnexpectedSnafu { + err_msg: format!( + "Invalid flow option {FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY}: {value}" + ), + } + .build() + })?; + } + defer_on_missing_source(flow_task)?; get_flow_type_from_options(flow_task)?; Ok(()) @@ -770,6 +785,9 @@ pub enum FlowType { pub const FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY: &str = "experimental_enable_incremental_read"; +/// Reserved internal value for flows requiring exact sequence ranges. +pub const FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_SEQUENCE_RANGE: &str = + "__greptime_internal_exact_sequence_range"; impl FlowType { pub const BATCHING: &str = "batching"; diff --git a/src/flow/src/batching_mode/engine.rs b/src/flow/src/batching_mode/engine.rs index d05eec92518..3532b38d366 100644 --- a/src/flow/src/batching_mode/engine.rs +++ b/src/flow/src/batching_mode/engine.rs @@ -21,7 +21,10 @@ 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, + FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_SEQUENCE_RANGE, FlowType, +}; use common_meta::key::TableMetadataManagerRef; use common_meta::key::flow::FlowMetadataManagerRef; use common_meta::key::flow::flow_state::FlowStat; @@ -465,21 +468,24 @@ impl BatchingEngine { fn batch_opts_for_flow_options( &self, flow_options: &HashMap, + exact_sequence_range_required: bool, ) -> Result, Error> { let mut batch_opts = (*self.batch_opts).clone(); if let Some(enable_incremental_read) = flow_options.get(FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY) { - batch_opts.experimental_enable_incremental_read = enable_incremental_read - .parse::() - .map_err(|_| { + batch_opts.experimental_enable_incremental_read = if exact_sequence_range_required { + true + } else { + enable_incremental_read.parse::().map_err(|_| { InvalidQuerySnafu { reason: format!( "Invalid flow option {FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY}: {enable_incremental_read}" ), } .build() - })?; + })? + }; } Ok(Arc::new(batch_opts)) @@ -594,7 +600,14 @@ impl BatchingEngine { } ); - let batch_opts = self.batch_opts_for_flow_options(&flow_options)?; + // The meta layer validates this reserved sentinel before it reaches the + // flownode. Derive the requirement once and pass it directly to task + // config; query-context extensions are irrelevant. + let exact_sequence_range_required = flow_options + .get(FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY) + .is_some_and(|value| value == FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_SEQUENCE_RANGE); + let batch_opts = + self.batch_opts_for_flow_options(&flow_options, exact_sequence_range_required)?; let mut source_table_names = Vec::with_capacity(2); for src_id in source_table_ids { @@ -697,7 +710,21 @@ impl BatchingEngine { eval_schedule, }; - let task = BatchingTask::try_new(task_args)?; + let task = BatchingTask::try_new_with_exact_sequence_range_required( + task_args, + exact_sequence_range_required, + )?; + + if task.config.exact_sequence_range_required { + ensure!( + task.sequence_range_capable().await?, + UnsupportedSnafu { + reason: format!( + "Flow {flow_id} requires exact sequence-range reads, but a source table lacks the Mito preserve_row_sequence capability" + ), + } + ); + } let task_inner = task.clone(); let engine = self.query_engine.clone(); @@ -1087,16 +1114,32 @@ mod tests { async fn test_flow_option_overrides_incremental_read_switch() { let engine = new_test_engine().await; - let default_opts = engine.batch_opts_for_flow_options(&HashMap::new()).unwrap(); + let default_opts = engine + .batch_opts_for_flow_options(&HashMap::new(), false) + .unwrap(); assert!(!default_opts.experimental_enable_incremental_read); let enabled_opts = engine - .batch_opts_for_flow_options(&HashMap::from([( - FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY.to_string(), - "true".to_string(), - )])) + .batch_opts_for_flow_options( + &HashMap::from([( + FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY.to_string(), + "true".to_string(), + )]), + false, + ) .unwrap(); assert!(enabled_opts.experimental_enable_incremental_read); + + let exact_opts = engine + .batch_opts_for_flow_options( + &HashMap::from([( + FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY.to_string(), + FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_SEQUENCE_RANGE.to_string(), + )]), + true, + ) + .unwrap(); + assert!(exact_opts.experimental_enable_incremental_read); } #[test] diff --git a/src/flow/src/batching_mode/task.rs b/src/flow/src/batching_mode/task.rs index fb64dc9c20e..95b989851f7 100644 --- a/src/flow/src/batching_mode/task.rs +++ b/src/flow/src/batching_mode/task.rs @@ -149,6 +149,7 @@ pub struct TaskConfig { pub catalog_manager: CatalogManagerRef, pub query_type: QueryType, pub batch_opts: Arc, + pub exact_sequence_range_required: bool, pub flow_eval_interval: Option, /// Typed schedule configuration, pre-parsed at task creation time. pub eval_schedule: Option, @@ -292,7 +293,11 @@ struct ExecuteOnceOutcome { impl BatchingTask { #[allow(clippy::too_many_arguments)] - pub fn try_new( + pub fn try_new(args: TaskArgs<'_>) -> Result { + Self::try_new_with_exact_sequence_range_required(args, false) + } + + pub fn try_new_with_exact_sequence_range_required( TaskArgs { flow_id, query, @@ -308,6 +313,7 @@ impl BatchingTask { flow_eval_interval, eval_schedule, }: TaskArgs<'_>, + exact_sequence_range_required: bool, ) -> Result { let mut state = TaskState::with_dirty_time_windows( query_ctx.clone(), @@ -332,6 +338,7 @@ impl BatchingTask { catalog_manager, output_schema: plan.schema().clone(), query_type: determine_query_type(query, &query_ctx)?, + exact_sequence_range_required, batch_opts, flow_eval_interval, eval_schedule, diff --git a/src/flow/src/batching_mode/task/inc.rs b/src/flow/src/batching_mode/task/inc.rs index fbea3abd259..8c090d9814e 100644 --- a/src/flow/src/batching_mode/task/inc.rs +++ b/src/flow/src/batching_mode/task/inc.rs @@ -20,9 +20,10 @@ 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 store_api::mito_engine_options::PRESERVE_ROW_SEQUENCE; use table::metadata::TableId; use crate::Error; @@ -61,6 +62,50 @@ impl BatchingTask { Ok(table.table_info().table_id()) } + /// Whether every source table provably supports exact sequence-range + /// scans: it must be the canonical mito engine and declare the + /// `preserve_row_sequence` capability (a mito region option enforced to + /// require append-only mode). When the capability is unknown — a source + /// table that cannot be resolved, is not the mito engine, or lacks the + /// option — this returns `false` so the caller keeps the historical + /// `memtable_only` mode instead of upgrading. + pub async fn sequence_range_capable(&self) -> Result { + for name in &self.config.source_table_names { + let table = match self + .config + .catalog_manager + .table(&name[0], &name[1], &name[2], None) + .await + { + Ok(Some(table)) => table, + Ok(None) => { + return Err(UnexpectedSnafu { + reason: format!( + "Flow {} source table {} not found for sequence_range capability check", + self.config.flow_id, + name.join(".") + ), + } + .build()); + } + Err(err) => Err(BoxedError::new(err)).context(ExternalSnafu)?, + }; + + let info = table.table_info(); + let preserves = info.meta.engine == "mito" + && info + .meta + .options + .extra_options + .get(PRESERVE_ROW_SEQUENCE) + .is_some_and(|value| value.eq_ignore_ascii_case("true")); + if !preserves { + return Ok(false); + } + } + Ok(!self.config.source_table_names.is_empty()) + } + /// For incremental-mode SQL queries, attempt to prepare an executable plan /// that is safe for incremental scan extensions. /// @@ -109,6 +154,15 @@ impl BatchingTask { permanently disabling incremental for this flow", self.config.flow_id ); + if self.config.exact_sequence_range_required { + return Err(UnexpectedSnafu { + reason: format!( + "Flow {} requires exact sequence-range reads, but its incremental plan is unsupported", + self.config.flow_id + ), + } + .build()); + } self.state.write().unwrap().disable_incremental(); return Ok(None); }; @@ -119,6 +173,15 @@ impl BatchingTask { permanently disabling incremental for this flow", self.config.flow_id, analysis.unsupported_exprs ); + if self.config.exact_sequence_range_required { + return Err(UnexpectedSnafu { + reason: format!( + "Flow {} requires exact sequence-range reads, but its incremental aggregate is unsupported: {:?}", + self.config.flow_id, analysis.unsupported_exprs + ), + } + .build()); + } self.state.write().unwrap().disable_incremental(); return Ok(None); } @@ -219,12 +282,42 @@ impl BatchingTask { }; if let Some(checkpoints_json) = incremental_checkpoints_json { + // Select `sequence_range` only when every append-only source table + // proves the `preserve_row_sequence` capability; otherwise retain + // the historical `memtable_only` mode. The `sequence_range` scan + // keeps SSTs and reads the exact (checkpoint, scan-open snapshot] + // row-level delta; the engine fails closed when the capability + // does not hold at scan time. + let capable = match self.sequence_range_capable().await { + Ok(capable) => capable, + Err(err) => { + if self.config.exact_sequence_range_required { + return Err(err); + } + false + } + }; + if self.config.exact_sequence_range_required && !capable { + return Err(UnexpectedSnafu { + reason: format!( + "Flow {} requires exact sequence-range reads, but source capability was revoked", + self.config.flow_id + ), + } + .build()); + } let sink_table_id = self.sink_table_id().await?; + let incremental_mode = if capable { + debug!( + "Flow {} selected sequence_range incremental mode", + self.config.flow_id + ); + FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE + } else { + FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY + }; extensions.push((FLOW_SINK_TABLE_ID, sink_table_id.to_string())); - extensions.push(( - FLOW_INCREMENTAL_MODE, - FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY.to_string(), - )); + 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 acb872d8fea..fbb8e189917 100644 --- a/src/flow/src/batching_mode/task/test.rs +++ b/src/flow/src/batching_mode/task/test.rs @@ -15,8 +15,8 @@ use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::sync::Arc; -use catalog::RegisterTableRequest; use catalog::memory::MemoryCatalogManager; +use catalog::{DeregisterTableRequest, RegisterTableRequest}; use client::OutputWithMetrics; use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME}; use common_error::ext::BoxedError; @@ -32,11 +32,13 @@ 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_SCHEDULED_TIME_MILLIS, FLOW_SINK_TABLE_ID, QueryOptions, }; use session::context::QueryContext; use snafu::ResultExt; +use table::Table; +use table::metadata::FilterPushDownType; use table::test_util::MemTable; use super::*; @@ -86,6 +88,18 @@ async fn new_test_task_engine_and_plan_with_query_and_opts( query: &str, sink_table: &str, batch_opts: Arc, +) -> TestTaskParts { + new_test_task_engine_and_plan_with_query_and_opts_and_required( + query, sink_table, batch_opts, false, + ) + .await +} + +async fn new_test_task_engine_and_plan_with_query_and_opts_and_required( + query: &str, + sink_table: &str, + batch_opts: Arc, + exact_sequence_range_required: bool, ) -> TestTaskParts { let query_engine = create_test_query_engine(); let ctx = QueryContext::arc(); @@ -99,29 +113,32 @@ async fn new_test_task_engine_and_plan_with_query_and_opts( .unwrap(); let (_tx, rx) = tokio::sync::oneshot::channel(); - let task = BatchingTask::try_new(TaskArgs { - flow_id: 1, - query, - plan: plan.clone(), - time_window_expr: None, - expire_after: None, - sink_table_name: [ - "greptime".to_string(), - "public".to_string(), - sink_table.to_string(), - ], - source_table_names: vec![[ - "greptime".to_string(), - "public".to_string(), - "numbers_with_ts".to_string(), - ]], - query_ctx: ctx, - catalog_manager: query_engine.engine_state().catalog_manager().clone(), - shutdown_rx: rx, - batch_opts, - flow_eval_interval: None, - eval_schedule: None, - }) + let task = BatchingTask::try_new_with_exact_sequence_range_required( + TaskArgs { + flow_id: 1, + query, + plan: plan.clone(), + time_window_expr: None, + expire_after: None, + sink_table_name: [ + "greptime".to_string(), + "public".to_string(), + sink_table.to_string(), + ], + source_table_names: vec![[ + "greptime".to_string(), + "public".to_string(), + "numbers_with_ts".to_string(), + ]], + query_ctx: ctx, + catalog_manager: query_engine.engine_state().catalog_manager().clone(), + shutdown_rx: rx, + batch_opts, + flow_eval_interval: None, + eval_schedule: None, + }, + exact_sequence_range_required, + ) .unwrap(); TestTaskParts { @@ -380,6 +397,62 @@ fn register_auto_created_aggregate_sink(query_engine: &QueryEngineRef, table_nam memory_catalog.register_table_sync(request).unwrap(); } +async fn configure_source_capability( + query_engine: &QueryEngineRef, + engine: &str, + preserve_row_sequence: bool, +) { + let catalog_manager = query_engine.engine_state().catalog_manager(); + let source = catalog_manager + .table( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + "numbers_with_ts", + None, + ) + .await + .unwrap() + .unwrap(); + let mut info = (*source.table_info()).clone(); + info.meta.engine = engine.to_string(); + if preserve_row_sequence { + info.meta + .options + .extra_options + .insert("preserve_row_sequence".to_string(), "true".to_string()); + } else { + info.meta + .options + .extra_options + .remove("preserve_row_sequence"); + } + let source = Arc::new(Table::new( + Arc::new(info), + FilterPushDownType::Unsupported, + source.data_source(), + )); + let memory_catalog = catalog_manager + .as_any() + .downcast_ref::() + .unwrap(); + memory_catalog + .deregister_table_sync(DeregisterTableRequest { + catalog: DEFAULT_CATALOG_NAME.to_string(), + schema: DEFAULT_SCHEMA_NAME.to_string(), + table_name: "numbers_with_ts".to_string(), + }) + .unwrap(); + memory_catalog + .register_table_sync(RegisterTableRequest { + catalog: DEFAULT_CATALOG_NAME.to_string(), + schema: DEFAULT_SCHEMA_NAME.to_string(), + table_name: "numbers_with_ts".to_string(), + table_id: source.table_info().table_id(), + table: source, + }) + .unwrap(); +} + fn dirty_marker() -> DirtyTimeWindows { let mut dirty = DirtyTimeWindows::default(); dirty.set_dirty(); @@ -504,6 +577,23 @@ fn register_scheduled_now_sink(query_engine: &QueryEngineRef, table_name: &str, memory_catalog.register_table_sync(request).unwrap(); } +struct ExactDeltaFailureHandler; + +#[async_trait::async_trait] +impl crate::batching_mode::frontend_client::GrpcQueryHandlerWithBoxedError + for ExactDeltaFailureHandler +{ + async fn do_query( + &self, + _query: api::v1::greptime_request::Request, + ctx: QueryContextRef, + ) -> std::result::Result { + assert_eq!(ctx.extension(FLOW_INCREMENTAL_MODE), Some("sequence_range")); + assert!(ctx.extension(FLOW_INCREMENTAL_AFTER_SEQS).is_some()); + Err(BoxedError::new(MockError::new(StatusCode::RequestOutdated))) + } +} + struct CaptureScheduledNowHandler { expected_extension: String, captured_sql: Arc>>, @@ -1601,6 +1691,101 @@ fn test_checkpoint_decision_labels_are_stable() { ); } +#[tokio::test] +async fn test_exact_required_attempt_rejects_revoked_capability_without_extensions() { + let task = new_test_task_engine_and_plan_with_query_and_opts_and_required( + "SELECT number, ts FROM numbers_with_ts", + "exact_required_revoked", + incremental_batch_opts(), + true, + ) + .await + .into_task_and_plan() + .0; + + task.state + .write() + .unwrap() + .advance_checkpoints(HashMap::from([(1_u64, 10_u64)])); + let checkpoints_before = task.state.read().unwrap().checkpoints().clone(); + + let err = task + .build_flow_query_extensions(true, true) + .await + .unwrap_err(); + assert!(err.to_string().contains("requires exact sequence-range")); + assert_eq!( + task.state.read().unwrap().checkpoints(), + &checkpoints_before + ); +} + +#[tokio::test] +async fn test_sequence_range_producer_emits_capable_source_extensions() { + let parts = new_test_task_engine_and_plan_with_query_and_opts_and_required( + "SELECT number, ts FROM numbers_with_ts", + "numbers_with_ts", + incremental_batch_opts(), + true, + ) + .await; + configure_source_capability(&parts.query_engine, "mito", true).await; + let task = parts.task; + task.state + .write() + .unwrap() + .advance_checkpoints(HashMap::from([(1024_u64, 10_u64), (2048_u64, 20_u64)])); + + let extensions = task.build_flow_query_extensions(true, true).await.unwrap(); + + assert_eq!( + extensions, + vec![ + ("flow.return_region_seq", "true".to_string()), + (FLOW_SINK_TABLE_ID, "1".to_string()), + (FLOW_INCREMENTAL_MODE, "sequence_range".to_string()), + ( + FLOW_INCREMENTAL_AFTER_SEQS, + serde_json::json!({"1024": 10, "2048": 20}).to_string(), + ), + ] + ); +} + +#[tokio::test] +async fn test_sequence_range_producer_keeps_memtable_only_for_non_mito_source() { + let parts = new_test_task_engine_and_plan_with_query_and_opts( + "SELECT number, ts FROM numbers_with_ts", + "numbers_with_ts", + incremental_batch_opts(), + ) + .await; + configure_source_capability(&parts.query_engine, "file", true).await; + let task = parts.task; + task.state + .write() + .unwrap() + .advance_checkpoints(HashMap::from([(1024_u64, 10_u64)])); + + let extensions = task.build_flow_query_extensions(true, true).await.unwrap(); + + assert_eq!( + extensions, + vec![ + ("flow.return_region_seq", "true".to_string()), + (FLOW_SINK_TABLE_ID, "1".to_string()), + ( + FLOW_INCREMENTAL_MODE, + FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY.to_string(), + ), + ( + FLOW_INCREMENTAL_AFTER_SEQS, + serde_json::json!({"1024": 10}).to_string(), + ), + ] + ); +} + #[tokio::test] async fn test_build_flow_query_extensions_switches_with_checkpoint_mode() { let (task, _) = new_test_task_engine_and_plan_with_query( @@ -1956,15 +2141,82 @@ async fn test_incremental_delta_applies_expire_after_retention_filter() { } #[tokio::test] -async fn test_successful_incremental_checkpoint_fallback_consumes_unscoped_dirty_signal() { +async fn test_exact_required_executed_failure_selects_full_snapshot_repair() { let TestTaskParts { - task, + mut task, query_engine, .. } = new_time_window_test_task_with_query( "SELECT max(number) AS number, date_bin(INTERVAL '5 second', ts) AS time_window FROM numbers_with_ts GROUP BY time_window", ) .await; + configure_source_capability(&query_engine, "mito", true).await; + register_twe_sink(&query_engine, "missing_sink", 9201); + Arc::get_mut(&mut task.config) + .expect("test task config should be uniquely owned") + .exact_sequence_range_required = true; + { + let mut state = task.state.write().unwrap(); + state.advance_checkpoints(HashMap::from([(1_u64, 10_u64)])); + state + .dirty_time_windows + .add_window(Timestamp::new_second(0), Some(Timestamp::new_second(5))); + } + let sink_schema = aggregate_time_window_sink_schema(); + let plan_info = task + .gen_query_with_time_window(query_engine.clone(), &sink_schema, &[], false, Some(1)) + .await + .unwrap() + .unwrap(); + assert!(matches!(plan_info.coverage, QueryCoverage::IncrementalDelta)); + + let handler: Arc = + Arc::new(ExactDeltaFailureHandler); + let frontend_client = Arc::new(FrontendClient::from_grpc_handler( + Arc::downgrade(&handler), + QueryOptions::default(), + )); + task.execute_logical_plan_unlocked( + &frontend_client, + &plan_info.plan, + &plan_info.dirty_restore, + &plan_info.coverage, + ) + .await + .expect_err("the dispatched exact delta must fail without a frontend handler"); + task.handle_executed_query_failure(Some(&plan_info)); + + { + let state = task.state.read().unwrap(); + assert_eq!(state.checkpoint_mode(), CheckpointMode::FullSnapshot); + assert_eq!(state.checkpoints(), &BTreeMap::from([(1_u64, 10_u64)])); + assert_eq!(state.dirty_time_windows.len(), 1); + } + + let repair = task + .gen_query_with_time_window(query_engine, &sink_schema, &[], false, Some(1)) + .await + .unwrap() + .expect("failed exact delta must retry through existing base repair"); + assert!(matches!(repair.coverage, QueryCoverage::ScopedBaseRepair)); + assert!(repair.plan.to_string().contains("Filter:")); + assert!(!repair.plan.to_string().contains("Left Join")); +} + +#[tokio::test] +async fn test_exact_required_incomplete_proof_selects_base_recomputation() { + let TestTaskParts { + mut task, + query_engine, + .. + } = new_time_window_test_task_with_query( + "SELECT max(number) AS number, date_bin(INTERVAL '5 second', ts) AS time_window FROM numbers_with_ts GROUP BY time_window", + ) + .await; + configure_source_capability(&query_engine, "mito", true).await; + Arc::get_mut(&mut task.config) + .expect("test task config should be uniquely owned") + .exact_sequence_range_required = true; { let mut state = task.state.write().unwrap(); state.advance_checkpoints(HashMap::from([(1_u64, 10_u64), (2_u64, 20_u64)])); @@ -2014,13 +2266,24 @@ async fn test_successful_incremental_checkpoint_fallback_consumes_unscoped_dirty assert!(state.dirty_time_windows.is_empty()); } + task.state + .write() + .unwrap() + .dirty_time_windows + .add_window(Timestamp::new_second(5), Some(Timestamp::new_second(10))); let followup = task .gen_query_with_time_window(query_engine, &sink_schema, &[], false, Some(1)) .await - .unwrap(); + .unwrap() + .expect("new dirty work must use base recomputation after incomplete delta proof"); + assert!(matches!(followup.coverage, QueryCoverage::ScopedBaseRepair)); assert!( - followup.is_none(), - "successful fallback consumes the dirty signal instead of re-running it" + followup.plan.to_string().contains("Filter:"), + "base recomputation must retain its dirty-window filter" + ); + assert!( + !followup.plan.to_string().contains("Left Join"), + "base recomputation must not replay the additive incremental sink merge" ); } @@ -2101,6 +2364,96 @@ async fn test_executed_query_failure_restores_scoped_dirty_windows_for_flush_pat ); } +#[tokio::test] +async fn test_exact_required_unsupported_plan_keeps_exact_retry_state() { + let query_engine = create_test_query_engine(); + configure_source_capability(&query_engine, "mito", true).await; + let ctx = QueryContext::arc(); + let plan = sql_to_df_plan( + ctx.clone(), + query_engine.clone(), + "SELECT number, ts FROM numbers_with_ts", + true, + ) + .await + .unwrap(); + let (sink_table, _) = get_table_info_df_schema( + query_engine.engine_state().catalog_manager().clone(), + [ + "greptime".to_string(), + "public".to_string(), + "numbers_with_ts".to_string(), + ], + ) + .await + .unwrap(); + let dml_plan = LogicalPlan::Dml(DmlStatement::new( + datafusion_common::TableReference::bare("test"), + Arc::new(DefaultTableSource::new(Arc::new(DfTableProviderAdapter::new( + sink_table, + )))), + WriteOp::Insert(datafusion_expr::dml::InsertOp::Append), + Arc::new(plan), + )); + let (_tx, rx) = tokio::sync::oneshot::channel(); + let task = BatchingTask::try_new_with_exact_sequence_range_required( + TaskArgs { + flow_id: 1, + query: "SELECT number, ts FROM numbers_with_ts", + plan: dml_plan.clone(), + time_window_expr: None, + expire_after: None, + sink_table_name: [ + "greptime".to_string(), + "public".to_string(), + "numbers_with_ts".to_string(), + ], + source_table_names: vec![[ + "greptime".to_string(), + "public".to_string(), + "numbers_with_ts".to_string(), + ]], + query_ctx: ctx, + catalog_manager: query_engine.engine_state().catalog_manager().clone(), + shutdown_rx: rx, + batch_opts: incremental_batch_opts(), + flow_eval_interval: None, + eval_schedule: None, + }, + true, + ) + .unwrap(); + task.state + .write() + .unwrap() + .advance_checkpoints(HashMap::from([(1_u64, 10_u64)])); + task.state + .write() + .unwrap() + .dirty_time_windows + .add_window(Timestamp::new_second(10), Some(Timestamp::new_second(15))); + + let (frontend_client, _) = FrontendClient::from_empty_grpc_handler(QueryOptions::default()); + let frontend_client = Arc::new(frontend_client); + for _ in 0..2 { + let err = task + .execute_logical_plan_unlocked( + &frontend_client, + &dml_plan, + &DirtyRestore::Unscoped(dirty_range(10, 15)), + &QueryCoverage::IncrementalDelta, + ) + .await + .unwrap_err(); + assert!(err.to_string().contains("requires exact sequence-range")); + + let state = task.state.read().unwrap(); + assert_eq!(state.checkpoint_mode(), CheckpointMode::Incremental); + assert_eq!(state.checkpoints(), &BTreeMap::from([(1_u64, 10_u64)])); + assert_eq!(state.dirty_time_windows.len(), 1); + } +} + #[tokio::test] async fn test_prepare_plan_for_incremental_disables_on_non_aggregate() { let query_engine = create_test_query_engine(); diff --git a/src/query/src/dummy_catalog.rs b/src/query/src/dummy_catalog.rs index 3d1dfe5f375..90acf1b46d4 100644 --- a/src/query/src/dummy_catalog.rs +++ b/src/query/src/dummy_catalog.rs @@ -382,6 +382,8 @@ struct FlowScanDecision { memtable_max_sequence: Option, /// Whether to skip SST files for memtable-only incremental source scans. skip_sst_files: bool, + /// Whether this source scan must enforce the exact sequence range. + exact_sequence_range: bool, } impl FlowScanDecision { @@ -392,6 +394,7 @@ impl FlowScanDecision { memtable_min_sequence: None, memtable_max_sequence: None, skip_sst_files: false, + exact_sequence_range: false, } } } @@ -406,6 +409,7 @@ fn decide_flow_scan(query_ctx: &QueryContext, region_id: RegionId) -> Result Result RegionId { @@ -754,6 +761,64 @@ mod tests { assert_eq!(request.memtable_max_sequence, None); } + #[test] + fn test_scan_request_from_sequence_range_context_uses_exact_source_scan() { + let region_id = test_region_id(); + let query_ctx = QueryContextBuilder::default() + .extensions(HashMap::from([ + ( + FLOW_INCREMENTAL_MODE.to_string(), + FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE.to_string(), + ), + ( + 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, + )])))) + .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.exact_sequence_range); + assert!(!request.skip_sst_files); + assert_eq!(request.memtable_min_sequence, Some(10)); + assert_eq!(request.memtable_max_sequence, Some(42)); + assert_eq!(request.sst_min_sequence, None); + } + + #[test] + fn test_scan_request_from_sequence_range_context_binds_snapshot_on_scan() { + let region_id = test_region_id(); + let query_ctx = QueryContextBuilder::default() + .extensions(HashMap::from([ + ( + FLOW_INCREMENTAL_MODE.to_string(), + FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE.to_string(), + ), + ( + 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.exact_sequence_range); + assert!(request.snapshot_on_scan); + assert_eq!(request.memtable_min_sequence, Some(10)); + assert_eq!(request.memtable_max_sequence, None); + assert!(!request.skip_sst_files); + } + #[test] fn test_scan_request_from_query_context_keeps_snapshot_fields() { let region_id = test_region_id(); @@ -824,6 +889,32 @@ mod tests { assert!(!request.snapshot_on_scan); } + #[test] + fn test_apply_cached_snapshot_to_request_preserves_exact_sequence_range() { + let region_id = test_region_id(); + let query_ctx = QueryContextBuilder::default() + .snapshot_seqs(Arc::new(RwLock::new(HashMap::from([( + region_id.as_u64(), + 10_u64, + )])))) + .build(); + let mut request = ScanRequest { + memtable_min_sequence: Some(10), + snapshot_on_scan: true, + exact_sequence_range: true, + ..Default::default() + }; + + apply_cached_snapshot_to_request(&query_ctx, region_id, false, &mut request); + + assert_eq!(request.memtable_min_sequence, Some(10)); + assert_eq!(request.memtable_max_sequence, Some(10)); + assert!(request.exact_sequence_range); + assert!(!request.skip_sst_files); + assert_eq!(request.sst_min_sequence, None); + assert!(!request.snapshot_on_scan); + } + #[test] fn test_apply_cached_snapshot_to_request_skips_sink_scan() { let region_id = test_region_id(); @@ -891,6 +982,44 @@ mod tests { assert_eq!(request.memtable_min_sequence, Some(55)); assert_eq!(request.sst_min_sequence, None); assert!(request.skip_sst_files); + assert!(!request.exact_sequence_range); + } + + #[test] + fn test_scan_request_from_sequence_range_context_excludes_sink_scan() { + let region_id = test_region_id(); + let query_ctx = QueryContextBuilder::default() + .extensions(HashMap::from([ + ( + FLOW_INCREMENTAL_MODE.to_string(), + FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE.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!(!request.exact_sequence_range); + assert!(!request.skip_sst_files); + assert_eq!(request.memtable_min_sequence, None); + assert_eq!(request.memtable_max_sequence, None); + assert_eq!(request.sst_min_sequence, None); } #[test] @@ -949,6 +1078,27 @@ mod tests { assert!(matches!(err, Error::InvalidQueryContextExtension { .. })); } + #[test] + fn test_scan_request_from_sequence_range_rejects_missing_source_bound() { + let region_id = test_region_id(); + let query_ctx = QueryContextBuilder::default() + .extensions(HashMap::from([ + ( + FLOW_INCREMENTAL_MODE.to_string(), + FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE.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 { .. })); + assert_eq!(err.status_code(), StatusCode::InvalidArguments); + } + #[test] fn test_scan_request_from_query_context_rejects_invalid_incremental_json() { let region_id = test_region_id(); diff --git a/src/query/src/options.rs b/src/query/src/options.rs index 24cd3853f07..a49021f2f15 100644 --- a/src/query/src/options.rs +++ b/src/query/src/options.rs @@ -39,6 +39,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 spill mode controlling disk manager behavior. #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] @@ -131,6 +132,7 @@ impl Default for QueryOptions { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum FlowIncrementalMode { MemtableOnly, + SequenceRange, } #[derive(Debug, Clone, PartialEq, Eq, Default)] @@ -167,6 +169,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 @@ -197,21 +202,25 @@ impl FlowQueryExtensions { }) .transpose()?; - if matches!(incremental_mode, Some(FlowIncrementalMode::MemtableOnly)) { + if matches!( + incremental_mode, + Some(FlowIncrementalMode::MemtableOnly | FlowIncrementalMode::SequenceRange) + ) { + let mode = if incremental_mode == Some(FlowIncrementalMode::MemtableOnly) { + FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY + } else { + FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE + }; 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 + FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE, mode )) })?; if after_seqs.is_empty() { return Err(invalid_query_context_extension(format!( "{} must not be empty when {}={}.", - FLOW_INCREMENTAL_AFTER_SEQS, - FLOW_INCREMENTAL_MODE, - FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY + FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE, mode ))); } } @@ -231,19 +240,24 @@ impl FlowQueryExtensions { if matches!( self.incremental_mode, - Some(FlowIncrementalMode::MemtableOnly) + Some(FlowIncrementalMode::MemtableOnly | FlowIncrementalMode::SequenceRange) ) { + let mode = if self.incremental_mode == Some(FlowIncrementalMode::MemtableOnly) { + FLOW_INCREMENTAL_MODE_MEMTABLE_ONLY + } else { + FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE + }; let after_seqs = self.incremental_after_seqs.as_ref().ok_or_else(|| { invalid_query_context_extension(format!( - "{} is required when {}=memtable_only.", - FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE + "{} is required when {}={}.", + FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE, mode )) })?; if !after_seqs.contains_key(&source_region_id.as_u64()) { return Err(invalid_query_context_extension(format!( - "Missing region {} in {} when {}=memtable_only.", - source_region_id, FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE + "Missing region {} in {} when {}={}.", + source_region_id, FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE, mode ))); } } @@ -510,6 +524,57 @@ 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}"#.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, + Some(HashMap::from([(1, 10)])) + ); + } + + #[test] + fn test_parse_flow_extensions_sequence_range_rejects_empty_after_seqs() { + let exts = HashMap::from([ + ( + FLOW_INCREMENTAL_MODE.to_string(), + FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE.to_string(), + ), + (FLOW_INCREMENTAL_AFTER_SEQS.to_string(), "{}".to_string()), + ]); + + let err = FlowQueryExtensions::parse_flow_extensions(&exts).unwrap_err(); + assert!(format!("{err}").contains(FLOW_INCREMENTAL_AFTER_SEQS)); + } + + #[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_parse_flow_extensions_mode_requires_after_seqs() { let exts = HashMap::from([( diff --git a/src/servers/src/grpc/flight.rs b/src/servers/src/grpc/flight.rs index 0b7a9dd5704..f0035ed3f8c 100644 --- a/src/servers/src/grpc/flight.rs +++ b/src/servers/src/grpc/flight.rs @@ -632,7 +632,10 @@ fn output_to_flight_record_batch_source( #[cfg(test)] mod tests { - use query::options::FLOW_SCHEDULED_TIME_MILLIS; + use query::options::{ + FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE, FLOW_RETURN_REGION_SEQ, + FLOW_SCHEDULED_TIME_MILLIS, FLOW_SINK_TABLE_ID, FlowIncrementalMode, + }; use tonic::metadata::{AsciiMetadataValue, MetadataMap}; use super::*; @@ -680,6 +683,47 @@ mod tests { ); } + #[test] + fn test_flow_extensions_forward_sequence_range_to_query_context() { + let mut metadata = MetadataMap::new(); + metadata.insert( + FLOW_EXTENSIONS_METADATA_KEY, + AsciiMetadataValue::try_from( + r#"[["flow.return_region_seq","true"],["flow.incremental_mode","sequence_range"],["flow.incremental_after_seqs","{\"1\":10,\"2\":20}"],["flow.sink_table_id","42"]]"#, + ) + .unwrap(), + ); + + let flow_extensions = extract_flow_extensions(&metadata).unwrap(); + let query_ctx = + create_query_context(Channel::Grpc, None, flow_extensions, HashMap::new()).unwrap(); + let parsed = + query::options::FlowQueryExtensions::parse_flow_extensions(&query_ctx.extensions()) + .unwrap() + .unwrap(); + + assert_eq!( + parsed.incremental_mode, + Some(FlowIncrementalMode::SequenceRange) + ); + assert_eq!( + parsed.incremental_after_seqs, + Some(HashMap::from([(1, 10), (2, 20)])) + ); + assert!(parsed.return_region_seq); + assert_eq!(parsed.sink_table_id, Some(42)); + assert_eq!( + query_ctx.extension(FLOW_INCREMENTAL_MODE), + Some("sequence_range") + ); + assert_eq!( + query_ctx.extension(FLOW_INCREMENTAL_AFTER_SEQS), + Some(r#"{"1":10,"2":20}"#) + ); + assert_eq!(query_ctx.extension(FLOW_RETURN_REGION_SEQ), Some("true")); + assert_eq!(query_ctx.extension(FLOW_SINK_TABLE_ID), Some("42")); + } + #[test] fn test_extract_flow_extensions_rejects_invalid_json() { let mut metadata = MetadataMap::new(); diff --git a/tests/cases/standalone/common/flow/flow_incremental_sequence_range.sql b/tests/cases/standalone/common/flow/flow_incremental_sequence_range.sql new file mode 100644 index 00000000000..ffe8aedf263 --- /dev/null +++ b/tests/cases/standalone/common/flow/flow_incremental_sequence_range.sql @@ -0,0 +1,71 @@ +-- Validate incremental sequence_range reads for an append-only source whose rows +-- preserve insertion sequence across SST flushes. +CREATE TABLE flow_incr_seq_range_input ( + host_id INT, + n INT, + ts TIMESTAMP TIME INDEX, + PRIMARY KEY(host_id) +) WITH ( + append_mode = 'true', + preserve_row_sequence = 'true' +); + +CREATE FLOW flow_incr_seq_range SINK TO flow_incr_seq_range_sink +WITH (experimental_enable_incremental_read = 'true') +AS +SELECT + sum(n) AS total, + count(n) AS row_count, + min(n) AS min_n, + max(n) AS max_n, + date_bin(INTERVAL '1 minute', ts, '2024-01-01 00:00:00') AS time_window +FROM + flow_incr_seq_range_input +GROUP BY + time_window; + +-- ==== Phase 1: initial insert + checkpoint ==== +INSERT INTO flow_incr_seq_range_input VALUES + (1, 10, '2024-01-01 00:00:00'), + (2, 20, '2024-01-01 00:00:15'), + (3, 30, '2024-01-01 00:00:30'); + +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('flow_incr_seq_range'); + +SELECT total, row_count, min_n, max_n, time_window +FROM flow_incr_seq_range_sink +ORDER BY time_window; + +-- Move the checkpointed source and sink state into SST files. +ADMIN FLUSH_TABLE('flow_incr_seq_range_sink'); +ADMIN FLUSH_TABLE('flow_incr_seq_range_input'); + +-- ==== Phase 2: flushed delta in the same window ==== +INSERT INTO flow_incr_seq_range_input VALUES + (4, 40, '2024-01-01 00:00:45'), + (5, 50, '2024-01-01 00:00:55'); + +-- Flush the delta to SST before the incremental flow run. The sequence range +-- scan must read only these new rows, not the already-checkpointed SST. +ADMIN FLUSH_TABLE('flow_incr_seq_range_input'); + +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('flow_incr_seq_range'); + +SELECT total, row_count, min_n, max_n, time_window +FROM flow_incr_seq_range_sink +ORDER BY time_window; + +-- ==== Empty incremental run ==== +-- No new source rows must leave the aggregate unchanged. +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('flow_incr_seq_range'); + +SELECT total, row_count, min_n, max_n, time_window +FROM flow_incr_seq_range_sink +ORDER BY time_window; + +DROP FLOW flow_incr_seq_range; +DROP TABLE flow_incr_seq_range_input; +DROP TABLE flow_incr_seq_range_sink;