feat: don't allow creating flow with the same sink and source table (#6435)

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

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?;
         }

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"));
+    }
+}

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

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;