feat: validate public query hints across protocols

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

src/servers/src/grpc/flight.rs

@@ -56,7 +56,7 @@ pub use crate::grpc::flight::stream::{
     FlightRecordBatchSource, FlightRecordBatchStream, FlightRecordBatchStreamInput,
 };
 use crate::grpc::greptime_handler::{
-    GreptimeRequestHandler, create_query_context, get_request_type,
+    GreptimeRequestHandler, create_query_context_with_internal_metadata, get_request_type,
 };
 use crate::grpc::{FlightCompression, TonicResult, context_auth};
 use crate::request_memory_limiter::ServerMemoryLimiter;
@@ -196,20 +196,35 @@ impl FlightCraft for GreptimeRequestHandler {
         &self,
         request: Request<Ticket>,
     ) -> TonicResult<Response<TonicStream<FlightData>>> {
-        let mut hints = hint_headers::extract_hints(request.metadata());
-        hints.extend(extract_flow_extensions(request.metadata())?);
-        let snapshot_seqs = extract_snapshot_seqs(request.metadata())?;
+        let hints = hint_headers::extract_hints(request.metadata());
         let channel = request
             .extensions()
             .get::<Channel>()
             .copied()
             .unwrap_or(Channel::Grpc);
+        let flow_extensions = extract_flow_extensions(request.metadata())?;
+        let snapshot_seqs = extract_snapshot_seqs(request.metadata())?;
+        validate_dedicated_metadata_channel(
+            channel,
+            !flow_extensions.is_empty(),
+            !snapshot_seqs.is_empty(),
+        )?;
+        let (internal_flow, snapshot_seqs) = if channel == Channel::Internal {
+            (flow_extensions, snapshot_seqs)
+        } else {
+            (Vec::new(), HashMap::new())
+        };
 
         let ticket = request.into_inner().ticket;
         let request =
             GreptimeRequest::decode(ticket.as_ref()).context(error::InvalidFlightTicketSnafu)?;
-        let query_ctx =
-            create_query_context(channel, request.header.as_ref(), hints, snapshot_seqs)?;
+        let query_ctx = create_query_context_with_internal_metadata(
+            channel,
+            request.header.as_ref(),
+            hints,
+            internal_flow,
+            snapshot_seqs,
+        )?;
         // Validate flow hint syntax at the transport boundary before dispatching the request.
         // This does not authorize or execute anything; `handle_request()` below still performs
         // the normal frontend handling and auth checks before query execution.
@@ -554,6 +569,19 @@ impl Stream for PutRecordBatchRequestStream {
     }
 }
 
+fn validate_dedicated_metadata_channel(
+    channel: Channel,
+    flow_metadata_present: bool,
+    snapshot_metadata_present: bool,
+) -> TonicResult<()> {
+    if channel != Channel::Internal && (flow_metadata_present || snapshot_metadata_present) {
+        return Err(Status::invalid_argument(
+            "Flow and snapshot metadata are only accepted on internal channels",
+        ));
+    }
+    Ok(())
+}
+
 fn extract_flow_extensions(
     metadata: &tonic::metadata::MetadataMap,
 ) -> TonicResult<Vec<(String, String)>> {
@@ -638,14 +666,106 @@ fn output_to_flight_record_batch_source(
 
 #[cfg(test)]
 mod tests {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    use api::v1::{QueryRequest, greptime_request};
+    use common_grpc::flight::do_put::DoPutResponse;
+    use futures::stream;
     use query::options::{
         FLOW_INCREMENTAL_AFTER_SEQS, FLOW_INCREMENTAL_MODE, FLOW_RETURN_REGION_SEQ,
         FLOW_SCHEDULED_TIME_MILLIS, FLOW_SINK_TABLE_ID, FlowIncrementalMode,
     };
+    use session::context::QueryContextRef;
     use tonic::metadata::{AsciiMetadataValue, MetadataMap};
 
     use super::*;
 
+    struct FlightQueryHandler {
+        calls: Arc<AtomicUsize>,
+    }
+
+    #[async_trait]
+    impl crate::query_handler::grpc::GrpcQueryHandler for FlightQueryHandler {
+        async fn do_query(
+            &self,
+            _query: greptime_request::Request,
+            _ctx: QueryContextRef,
+        ) -> Result<Output> {
+            self.calls.fetch_add(1, Ordering::SeqCst);
+            Err(crate::error::InternalSnafu { err_msg: "test" }.build())
+        }
+
+        fn handle_put_record_batch_stream(
+            &self,
+            _stream: PutRecordBatchRequestStream,
+            _ctx: QueryContextRef,
+        ) -> Pin<Box<dyn Stream<Item = Result<DoPutResponse>> + Send>> {
+            Box::pin(stream::empty())
+        }
+    }
+
+    fn flight_handler(calls: Arc<AtomicUsize>) -> GreptimeRequestHandler {
+        GreptimeRequestHandler::new(
+            Arc::new(FlightQueryHandler { calls }),
+            None,
+            None,
+            FlightCompression::None,
+        )
+    }
+
+    fn sql_ticket() -> Ticket {
+        let request = GreptimeRequest {
+            header: None,
+            request: Some(greptime_request::Request::Query(QueryRequest {
+                query: Some(api::v1::query_request::Query::Sql("SELECT 1".into())),
+            })),
+        };
+        Ticket {
+            ticket: request.encode_to_vec().into(),
+        }
+    }
+
+    #[tokio::test]
+    async fn test_do_get_rejects_public_dedicated_metadata_before_dispatch() {
+        let calls = Arc::new(AtomicUsize::new(0));
+        for (key, value) in [
+            (
+                FLOW_EXTENSIONS_METADATA_KEY,
+                r#"[["flow.return_region_seq","true"]]"#,
+            ),
+            (SNAPSHOT_SEQS_METADATA_KEY, r#"{"1":10}"#),
+            (FLOW_EXTENSIONS_METADATA_KEY, "not-json"),
+        ] {
+            let mut request = Request::new(sql_ticket());
+            request.metadata_mut().insert(key, value.parse().unwrap());
+            let error = match FlightCraft::do_get(&flight_handler(calls.clone()), request).await {
+                Ok(_) => panic!("Expected dedicated external metadata to be rejected"),
+                Err(error) => error,
+            };
+            assert_eq!(error.code(), tonic::Code::InvalidArgument);
+            assert_eq!(calls.load(Ordering::SeqCst), 0);
+        }
+    }
+
+    #[tokio::test]
+    async fn test_do_get_internal_dedicated_flow_reaches_query_dispatch() {
+        let calls = Arc::new(AtomicUsize::new(0));
+        let mut request = Request::new(sql_ticket());
+        request.extensions_mut().insert(Channel::Internal);
+        request.metadata_mut().insert(
+            FLOW_EXTENSIONS_METADATA_KEY,
+            r#"[["flow.scheduled_time_millis","1700000000000"]]"#
+                .parse()
+                .unwrap(),
+        );
+        assert!(
+            FlightCraft::do_get(&flight_handler(calls.clone()), request)
+                .await
+                .is_ok()
+        );
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+    }
+
     #[test]
     fn test_extract_flow_extensions_preserves_comma_bearing_values() {
         let mut metadata = MetadataMap::new();
@@ -680,8 +800,14 @@ mod tests {
         );
 
         let flow_extensions = extract_flow_extensions(&metadata).unwrap();
-        let query_ctx =
-            create_query_context(Channel::Grpc, None, flow_extensions, HashMap::new()).unwrap();
+        let query_ctx = create_query_context_with_internal_metadata(
+            Channel::Internal,
+            None,
+            Vec::new(),
+            flow_extensions,
+            HashMap::new(),
+        )
+        .unwrap();
 
         assert_eq!(
             query_ctx.extension(FLOW_SCHEDULED_TIME_MILLIS),
@@ -701,8 +827,14 @@ mod tests {
         );
 
         let flow_extensions = extract_flow_extensions(&metadata).unwrap();
-        let query_ctx =
-            create_query_context(Channel::Grpc, None, flow_extensions, HashMap::new()).unwrap();
+        let query_ctx = create_query_context_with_internal_metadata(
+            Channel::Internal,
+            None,
+            Vec::new(),
+            flow_extensions,
+            HashMap::new(),
+        )
+        .unwrap();
         let parsed =
             query::options::FlowQueryExtensions::parse_flow_extensions(&query_ctx.extensions())
                 .unwrap()
@@ -730,6 +862,22 @@ mod tests {
         assert_eq!(query_ctx.extension(FLOW_SINK_TABLE_ID), Some("42"));
     }
 
+    #[test]
+    fn test_flight_external_flow_metadata_rejected() {
+        let mut metadata = MetadataMap::new();
+        metadata.insert(
+            FLOW_EXTENSIONS_METADATA_KEY,
+            AsciiMetadataValue::try_from(r#"[["flow.return_region_seq","true"]]"#).unwrap(),
+        );
+        let flow_extensions = extract_flow_extensions(&metadata).unwrap();
+        assert!(
+            validate_dedicated_metadata_channel(Channel::Grpc, !flow_extensions.is_empty(), false,)
+                .is_err()
+        );
+        assert!(validate_dedicated_metadata_channel(Channel::Grpc, false, true).is_err());
+        assert!(validate_dedicated_metadata_channel(Channel::Internal, true, true).is_ok());
+    }
+
     #[test]
     fn test_extract_flow_extensions_rejects_invalid_json() {
         let mut metadata = MetadataMap::new();

src/servers/src/grpc/greptime_handler.rs

@@ -205,6 +205,22 @@ pub(crate) fn create_query_context(
     header: Option<&RequestHeader>,
     extensions: Vec<(String, String)>,
     snapshot_seqs: HashMap<u64, u64>,
+) -> Result<QueryContextRef> {
+    create_query_context_with_internal_metadata(
+        channel,
+        header,
+        extensions,
+        Vec::new(),
+        snapshot_seqs,
+    )
+}
+
+pub(crate) fn create_query_context_with_internal_metadata(
+    channel: Channel,
+    header: Option<&RequestHeader>,
+    public_hints: Vec<(String, String)>,
+    internal_flow: Vec<(String, String)>,
+    snapshot_seqs: HashMap<u64, u64>,
 ) -> Result<QueryContextRef> {
     let (catalog, schema) = header
         .map(|header| {
@@ -234,6 +250,33 @@ pub(crate) fn create_query_context(
             )
         });
     let timezone = parse_timezone(header.map(|h| h.timezone.as_str()));
+    if channel != Channel::Internal && !internal_flow.is_empty() {
+        return crate::error::InvalidParameterSnafu {
+            reason: "Flow metadata is only accepted on internal channels".to_string(),
+        }
+        .fail();
+    }
+    let extensions = crate::hint_headers::validate_public_hints(public_hints)?;
+    let mut validated_internal = Vec::with_capacity(internal_flow.len());
+    for (key, value) in internal_flow {
+        let known_flow = matches!(
+            key.as_str(),
+            query::options::FLOW_INCREMENTAL_AFTER_SEQS
+                | query::options::FLOW_INCREMENTAL_MODE
+                | query::options::FLOW_RETURN_REGION_SEQ
+                | query::options::FLOW_SINK_TABLE_ID
+                | query::options::FLOW_SCHEDULED_TIME_MILLIS
+        );
+        if !known_flow && !is_reserved_extension_key(&key) {
+            return crate::error::InvalidParameterSnafu {
+                reason: format!("Unsupported internal metadata key `{key}`"),
+            }
+            .fail();
+        }
+        validated_internal.push((key, value));
+    }
+    validated_internal.extend(extensions);
+    let extensions = validated_internal;
     let mut ctx_builder = QueryContextBuilder::default()
         .current_catalog(catalog)
         .current_schema(schema)
@@ -261,7 +304,7 @@ pub(crate) fn create_query_context(
                 })?;
                 ctx_builder = ctx_builder.skip_wal(skip_wal);
             }
-            _ if is_reserved_extension_key(&key) => {
+            _ if is_reserved_extension_key(&key) && channel != Channel::Internal => {
                 debug!(
                     key = key.as_str(),
                     "Ignoring reserved external query context extension key"
@@ -412,8 +455,8 @@ mod tests {
             timezone: "+01:00".to_string(),
             ..Default::default()
         };
-        let query_context = create_query_context(
-            Channel::Unknown,
+        let query_context = create_query_context_with_internal_metadata(
+            Channel::Internal,
             Some(&header),
             vec![
                 ("auto_create_table".to_string(), "true".to_string()),
@@ -426,11 +469,11 @@ mod tests {
                     INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY.to_string(),
                     "spoofed-regs".to_string(),
                 ),
-                (
-                    FLOW_SCHEDULED_TIME_MILLIS.to_string(),
-                    "1700000000000".to_string(),
-                ),
             ],
+            vec![(
+                FLOW_SCHEDULED_TIME_MILLIS.to_string(),
+                "1700000000000".to_string(),
+            )],
             HashMap::from([(7, 88)]),
         )
         .unwrap();

src/servers/src/hint_headers.rs

@@ -16,6 +16,8 @@ use http::HeaderMap;
 use session::hints::{HINT_KEYS, HINTS_KEY, HINTS_KEY_PREFIX};
 use tonic::metadata::MetadataMap;
 
+use crate::error::{InvalidParameterSnafu, Result};
+
 pub(crate) fn extract_hints<T: ToHeaderMap>(headers: &T) -> Vec<(String, String)> {
     let mut hints = Vec::new();
     if let Some(value_str) = headers.get(HINTS_KEY) {
@@ -23,9 +25,13 @@ pub(crate) fn extract_hints<T: ToHeaderMap>(headers: &T) -> Vec<(String, String)
             let mut parts = hint.splitn(2, '=');
             if let (Some(key), Some(value)) = (parts.next(), parts.next()) {
                 hints.push((key.trim().to_string(), value.trim().to_string()));
+            } else {
+                let key = hint.trim();
+                if owns_hint_namespace(key) {
+                    hints.push((key.to_string(), String::new()));
+                }
             }
         });
-        // If hints are provided in the `x-greptime-hints` header, ignore the rest of the headers
         return hints;
     }
     for key in HINT_KEYS.iter() {
@@ -37,6 +43,81 @@ pub(crate) fn extract_hints<T: ToHeaderMap>(headers: &T) -> Vec<(String, String)
     hints
 }
 
+fn owns_hint_namespace(key: &str) -> bool {
+    let key = key.to_ascii_lowercase();
+    key.starts_with("query.")
+        || key.starts_with("datafusion.")
+        || key.starts_with("flow.")
+        || key == "query_parallelism"
+        || key == "query_fallback"
+        || key == "allow_query_fallback"
+}
+
+/// Normalize public hints before any QueryContext mutation.
+pub(crate) fn validate_public_hints(hints: Vec<(String, String)>) -> Result<Vec<(String, String)>> {
+    use std::collections::HashMap;
+
+    let mut normalized = Vec::with_capacity(hints.len());
+    let mut seen = HashMap::<String, String>::new();
+    for (key, value) in hints {
+        if session::hints::is_reserved_extension_key(&key) {
+            continue;
+        }
+        let is_read_preference = key.eq_ignore_ascii_case(session::hints::READ_PREFERENCE_HINT);
+        if is_read_preference && value.parse::<session::ReadPreference>().is_err() {
+            return InvalidParameterSnafu {
+                reason: format!("Invalid read preference `{value}`"),
+            }
+            .fail();
+        }
+        if key.to_ascii_lowercase().starts_with("flow.") {
+            return InvalidParameterSnafu {
+                reason: format!("Public hint `{key}` is reserved for internal use"),
+            }
+            .fail();
+        }
+        let pair = if is_read_preference {
+            let preference = value.parse::<session::ReadPreference>().unwrap();
+            (
+                session::hints::READ_PREFERENCE_HINT.to_string(),
+                preference.to_string().to_ascii_lowercase(),
+            )
+        } else {
+            match session::query_options::parse_query_option(&key, &value) {
+                Ok(Some(pair)) => pair,
+                Ok(None) => (key, value),
+                Err(error) => {
+                    return InvalidParameterSnafu {
+                        reason: error.to_string(),
+                    }
+                    .fail();
+                }
+            }
+        };
+        let canonical = pair.0.to_ascii_lowercase();
+        if canonical.starts_with("query.")
+            || canonical.starts_with("datafusion.")
+            || canonical == "query_parallelism"
+            || canonical == "query_fallback"
+            || canonical == "allow_query_fallback"
+            || canonical == session::hints::READ_PREFERENCE_HINT
+        {
+            if let Some(existing) = seen.get(&pair.0) {
+                if existing != &pair.1 {
+                    return InvalidParameterSnafu {
+                        reason: format!("Conflicting values for hint `{}`", pair.0),
+                    }
+                    .fail();
+                }
+                continue;
+            }
+            seen.insert(pair.0.clone(), pair.1.clone());
+        }
+        normalized.push(pair);
+    }
+    Ok(normalized)
+}
+
 pub(crate) trait ToHeaderMap {
     fn get(&self, key: &str) -> Option<&str>;
 }
@@ -54,11 +135,89 @@ impl ToHeaderMap for HeaderMap {
 }
 #[cfg(test)]
 mod tests {
+    use common_error::ext::ErrorExt;
     use http::header::{HeaderMap, HeaderValue};
     use tonic::metadata::{MetadataMap, MetadataValue};
 
     use super::*;
 
+    #[test]
+    fn test_query_style_missing_equals_is_preserved() {
+        let mut headers = HeaderMap::new();
+        headers.insert(HINTS_KEY, HeaderValue::from_static("query.parallelism"));
+        assert_eq!(
+            extract_hints(&headers),
+            vec![("query.parallelism".to_string(), String::new())]
+        );
+    }
+
+    #[test]
+    fn test_query_option_aliases_normalize_and_conflicts_fail() {
+        let hints = validate_public_hints(vec![
+            ("query_parallelism".into(), "04".into()),
+            ("query.parallelism".into(), "4".into()),
+            ("query_fallback".into(), "TRUE".into()),
+        ])
+        .unwrap();
+        assert_eq!(
+            hints,
+            vec![
+                ("query.parallelism".into(), "4".into()),
+                ("query.allow_query_fallback".into(), "true".into()),
+            ]
+        );
+        assert!(
+            validate_public_hints(vec![
+                ("query_parallelism".into(), "4".into()),
+                ("query.parallelism".into(), "5".into()),
+            ])
+            .is_err()
+        );
+    }
+
+    #[test]
+    fn test_read_preference_is_canonicalized_and_validated() {
+        assert_eq!(
+            validate_public_hints(vec![("READ_PREFERENCE".into(), "leader".into())]).unwrap(),
+            vec![("read_preference".into(), "leader".into())]
+        );
+        assert!(
+            validate_public_hints(vec![("read_preference".into(), "follower".into())]).is_err()
+        );
+    }
+
+    #[test]
+    fn test_unknown_or_malformed_query_options_fail() {
+        for (key, value) in [
+            ("query.parallelism", "0"),
+            ("query.unknown_option", "true"),
+            ("datafusion.execution.batch_size", "12"),
+            ("query.enable_remote_dynamic_filter_pushdown", "yes"),
+        ] {
+            assert!(validate_public_hints(vec![(key.into(), value.into())]).is_err());
+        }
+    }
+
+    #[test]
+    fn test_unrelated_duplicate_hints_remain_last_write_wins() {
+        let hints = validate_public_hints(vec![
+            ("ttl".into(), "1d".into()),
+            ("ttl".into(), "2d".into()),
+        ])
+        .unwrap();
+        assert_eq!(hints.len(), 2);
+    }
+
+    #[test]
+    fn test_public_query_options_are_validated_before_apply() {
+        let err = validate_public_hints(vec![("flow.scheduled_time_millis".into(), "1".into())])
+            .unwrap_err();
+        assert_eq!(
+            err.status_code(),
+            common_error::status_code::StatusCode::InvalidArguments
+        );
+    }
+
     #[test]
     fn test_extract_skip_wal_hint() {
         use session::hints::INSERT_SKIP_WAL_HINT;

src/servers/src/http/hints.rs

@@ -15,7 +15,7 @@
 use axum::body::Body;
 use axum::http::Request;
 use axum::middleware::Next;
-use axum::response::Response;
+use axum::response::{IntoResponse, Response};
 use common_telemetry::debug;
 use session::context::QueryContext;
 use session::hints::is_reserved_extension_key;
@@ -24,13 +24,24 @@ use crate::hint_headers;
 
 pub async fn extract_hints(mut request: Request<Body>, next: Next) -> Response {
     let hints = hint_headers::extract_hints(request.headers());
+    let hints = match hint_headers::validate_public_hints(hints) {
+        Ok(hints) => hints,
+        Err(error) => return error.into_response(),
+    };
+    let dedicated_read_preference = request
+        .headers()
+        .contains_key(&crate::http::header::GREPTIME_DB_HEADER_READ_PREFERENCE);
     if let Some(query_ctx) = request.extensions_mut().get_mut::<QueryContext>() {
-        apply_hints(query_ctx, hints);
+        apply_hints(query_ctx, hints, dedicated_read_preference);
     }
     next.run(request).await
 }
 
-fn apply_hints(query_ctx: &mut QueryContext, hints: Vec<(String, String)>) {
+fn apply_hints(
+    query_ctx: &mut QueryContext,
+    hints: Vec<(String, String)>,
+    dedicated_read_preference: bool,
+) {
     for (key, value) in hints {
         if is_reserved_extension_key(&key) {
             debug!(
@@ -39,14 +50,21 @@ fn apply_hints(query_ctx: &mut QueryContext, hints: Vec<(String, String)>) {
             );
             continue;
         }
-        query_ctx.set_extension(key, value);
+        if key.eq_ignore_ascii_case(session::hints::READ_PREFERENCE_HINT) {
+            if !dedicated_read_preference {
+                if let Ok(preference) = value.parse::<session::ReadPreference>() {
+                    query_ctx.set_read_preference(preference);
+                }
+            }
+        } else {
+            query_ctx.set_extension(key, value);
+        }
     }
 }
 
 #[cfg(test)]
 mod tests {
     use common_query::request::INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY as COMMON_INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY;
-    use query::options::FLOW_SCHEDULED_TIME_MILLIS;
     use session::context::{QueryContextBuilder, generate_remote_query_id};
     use session::hints::{
         INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY, REMOTE_QUERY_ID_EXTENSION_KEY,
@@ -75,12 +93,10 @@ mod tests {
                     INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY.to_string(),
                     "spoofed-regs".to_string(),
                 ),
-                (
-                    FLOW_SCHEDULED_TIME_MILLIS.to_string(),
-                    "1700000000000".to_string(),
-                ),
+                ("custom.extension".to_string(), "value".to_string()),
                 ("ttl".to_string(), "7d".to_string()),
             ],
+            false,
         );
 
         assert_eq!(
@@ -92,10 +108,7 @@ mod tests {
                 .extension(INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY)
                 .is_none()
         );
-        assert_eq!(
-            query_ctx.extension(FLOW_SCHEDULED_TIME_MILLIS),
-            Some("1700000000000")
-        );
+        assert_eq!(query_ctx.extension("custom.extension"), Some("value"));
         assert_eq!(query_ctx.extension("ttl"), Some("7d"));
     }
 

src/servers/src/http/read_preference.rs

@@ -12,12 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::str::FromStr;
-
 use axum::body::Body;
 use axum::http::Request;
 use axum::middleware::Next;
-use axum::response::Response;
+use axum::response::{IntoResponse, Response};
 use session::ReadPreference;
 use session::context::QueryContext;
 
@@ -25,16 +23,22 @@ use crate::http::header::GREPTIME_DB_HEADER_READ_PREFERENCE;
 
 /// Extract read preference from the request headers.
 pub async fn extract_read_preference(mut request: Request<Body>, next: Next) -> Response {
-    let read_preference = request
-        .headers()
-        .get(&GREPTIME_DB_HEADER_READ_PREFERENCE)
-        .and_then(|header| header.to_str().ok())
-        .and_then(|s| ReadPreference::from_str(s).ok())
-        .unwrap_or_default();
-
-    if let Some(query_ctx) = request.extensions_mut().get_mut::<QueryContext>() {
-        common_telemetry::debug!("Setting read preference to {}", read_preference);
-        query_ctx.set_read_preference(read_preference);
+    if let Some(header) = request.headers().get(&GREPTIME_DB_HEADER_READ_PREFERENCE) {
+        let read_preference = header
+            .to_str()
+            .ok()
+            .and_then(|s| s.parse::<ReadPreference>().ok());
+        let Some(read_preference) = read_preference else {
+            return crate::error::InvalidParameterSnafu {
+                reason: "Invalid read preference header".to_string(),
+            }
+            .build()
+            .into_response();
+        };
+        if let Some(query_ctx) = request.extensions_mut().get_mut::<QueryContext>() {
+            common_telemetry::debug!("Setting read preference to {}", read_preference);
+            query_ctx.set_read_preference(read_preference);
+        }
     }
     next.run(request).await
 }

src/servers/tests/http/http_handler_test.rs

@@ -1048,6 +1048,13 @@ async fn test_http_skip_wal_header() {
     let client = TestClient::new(server.build(server.make_app()).unwrap()).await;
     for (header, expected) in [
         (Some(("x-greptime-insert-skip-wal", "true")), Some(true)),
+        (
+            Some(("x-greptime-hints", "query.parallelism=4")),
+            Some(false),
+        ),
+        (Some(("x-greptime-hints", "query.parallelism=0")), None),
+        (Some(("x-greptime-hints", "query.unknown=true")), None),
+        (Some(("x-greptime-hints", "query.parallelism")), None),
         (Some(("x-greptime-insert-skip-wal", "false")), Some(false)),
         (None, Some(false)),
         (Some(("x-greptime-insert-skip-wal", "yes")), None),