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..ad334e29863 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 [ @@ -346,9 +368,10 @@ fn test_validate_flow_options_rejects_schedule_and_internal_keys_as_unknown() { .insert(key.to_string(), "value".to_string()); let err = validate_flow_options(&task).unwrap_err(); + assert_eq!(StatusCode::InvalidArguments, err.status_code()); assert!( err.to_string() - .contains(&format!("Unknown flow option '{key}'")), + .contains(&format!("unknown flow option '{key}'")), "unexpected error for {key}: {err}" ); } @@ -884,10 +907,10 @@ async fn test_create_flow_rejects_unknown_option_in_meta_task() { let mut procedure = CreateFlowProcedure::new(task, query_ctx, ddl_context); let err = procedure.on_prepare().await.unwrap_err(); - assert_matches!(err, error::Error::Unexpected { .. }); + assert_eq!(StatusCode::InvalidArguments, err.status_code()); assert!( err.to_string() - .contains("Unknown flow option 'unknown_option'") + .contains("unknown flow option 'unknown_option'") ); } diff --git a/src/common/meta/src/error.rs b/src/common/meta/src/error.rs index 62559fd6f1b..5c1febb9b4d 100644 --- a/src/common/meta/src/error.rs +++ b/src/common/meta/src/error.rs @@ -515,6 +515,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>, @@ -1247,6 +1254,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 c98510bd7d5..5b15ad0a93c 100644 --- a/src/flow/src/adapter.rs +++ b/src/flow/src/adapter.rs @@ -141,8 +141,9 @@ fn relation_desc_from_output( columns: &[ColumnSchema], lineage: &[Option], source_primary_key_indices: &[usize], + plan: &LogicalPlan, ) -> RelationDesc { - let keys = source_primary_key_indices + let mut keys = source_primary_key_indices .iter() .filter_map(|source_index| { lineage @@ -150,6 +151,14 @@ fn relation_desc_from_output( .position(|index| index == &Some(*source_index)) }) .collect_vec(); + // 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()) + .collect(); + } let time_index = columns.iter().position(ColumnSchema::is_time_index); RelationDesc { typ: crate::repr::RelationType { @@ -849,8 +858,12 @@ impl StreamingEngine { let source_primary_key_indices = source_meta.primary_key_indices; stateless::validate_source_scan(&flow_plan, source_table_id, &source_schema)?; let (inferred_schema, lineage) = output_column_schemas(&flow_plan, &source_schema)?; - let inferred_relation = - relation_desc_from_output(&inferred_schema, &lineage, &source_primary_key_indices); + let inferred_relation = relation_desc_from_output( + &inferred_schema, + &lineage, + &source_primary_key_indices, + &flow_plan, + ); let sink_exists = self.fetch_table_pk_schema(sink_table_name).await?.is_some(); if !sink_exists { validate_auto_column_names(&inferred_schema)?; diff --git a/src/flow/src/adapter/tests.rs b/src/flow/src/adapter/tests.rs index 6bc02f60786..6a2bf5ed816 100644 --- a/src/flow/src/adapter/tests.rs +++ b/src/flow/src/adapter/tests.rs @@ -20,7 +20,7 @@ use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, TimeUnit as use datafusion::catalog::MemTable; use datafusion::datasource::provider_as_source; use datafusion_common::TableReference; -use datafusion_expr::LogicalPlanBuilder; +use datafusion_expr::{LogicalPlan, LogicalPlanBuilder}; use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema, Schema, SchemaBuilder}; use store_api::storage::{ConcreteDataType, TableId}; use table::metadata::{TableInfo, TableInfoBuilder, TableMetaBuilder}; @@ -194,10 +194,133 @@ fn stateless_distinct_preserves_direct_column_lineage() { let (output, lineage) = super::output_column_schemas(&plan, &source).unwrap(); assert_eq!(output[0].name, "dis"); assert_eq!(lineage, vec![Some(0)]); - let relation = super::relation_desc_from_output(&output, &lineage, &[0]); + let relation = super::relation_desc_from_output(&output, &lineage, &[0], &plan); assert_eq!(relation.typ.keys[0].column_indices, vec![0]); } +#[test] +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() + } + }; + 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] fn stateless_normalizes_dictionary_output_type() { let field = Field::new_dictionary("host", ArrowDataType::UInt32, ArrowDataType::Utf8, true); @@ -220,7 +343,7 @@ fn stateless_normalizes_dictionary_output_type() { assert_eq!(output[0].data_type, ConcreteDataType::string_datatype()); assert_eq!(lineage, vec![Some(0)]); - let relation = super::relation_desc_from_output(&output, &lineage, &[0]); + let relation = super::relation_desc_from_output(&output, &lineage, &[0], &plan); assert_eq!(relation.typ.keys[0].column_indices, vec![0]); } diff --git a/tests/cases/standalone/common/flow/flow_advance_ttl.result b/tests/cases/standalone/common/flow/flow_advance_ttl.result index eae65328a6b..27a28babebf 100644 --- a/tests/cases/standalone/common/flow/flow_advance_ttl.result +++ b/tests/cases/standalone/common/flow/flow_advance_ttl.result @@ -414,3 +414,110 @@ DROP TABLE out_distinct_basic; Affected Rows: 0 +-- 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'); + +Affected Rows: 0 + +CREATE FLOW test_distinct_auto_sink SINK TO out_distinct_auto_sink AS +SELECT DISTINCT v FROM distinct_auto_sink; + +Affected Rows: 0 + +INSERT INTO distinct_auto_sink (v) +SELECT number FROM numbers LIMIT 1000; + +Affected Rows: 0 + +-- 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'); + ++---------------------------------------------+ +| ADMIN FLUSH_FLOW('test_distinct_auto_sink') | ++---------------------------------------------+ +| FLOW_FLUSHED | ++---------------------------------------------+ + +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; + ++------+-----------------+-------+-------+ +| rows | distinct_values | min_v | max_v | ++------+-----------------+-------+-------+ +| 1000 | 1000 | 0 | 999 | ++------+-----------------+-------+-------+ + +DROP FLOW test_distinct_auto_sink; + +Affected Rows: 0 + +DROP TABLE distinct_auto_sink; + +Affected Rows: 0 + +DROP TABLE out_distinct_auto_sink; + +Affected Rows: 0 + +CREATE TABLE distinct_auto_sink_pk ( + v INT, + k INT, + extra_value INT, + ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP TIME INDEX, + PRIMARY KEY (v, k) +) WITH ('ttl' = 'instant'); + +Affected Rows: 0 + +CREATE FLOW test_distinct_auto_sink_pk SINK TO out_distinct_auto_sink_pk AS +SELECT DISTINCT v, extra_value FROM distinct_auto_sink_pk; + +Affected Rows: 0 + +INSERT INTO distinct_auto_sink_pk (v, k, extra_value) +SELECT number % 10 AS v, number AS k, number AS extra_value FROM numbers LIMIT 1000; + +Affected Rows: 0 + +-- 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'); + ++------------------------------------------------+ +| ADMIN FLUSH_FLOW('test_distinct_auto_sink_pk') | ++------------------------------------------------+ +| FLOW_FLUSHED | ++------------------------------------------------+ + +SELECT + count(*) AS rows, + count(DISTINCT extra_value) AS distinct_values, + min(extra_value) AS min_value, + max(extra_value) AS max_value, + sum(CASE WHEN v = extra_value % 10 THEN 0 ELSE 1 END) AS mismatched_keys +FROM out_distinct_auto_sink_pk; + ++------+-----------------+-----------+-----------+-----------------+ +| rows | distinct_values | min_value | max_value | mismatched_keys | ++------+-----------------+-----------+-----------+-----------------+ +| 1000 | 1000 | 0 | 999 | 0 | ++------+-----------------+-----------+-----------+-----------------+ + +DROP FLOW test_distinct_auto_sink_pk; + +Affected Rows: 0 + +DROP TABLE distinct_auto_sink_pk; + +Affected Rows: 0 + +DROP TABLE out_distinct_auto_sink_pk; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/flow/flow_advance_ttl.sql b/tests/cases/standalone/common/flow/flow_advance_ttl.sql index 2aacc0f9458..72356d96372 100644 --- a/tests/cases/standalone/common/flow/flow_advance_ttl.sql +++ b/tests/cases/standalone/common/flow/flow_advance_ttl.sql @@ -179,3 +179,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, + extra_value 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, extra_value FROM distinct_auto_sink_pk; + +INSERT INTO distinct_auto_sink_pk (v, k, extra_value) +SELECT number % 10 AS v, number AS k, number AS extra_value 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 extra_value) AS distinct_values, + min(extra_value) AS min_value, + max(extra_value) AS max_value, + sum(CASE WHEN v = extra_value % 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;