fix: unify unknown flow option rejection across OSS and enterprise builds (#9319)

* fix(operator): unify unknown flow option rejection across OSS and enterprise builds Remove the enterprise-gated passthrough that let unknown flow options through SQL-layer validation. Both OSS and enterprise binaries now reject unknown flow options with a single 1004(InvalidArguments) error at the SQL layer, producing identical sqlness output. The metasrv-side validate_flow_options remains as a defensive check. Signed-off-by: discord9 <discord9@163.com> * fix(flow): use DISTINCT output columns as sink key for streaming flows When a streaming flow's source table has no primary key, sink key inference produced an empty key. For SELECT DISTINCT v, the DISTINCT establishes v as the output key, but the inference only looked at source primary keys. Combined with the zero-valued placeholder timestamp, all output rows shared the same storage identity and were deduplicated into a single row. Use all non-time-index output columns as the sink key when the plan is a DISTINCT and no source primary key was inherited. Signed-off-by: discord9 <discord9@163.com> * fix(flow): preserve extension options and complete DISTINCT sink identity Keep enterprise CREATE FLOW options available to the existing handler and report unknown ordinary options as InvalidArguments with the OSS SQL error text. Infer DISTINCT sink keys from every selected non-time-index output column, including when source key columns survive. Wait for asynchronous mirrors in show_create_flow and exercise 1000 distinct sink rows in the SQL regression. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(flow): align meta unknown option status assertions Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(flow): use valid aliases in DISTINCT sink regression SQL Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(flow): record complete DISTINCT sink SQL results Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <discord9@163.com> Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
2026-10-12 06:52:34 +00:00 · 2026-10-10 06:45:11 +00:00
parent a8e293fd79
commit 178a0021b8
7 changed files with 341 additions and 12 deletions
@@ -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();
@@ -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'")
    );
 }

@@ -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<Option<api::v1::flow::flow_request::Body>>,
@@ -1247,6 +1254,7 @@ impl ErrorExt for Error {

            ProcedureNotFound { .. }
            | InvalidViewInfo { .. }
+            | InvalidFlowOption { .. }
            | PrimaryKeyNotFound { .. }
            | EmptyKey { .. }
            | AlterLogicalTablesInvalidArguments { .. }
@@ -141,8 +141,9 @@ fn relation_desc_from_output(
    columns: &[ColumnSchema],
    lineage: &[Option<usize>],
    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)?;
@@ -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<datafusion_expr::Expr>, 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::<Vec<_>>(),
+            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]);
 }

@@ -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
+
@@ -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;