rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-17 17:32:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(flow): preserve extension options and complete DISTINCT sink identity

Browse Source 
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>
This commit is contained in:
discord9
2026-10-08 18:04:46 +08:00
parent cc5b894651
commit c49219b799
8 changed files with 265 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/meta/src/ddl/create_flow.rs
+3 -3
View File
@@ -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();
src/common/meta/src/ddl/tests/create_flow.rs
+22
View File
@@ -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 [
src/common/meta/src/error.rs
+8
View File
@@ -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<Option<api::v1::flow::flow_request::Body>>,
@@ -1255,6 +1262,7 @@ impl ErrorExt for Error {
ProcedureNotFound { .. }
| InvalidViewInfo { .. }
| InvalidFlowOption { .. }
| PrimaryKeyNotFound { .. }
| EmptyKey { .. }
| AlterLogicalTablesInvalidArguments { .. }
src/flow/src/adapter.rs
+3 -4
View File
@@ -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())
src/flow/src/adapter/tests.rs
+118 -40
View File
@@ -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<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()
}
_ => 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::<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]
src/operator/src/statement/ddl.rs
+44 -4
View File
@@ -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<DdlSubmitOptions> {
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 [
tests/cases/standalone/common/flow/flow_advance_ttl.sql
+55
View File
@@ -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;
tests/cases/standalone/common/flow/show_create_flow.sql
+12
View File
@@ -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');