diff --git a/src/common/meta/src/ddl/create_flow.rs b/src/common/meta/src/ddl/create_flow.rs index 647f78c3032..d103d9c862e 100644 --- a/src/common/meta/src/ddl/create_flow.rs +++ b/src/common/meta/src/ddl/create_flow.rs @@ -167,6 +167,25 @@ impl CreateFlowProcedure { } self.collect_source_tables().await?; + + // Validate that source and sink tables are not the same + let sink_table_name = &self.data.task.sink_table_name; + if self + .data + .task + .source_table_names + .iter() + .any(|source| source == sink_table_name) + { + return error::UnsupportedSnafu { + operation: format!( + "Creating flow with source and sink table being the same: {}", + sink_table_name + ), + } + .fail(); + } + if self.data.flow_id.is_none() { self.allocate_flow_id().await?; } diff --git a/src/common/meta/src/ddl/tests/create_flow.rs b/src/common/meta/src/ddl/tests/create_flow.rs index 3b24e86400b..a5a7688924a 100644 --- a/src/common/meta/src/ddl/tests/create_flow.rs +++ b/src/common/meta/src/ddl/tests/create_flow.rs @@ -141,3 +141,41 @@ async fn test_create_flow() { let err = procedure.on_prepare().await.unwrap_err(); assert_matches!(err, error::Error::FlowAlreadyExists { .. }); } + +#[tokio::test] +async fn test_create_flow_same_source_and_sink_table() { + let table_id = 1024; + let table_name = TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "same_table"); + + // Use the same table for both source and sink + let source_table_names = vec![table_name.clone()]; + let sink_table_name = table_name.clone(); + + let node_manager = Arc::new(MockFlownodeManager::new(NaiveFlownodeHandler)); + let ddl_context = new_ddl_context(node_manager); + + // Create the table first so it exists + let task = test_create_table_task("same_table", table_id); + ddl_context + .table_metadata_manager + .create_table_metadata( + task.table_info.clone(), + TableRouteValue::physical(vec![]), + HashMap::new(), + ) + .await + .unwrap(); + + // Try to create a flow with same source and sink table - should fail + let task = test_create_flow_task("my_flow", source_table_names, sink_table_name, false); + let query_ctx = QueryContext::arc().into(); + let mut procedure = CreateFlowProcedure::new(task, query_ctx, ddl_context); + let err = procedure.on_prepare().await.unwrap_err(); + assert_matches!(err, error::Error::Unsupported { .. }); + + // Verify the error message contains information about the same table + if let error::Error::Unsupported { operation, .. } = &err { + assert!(operation.contains("source and sink table being the same")); + assert!(operation.contains("same_table")); + } +} diff --git a/tests/cases/standalone/common/flow/flow_basic.result b/tests/cases/standalone/common/flow/flow_basic.result index 20ce8bafc48..e089af17812 100644 --- a/tests/cases/standalone/common/flow/flow_basic.result +++ b/tests/cases/standalone/common/flow/flow_basic.result @@ -1858,6 +1858,15 @@ SELECT device_model, | STM51 | 1 | 1 | 0.5 | 1 | 0 | 1 | 1.0 | 0.1 | 0.1 | 1.0 | 1970-01-01T00:00:00 | +--------------+------------------+--------------+-------------------------+-------------------------+---------------------------+---------------+----------------+----------------------+----------------------+-------------------+---------------------+ +-- Create a flow with same source and sink table +CREATE FLOW same_source_and_sink_table SINK TO live_connection_log AS +SELECT + * +FROM + live_connection_log; + +Error: 1001(Unsupported), Unsupported operation Creating flow with source and sink table being the same: greptime.public.live_connection_log + DROP FLOW live_connection_aggregation_detail; Affected Rows: 0 diff --git a/tests/cases/standalone/common/flow/flow_basic.sql b/tests/cases/standalone/common/flow/flow_basic.sql index 99c54db623e..44267edaae2 100644 --- a/tests/cases/standalone/common/flow/flow_basic.sql +++ b/tests/cases/standalone/common/flow/flow_basic.sql @@ -907,6 +907,13 @@ SELECT device_model, ok_conection_rate, record_time_window FROM live_connection_statistics_detail; +-- Create a flow with same source and sink table +CREATE FLOW same_source_and_sink_table SINK TO live_connection_log AS +SELECT + * +FROM + live_connection_log; + DROP FLOW live_connection_aggregation_detail; DROP TABLE live_connection_log; DROP TABLE live_connection_statistics_detail;