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

@@ -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(),

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(),
                 }

src/servers/src/hint_headers.rs

@@ -64,7 +64,11 @@ pub(crate) fn validate_public_hints(hints: Vec<(String, String)>) -> Result<Vec<
         if key.eq_ignore_ascii_case("query_parallelism") && value == "0" {
             continue;
         }
-        match session::query_options::parse_query_option(&key, &value) {
+        match session::query_options::parse_query_option(
+            &key,
+            &value,
+            session::query_options::MAX_PUBLIC_PARALLELISM,
+        ) {
             Ok(Some((canonical, normalized_value))) => {
                 if let Some(existing) = seen.insert(canonical.clone(), normalized_value.clone())
                     && existing != normalized_value

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<Router> {
-        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(

src/servers/src/mysql/federated.rs

@@ -288,7 +288,7 @@ fn check_show_variables(query: &str) -> Option<Output> {
         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<Output> {
                         .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));

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(

src/session/src/query_options.rs

@@ -30,8 +30,15 @@ pub fn canonical_query_option_name(key: &str) -> Option<String> {
     }
 }
 
+/// 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<Option<(String, String)>, Error> {
+pub fn parse_query_option(
+    key: &str,
+    value: &str,
+    max_parallelism: u16,
+) -> Result<Option<(String, String)>, 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<Option<(String, Stri
 
     let normalized = if canonical == "query.parallelism" {
         match value.parse::<u16>() {
-            Ok(n @ 1..=1024) => n.to_string(),
+            Ok(n) if (1..=max_parallelism).contains(&n) => n.to_string(),
             _ => {
                 return InvalidConfigValueSnafu {
                     name: key,