fix: record timed out EXPLAIN ANALYZE VERBOSE over HTTP SQL

Statement timeouts were only applied to the MySQL and Postgres channels,
so EXPLAIN ANALYZE VERBOSE executed via the HTTP `/v1/sql` endpoint never
got the `attach_timeout` wrapper and its timeout metrics were never
recorded to slow queries (follow-up of #8668).

- `src/frontend/src/instance.rs`: match `Channel::HttpSql` in
  `derive_timeout` and `derive_timeout_for_plan` for readonly
  statements/plans; pass a real `SlowQueryRecorder` instead of `None`
  in `do_analyze_stream_query_inner` so the `/v1/sql/analyze/stream`
  endpoint also records on timeout; add
  `test_derive_timeout_for_http_sql_channel`.
- `src/operator/src/statement.rs`: accept `SET statement_timeout` and
  `SET max_execution_time` for the HTTP SQL channel so HTTP requests can
  actually set a query timeout.
- `src/operator/src/statement/set.rs`: allow PG-style time-unit strings
  (e.g. `SET statement_timeout = '50ms'`) for the HTTP SQL channel,
  which defaults to the PostgreSQL dialect.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

src/frontend/src/instance.rs

@@ -567,7 +567,7 @@ fn derive_timeout(stmt: &Statement, query_ctx: &QueryContextRef) -> Option<Durat
         return None;
     }
     match query_ctx.channel() {
-        Channel::Mysql if stmt.is_readonly() => Some(query_timeout),
+        Channel::Mysql | Channel::HttpSql if stmt.is_readonly() => Some(query_timeout),
         Channel::Postgres => Some(query_timeout),
         _ => None,
     }
@@ -580,7 +580,7 @@ fn derive_timeout_for_plan(plan: &LogicalPlan, query_ctx: &QueryContextRef) -> O
         return None;
     }
     match query_ctx.channel() {
-        Channel::Mysql if is_readonly_plan(plan) => Some(query_timeout),
+        Channel::Mysql | Channel::HttpSql if is_readonly_plan(plan) => Some(query_timeout),
         Channel::Postgres => Some(query_timeout),
         _ => None,
     }
@@ -721,6 +721,9 @@ impl Instance {
         let catalog_name = query_ctx.current_catalog().to_string();
         let schema_name = query_ctx.current_schema();
         let slow_query_timer = self.statement_slow_query_timer(&stmt, schema_name.clone());
+        // The statement is guaranteed to be EXPLAIN ANALYZE VERBOSE here, so
+        // attach a recorder to capture plan metrics if the stream times out.
+        let timeout_recorder = slow_query_timer.as_ref().map(SlowQueryTimer::recorder);
         let ticket = self.process_manager.register_query(
             catalog_name,
             vec![schema_name],
@@ -729,8 +732,12 @@ impl Instance {
             Some(query_ctx.process_id()),
             slow_query_timer,
         );
-        let query_fut =
-            self.exec_statement_with_timeout(stmt, query_ctx.clone(), query_interceptor, None);
+        let query_fut = self.exec_statement_with_timeout(
+            stmt,
+            query_ctx.clone(),
+            query_interceptor,
+            timeout_recorder,
+        );
         let output = CancellableFuture::new(query_fut, ticket.cancellation_handle.clone())
             .await
             .map_err(|_| error::CancelledSnafu.build())??;
@@ -2249,6 +2256,31 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_derive_timeout_for_http_sql_channel() {
+        let query_ctx = Arc::new(
+            QueryContextBuilder::default()
+                .channel(Channel::HttpSql)
+                .build(),
+        );
+        query_ctx.set_query_timeout(Duration::from_secs(10));
+
+        // Readonly statements are wrapped with the query timeout.
+        let explain = parse_one_sql("EXPLAIN ANALYZE VERBOSE SELECT 1");
+        assert_eq!(
+            derive_timeout(&explain, &query_ctx),
+            Some(Duration::from_secs(10))
+        );
+
+        // Non-readonly statements are not wrapped.
+        let insert = parse_one_sql("INSERT INTO demo VALUES (1)");
+        assert_eq!(derive_timeout(&insert, &query_ctx), None);
+
+        // Zero timeout disables the wrapper.
+        query_ctx.set_query_timeout(Duration::ZERO);
+        assert_eq!(derive_timeout(&explain, &query_ctx), None);
+    }
+
     struct PendingDataSource {
         schema: GtSchemaRef,
         polled_tx: std::sync::Mutex<Option<oneshot::Sender<()>>>,

src/operator/src/statement.rs

@@ -552,7 +552,7 @@ impl StatementExecutor {
 
             "CLIENT_ENCODING" => validate_client_encoding(set_var)?,
             "@@SESSION.MAX_EXECUTION_TIME" | "MAX_EXECUTION_TIME" => match query_ctx.channel() {
-                Channel::Mysql => set_query_timeout(set_var.value, query_ctx)?,
+                Channel::Mysql | Channel::HttpSql => set_query_timeout(set_var.value, query_ctx)?,
                 Channel::Postgres => {
                     warn!(
                         "Unsupported set variable {} for channel {:?}",
@@ -569,7 +569,9 @@ impl StatementExecutor {
                 }
             },
             "STATEMENT_TIMEOUT" => match query_ctx.channel() {
-                Channel::Postgres => set_query_timeout(set_var.value, query_ctx)?,
+                Channel::Postgres | Channel::HttpSql => {
+                    set_query_timeout(set_var.value, query_ctx)?
+                }
                 Channel::Mysql => {
                     warn!(
                         "Unsupported set variable {} for channel {:?}",

src/operator/src/statement/set.rs

@@ -19,7 +19,7 @@ use common_time::Timezone;
 use lazy_static::lazy_static;
 use regex::Regex;
 use session::ReadPreference;
-use session::context::Channel::Postgres;
+use session::context::Channel::{HttpSql, Postgres};
 use session::context::QueryContextRef;
 use session::session_config::{PGByteaOutputValue, PGDateOrder, PGDateTimeStyle, PGIntervalStyle};
 use snafu::{OptionExt, ResultExt, ensure};
@@ -328,7 +328,9 @@ pub fn set_query_timeout(exprs: Vec<Expr>, ctx: QueryContextRef) -> Result<()> {
             value: Value::DoubleQuotedString(timeout),
             ..
         }) => {
-            if ctx.channel() != Postgres {
+            // HTTP SQL defaults to the PostgreSQL dialect, so it accepts the
+            // same time-unit strings as the Postgres channel.
+            if ctx.channel() != Postgres && ctx.channel() != HttpSql {
                 return NotSupportedSnafu {
                     feat: format!("Invalid timeout expr {} in set variable statement", timeout),
                 }