mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-09-13 00:42:14 +00:00
fix(flow): restore exact sequence range query wiring
Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
@@ -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::<MemoryCatalogManager>()
|
||||
.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(
|
||||
|
||||
@@ -371,6 +371,8 @@ struct FlowScanDecision {
|
||||
memtable_max_sequence: Option<u64>,
|
||||
/// 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<Flo
|
||||
memtable_min_sequence: None,
|
||||
memtable_max_sequence: query_ctx.get_snapshot(region_id.as_u64()),
|
||||
skip_sst_files: false,
|
||||
exact_sequence_range: false,
|
||||
});
|
||||
};
|
||||
|
||||
@@ -437,6 +441,8 @@ fn decide_flow_scan(query_ctx: &QueryContext, region_id: RegionId) -> Result<Flo
|
||||
memtable_min_sequence,
|
||||
memtable_max_sequence,
|
||||
skip_sst_files,
|
||||
exact_sequence_range: apply_incremental
|
||||
&& flow_extensions.incremental_mode == Some(FlowIncrementalMode::SequenceRange),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -449,11 +455,12 @@ fn build_scan_request(
|
||||
// time. A later scan may still refresh `memtable_max_sequence` if another source scan
|
||||
// has bound a snapshot into `query_ctx` after this provider was created.
|
||||
ScanRequest {
|
||||
sst_min_sequence: (!decision.is_sink_scan)
|
||||
sst_min_sequence: (!decision.is_sink_scan && !decision.exact_sequence_range)
|
||||
.then(|| query_ctx.sst_min_sequence(region_id.as_u64()))
|
||||
.flatten(),
|
||||
skip_sst_files: decision.skip_sst_files,
|
||||
snapshot_on_scan: decision.snapshot_on_scan,
|
||||
exact_sequence_range: decision.exact_sequence_range,
|
||||
memtable_min_sequence: decision.memtable_min_sequence,
|
||||
memtable_max_sequence: decision.memtable_max_sequence,
|
||||
..Default::default()
|
||||
@@ -650,8 +657,8 @@ mod tests {
|
||||
use super::*;
|
||||
use crate::error::Error;
|
||||
use crate::options::{
|
||||
FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE, FLOW_RETURN_REGION_SEQ,
|
||||
FLOW_SINK_TABLE_ID,
|
||||
FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE, FLOW_INCREMENTAL_MODE_SEQUENCE_RANGE,
|
||||
FLOW_RETURN_REGION_SEQ, FLOW_SINK_TABLE_ID,
|
||||
};
|
||||
|
||||
fn test_region_id() -> 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();
|
||||
|
||||
+77
-12
@@ -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([(
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
Reference in New Issue
Block a user