diff --git a/src/common/meta/src/ddl/create_flow.rs b/src/common/meta/src/ddl/create_flow.rs index e36c89fbb34..89ade1b5cbe 100644 --- a/src/common/meta/src/ddl/create_flow.rs +++ b/src/common/meta/src/ddl/create_flow.rs @@ -42,7 +42,7 @@ use crate::cache_invalidator::Context; use crate::ddl::DdlContext; use crate::ddl::event::flow::{CREATE_FLOW_EVENT_TYPE, CreateFlowEventIntent, FlowDdlEvent}; use crate::ddl::utils::{add_peer_context_if_needed, map_to_procedure_error}; -use crate::error::{self, Result, UnexpectedSnafu}; +use crate::error::{self, InvalidFlowOptionSnafu, Result, UnexpectedSnafu}; use crate::instruction::{CacheIdent, CreateFlow, DropFlow}; use crate::key::flow::flow_info::{FlowInfoValue, FlowScheduleConfig, FlowStatus}; use crate::key::flow::flow_route::FlowRouteValue; @@ -558,9 +558,9 @@ pub fn validate_flow_options(flow_task: &CreateFlowTask) -> Result<()> { | FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY | FlowType::FLOW_TYPE_KEY => {} unknown => { - return UnexpectedSnafu { + return InvalidFlowOptionSnafu { err_msg: format!( - "Unknown flow option '{unknown}', supported user options: {DEFER_ON_MISSING_SOURCE_KEY}, {FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY}" + "unknown flow option '{unknown}', supported options: {DEFER_ON_MISSING_SOURCE_KEY}, {FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY}" ), } .fail(); diff --git a/src/common/meta/src/ddl/tests/create_flow.rs b/src/common/meta/src/ddl/tests/create_flow.rs index e7004656fe9..e2b313c9066 100644 --- a/src/common/meta/src/ddl/tests/create_flow.rs +++ b/src/common/meta/src/ddl/tests/create_flow.rs @@ -18,6 +18,8 @@ use std::sync::Arc; use api::v1::flow::CreateRequest; use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME}; +use common_error::ext::ErrorExt; +use common_error::status_code::StatusCode; use common_procedure::{Procedure, Status}; use common_procedure_test::execute_procedure_until_done; use table::table_name::TableName; @@ -326,6 +328,26 @@ fn test_validate_flow_options_allows_incremental_read_option() { validate_flow_options(&task).unwrap(); } +#[test] +fn test_validate_flow_options_unknown_option_matches_invalid_sql_error() { + let mut task = test_create_flow_task( + "my_flow", + vec![], + TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_sink_table"), + false, + ); + task.flow_options + .insert("foo".to_string(), "sensitive-value".to_string()); + + let err = validate_flow_options(&task).unwrap_err(); + assert_eq!(StatusCode::InvalidArguments, err.status_code()); + assert_eq!( + "Invalid SQL, error: unknown flow option 'foo', supported options: defer_on_missing_source, experimental_enable_incremental_read", + err.output_msg() + ); + assert!(!err.output_msg().contains("sensitive-value")); +} + #[test] fn test_validate_flow_options_rejects_schedule_and_internal_keys_as_unknown() { for key in [ diff --git a/src/common/meta/src/error.rs b/src/common/meta/src/error.rs index 3f4868ffba0..c281ac9e83e 100644 --- a/src/common/meta/src/error.rs +++ b/src/common/meta/src/error.rs @@ -522,6 +522,13 @@ pub enum Error { location: Location, }, + #[snafu(display("Invalid SQL, error: {err_msg}"))] + InvalidFlowOption { + err_msg: String, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Invalid flow request body: {:?}", body))] InvalidFlowRequestBody { body: Box>, @@ -1255,6 +1262,7 @@ impl ErrorExt for Error { ProcedureNotFound { .. } | InvalidViewInfo { .. } + | InvalidFlowOption { .. } | PrimaryKeyNotFound { .. } | EmptyKey { .. } | AlterLogicalTablesInvalidArguments { .. } diff --git a/src/flow/src/adapter.rs b/src/flow/src/adapter.rs index 7e17032e2e0..5b15ad0a93c 100644 --- a/src/flow/src/adapter.rs +++ b/src/flow/src/adapter.rs @@ -151,10 +151,9 @@ fn relation_desc_from_output( .position(|index| index == &Some(*source_index)) }) .collect_vec(); - // A `DISTINCT` establishes the output key by itself: every non-time-index output - // column is part of the distinct key. When no source primary key is inherited, - // fall back to the DISTINCT output key so the sink relation keeps row identity. - if keys.is_empty() && matches!(plan, LogicalPlan::Distinct(Distinct::All(_))) { + // DISTINCT defines identity from every output tuple; the source timestamp participates + // separately as the relation time index, so it is excluded from the tag key. + if matches!(plan, LogicalPlan::Distinct(Distinct::All(_))) { keys = columns .iter() .positions(|column| !column.is_time_index()) diff --git a/src/flow/src/adapter/tests.rs b/src/flow/src/adapter/tests.rs index dba7a950bb6..6a2bf5ed816 100644 --- a/src/flow/src/adapter/tests.rs +++ b/src/flow/src/adapter/tests.rs @@ -199,48 +199,126 @@ fn stateless_distinct_preserves_direct_column_lineage() { } #[test] -fn stateless_distinct_without_source_pk_uses_distinct_output_as_key() { - let source = Arc::new(Schema::new(vec![ColumnSchema::new( - "v", - ConcreteDataType::int32_datatype(), - true, - )])); - let provider = MemTable::try_new( - Arc::new(datafusion::arrow::datatypes::Schema::new(vec![Field::new( - "v", - ArrowDataType::Int32, - true, - )])), - vec![vec![]], - ) - .unwrap(); - let plan = LogicalPlanBuilder::scan( - TableReference::bare("source"), - provider_as_source(Arc::new(provider)), - None, - ) - .unwrap() - .project(vec![datafusion_expr::col("v")]) - .unwrap() - .distinct() - .unwrap() - .build() - .unwrap(); - - let (output, lineage) = super::output_column_schemas(&plan, &source).unwrap(); - let relation = super::relation_desc_from_output(&output, &lineage, &[], &plan); - assert_eq!(relation.typ.keys.len(), 1); - assert_eq!(relation.typ.keys[0].column_indices, vec![0]); - - // Non-DISTINCT plans keep the empty-key behavior. - let project_plan = match &plan { - LogicalPlan::Distinct(datafusion_expr::logical_plan::Distinct::All(input)) => { - input.as_ref().clone() +fn stateless_distinct_output_is_always_the_sink_key() { + let source = Arc::new(Schema::new(vec![ + ColumnSchema::new("v", ConcreteDataType::int32_datatype(), true), + ColumnSchema::new("k", ConcreteDataType::int32_datatype(), true), + ColumnSchema::new("other", ConcreteDataType::int32_datatype(), true), + ColumnSchema::new( + "ts", + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ) + .with_time_index(true), + ])); + // All cases share one provider so the key assertions exercise only plan/output metadata. + let provider = Arc::new( + MemTable::try_new( + Arc::new(datafusion::arrow::datatypes::Schema::new(vec![ + Field::new("v", ArrowDataType::Int32, true), + Field::new("k", ArrowDataType::Int32, true), + Field::new("other", ArrowDataType::Int32, true), + Field::new( + "ts", + ArrowDataType::Timestamp(ArrowTimeUnit::Millisecond, None), + false, + ), + ])), + vec![vec![]], + ) + .unwrap(), + ); + let make_plan = |expressions: Vec, distinct: bool| { + let builder = LogicalPlanBuilder::scan( + TableReference::bare("source"), + provider_as_source(provider.clone()), + None, + ) + .unwrap() + .project(expressions) + .unwrap(); + if distinct { + builder.distinct().unwrap().build().unwrap() + } else { + builder.build().unwrap() } - _ => panic!("expected distinct plan"), }; - let relation = super::relation_desc_from_output(&output, &lineage, &[], &project_plan); - assert!(relation.typ.keys.is_empty()); + let assert_key = |plan: &LogicalPlan, source_keys: &[usize], expected: &[usize]| { + let (output, lineage) = super::output_column_schemas(plan, &source).unwrap(); + let relation = super::relation_desc_from_output(&output, &lineage, source_keys, plan); + assert_eq!( + relation + .typ + .keys + .iter() + .map(|key| key.column_indices.clone()) + .collect::>(), + if expected.is_empty() { + vec![] + } else { + vec![expected.to_vec()] + } + ); + relation + }; + + // No source PK: all distinct output fields form the key. + assert_key(&make_plan(vec![datafusion_expr::col("v")], true), &[], &[0]); + // A surviving subset of a composite source PK cannot identify projected DISTINCT tuples. + assert_key( + &make_plan( + vec![datafusion_expr::col("v"), datafusion_expr::col("other")], + true, + ), + &[0, 1], + &[0, 1], + ); + // Even a retained source PK needs the additional selected fields in the DISTINCT key. + assert_key( + &make_plan( + vec![datafusion_expr::col("v"), datafusion_expr::col("other")], + true, + ), + &[0], + &[0, 1], + ); + // Aliases and computed output expressions are included by output position. + let alias_plan = make_plan( + vec![ + datafusion_expr::col("v").alias("renamed"), + (datafusion_expr::col("v") + datafusion_expr::col("other")).alias("computed"), + ], + true, + ); + let (output, lineage) = super::output_column_schemas(&alias_plan, &source).unwrap(); + let relation = super::relation_desc_from_output(&output, &lineage, &[0], &alias_plan); + assert_eq!( + output + .iter() + .map(|column| column.name.as_str()) + .collect_vec(), + ["renamed", "computed"] + ); + assert_eq!(relation.typ.keys[0].column_indices, vec![0, 1]); + + // A selected time index remains the relation time index, not part of the tag key. + let relation = assert_key( + &make_plan( + vec![datafusion_expr::col("v"), datafusion_expr::col("ts")], + true, + ), + &[0], + &[0], + ); + assert_eq!(relation.typ.time_index, Some(1)); + + // Non-DISTINCT plans retain only inherited source-key lineage (or no key if none survives). + let ordinary = make_plan( + vec![datafusion_expr::col("v"), datafusion_expr::col("other")], + false, + ); + assert_key(&ordinary, &[0, 1], &[0]); + assert_key(&ordinary, &[], &[]); } #[test] diff --git a/src/operator/src/statement/ddl.rs b/src/operator/src/statement/ddl.rs index 21376fb725a..772bff4aad9 100644 --- a/src/operator/src/statement/ddl.rs +++ b/src/operator/src/statement/ddl.rs @@ -128,6 +128,7 @@ struct DdlSubmitOptions { timeout: Duration, } +#[cfg(not(feature = "enterprise"))] const ALLOWED_FLOW_OPTIONS: [&str; 2] = [ DEFER_ON_MISSING_SOURCE_KEY, FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY, @@ -172,6 +173,7 @@ fn parse_ddl_options(options: &OptionMap) -> Result { Ok(DdlSubmitOptions { wait, timeout }) } +#[cfg(not(feature = "enterprise"))] fn supported_flow_options() -> String { ALLOWED_FLOW_OPTIONS.join(", ") } @@ -221,6 +223,9 @@ fn validate_and_normalize_flow_options( DEFER_ON_MISSING_SOURCE_KEY | FLOW_EXPERIMENTAL_ENABLE_INCREMENTAL_READ_KEY => { normalize_flow_bool_option(&key, &value)? } + #[cfg(feature = "enterprise")] + _ => value, + #[cfg(not(feature = "enterprise"))] _ => { return InvalidSqlSnafu { err_msg: format!( @@ -3281,6 +3286,7 @@ mod test { ); } + #[cfg(not(feature = "enterprise"))] #[test] fn test_validate_and_normalize_flow_options_unknown_option() { let err = validate_and_normalize_flow_options( @@ -3349,10 +3355,18 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;"; let expr = expr_helper::to_create_flow_task_expr(create_flow, &QueryContext::arc()).unwrap(); - let err = validate_and_normalize_flow_options(expr.flow_options, None).unwrap_err(); - assert!( - err.to_string() - .contains("unknown flow option 'access_key_id'") + #[cfg(not(feature = "enterprise"))] + { + let err = validate_and_normalize_flow_options(expr.flow_options, None).unwrap_err(); + assert!( + err.to_string() + .contains("unknown flow option 'access_key_id'") + ); + } + #[cfg(feature = "enterprise")] + assert_eq!( + validate_and_normalize_flow_options(expr.flow_options, None).unwrap(), + HashMap::from([("access_key_id".to_string(), "['true']".to_string())]) ); } @@ -3454,6 +3468,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;"; assert!(err.to_string().contains("reserved for internal use")); } + #[cfg(not(feature = "enterprise"))] #[test] fn test_schedule_keys_rejected_as_unknown_options() { for key in [ @@ -3475,6 +3490,31 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;"; } } + #[cfg(feature = "enterprise")] + #[test] + fn test_schedule_keys_preserved() { + let options = HashMap::from([ + ("eval_interval_anchor".to_string(), "anchor".to_string()), + ("eval_interval_start".to_string(), "start".to_string()), + ( + "eval_interval_missed_tick_policy".to_string(), + "missed".to_string(), + ), + ( + "eval_interval_catchup_max_runs".to_string(), + "runs".to_string(), + ), + ( + "eval_interval_catchup_max_lag".to_string(), + "lag".to_string(), + ), + ]); + assert_eq!( + validate_and_normalize_flow_options(options.clone(), Some(300)).unwrap(), + options + ); + } + #[test] fn test_internal_transport_keys_rejected_as_reserved() { for key in [ diff --git a/tests/cases/standalone/common/flow/flow_advance_ttl.sql b/tests/cases/standalone/common/flow/flow_advance_ttl.sql index 653fa16637f..1c3739a1979 100644 --- a/tests/cases/standalone/common/flow/flow_advance_ttl.sql +++ b/tests/cases/standalone/common/flow/flow_advance_ttl.sql @@ -181,3 +181,58 @@ SELECT number FROM distinct_basic; DROP FLOW test_distinct_basic; DROP TABLE distinct_basic; DROP TABLE out_distinct_basic; + +-- Streaming DISTINCT auto-sink retains all 1000 output tuples. +CREATE TABLE distinct_auto_sink ( + v INT, + ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP TIME INDEX +) WITH ('ttl' = 'instant'); + +CREATE FLOW test_distinct_auto_sink SINK TO out_distinct_auto_sink AS +SELECT DISTINCT v FROM distinct_auto_sink; + +INSERT INTO distinct_auto_sink (v) +SELECT number FROM numbers LIMIT 1000; + +-- Mirror inserts reach the flownode asynchronously; wait before flushing. +-- SQLNESS SLEEP 3s +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('test_distinct_auto_sink'); + +SELECT count(*) AS rows, count(DISTINCT v) AS distinct_values, min(v) AS min_v, max(v) AS max_v +FROM out_distinct_auto_sink; + +DROP FLOW test_distinct_auto_sink; +DROP TABLE distinct_auto_sink; +DROP TABLE out_distinct_auto_sink; + +CREATE TABLE distinct_auto_sink_pk ( + v INT, + k INT, + other INT, + ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP TIME INDEX, + PRIMARY KEY (v, k) +) WITH ('ttl' = 'instant'); + +CREATE FLOW test_distinct_auto_sink_pk SINK TO out_distinct_auto_sink_pk AS +SELECT DISTINCT v, other FROM distinct_auto_sink_pk; + +INSERT INTO distinct_auto_sink_pk (v, k, other) +SELECT number % 10, number, number FROM numbers LIMIT 1000; + +-- Mirror inserts reach the flownode asynchronously; wait before flushing. +-- SQLNESS SLEEP 3s +-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | +ADMIN FLUSH_FLOW('test_distinct_auto_sink_pk'); + +SELECT + count(*) AS rows, + count(DISTINCT other) AS distinct_values, + min(other) AS min_value, + max(other) AS max_value, + sum(CASE WHEN v = other % 10 THEN 0 ELSE 1 END) AS mismatched_keys +FROM out_distinct_auto_sink_pk; + +DROP FLOW test_distinct_auto_sink_pk; +DROP TABLE distinct_auto_sink_pk; +DROP TABLE out_distinct_auto_sink_pk; diff --git a/tests/cases/standalone/common/flow/show_create_flow.sql b/tests/cases/standalone/common/flow/show_create_flow.sql index 9856a6041f8..a2bcc4c06d9 100644 --- a/tests/cases/standalone/common/flow/show_create_flow.sql +++ b/tests/cases/standalone/common/flow/show_create_flow.sql @@ -45,6 +45,8 @@ SELECT flow_name, table_catalog, flow_definition, source_table_names FROM INFORM -- makesure it's not replaced in flownode INSERT INTO numbers_input_show VALUES (10, 0),(15, 1),(16, 2); +-- Mirror inserts reach the flownode asynchronously; wait before flushing. +-- SQLNESS SLEEP 3s -- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | ADMIN FLUSH_FLOW('filter_numbers_show'); @@ -59,6 +61,8 @@ SELECT flow_name, table_catalog, flow_definition, source_table_names FROM INFORM -- makesure it's not replaced in flownode INSERT INTO numbers_input_show VALUES (4,4),(5,4),(10, 3),(11, 4); +-- Mirror inserts reach the flownode asynchronously; wait before flushing. +-- SQLNESS SLEEP 3s -- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | ADMIN FLUSH_FLOW('filter_numbers_show'); @@ -73,6 +77,8 @@ SELECT flow_name, table_catalog, flow_definition, source_table_names FROM INFORM -- makesure it's replaced in flownode INSERT INTO numbers_input_show VALUES (3, 1),(4, 2),(10, 3),(11, 4); +-- Mirror inserts reach the flownode asynchronously; wait before flushing. +-- SQLNESS SLEEP 3s -- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | ADMIN FLUSH_FLOW('filter_numbers_show'); @@ -118,6 +124,8 @@ SELECT * FROM out_num_cnt_show; INSERT INTO numbers_input_show VALUES(-4,0), (-3,1), (-2,2), (-1,3); +-- Mirror inserts reach the flownode asynchronously; wait before flushing. +-- SQLNESS SLEEP 3s -- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | ADMIN FLUSH_FLOW('filter_numbers_show'); @@ -145,6 +153,8 @@ CREATE FLOW filter_numbers_show SINK TO out_num_cnt_show AS SELECT number as n1 INSERT INTO numbers_input_show VALUES (10, 0),(15, 1),(16, 2); +-- Mirror inserts reach the flownode asynchronously; wait before flushing. +-- SQLNESS SLEEP 3s -- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | ADMIN FLUSH_FLOW('filter_numbers_show'); @@ -160,6 +170,8 @@ SELECT flow_definition, source_table_names FROM INFORMATION_SCHEMA.FLOWS WHERE f INSERT INTO numbers_input_show VALUES (10, 6),(11, 8),(15, 7),(18, 3); +-- Mirror inserts reach the flownode asynchronously; wait before flushing. +-- SQLNESS SLEEP 3s -- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED | ADMIN FLUSH_FLOW('filter_numbers_show');