From 4e0df3a2b8d015cb231c4a37c1b1aa503920731b Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Tue, 13 Oct 2026 17:05:03 +0800 Subject: [PATCH] fix: address review on query options consistency - federated: absorb SHOW VARIABLES with a WHERE/LIKE filter or a non-identifier remainder as an empty table; only a bare or dotted-identifier remainder falls back to the parser so named variables like query.parallelism reach the real handler (restores Connector/J 5.1 loadServerVariables behavior). - datanode: report engine_context query-option failures as InvalidQueryContext instead of NewPlanDecoder. - http: drop the redundant Option around the always-present timeout layer. - session: split the query.parallelism 1024 cap to public entry points (SQL SET, HTTP hints) and keep wire decoding at u16::MAX so a legacy frontend's nonzero value such as 2048 still restores. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/datanode/src/region_server.rs | 7 +++- src/query/src/sql.rs | 7 +++- src/servers/src/hint_headers.rs | 6 +++- src/servers/src/http.rs | 10 +++--- src/servers/src/mysql/federated.rs | 43 ++++++++++++++++++++++-- src/session/src/context.rs | 52 +++++++++++++++++++++++++----- src/session/src/query_options.rs | 11 +++++-- 7 files changed, 115 insertions(+), 21 deletions(-) diff --git a/src/datanode/src/region_server.rs b/src/datanode/src/region_server.rs index f8261f03e11..fdc12d08e7b 100644 --- a/src/datanode/src/region_server.rs +++ b/src/datanode/src/region_server.rs @@ -301,7 +301,12 @@ impl RegionServer { .inner .query_engine .engine_context(query_ctx) - .context(NewPlanDecoderSnafu)?; + .map_err(|error| { + crate::error::InvalidQueryContextSnafu { + reason: error.to_string(), + } + .build() + })?; let query_ctx = engine_ctx.query_ctx(); let catalog_list = Arc::new(NameAwareCatalogList::new( self.clone(), diff --git a/src/query/src/sql.rs b/src/query/src/sql.rs index a8a45b49426..8d4a4bc99d4 100644 --- a/src/query/src/sql.rs +++ b/src/query/src/sql.rs @@ -855,7 +855,12 @@ pub fn show_query_option( if name.to_ascii_lowercase().starts_with("query.") || name.to_ascii_lowercase().starts_with("datafusion.") { - session::query_options::parse_query_option(&name, "").map_err(|error| { + session::query_options::parse_query_option( + &name, + "", + session::query_options::MAX_PUBLIC_PARALLELISM, + ) + .map_err(|error| { error::InvalidQueryContextExtensionSnafu { reason: error.to_string(), } diff --git a/src/servers/src/hint_headers.rs b/src/servers/src/hint_headers.rs index 589bcd4421b..6c14aecdc87 100644 --- a/src/servers/src/hint_headers.rs +++ b/src/servers/src/hint_headers.rs @@ -64,7 +64,11 @@ pub(crate) fn validate_public_hints(hints: Vec<(String, String)>) -> Result { if let Some(existing) = seen.insert(canonical.clone(), normalized_value.clone()) && existing != normalized_value diff --git a/src/servers/src/http.rs b/src/servers/src/http.rs index 30f2ba6ec2e..56ff2a1af1e 100644 --- a/src/servers/src/http.rs +++ b/src/servers/src/http.rs @@ -1037,11 +1037,9 @@ impl HttpServer { /// Attaches middlewares and debug routes to the router. /// Callers should call this method after [HttpServer::make_app()]. pub fn build(&self, router: Router) -> Result { - let timeout_layer = Some( - ServiceBuilder::new().layer( - DynamicTimeoutLayer::new(self.options.timeout) - .with_status_code_fn(Self::request_timeout_status_code), - ), + let timeout_layer = ServiceBuilder::new().layer( + DynamicTimeoutLayer::new(self.options.timeout) + .with_status_code_fn(Self::request_timeout_status_code), ); let body_limit_layer = if self.options.body_limit != ReadableSize(0) { Some( @@ -1078,7 +1076,7 @@ impl HttpServer { // and we have impl IntoResponse for Error. It will print out more detailed error messages .layer(TraceLayer::new_for_http().on_failure(())) .option_layer(cors_layer) - .option_layer(timeout_layer) + .layer(timeout_layer) .option_layer(body_limit_layer) // memory limit layer - must be before body is consumed .layer(middleware::from_fn_with_state( diff --git a/src/servers/src/mysql/federated.rs b/src/servers/src/mysql/federated.rs index b62174679d9..2402be334e3 100644 --- a/src/servers/src/mysql/federated.rs +++ b/src/servers/src/mysql/federated.rs @@ -288,7 +288,7 @@ fn check_show_variables(query: &str) -> Option { if remainder.is_empty() { Some(show_variables("", "")) } else { - let remainder = remainder.trim_start(); + let remainder = remainder.trim(); // Filter keywords need an identifier boundary; named variables do not. let looks_like_filter = ["WHERE", "LIKE"].iter().any(|keyword| { remainder @@ -299,7 +299,15 @@ fn check_show_variables(query: &str) -> Option { .next() .is_none_or(|c| !(c == '_' || c == '$' || c.is_alphanumeric())) }); - if looks_like_filter { + let is_dotted_identifier = remainder.split('.').all(|part| { + let bytes = part.as_bytes(); + !bytes.is_empty() + && (bytes[0].is_ascii_alphabetic() || bytes[0] == b'_' || bytes[0] == b'$') + && bytes[1..] + .iter() + .all(|byte| byte.is_ascii_alphanumeric() || *byte == b'_' || *byte == b'$') + }); + if looks_like_filter || !is_dotted_identifier { Some(show_variables("", "")) } else { None @@ -577,6 +585,37 @@ mod test { use super::*; + #[test] + fn test_show_variables_where_filter_is_absorbed() { + let session = Arc::new(Session::new(None, Channel::Mysql, Default::default(), 0)); + let output = check( + "show variables where Variable_name = 'sql_mode'", + QueryContext::arc(), + session, + ) + .expect("WHERE-filtered SHOW VARIABLES should be absorbed"); + assert!(matches!(output.data, OutputData::RecordBatches(_))); + } + + #[test] + fn test_show_variables_named_identifier_falls_back() { + let session = Arc::new(Session::new(None, Channel::Mysql, Default::default(), 0)); + assert!(check("show variables some_name", QueryContext::arc(), session).is_none()); + } + + #[test] + fn test_show_variables_dotted_identifier_falls_back() { + let session = Arc::new(Session::new(None, Channel::Mysql, Default::default(), 0)); + assert!( + check( + "show variables query.parallelism", + QueryContext::arc(), + session + ) + .is_none() + ); + } + #[test] fn test_check_abnormal() { let session = Arc::new(Session::new(None, Channel::Mysql, Default::default(), 0)); diff --git a/src/session/src/context.rs b/src/session/src/context.rs index 908be103276..1792f592a7d 100644 --- a/src/session/src/context.rs +++ b/src/session/src/context.rs @@ -495,8 +495,9 @@ impl QueryContext { if key.eq_ignore_ascii_case("query_parallelism") && value == "0" { continue; } + // Wire decoding preserves legacy frontend nonzero values; public entry points enforce the limit. if let Some((canonical, normalized)) = - crate::query_options::parse_query_option(key, value)? + crate::query_options::parse_query_option(key, value, u16::MAX)? { if request_options .get(&canonical) @@ -862,13 +863,17 @@ impl ConfigurationVariables { key: &str, value: &str, ) -> Result<(), crate::session_config::Error> { - let (canonical, normalized) = crate::query_options::parse_query_option(key, value)? - .ok_or_else(|| crate::session_config::Error::InvalidConfigValue { - name: key.to_string(), - value: value.to_string(), - hint: "Not a supported query option".to_string(), - location: snafu::Location::new(file!(), line!(), column!()), - })?; + let (canonical, normalized) = crate::query_options::parse_query_option( + key, + value, + crate::query_options::MAX_PUBLIC_PARALLELISM, + )? + .ok_or_else(|| crate::session_config::Error::InvalidConfigValue { + name: key.to_string(), + value: value.to_string(), + hint: "Not a supported query option".to_string(), + location: snafu::Location::new(file!(), line!(), column!()), + })?; self.query_options .write() .unwrap() @@ -891,6 +896,37 @@ mod test { use crate::Session; use crate::context::{Channel, *}; + #[test] + fn test_query_parallelism_wire_accepts_legacy_value() { + let mut context = QueryContext::with(DEFAULT_CATALOG_NAME, "public"); + context.set_extension("query_parallelism", "2048"); + + assert_eq!( + context + .effective_query_options() + .unwrap() + .get("query.parallelism"), + Some(&"2048".to_string()) + ); + } + + #[test] + fn test_set_query_parallelism_public_limit() { + let context = QueryContext::with(DEFAULT_CATALOG_NAME, "public"); + assert!( + context + .configuration_parameter() + .set_query_option("query.parallelism", "1025") + .is_err() + ); + assert!( + context + .configuration_parameter() + .set_query_option("query.parallelism", "1024") + .is_ok() + ); + } + #[test] fn test_session() { let session = Session::new( diff --git a/src/session/src/query_options.rs b/src/session/src/query_options.rs index e632a64aa99..521123718c2 100644 --- a/src/session/src/query_options.rs +++ b/src/session/src/query_options.rs @@ -30,8 +30,15 @@ pub fn canonical_query_option_name(key: &str) -> Option { } } +/// Maximum query parallelism accepted through public query-option inputs. +pub const MAX_PUBLIC_PARALLELISM: u16 = 1024; + /// Validate a query option, returning its canonical name and normalized value. -pub fn parse_query_option(key: &str, value: &str) -> Result, Error> { +pub fn parse_query_option( + key: &str, + value: &str, + max_parallelism: u16, +) -> Result, Error> { let key_lower = key.to_ascii_lowercase(); let canonical = match canonical_query_option_name(key) { Some(name) => name, @@ -48,7 +55,7 @@ pub fn parse_query_option(key: &str, value: &str) -> Result() { - Ok(n @ 1..=1024) => n.to_string(), + Ok(n) if (1..=max_parallelism).contains(&n) => n.to_string(), _ => { return InvalidConfigValueSnafu { name: key,