diff --git a/src/common/meta/src/key/view_info.rs b/src/common/meta/src/key/view_info.rs index 5efc2c9bb78..a5d53580c25 100644 --- a/src/common/meta/src/key/view_info.rs +++ b/src/common/meta/src/key/view_info.rs @@ -114,7 +114,9 @@ impl ViewInfoValue { } } - pub(crate) fn update( + /// Builds a replacement [`ViewInfoValue`] with the next version, leaving this + /// value unchanged for a caller's compare-and-swap. + pub fn update( &self, new_view_info: Bytes, table_names: HashSet, @@ -315,6 +317,73 @@ mod tests { assert_eq!(value, deserialized); } + #[test] + fn test_update_replaces_fields_and_increments_version() { + let table_names = |table_name: &str| { + let mut set = HashSet::new(); + set.insert(TableName { + catalog_name: "greptime".to_string(), + schema_name: "public".to_string(), + table_name: table_name.to_string(), + }); + set + }; + + let view_info = Bytes::from([1, 2, 3].as_ref()); + let names = table_names("a_table"); + let columns = vec!["a".to_string()]; + let plan_columns = vec!["number".to_string()]; + let definition = "CREATE VIEW test AS SELECT * FROM a_table".to_string(); + let original = ViewInfoValue::new( + view_info.clone(), + names.clone(), + columns.clone(), + plan_columns.clone(), + definition.clone(), + ); + + let new_view_info = Bytes::from([4, 5, 6].as_ref()); + let new_names = table_names("b_table"); + let new_columns = vec!["b".to_string()]; + let new_plan_columns = vec!["number2".to_string()]; + let new_definition = "CREATE VIEW test AS SELECT * FROM b_table".to_string(); + let updated = original.update( + new_view_info.clone(), + new_names.clone(), + new_columns.clone(), + new_plan_columns.clone(), + new_definition.clone(), + ); + + assert_eq!(updated.view_info, new_view_info); + assert_eq!(updated.table_names, new_names); + assert_eq!(updated.columns, new_columns); + assert_eq!(updated.plan_columns, new_plan_columns); + assert_eq!(updated.definition, new_definition); + assert_eq!(original.version, 0); + assert_eq!(updated.version, original.version + 1); + + assert_eq!(original.view_info, view_info); + assert_eq!(original.table_names, names); + assert_eq!(original.columns, columns); + assert_eq!(original.plan_columns, plan_columns); + assert_eq!(original.definition, definition); + + let newest = updated.update( + new_view_info.clone(), + new_names.clone(), + new_columns.clone(), + new_plan_columns.clone(), + new_definition.clone(), + ); + assert_eq!(newest.version, updated.version + 1); + assert_eq!(newest.version, 2); + + let serialized = newest.try_as_raw_value().unwrap(); + let deserialized = ViewInfoValue::try_from_raw_value(&serialized).unwrap(); + assert_eq!(newest, deserialized); + } + #[test] fn test_deserialize_view_info_value_with_vec_u8() { #[derive(Serialize)] diff --git a/src/flow/src/batching_mode/task.rs b/src/flow/src/batching_mode/task.rs index c2c3082995a..48b918400c7 100644 --- a/src/flow/src/batching_mode/task.rs +++ b/src/flow/src/batching_mode/task.rs @@ -340,20 +340,33 @@ pub enum QueryCoverage { FencedRepairChunk { high: BTreeMap }, /// Incremental delta query over `(checkpoint, scan-open snapshot]`. IncrementalDelta, + /// A caller-fenced incremental delta over `(checkpoints, high]`, where + /// `high` is sent as the snapshot read bounds for every source region. + /// + /// Raw [`BatchingTask::execute_plan_unlocked`] only: the caller owns the + /// execution guard and every checkpoint transition. Default execution + /// rejects it, and its bounds are validated before dispatch. + FencedIncrementalDelta { high: BTreeMap }, } impl QueryCoverage { - /// Whether this query should use incremental scan extensions and - /// incremental checkpoint advancement rules. + /// Whether this query uses incremental delta planning and scan extensions. + /// Ordinary deltas may still scan `memtable_only`; the fenced delta may not. fn is_incremental_delta(&self) -> bool { - matches!(self, Self::IncrementalDelta) + matches!( + self, + Self::IncrementalDelta | Self::FencedIncrementalDelta { .. } + ) } /// Snapshot upper bounds requested from the storage layer. Only fenced - /// repair chunks carry bounds; all other coverage relies on normal scans. + /// repair chunks and fenced incremental deltas carry bounds; all other + /// coverage relies on normal scans. fn snapshot_seqs(&self) -> HashMap { match self { - Self::FencedRepairChunk { high } => high.iter().map(|(k, v)| (*k, *v)).collect(), + Self::FencedRepairChunk { high } | Self::FencedIncrementalDelta { high } => { + high.iter().map(|(k, v)| (*k, *v)).collect() + } _ => HashMap::new(), } } @@ -1164,6 +1177,7 @@ impl BatchingTask { } /// Applies checkpoint state updates around the raw insert-plan execution. + /// A fenced incremental delta is rejected before dispatch: it is raw-only. async fn execute_logical_plan_unlocked( &self, engine: &QueryEngineRef, @@ -1173,6 +1187,14 @@ impl BatchingTask { coverage: &QueryCoverage, ) -> Result, Error> { let flow_id = self.config.flow_id; + if matches!(coverage, QueryCoverage::FencedIncrementalDelta { .. }) { + return InvalidQuerySnafu { + reason: format!( + "Flow {flow_id} fenced incremental delta is raw-only: its caller owns checkpoint transitions" + ), + } + .fail(); + } let Some((res, elapsed)) = self .execute_plan_unlocked(engine, frontend_client, plan, dirty_restore, coverage) .await? @@ -1204,6 +1226,10 @@ impl BatchingTask { /// Executes a fully constructed insert plan without checkpoint-state mutation. /// Callers must own the serialized execution guard. Its unsafe-incremental safety fallback /// still restores the supplied dirty work before returning `Ok(None)`. + /// + /// This is the only entry point that accepts + /// [`QueryCoverage::FencedIncrementalDelta`]; it validates the fence against + /// the live checkpoints before dispatch and leaves advancement to its caller. pub async fn execute_plan_unlocked( &self, engine: &QueryEngineRef, @@ -1215,6 +1241,37 @@ impl BatchingTask { let instant = Instant::now(); let flow_id = self.config.flow_id; + // A caller-fenced incremental delta reads `(checkpoints, high]` on every + // source region, so its fence is proved before any rewrite or dispatch. + if let QueryCoverage::FencedIncrementalDelta { high } = coverage { + let state = self.state.read().unwrap(); + if state.checkpoint_mode() != CheckpointMode::Incremental + || state.is_incremental_disabled() + { + return InvalidQuerySnafu { + reason: format!( + "Flow {flow_id} fenced incremental delta requires an enabled incremental flow in incremental mode" + ), + } + .fail(); + } + let checkpoints = state.checkpoints(); + if checkpoints.is_empty() + || !high.keys().eq(checkpoints.keys()) + || high + .values() + .zip(checkpoints.values()) + .any(|(upper, lower)| upper < lower) + { + return InvalidQuerySnafu { + reason: format!( + "Flow {flow_id} fenced incremental delta bounds {high:?} must extend the checkpoint bounds {checkpoints:?} over the same regions" + ), + } + .fail(); + } + } + debug!( "Executing flow {flow_id}(expire_after={:?} secs) with query {}", self.config.expire_after, &plan @@ -1265,6 +1322,20 @@ impl BatchingTask { let extensions = self .build_flow_query_extensions(incremental_safe, coverage.is_incremental_delta()) .await?; + // The builder owns capability discovery; an exact fence additionally + // requires the extensions it actually emitted to scan a sequence range. + if matches!(coverage, QueryCoverage::FencedIncrementalDelta { .. }) + && !extensions.iter().any(|(key, value)| { + *key == FLOW_INCREMENTAL_MODE && *value == FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE + }) + { + return InvalidQuerySnafu { + reason: format!( + "Flow {flow_id} fenced incremental delta requires {FLOW_INCREMENTAL_MODE}={FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE}" + ), + } + .fail(); + } let frontend_extensions = self.frontend_extensions(); let extension_refs = extensions .iter() diff --git a/src/flow/src/batching_mode/task/ckpt.rs b/src/flow/src/batching_mode/task/ckpt.rs index c736a0fdbdf..690cfcf837f 100644 --- a/src/flow/src/batching_mode/task/ckpt.rs +++ b/src/flow/src/batching_mode/task/ckpt.rs @@ -228,6 +228,9 @@ impl BatchingTask { } } } + QueryCoverage::FencedIncrementalDelta { .. } => unreachable!( + "FencedIncrementalDelta is rejected by the default wrapper; only the raw execute_plan_unlocked dispatches it under a caller-owned guard" + ), } } else { if matches!(coverage, QueryCoverage::FencedRepairChunk { .. }) { diff --git a/src/flow/src/batching_mode/task/test.rs b/src/flow/src/batching_mode/task/test.rs index 29a15600f15..a521567dc2b 100644 --- a/src/flow/src/batching_mode/task/test.rs +++ b/src/flow/src/batching_mode/task/test.rs @@ -2720,9 +2720,9 @@ fn test_query_failure_reason_distinguishes_fenced_repair_stale_fence() { #[test] fn test_fenced_repair_coverage_produces_snapshot_seq_map_for_distributed_metadata_path() { // Covers the metadata boundary between QueryCoverage and the - // frontend/distributed client API: only FencedRepairChunk carries a + // frontend/distributed client API: only the fenced variants carry a // non-empty snapshot_seqs map so the datanode can bind per-region - // snapshot upper bounds against the frozen high H. Other coverage + // snapshot upper bounds against the frozen fence. Other coverage // variants must produce an empty map. let high = BTreeMap::from([(1_u64, 10_u64), (2_u64, 20_u64)]); let coverage = QueryCoverage::FencedRepairChunk { high: high.clone() }; @@ -2734,6 +2734,17 @@ fn test_fenced_repair_coverage_produces_snapshot_seq_map_for_distributed_metadat assert!(QueryCoverage::UnfilteredFull.snapshot_seqs().is_empty()); assert!(QueryCoverage::ScopedBaseRepair.snapshot_seqs().is_empty()); assert!(QueryCoverage::IncrementalDelta.snapshot_seqs().is_empty()); + + // A caller-fenced incremental delta binds its fence as the same per-region + // snapshot upper bounds and counts as an incremental delta. + let fenced = QueryCoverage::FencedIncrementalDelta { high: high.clone() }; + assert_eq!( + fenced.snapshot_seqs(), + HashMap::from([(1_u64, 10_u64), (2_u64, 20_u64)]) + ); + assert!(fenced.is_incremental_delta()); + assert!(QueryCoverage::IncrementalDelta.is_incremental_delta()); + assert!(!QueryCoverage::FencedRepairChunk { high }.is_incremental_delta()); } #[tokio::test] @@ -4188,3 +4199,511 @@ async fn test_insert_plan_matching_failure_restores_consumed_dirty_marker() { std::time::Duration::from_secs(5) ); } + +/// Task that can dispatch a fenced incremental delta: sequence-range-capable +/// source and a registered sink table. +async fn new_fenced_incremental_test_task() -> TestTaskParts { + let parts = 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(&parts.query_engine, "mito", true).await; + register_twe_sink(&parts.query_engine, "missing_sink", 9210); + parts +} + +/// Seeds incremental checkpoints plus a dirty signal and returns the insert plan +/// of an ordinary incremental-delta round. +async fn seed_fenced_incremental_plan( + task: &BatchingTask, + engine: &QueryEngineRef, + checkpoints: HashMap, +) -> PlanInfo { + { + let mut state = task.state.write().unwrap(); + state.advance_checkpoints(checkpoints); + state + .dirty_time_windows + .add_window(Timestamp::new_second(0), Some(Timestamp::new_second(5))); + } + let plan = task + .gen_insert_plan_unlocked(engine, Some(1)) + .await + .expect("the dirty signal must generate an insert plan") + .expect("the dirty signal must produce a plan"); + assert!(matches!(plan.coverage, QueryCoverage::IncrementalDelta)); + plan +} + +/// Records the real frontend request of a fenced incremental delta and succeeds +/// with an affected-rows output. +#[derive(Default)] +struct FencedIncrementalHandler { + calls: std::sync::atomic::AtomicUsize, + mode: std::sync::Mutex>, + lower: std::sync::Mutex>, + upper: std::sync::Mutex>>, +} + +#[async_trait::async_trait] +impl crate::batching_mode::frontend_client::GrpcQueryHandlerWithBoxedError + for FencedIncrementalHandler +{ + async fn do_query( + &self, + query: api::v1::greptime_request::Request, + ctx: QueryContextRef, + ) -> std::result::Result { + self.calls.fetch_add(1, Ordering::SeqCst); + *self.mode.lock().unwrap() = ctx.extension(FLOW_INCREMENTAL_MODE).map(str::to_string); + *self.lower.lock().unwrap() = ctx + .extension(FLOW_INCREMENTAL_AFTER_SEQS) + .map(str::to_string); + *self.upper.lock().unwrap() = Some(ctx.snapshots()); + let api::v1::greptime_request::Request::Query(request) = query else { + panic!("fenced incremental delta must issue a query request"); + }; + assert!( + matches!( + request.query, + Some(api::v1::query_request::Query::InsertIntoPlan(_)) + ), + "the rewritten delta-sink merge must dispatch as an insert plan" + ); + Ok(Output::new_with_affected_rows(0)) + } +} + +fn fenced_incremental_frontend( + recorded: &Arc, +) -> ( + Arc, + Arc, +) { + let handler: Arc = + recorded.clone(); + let frontend = Arc::new(FrontendClient::from_grpc_handler( + Arc::downgrade(&handler), + QueryOptions::default(), + )); + (frontend, handler) +} + +/// A valid caller-fenced delta dispatches the rewritten delta-sink merge with +/// the checkpoint map as lower cursor and the fence as per-region snapshot +/// upper bound, and the raw executor never advances checkpoints for it. Both a +/// strictly advancing fence `F < C` and an equal fence `F == C` are dispatched. +#[tokio::test] +async fn test_fenced_incremental_delta_raw_dispatch_never_advances_checkpoints() { + let checkpoints = [(1_u64, 10_u64), (2_u64, 20_u64)]; + for (case, high) in [ + ( + "advancing fence", + BTreeMap::from([(1_u64, 20_u64), (2_u64, 20_u64)]), + ), + ("equal fence", BTreeMap::from(checkpoints)), + ] { + let TestTaskParts { + task, query_engine, .. + } = new_fenced_incremental_test_task().await; + let plan_info = + seed_fenced_incremental_plan(&task, &query_engine, HashMap::from(checkpoints)).await; + + let recording = Arc::new(PlanRecordingExecution::default()); + let task = task.with_execution(Some(recording.clone())); + let recorded = Arc::new(FencedIncrementalHandler::default()); + let (frontend, _handler) = fenced_incremental_frontend(&recorded); + + let raw = task + .execute_plan_unlocked( + &query_engine, + &frontend, + &plan_info.plan, + &plan_info.dirty_restore, + &QueryCoverage::FencedIncrementalDelta { high: high.clone() }, + ) + .await + .unwrap() + .expect("the raw executor must dispatch the fenced incremental delta"); + assert!(raw.0.is_ok(), "{case}: the dispatched query must succeed"); + + assert_eq!(recorded.calls.load(Ordering::SeqCst), 1, "{case}"); + assert_eq!( + *recorded.mode.lock().unwrap(), + Some(FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE.to_string()), + "{case}: the request must carry the exact sequence_range mode" + ); + assert_eq!( + *recorded.lower.lock().unwrap(), + Some(serde_json::json!({"1": 10, "2": 20}).to_string()), + "{case}: the request must read from the checkpoint map" + ); + assert_eq!( + *recorded.upper.lock().unwrap(), + Some( + high.iter() + .map(|(region, seq)| (*region, *seq)) + .collect::>() + ), + "{case}: the request must bind the fence as snapshot upper bounds" + ); + + let recorded_plans = recording.plans.lock().unwrap(); + let merge_plan = recorded_plans + .first() + .expect("the execution hook must see the dispatched plan") + .to_string(); + assert!( + merge_plan.contains("Left Join"), + "{case}: the dispatched plan must be the delta-sink merge rewrite:\n{merge_plan}" + ); + + let state = task.state.read().unwrap(); + assert_eq!( + state.checkpoints(), + &BTreeMap::from(checkpoints), + "{case}: the raw executor must leave checkpoint advancement to its caller" + ); + assert_eq!( + state.checkpoint_mode(), + CheckpointMode::Incremental, + "{case}: the raw executor must not change the checkpoint mode" + ); + } +} + +/// Invalid flow state and invalid lower/upper cursors are rejected as invalid +/// queries before any dispatch or state write. +#[tokio::test] +async fn test_fenced_incremental_delta_rejects_invalid_state_or_bounds() { + struct Case { + name: &'static str, + reason: &'static str, + setup: fn(&mut TaskState), + high: BTreeMap, + } + + let cases = [ + Case { + name: "full snapshot mode", + reason: "requires an enabled incremental flow", + setup: |state| { + state.advance_checkpoints(HashMap::from([(1_u64, 10_u64)])); + state.mark_full_snapshot(); + }, + high: BTreeMap::from([(1_u64, 20_u64)]), + }, + Case { + name: "incremental disabled", + reason: "requires an enabled incremental flow", + setup: |state| { + state.advance_checkpoints(HashMap::from([(1_u64, 10_u64)])); + state.disable_incremental(); + }, + high: BTreeMap::from([(1_u64, 20_u64)]), + }, + Case { + name: "empty checkpoints", + reason: "must extend the checkpoint bounds", + setup: |state| state.advance_checkpoints(HashMap::new()), + high: BTreeMap::from([(1_u64, 10_u64)]), + }, + Case { + name: "empty high", + reason: "must extend the checkpoint bounds", + setup: |state| state.advance_checkpoints(HashMap::from([(1_u64, 10_u64)])), + high: BTreeMap::new(), + }, + Case { + name: "missing region", + reason: "must extend the checkpoint bounds", + setup: |state| { + state.advance_checkpoints(HashMap::from([(1_u64, 10_u64), (2_u64, 20_u64)])) + }, + high: BTreeMap::from([(1_u64, 20_u64)]), + }, + Case { + name: "extra region", + reason: "must extend the checkpoint bounds", + setup: |state| state.advance_checkpoints(HashMap::from([(1_u64, 10_u64)])), + high: BTreeMap::from([(1_u64, 20_u64), (2_u64, 20_u64)]), + }, + Case { + name: "backward high", + reason: "must extend the checkpoint bounds", + setup: |state| { + state.advance_checkpoints(HashMap::from([(1_u64, 10_u64), (2_u64, 20_u64)])) + }, + high: BTreeMap::from([(1_u64, 9_u64), (2_u64, 20_u64)]), + }, + ]; + + for case in cases { + let TestTaskParts { + task, + query_engine, + plan, + .. + } = new_fenced_incremental_test_task().await; + (case.setup)(&mut task.state.write().unwrap()); + let checkpoints_before = task.state.read().unwrap().checkpoints().clone(); + let mode_before = task.state.read().unwrap().checkpoint_mode(); + + let recorded = Arc::new(FencedIncrementalHandler::default()); + let (frontend, _handler) = fenced_incremental_frontend(&recorded); + let result = task + .execute_plan_unlocked( + &query_engine, + &frontend, + &plan, + &DirtyRestore::Unscoped(dirty_marker()), + &QueryCoverage::FencedIncrementalDelta { + high: case.high.clone(), + }, + ) + .await; + let err = match result { + Ok(_) => panic!( + "{}: an invalid fence must be rejected before dispatch", + case.name + ), + Err(err) => err, + }; + + assert!( + matches!(err, Error::InvalidQuery { .. }), + "{}: expected an invalid query, got {err:?}", + case.name + ); + assert!( + err.to_string().contains(case.reason), + "{}: expected {:?} in {err}", + case.name, + case.reason + ); + assert_eq!( + recorded.calls.load(Ordering::SeqCst), + 0, + "{}: an invalid fence must not dispatch a query", + case.name + ); + let state = task.state.read().unwrap(); + assert_eq!( + state.checkpoints(), + &checkpoints_before, + "{}: an invalid fence must not write checkpoints", + case.name + ); + assert_eq!( + state.checkpoint_mode(), + mode_before, + "{}: an invalid fence must not change the checkpoint mode", + case.name + ); + } +} + +/// Sources that cannot prove an exact sequence range — present but incapable, or +/// not resolvable at all — never fall back to `memtable_only`: the emitted +/// extensions fail the exact-mode guard and nothing is dispatched. Each case +/// degrades exactly one source, so no capability-loop ordering can mask it. +#[tokio::test] +async fn test_fenced_incremental_delta_requires_sequence_range_mode_before_dispatch() { + for (case, incapable_source, missing_source) in [ + ("incapable source", true, false), + ("unresolvable source", false, true), + ] { + let TestTaskParts { + mut task, + query_engine, + .. + } = new_fenced_incremental_test_task().await; + assert!( + !task.config.exact_sequence_range_required, + "{case}: the exact read requirement must not depend on the config flag" + ); + let plan_info = + seed_fenced_incremental_plan(&task, &query_engine, HashMap::from([(1_u64, 10_u64)])) + .await; + + if incapable_source { + // A present but non-mito source cannot prove `preserve_row_sequence`. + configure_source_capability(&query_engine, "file", true).await; + } + if missing_source { + // An unresolvable source makes the capability lookup fail, which the + // builder treats as incapable when exact reads are not required. + Arc::get_mut(&mut task.config) + .expect("test task config should be uniquely owned") + .source_table_names + .insert([ + DEFAULT_CATALOG_NAME.to_string(), + DEFAULT_SCHEMA_NAME.to_string(), + "missing_source".to_string(), + ]); + } + + let recorded = Arc::new(FencedIncrementalHandler::default()); + let (frontend, _handler) = fenced_incremental_frontend(&recorded); + let result = task + .execute_plan_unlocked( + &query_engine, + &frontend, + &plan_info.plan, + &plan_info.dirty_restore, + &QueryCoverage::FencedIncrementalDelta { + high: BTreeMap::from([(1_u64, 20_u64)]), + }, + ) + .await; + let err = match result { + Ok(_) => panic!("{case}: a non-capable source must not dispatch a fenced delta"), + Err(err) => err, + }; + assert!(matches!(err, Error::InvalidQuery { .. }), "{case}: {err:?}"); + assert!(err.to_string().contains("sequence_range"), "{case}: {err}"); + assert_eq!( + recorded.calls.load(Ordering::SeqCst), + 0, + "{case}: a non-capable source must not dispatch a query" + ); + let state = task.state.read().unwrap(); + assert_eq!( + state.checkpoints(), + &BTreeMap::from([(1_u64, 10_u64)]), + "{case}: a rejected fence must not write checkpoints" + ); + assert_eq!( + state.checkpoint_mode(), + CheckpointMode::Incremental, + "{case}: a rejected fence must not change the checkpoint mode" + ); + } +} + +/// An incremental-unsafe plan is skipped instead of degrading to an unfiltered +/// full snapshot, and the pre-existing permanent incremental disable stays. +#[tokio::test] +async fn test_fenced_incremental_delta_unsafe_plan_skips_without_full_fallback() { + let TestTaskParts { + task, query_engine, .. + } = new_time_window_test_task_with_query("SELECT number FROM numbers_with_ts").await; + configure_source_capability(&query_engine, "mito", true).await; + register_number_only_sink(&query_engine, "missing_sink"); + let plan_info = + seed_fenced_incremental_plan(&task, &query_engine, HashMap::from([(1_u64, 10_u64)])).await; + + let recorded = Arc::new(FencedIncrementalHandler::default()); + let (frontend, _handler) = fenced_incremental_frontend(&recorded); + let result = task + .execute_plan_unlocked( + &query_engine, + &frontend, + &plan_info.plan, + &plan_info.dirty_restore, + &QueryCoverage::FencedIncrementalDelta { + high: BTreeMap::from([(1_u64, 20_u64)]), + }, + ) + .await + .expect("an unsafe incremental plan must skip the round"); + + assert!( + result.is_none(), + "an unsafe incremental plan must not dispatch anything" + ); + assert_eq!(recorded.calls.load(Ordering::SeqCst), 0); + let state = task.state.read().unwrap(); + assert_eq!(state.checkpoints(), &BTreeMap::from([(1_u64, 10_u64)])); + assert_eq!(state.dirty_time_windows.len(), 1); + // Existing behavior: an unsafe plan permanently disables incremental reads. + assert!(state.is_incremental_disabled()); +} + +/// A dispatched frontend error is returned to the caller without touching +/// checkpoints. +#[tokio::test] +async fn test_fenced_incremental_delta_frontend_error_leaves_checkpoints_unchanged() { + let TestTaskParts { + task, query_engine, .. + } = new_fenced_incremental_test_task().await; + let plan_info = + seed_fenced_incremental_plan(&task, &query_engine, HashMap::from([(1_u64, 10_u64)])).await; + + let handler_invoked = Arc::new(AtomicBool::new(false)); + let handler: Arc = + Arc::new(ExactDeltaFailureHandler { + invoked: handler_invoked.clone(), + }); + let frontend_client = Arc::new(FrontendClient::from_grpc_handler( + Arc::downgrade(&handler), + QueryOptions::default(), + )); + + let raw = task + .execute_plan_unlocked( + &query_engine, + &frontend_client, + &plan_info.plan, + &plan_info.dirty_restore, + &QueryCoverage::FencedIncrementalDelta { + high: BTreeMap::from([(1_u64, 20_u64)]), + }, + ) + .await + .expect("the raw executor must return the dispatched execution result") + .expect("the raw executor must dispatch the fenced incremental delta"); + assert!( + raw.0.is_err(), + "the dispatched exact delta must fail through the injected frontend handler" + ); + assert!(handler_invoked.load(Ordering::SeqCst)); + + let state = task.state.read().unwrap(); + assert_eq!(state.checkpoints(), &BTreeMap::from([(1_u64, 10_u64)])); + assert_eq!(state.checkpoint_mode(), CheckpointMode::Incremental); +} + +/// The default execution wrapper rejects a fenced incremental delta before the +/// raw executor runs, so it can neither dispatch a query nor touch checkpoints. +#[tokio::test] +async fn test_fenced_incremental_delta_default_wrapper_rejects_before_dispatch() { + let TestTaskParts { + task, + query_engine, + plan, + .. + } = new_fenced_incremental_test_task().await; + task.state + .write() + .unwrap() + .advance_checkpoints(HashMap::from([(1_u64, 10_u64)])); + + let recorded = Arc::new(FencedIncrementalHandler::default()); + let (frontend, _handler) = fenced_incremental_frontend(&recorded); + let result = task + .execute_logical_plan_unlocked( + &query_engine, + &frontend, + &plan, + &DirtyRestore::Unscoped(dirty_marker()), + &QueryCoverage::FencedIncrementalDelta { + high: BTreeMap::from([(1_u64, 20_u64)]), + }, + ) + .await; + let err = match result { + Ok(_) => panic!("the default wrapper must reject a fenced incremental delta"), + Err(err) => err, + }; + assert!(matches!(err, Error::InvalidQuery { .. }), "{err:?}"); + assert!(err.to_string().contains("raw-only"), "{err}"); + assert_eq!( + recorded.calls.load(Ordering::SeqCst), + 0, + "the default wrapper must reject before dispatch" + ); + let state = task.state.read().unwrap(); + assert_eq!(state.checkpoints(), &BTreeMap::from([(1_u64, 10_u64)])); + assert_eq!(state.checkpoint_mode(), CheckpointMode::Incremental); +}