rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-12 15:02:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: use exact sequence ranges for capable incremental flow sources

Browse Source 
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-09-15 17:31:02 +08:00
parent 8a19410be5
commit c9c86eba44
9 changed files with 909 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/meta/src/ddl/create_flow.rs
+18
View File
@@ -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::<bool>().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";
src/flow/src/batching_mode/engine.rs
+55 -12
View File
@@ -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<String, String>,
exact_sequence_range_required: bool,
) -> Result<Arc<BatchingModeOptions>, 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::<bool>()
.map_err(|_| {
batch_opts.experimental_enable_incremental_read = if exact_sequence_range_required {
true
} else {
enable_incremental_read.parse::<bool>().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]
src/flow/src/batching_mode/task.rs
+8 -1
View File
@@ -149,6 +149,7 @@ pub struct TaskConfig {
pub catalog_manager: CatalogManagerRef,
pub query_type: QueryType,
pub batch_opts: Arc<BatchingModeOptions>,
pub exact_sequence_range_required: bool,
pub flow_eval_interval: Option<Duration>,
/// Typed schedule configuration, pre-parsed at task creation time.
pub eval_schedule: Option<EvalSchedule>,
@@ -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, Error> {
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<Self, Error> {
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,
src/flow/src/batching_mode/task/inc.rs
+98 -5
View File
@@ -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<bool, Error> {
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));
}
src/flow/src/batching_mode/task/test.rs
+384 -31
View File
@@ -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<BatchingModeOptions>,
) -> 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<BatchingModeOptions>,
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::<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();
@@ -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<Output, BoxedError> {
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<std::sync::Mutex<Option<String>>>,
@@ -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<dyn crate::batching_mode::frontend_client::GrpcQueryHandlerWithBoxedError> =
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();
src/query/src/dummy_catalog.rs
+153 -3
View File
@@ -382,6 +382,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 {
@@ -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<Flo
memtable_min_sequence: None,
memtable_max_sequence: query_ctx.get_snapshot(region_id.as_u64()),
skip_sst_files: false,
exact_sequence_range: false,
});
};
@@ -448,6 +452,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),
})
}
@@ -460,11 +466,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()
@@ -661,8 +668,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 {
@@ -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();
src/query/src/options.rs
+77 -12
View File
@@ -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([(
src/servers/src/grpc/flight.rs
+45 -1
View File
@@ -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();
tests/cases/standalone/common/flow/flow_incremental_sequence_range.sql
+71
View File
@@ -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;