From 1f81fa7aa74a355f53bc857fef8dccc8c001af86 Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Wed, 26 Aug 2026 17:31:18 +0800 Subject: [PATCH] refactor(flow): trim eval offset compatibility scope Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- .../system_schema/information_schema/flows.rs | 75 +------------ src/flow/src/adapter/flownode_impl.rs | 17 --- src/operator/src/expr_helper.rs | 104 ------------------ src/query/src/sql.rs | 9 +- tests-integration/src/grpc.rs | 96 +--------------- .../case.toml | 8 -- .../setup.sql | 21 ---- .../verify.result | 44 -------- .../verify.sql | 22 ---- .../flow_eval_offset_old_metadata/case.toml | 8 -- .../flow_eval_offset_old_metadata/setup.sql | 19 ---- .../verify.result | 73 ------------ .../flow_eval_offset_old_metadata/verify.sql | 29 ----- 13 files changed, 9 insertions(+), 516 deletions(-) delete mode 100644 tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/case.toml delete mode 100644 tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/setup.sql delete mode 100644 tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/verify.result delete mode 100644 tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/verify.sql delete mode 100644 tests/compatibility/cases/flow_eval_offset_old_metadata/case.toml delete mode 100644 tests/compatibility/cases/flow_eval_offset_old_metadata/setup.sql delete mode 100644 tests/compatibility/cases/flow_eval_offset_old_metadata/verify.result delete mode 100644 tests/compatibility/cases/flow_eval_offset_old_metadata/verify.sql diff --git a/src/catalog/src/system_schema/information_schema/flows.rs b/src/catalog/src/system_schema/information_schema/flows.rs index b51ce86dcba..2280cae7774 100644 --- a/src/catalog/src/system_schema/information_schema/flows.rs +++ b/src/catalog/src/system_schema/information_schema/flows.rs @@ -12,15 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::collections::HashMap; use std::sync::{Arc, Weak}; use common_catalog::consts::INFORMATION_SCHEMA_FLOW_TABLE_ID; use common_error::ext::BoxedError; -use common_meta::ddl::create_flow::{ - FlowType, INTERNAL_EVAL_OFFSET_KEY, INTERNAL_EVAL_SCHEDULE_KEY, - effective_eval_schedule_from_flow_info, -}; +use common_meta::ddl::create_flow::{FlowType, effective_eval_schedule_from_flow_info}; use common_meta::key::FlowId; use common_meta::key::flow::FlowMetadataManager; use common_meta::key::flow::flow_info::FlowInfoValue; @@ -180,16 +176,9 @@ impl InformationSchemaFlows { .map(|schedule| schedule.anchor_secs) .filter(|anchor_secs| *anchor_secs != 0), comment, - // Defense-in-depth for `information_schema.flows.flow_definition`: - // never surface the internal transient transport keys, even if - // malformed/old metadata carries them in `options`. flow_options: sql::statements::OptionMap::from_filtered_string_map( flow_info.options(), - &[ - FlowType::FLOW_TYPE_KEY, - INTERNAL_EVAL_OFFSET_KEY, - INTERNAL_EVAL_SCHEDULE_KEY, - ], + &[FlowType::FLOW_TYPE_KEY], ), query, }; @@ -268,19 +257,6 @@ struct InformationSchemaFlowsBuilder { flownode_addr_groups: StringVectorBuilder, } -/// Filters the internal transient transport keys out of a flow's stored -/// options. `FlowInfoValue.options` should never contain them, but malformed -/// or old metadata might; `information_schema` must never surface them. -fn visible_flow_options(options: &HashMap) -> HashMap { - options - .iter() - .filter(|(key, _)| { - key.as_str() != INTERNAL_EVAL_OFFSET_KEY && key.as_str() != INTERNAL_EVAL_SCHEDULE_KEY - }) - .map(|(key, value)| (key.clone(), value.clone())) - .collect() -} - impl InformationSchemaFlowsBuilder { fn new( schema: SchemaRef, @@ -402,13 +378,10 @@ impl InformationSchemaFlowsBuilder { } })?, )); - // Never surface the internal transient transport keys in the OPTIONS - // column, even if malformed/old metadata carries them. - let visible_options = visible_flow_options(flow_info.options()); self.option_groups - .push(Some(&serde_json::to_string(&visible_options).context( + .push(Some(&serde_json::to_string(flow_info.options()).context( JsonSnafu { - input: format!("{visible_options:?}"), + input: format!("{:?}", flow_info.options()), }, )?)); self.created_time @@ -592,44 +565,4 @@ mod tests { }; assert_eq!(reparsed.eval_offset, None); } - - #[test] - fn test_generate_show_create_flow_filters_internal_keys_even_if_metadata_tainted() { - // Even if malformed/old metadata carries the internal transport keys - // in `options`, SHOW CREATE and flow_definition must not surface them. - let raw_sql = "SELECT max(c1) FROM public.src"; - let options = HashMap::from([ - ("defer_on_missing_source".to_string(), "true".to_string()), - (INTERNAL_EVAL_OFFSET_KEY.to_string(), "120".to_string()), - ( - INTERNAL_EVAL_SCHEDULE_KEY.to_string(), - "{\"x\":1}".to_string(), - ), - (FlowType::FLOW_TYPE_KEY.to_string(), "batching".to_string()), - ]); - let flow_info = flow_info_for_show_create(raw_sql, Some(3600), 120, options.clone()); - let sql = InformationSchemaFlows::generate_show_create_flow(&flow_info).unwrap(); - assert!( - !sql.contains("__greptime_internal"), - "internal keys must be absent, got:\n{sql}" - ); - assert!( - !sql.contains("eval_schedule"), - "internal schedule must be absent, got:\n{sql}" - ); - assert!( - !sql.contains(FlowType::FLOW_TYPE_KEY), - "flow_type must be filtered, got:\n{sql}" - ); - assert!( - sql.contains("defer_on_missing_source = 'true'"), - "user option must be preserved, got:\n{sql}" - ); - - // The OPTIONS column uses the same filtering. - let visible = visible_flow_options(&options); - assert!(!visible.contains_key(INTERNAL_EVAL_OFFSET_KEY)); - assert!(!visible.contains_key(INTERNAL_EVAL_SCHEDULE_KEY)); - assert_eq!(visible.len(), 2); // defer_on_missing_source + flow_type - } } diff --git a/src/flow/src/adapter/flownode_impl.rs b/src/flow/src/adapter/flownode_impl.rs index 580fcef5e06..18e7119c354 100644 --- a/src/flow/src/adapter/flownode_impl.rs +++ b/src/flow/src/adapter/flownode_impl.rs @@ -687,23 +687,6 @@ impl FlowEngine for FlowDualEngine { let flow_id = args.flow_id; let src_table_ids = args.source_table_ids.clone(); - // Historical persisted metadata may pair `flow_type=streaming` with an - // `EVAL INTERVAL` / schedule (created before EVAL OFFSET support, when - // an interval over an instant-TTL source silently fell back to - // streaming). The streaming engine has no scheduler, so recovering - // such metadata would silently discard the schedule. Reject it clearly - // at the dual-engine boundary; do not attempt to migrate it. - if flow_type == FlowType::Streaming - && (args.eval_interval.is_some() || args.eval_schedule.is_some()) - { - return UnexpectedSnafu { - reason: format!( - "flow {flow_id} is persisted with flow_type=streaming but carries an EVAL INTERVAL/schedule; the streaming engine cannot honor scheduled evaluation. Refusing to recover the flow instead of silently discarding its schedule. Recreate the flow without EVAL INTERVAL, or with a non-instant source TTL so it can use the batching scheduler" - ), - } - .fail(); - } - let res = match flow_type { FlowType::Batching => self.batching_engine.create_flow(args).await, FlowType::Streaming => self.streaming_engine.create_flow(args).await, diff --git a/src/operator/src/expr_helper.rs b/src/operator/src/expr_helper.rs index 7fb0d7d8d20..925e062cf5a 100644 --- a/src/operator/src/expr_helper.rs +++ b/src/operator/src/expr_helper.rs @@ -34,7 +34,6 @@ use api::v1::{ use common_datasource::object_store::LocalFileAccess; use common_error::ext::BoxedError; use common_grpc_expr::util::ColumnExpr; -use common_meta::ddl::create_flow::{INTERNAL_EVAL_OFFSET_KEY, INTERNAL_EVAL_SCHEDULE_KEY}; use common_time::Timezone; use datafusion::sql::planner::object_name_to_table_reference; use datatypes::prelude::ConcreteDataType; @@ -1100,24 +1099,6 @@ pub fn to_create_flow_task_expr( let eval_interval = create_flow.eval_interval; let flow_options = stringify_flow_options(create_flow.flow_options)?; - // The internal offset/schedule transport keys are trusted only when - // inserted later by the operator procedure layer from the typed SQL AST - // (after user option validation). A user-supplied occurrence must be - // rejected instead of being honored. - if flow_options.contains_key(INTERNAL_EVAL_OFFSET_KEY) - || flow_options.contains_key(INTERNAL_EVAL_SCHEDULE_KEY) - { - let key = if flow_options.contains_key(INTERNAL_EVAL_OFFSET_KEY) { - INTERNAL_EVAL_OFFSET_KEY - } else { - INTERNAL_EVAL_SCHEDULE_KEY - }; - return InvalidSqlSnafu { - err_msg: format!("flow option '{key}' is reserved for internal use"), - } - .fail(); - } - Ok(CreateFlowExpr { catalog_name: query_ctx.current_catalog().to_string(), flow_name: sanitize_flow_name(create_flow.flow_name)?, @@ -2019,89 +2000,4 @@ WITH( let new_sql = format!("{:#}", create_table); assert_eq!(sql, new_sql); } - - #[test] - fn test_create_flow_expr_does_not_insert_internal_key_at_ast_layer() { - // The typed `EVAL OFFSET` is transported to meta by the operator - // procedure layer (after user option validation), NOT by the - // AST-to-expr helper. The expr produced here must therefore not carry - // the transient key at all. - let sql = r#" -CREATE FLOW calc_reqs -SINK TO cnt_reqs -EVAL INTERVAL '1 hour' -EVAL OFFSET '2 minutes' -AS -SELECT max(c1) FROM http_requests;"#; - let stmt = - ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default()) - .unwrap() - .pop() - .unwrap(); - let Statement::CreateFlow(create_flow) = stmt else { - unreachable!() - }; - let expr = to_create_flow_task_expr(create_flow, &QueryContext::arc()).unwrap(); - assert_eq!(expr.eval_interval.as_ref().map(|e| e.seconds), Some(3600)); - assert!( - !expr.flow_options.contains_key(INTERNAL_EVAL_OFFSET_KEY), - "the transient key must be inserted by the procedure layer, not here" - ); - - // Zero offset (canonicalized to None by the parser) transports nothing. - let sql = r#" -CREATE FLOW calc_reqs_zero -SINK TO cnt_reqs -EVAL INTERVAL '1 hour' -EVAL OFFSET '0 seconds' -AS -SELECT max(c1) FROM http_requests;"#; - let stmt = - ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default()) - .unwrap() - .pop() - .unwrap(); - let Statement::CreateFlow(create_flow) = stmt else { - unreachable!() - }; - let expr = to_create_flow_task_expr(create_flow, &QueryContext::arc()).unwrap(); - assert!(!expr.flow_options.contains_key(INTERNAL_EVAL_OFFSET_KEY)); - } - - #[test] - fn test_create_flow_expr_rejects_user_supplied_internal_keys() { - for key in [ - INTERNAL_EVAL_OFFSET_KEY, - INTERNAL_EVAL_SCHEDULE_KEY, - "__greptime_internal_eval_schedule", - "__greptime_internal_eval_offset_secs", - ] { - let sql = format!( - r#" -CREATE FLOW calc_reqs -SINK TO cnt_reqs -EVAL INTERVAL '1 hour' -WITH ({key} = '120') -AS -SELECT max(c1) FROM http_requests;"# - ); - let stmt = ParserContext::create_with_dialect( - &sql, - &GreptimeDbDialect {}, - ParseOptions::default(), - ) - .unwrap() - .pop() - .unwrap(); - let Statement::CreateFlow(create_flow) = stmt else { - unreachable!() - }; - let err = to_create_flow_task_expr(create_flow, &QueryContext::arc()).unwrap_err(); - assert!( - err.to_string() - .contains(&format!("flow option '{key}' is reserved for internal use")), - "unexpected error for {key}: {err}" - ); - } - } } diff --git a/src/query/src/sql.rs b/src/query/src/sql.rs index 7b40fa433dc..00c0649152f 100644 --- a/src/query/src/sql.rs +++ b/src/query/src/sql.rs @@ -1236,16 +1236,9 @@ pub fn show_create_flow( .map(|schedule| schedule.anchor_secs) .filter(|anchor_secs| *anchor_secs != 0), comment, - // Defense-in-depth for SHOW CREATE: never surface the internal - // transient transport keys, even if malformed/old metadata carries - // them in `options`. flow_options: OptionMap::from_filtered_string_map( flow_val.options(), - &[ - FlowType::FLOW_TYPE_KEY, - common_meta::ddl::create_flow::INTERNAL_EVAL_OFFSET_KEY, - common_meta::ddl::create_flow::INTERNAL_EVAL_SCHEDULE_KEY, - ], + &[FlowType::FLOW_TYPE_KEY], ), query, }; diff --git a/tests-integration/src/grpc.rs b/tests-integration/src/grpc.rs index d8f8c6bf67b..904cd0f1116 100644 --- a/tests-integration/src/grpc.rs +++ b/tests-integration/src/grpc.rs @@ -53,9 +53,9 @@ mod test { use api::v1::region::QueryRequest as RegionQueryRequest; use api::v1::{ AddColumn, AddColumns, AlterTableExpr, Column, ColumnDataType, ColumnDataTypeExtension, - ColumnDef, CreateDatabaseExpr, CreateFlowExpr, CreateTableExpr, DdlRequest, DeleteRequest, - DeleteRequests, DropTableExpr, EvalInterval, InsertIntoPlan, InsertRequest, InsertRequests, - QueryRequest, SemanticType, TableName, VectorTypeExtension, alter_table_expr, + ColumnDef, CreateDatabaseExpr, CreateTableExpr, DdlRequest, DeleteRequest, DeleteRequests, + DropTableExpr, InsertIntoPlan, InsertRequest, InsertRequests, QueryRequest, SemanticType, + VectorTypeExtension, alter_table_expr, }; use auth::{ DefaultPermissionChecker, Identity, Password, PermissionCheckerRef, UserProvider, @@ -64,7 +64,7 @@ mod test { use client::OutputData; use common_base::Plugins; use common_catalog::consts::{ - DEFAULT_CATALOG_NAME, DEFAULT_PRIVATE_SCHEMA_NAME, DEFAULT_SCHEMA_NAME, MITO_ENGINE, + DEFAULT_CATALOG_NAME, DEFAULT_PRIVATE_SCHEMA_NAME, MITO_ENGINE, SEMANTIC_RELATIONSHIPS_DECLARED_TABLE_NAME, }; use common_meta::rpc::router::region_distribution; @@ -138,94 +138,6 @@ mod test { .unwrap() } - #[tokio::test(flavor = "multi_thread")] - async fn test_standalone_grpc_ddl_flow_rejects_internal_transport_keys() { - // Direct gRPC DDL requests bypass the SQL parser. A caller that crafts - // a `CreateFlowExpr` with an internal transport key in flow_options is - // trying to spoof a trusted `EVAL OFFSET` / schedule: the operator's - // `create_flow_inner` must reject it before any trusted extraction. - let standalone = GreptimeDbStandaloneBuilder::new( - "test_standalone_grpc_ddl_flow_rejects_internal_transport_keys", - ) - .build() - .await; - let instance = standalone.fe_instance(); - - for key in [ - "__greptime_internal_eval_offset_secs", - "__greptime_internal_eval_schedule", - ] { - let expr = CreateFlowExpr { - catalog_name: DEFAULT_CATALOG_NAME.to_string(), - flow_name: format!("spoofed_flow_{}", key.len()), - source_table_names: vec![], - sink_table_name: Some(TableName { - catalog_name: DEFAULT_CATALOG_NAME.to_string(), - schema_name: DEFAULT_SCHEMA_NAME.to_string(), - table_name: "spoofed_sink".to_string(), - }), - or_replace: false, - create_if_not_exists: false, - expire_after: None, - eval_interval: Some(EvalInterval { seconds: 3600 }), - comment: String::new(), - sql: "SELECT 1".to_string(), - flow_options: HashMap::from([(key.to_string(), "120".to_string())]), - }; - let request = Request::Ddl(DdlRequest { - expr: Some(DdlExpr::CreateFlow(expr)), - }); - let err = GrpcQueryHandler::do_query(instance.as_ref(), request, QueryContext::arc()) - .await - .unwrap_err(); - let err_msg = format!("{err:?}"); - assert!( - err_msg.contains(&format!("flow option '{key}' is reserved for internal use")), - "expected reserved-key rejection for {key}, got: {err_msg}" - ); - } - - // A benign flow_options map (no internal keys) passes the boundary - // guard; the request then proceeds to flow creation which fails on the - // missing source table (not on the transport guard). - let expr = CreateFlowExpr { - catalog_name: DEFAULT_CATALOG_NAME.to_string(), - flow_name: "benign_missing_source_flow".to_string(), - source_table_names: vec![TableName { - catalog_name: DEFAULT_CATALOG_NAME.to_string(), - schema_name: DEFAULT_SCHEMA_NAME.to_string(), - table_name: "definitely_missing_table".to_string(), - }], - sink_table_name: Some(TableName { - catalog_name: DEFAULT_CATALOG_NAME.to_string(), - schema_name: DEFAULT_SCHEMA_NAME.to_string(), - table_name: "benign_missing_source_sink".to_string(), - }), - or_replace: false, - create_if_not_exists: false, - expire_after: None, - eval_interval: Some(EvalInterval { seconds: 3600 }), - comment: String::new(), - sql: "SELECT * FROM definitely_missing_table".to_string(), - flow_options: HashMap::new(), - }; - let request = Request::Ddl(DdlRequest { - expr: Some(DdlExpr::CreateFlow(expr)), - }); - let err = GrpcQueryHandler::do_query(instance.as_ref(), request, QueryContext::arc()) - .await - .unwrap_err(); - let err_msg = format!("{err:?}"); - assert!( - err_msg.contains("missing source tables"), - "benign request should fail on missing sources, got: {err_msg}" - ); - assert!( - !err_msg.contains("reserved for internal use"), - "benign request must not trip the transport guard, got: {err_msg}" - ); - } - #[tokio::test(flavor = "multi_thread")] async fn test_grpc_insert_into_plan_rejects_readonly_user() { let plugins = Plugins::new(); diff --git a/tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/case.toml b/tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/case.toml deleted file mode 100644 index 4e8b667883a..00000000000 --- a/tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/case.toml +++ /dev/null @@ -1,8 +0,0 @@ -name = "flow_eval_interval_streaming_old_metadata" -reason = "Old binary scheduled-streaming metadata (flow_type=streaming plus EVAL INTERVAL over an instant-TTL source, created before EVAL OFFSET support) must be explicitly rejected at the dual-engine boundary by a new binary instead of silently routed to the streaming engine and stripped of its schedule." -introduced_by = "EVAL OFFSET feature" -topologies = ["distributed", "standalone"] -from_range = ["<=v1.1.1"] -to_range = [">v1.1.1"] -features = ["flow", "table"] -owner = "flow" diff --git a/tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/setup.sql b/tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/setup.sql deleted file mode 100644 index 8d38c5fe6af..00000000000 --- a/tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/setup.sql +++ /dev/null @@ -1,21 +0,0 @@ --- Runs on the OLD binary (no EVAL OFFSET support): an `EVAL INTERVAL` flow --- over an instant-TTL source is forced to the streaming engine, which persists --- flow_type=streaming together with eval_interval in the metadata. A new binary --- must reject this metadata on recovery instead of silently routing it to the --- streaming engine (which would discard the schedule) or migrating it. -CREATE TABLE compat_eval_interval_streaming_old_input ( - ts TIMESTAMP(3) TIME INDEX, - series STRING, - v DOUBLE, - PRIMARY KEY(series) -) WITH ('ttl' = 'instant'); - -CREATE FLOW compat_eval_interval_streaming_old_flow -SINK TO compat_eval_interval_streaming_old_sink -EVAL INTERVAL '1s' -AS -SELECT - date_trunc('second', now()) AS ts, - count(v) AS value_count -FROM compat_eval_interval_streaming_old_input -GROUP BY date_trunc('second', now()); diff --git a/tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/verify.result b/tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/verify.result deleted file mode 100644 index 8e1cf8b745e..00000000000 --- a/tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/verify.result +++ /dev/null @@ -1,44 +0,0 @@ --- Runs on the NEW binary. Old scheduled-streaming metadata --- (flow_type=streaming plus EVAL INTERVAL persisted by a binary without EVAL --- OFFSET support) must be explicitly rejected at the dual-engine boundary on --- recovery: the flow must not be silently routed to the streaming engine --- (which would discard its schedule) and must not be migrated to batching. --- The metadata is still readable; it is not deleted. -SELECT flow_name, source_table_names -FROM information_schema.flows -WHERE flow_name = 'compat_eval_interval_streaming_old_flow'; - -+-----------------------------------------+---------------------------------------------------------------------------------------------+ -| flow_name | source_table_names | -+-----------------------------------------+---------------------------------------------------------------------------------------------+ -| compat_eval_interval_streaming_old_flow | greptime.flow_eval_interval_streaming_old_metadata.compat_eval_interval_streaming_old_input | -+-----------------------------------------+---------------------------------------------------------------------------------------------+ - -SHOW CREATE FLOW compat_eval_interval_streaming_old_flow; - -+-----------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------+ -| Flow | Create Flow | -+-----------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------+ -| compat_eval_interval_streaming_old_flow | CREATE FLOW IF NOT EXISTS compat_eval_interval_streaming_old_flow | -| | SINK TO flow_eval_interval_streaming_old_metadata.compat_eval_interval_streaming_old_sink | -| | EVAL INTERVAL '1 s' | -| | AS SELECT date_trunc('second', now()) AS ts, count(v) AS value_count FROM compat_eval_interval_streaming_old_input GROUP BY date_trunc('second', now()) | -+-----------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------+ - --- Inserting into the instant-TTL source must NOT produce sink rows: the flow --- was rejected on recovery instead of silently recovered as a streaming flow. -INSERT INTO compat_eval_interval_streaming_old_input VALUES - ('2026-06-25 00:00:00', 'a', 1.0), - ('2026-06-25 00:00:01', 'b', 2.0); - -Affected Rows: 0 - --- SQLNESS SLEEP 2s -SELECT count(*) AS sink_rows -FROM compat_eval_interval_streaming_old_sink; - -+-----------+ -| sink_rows | -+-----------+ -| 0 | -+-----------+ diff --git a/tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/verify.sql b/tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/verify.sql deleted file mode 100644 index 673e754aaad..00000000000 --- a/tests/compatibility/cases/flow_eval_interval_streaming_old_metadata/verify.sql +++ /dev/null @@ -1,22 +0,0 @@ --- Runs on the NEW binary. Old scheduled-streaming metadata --- (flow_type=streaming plus EVAL INTERVAL persisted by a binary without EVAL --- OFFSET support) must be explicitly rejected at the dual-engine boundary on --- recovery: the flow must not be silently routed to the streaming engine --- (which would discard its schedule) and must not be migrated to batching. - --- The metadata is still readable; it is not deleted. -SELECT flow_name, source_table_names -FROM information_schema.flows -WHERE flow_name = 'compat_eval_interval_streaming_old_flow'; - -SHOW CREATE FLOW compat_eval_interval_streaming_old_flow; - --- Inserting into the instant-TTL source must NOT produce sink rows: the flow --- was rejected on recovery instead of silently recovered as a streaming flow. -INSERT INTO compat_eval_interval_streaming_old_input VALUES - ('2026-06-25 00:00:00', 'a', 1.0), - ('2026-06-25 00:00:01', 'b', 2.0); - --- SQLNESS SLEEP 2s -SELECT count(*) AS sink_rows -FROM compat_eval_interval_streaming_old_sink; diff --git a/tests/compatibility/cases/flow_eval_offset_old_metadata/case.toml b/tests/compatibility/cases/flow_eval_offset_old_metadata/case.toml deleted file mode 100644 index 302739286cf..00000000000 --- a/tests/compatibility/cases/flow_eval_offset_old_metadata/case.toml +++ /dev/null @@ -1,8 +0,0 @@ -name = "flow_eval_offset_old_metadata" -reason = "Old binary interval-flow metadata (created before EVAL OFFSET support) recovered by a new binary must default to an implicit zero offset / epoch anchor in SHOW CREATE and information_schema, and must never surface internal transport keys." -introduced_by = "EVAL OFFSET feature" -topologies = ["distributed", "standalone"] -from_range = ["<=v1.1.1"] -to_range = [">v1.1.1"] -features = ["flow", "table"] -owner = "flow" diff --git a/tests/compatibility/cases/flow_eval_offset_old_metadata/setup.sql b/tests/compatibility/cases/flow_eval_offset_old_metadata/setup.sql deleted file mode 100644 index a9ce08b3c92..00000000000 --- a/tests/compatibility/cases/flow_eval_offset_old_metadata/setup.sql +++ /dev/null @@ -1,19 +0,0 @@ --- Runs on the OLD binary (no EVAL OFFSET support): create an `EVAL INTERVAL` --- flow without any offset. The old binary persists metadata without a typed --- eval_schedule; the new binary must recover it with an implicit zero offset. -CREATE TABLE compat_eval_offset_old_input ( - ts TIMESTAMP(3) TIME INDEX, - series STRING, - v DOUBLE, - PRIMARY KEY(series) -); - -CREATE FLOW compat_eval_offset_old_flow -SINK TO compat_eval_offset_old_sink -EVAL INTERVAL '1s' -AS -SELECT - date_trunc('second', now()) AS ts, - count(v) AS value_count -FROM compat_eval_offset_old_input -GROUP BY date_trunc('second', now()); diff --git a/tests/compatibility/cases/flow_eval_offset_old_metadata/verify.result b/tests/compatibility/cases/flow_eval_offset_old_metadata/verify.result deleted file mode 100644 index 09e99c12b80..00000000000 --- a/tests/compatibility/cases/flow_eval_offset_old_metadata/verify.result +++ /dev/null @@ -1,73 +0,0 @@ --- Runs on the NEW binary: old interval-flow metadata must recover with an --- implicit zero offset (no `EVAL OFFSET` line) and no internal transport keys. -SELECT flow_name, source_table_names -FROM information_schema.flows -WHERE flow_name = 'compat_eval_offset_old_flow'; - -+-----------------------------+---------------------------------------------------------------------+ -| flow_name | source_table_names | -+-----------------------------+---------------------------------------------------------------------+ -| compat_eval_offset_old_flow | greptime.flow_eval_offset_old_metadata.compat_eval_offset_old_input | -+-----------------------------+---------------------------------------------------------------------+ - -SHOW CREATE FLOW compat_eval_offset_old_flow; - -+-----------------------------+---------------------------------------------------------------------------------------------------------------------------------------------+ -| Flow | Create Flow | -+-----------------------------+---------------------------------------------------------------------------------------------------------------------------------------------+ -| compat_eval_offset_old_flow | CREATE FLOW IF NOT EXISTS compat_eval_offset_old_flow | -| | SINK TO flow_eval_offset_old_metadata.compat_eval_offset_old_sink | -| | EVAL INTERVAL '1 s' | -| | AS SELECT date_trunc('second', now()) AS ts, count(v) AS value_count FROM compat_eval_offset_old_input GROUP BY date_trunc('second', now()) | -+-----------------------------+---------------------------------------------------------------------------------------------------------------------------------------------+ - -SELECT flow_definition -FROM information_schema.flows -WHERE flow_name = 'compat_eval_offset_old_flow'; - -+---------------------------------------------------------------------------------------------------------------------------------------------+ -| flow_definition | -+---------------------------------------------------------------------------------------------------------------------------------------------+ -| CREATE FLOW IF NOT EXISTS compat_eval_offset_old_flow | -| SINK TO flow_eval_offset_old_metadata.compat_eval_offset_old_sink | -| EVAL INTERVAL '1 s' | -| AS SELECT date_trunc('second', now()) AS ts, count(v) AS value_count FROM compat_eval_offset_old_input GROUP BY date_trunc('second', now()) | -+---------------------------------------------------------------------------------------------------------------------------------------------+ - --- The recovered flow_definition must not surface internal transport keys. -SELECT count(*) AS leaked_internal_keys -FROM information_schema.flows -WHERE flow_name = 'compat_eval_offset_old_flow' - AND (flow_definition LIKE '%__greptime_internal%' - OR flow_definition LIKE '%eval_schedule%'); - -+----------------------+ -| leaked_internal_keys | -+----------------------+ -| 0 | -+----------------------+ - --- The old flow still executes on the new binary. -INSERT INTO compat_eval_offset_old_input VALUES - ('2026-06-25 00:00:00', 'a', 1.0), - ('2026-06-25 00:00:01', 'b', 2.0); - -Affected Rows: 2 - --- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | -ADMIN FLUSH_FLOW('compat_eval_offset_old_flow'); - -+-------------------------------------------------+ -| ADMIN FLUSH_FLOW('compat_eval_offset_old_flow') | -+-------------------------------------------------+ -| FLOW_FLUSHED | -+-------------------------------------------------+ - -SELECT count(*) > 0 AS old_flow_executed -FROM compat_eval_offset_old_sink; - -+-------------------+ -| old_flow_executed | -+-------------------+ -| true | -+-------------------+ diff --git a/tests/compatibility/cases/flow_eval_offset_old_metadata/verify.sql b/tests/compatibility/cases/flow_eval_offset_old_metadata/verify.sql deleted file mode 100644 index 5132607aae7..00000000000 --- a/tests/compatibility/cases/flow_eval_offset_old_metadata/verify.sql +++ /dev/null @@ -1,29 +0,0 @@ --- Runs on the NEW binary: old interval-flow metadata must recover with an --- implicit zero offset (no `EVAL OFFSET` line) and no internal transport keys. -SELECT flow_name, source_table_names -FROM information_schema.flows -WHERE flow_name = 'compat_eval_offset_old_flow'; - -SHOW CREATE FLOW compat_eval_offset_old_flow; - -SELECT flow_definition -FROM information_schema.flows -WHERE flow_name = 'compat_eval_offset_old_flow'; - --- The recovered flow_definition must not surface internal transport keys. -SELECT count(*) AS leaked_internal_keys -FROM information_schema.flows -WHERE flow_name = 'compat_eval_offset_old_flow' - AND (flow_definition LIKE '%__greptime_internal%' - OR flow_definition LIKE '%eval_schedule%'); - --- The old flow still executes on the new binary. -INSERT INTO compat_eval_offset_old_input VALUES - ('2026-06-25 00:00:00', 'a', 1.0), - ('2026-06-25 00:00:01', 'b', 2.0); - --- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | -ADMIN FLUSH_FLOW('compat_eval_offset_old_flow'); - -SELECT count(*) > 0 AS old_flow_executed -FROM compat_eval_offset_old_sink;