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

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

Browse Source 
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>
This commit is contained in:
discord9
2026-09-23 14:11:31 +08:00
parent 0f118bc5ba
commit 143930a361
1 changed files with 4 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/operator/src/statement/ddl.rs
+4 -44
View File
@@ -128,7 +128,6 @@ 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,
@@ -173,7 +172,6 @@ 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(", ")
}
@@ -223,9 +221,6 @@ 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!(
@@ -3286,7 +3281,6 @@ mod test {
);
}
#[cfg(not(feature = "enterprise"))]
#[test]
fn test_validate_and_normalize_flow_options_unknown_option() {
let err = validate_and_normalize_flow_options(
@@ -3355,18 +3349,10 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;";
let expr =
expr_helper::to_create_flow_task_expr(create_flow, &QueryContext::arc()).unwrap();
#[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())])
let err = validate_and_normalize_flow_options(expr.flow_options, None).unwrap_err();
assert!(
err.to_string()
.contains("unknown flow option 'access_key_id'")
);
}
@@ -3468,7 +3454,6 @@ 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 [
@@ -3490,31 +3475,6 @@ 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 [