diff --git a/src/flow/src/batching_mode/task/test.rs b/src/flow/src/batching_mode/task/test.rs index 937376c1d0..4e88d9a501 100644 --- a/src/flow/src/batching_mode/task/test.rs +++ b/src/flow/src/batching_mode/task/test.rs @@ -14,8 +14,8 @@ use std::collections::{BTreeMap, BTreeSet, HashMap}; -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; @@ -31,11 +31,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::*; @@ -394,6 +396,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(); @@ -1625,6 +1683,72 @@ async fn test_exact_required_attempt_rejects_revoked_capability_without_extensio ); } +#[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( diff --git a/src/query/src/dummy_catalog.rs b/src/query/src/dummy_catalog.rs index e38813b3e6..fd947b9c7d 100644 --- a/src/query/src/dummy_catalog.rs +++ b/src/query/src/dummy_catalog.rs @@ -371,6 +371,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 { @@ -381,6 +383,7 @@ impl FlowScanDecision { memtable_min_sequence: None, memtable_max_sequence: None, skip_sst_files: false, + exact_sequence_range: false, } } } @@ -395,6 +398,7 @@ fn decide_flow_scan(query_ctx: &QueryContext, region_id: RegionId) -> Result Result RegionId { @@ -743,6 +750,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(); @@ -813,6 +878,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(); @@ -880,6 +971,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] @@ -938,6 +1067,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 24cd3853f0..a49021f2f1 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 0b7a9dd570..f0035ed3f8 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.result b/tests/cases/standalone/common/flow/flow_incremental_sequence_range.result new file mode 100644 index 0000000000..a193bcd874 --- /dev/null +++ b/tests/cases/standalone/common/flow/flow_incremental_sequence_range.result @@ -0,0 +1,143 @@ +-- 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' +); + +Affected Rows: 0 + +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; + +Affected Rows: 0 + +-- ==== 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'); + +Affected Rows: 3 + +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('flow_incr_seq_range'); + ++-----------------------------------------+ +| ADMIN FLUSH_FLOW('flow_incr_seq_range') | ++-----------------------------------------+ +| FLOW_FLUSHED | ++-----------------------------------------+ + +SELECT total, row_count, min_n, max_n, time_window +FROM flow_incr_seq_range_sink +ORDER BY time_window; + ++-------+-----------+-------+-------+---------------------+ +| total | row_count | min_n | max_n | time_window | ++-------+-----------+-------+-------+---------------------+ +| 60 | 3 | 10 | 30 | 2024-01-01T00:00:00 | ++-------+-----------+-------+-------+---------------------+ + +-- 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_sink') | ++-----------------------------------------------+ +| 0 | ++-----------------------------------------------+ + +ADMIN FLUSH_TABLE('flow_incr_seq_range_input'); + ++------------------------------------------------+ +| ADMIN FLUSH_TABLE('flow_incr_seq_range_input') | ++------------------------------------------------+ +| 0 | ++------------------------------------------------+ + +-- ==== 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'); + +Affected Rows: 2 + +-- 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'); + ++------------------------------------------------+ +| ADMIN FLUSH_TABLE('flow_incr_seq_range_input') | ++------------------------------------------------+ +| 0 | ++------------------------------------------------+ + +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('flow_incr_seq_range'); + ++-----------------------------------------+ +| ADMIN FLUSH_FLOW('flow_incr_seq_range') | ++-----------------------------------------+ +| FLOW_FLUSHED | ++-----------------------------------------+ + +SELECT total, row_count, min_n, max_n, time_window +FROM flow_incr_seq_range_sink +ORDER BY time_window; + ++-------+-----------+-------+-------+---------------------+ +| total | row_count | min_n | max_n | time_window | ++-------+-----------+-------+-------+---------------------+ +| 150 | 5 | 10 | 50 | 2024-01-01T00:00:00 | ++-------+-----------+-------+-------+---------------------+ + +-- ==== 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'); + ++-----------------------------------------+ +| ADMIN FLUSH_FLOW('flow_incr_seq_range') | ++-----------------------------------------+ +| FLOW_FLUSHED | ++-----------------------------------------+ + +SELECT total, row_count, min_n, max_n, time_window +FROM flow_incr_seq_range_sink +ORDER BY time_window; + ++-------+-----------+-------+-------+---------------------+ +| total | row_count | min_n | max_n | time_window | ++-------+-----------+-------+-------+---------------------+ +| 150 | 5 | 10 | 50 | 2024-01-01T00:00:00 | ++-------+-----------+-------+-------+---------------------+ + +DROP FLOW flow_incr_seq_range; + +Affected Rows: 0 + +DROP TABLE flow_incr_seq_range_input; + +Affected Rows: 0 + +DROP TABLE flow_incr_seq_range_sink; + +Affected Rows: 0 + 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 0000000000..ffe8aedf26 --- /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;