test: cover query option effects and isolation across protocols

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

docs/how-to/query-options.md

@@ -0,0 +1,87 @@
+# Query options
+
+Query options let a caller adjust selected query behavior for a connection or a
+single request. They are not server-wide settings, execution/runtime controls,
+or a way to configure the server's resource pools.
+
+## Supported options
+
+| Canonical option | SQL aliases | Values |
+| --- | --- | --- |
+| `query.parallelism` | `query_parallelism`, `query.parallelism` | Integer `1`–`1024`; server value `0` continues to mean automatic CPU-based parallelism. Sets the query's parallelism/partition bound, not a server resource-pool limit. |
+| `query.allow_query_fallback` | `query_fallback`, `allow_query_fallback`, `query.allow_query_fallback` | `true` or `false` |
+| `query.enable_remote_dynamic_filter_pushdown` | same | `true` or `false` |
+| `datafusion.optimizer.*` | canonical name | Boolean, limited to: `repartition_joins`, `repartition_aggregations`, `repartition_sorts`, `repartition_windows`, `enable_round_robin_repartition`, `prefer_existing_sort`, `prefer_hash_join`, `join_reordering`, `enable_topk_aggregation`, `enable_dynamic_filter_pushdown` |
+
+For example:
+
+```sql
+SET query_parallelism = 4;
+SHOW VARIABLES query.parallelism;
+SET query.allow_query_fallback = true;
+SHOW VARIABLES query.allow_query_fallback;
+SET datafusion.optimizer.repartition_joins = false;
+```
+
+`SHOW VARIABLES <option>` shows the effective value. Boolean values must be
+`true` or `false`. Unknown, unsupported, and non-allowlisted DataFusion options
+are rejected; arbitrary DataFusion configuration is not enabled.
+
+The existing `ALLOW_QUERY_FALLBACK` SQL syntax and `query_fallback` hint remain
+available as aliases for `query.allow_query_fallback`.
+
+Unspecified options inherit the query engine's settings. With the default server
+configuration, parallelism is automatic, query fallback is `false`, and remote
+dynamic-filter pushdown is `true`. For the pinned DataFusion version, the listed
+optimizer switches default to `true` except `prefer_existing_sort`, which defaults
+to `false`. `SHOW VARIABLES <option>` reports the effective value rather than an
+unset session override.
+
+## Scope and precedence
+
+- SQL `SET` is connection-scoped. Later statements on that connection observe
+  the setting; a different connection does not inherit it.
+- HTTP `X-Greptime-Hints` supplies request-scoped options, for example
+  `X-Greptime-Hints: query.parallelism=4`.
+- A request hint takes precedence over a session `SET` for the same option. In a
+  multi-statement HTTP call it continues to win even if a statement executes
+  `SET`.
+- `SET LOCAL` transaction lifecycle semantics are not implemented; do not
+  rely on transaction-scoped restoration.
+
+HTTP SQL example (`/v1/sql` accepts form data):
+
+```sh
+curl 'http://localhost:4000/v1/sql' \
+  -H 'x-greptime-hints: query.parallelism=4,datafusion.optimizer.repartition_joins=false' \
+  --data-urlencode 'sql=EXPLAIN SELECT * FROM my_table'
+```
+
+For gRPC, hints are sent as `x-greptime-hints` request metadata, not as a
+protobuf `Hints` field. The Rust Flight client supports
+`Database::sql_with_hint(sql, &[...])` and `flight_request().with_hints(&[...])`;
+for example, pass `[("query_parallelism", "4")]` as the hints slice.
+
+## Dynamic filters
+
+`query.enable_remote_dynamic_filter_pushdown` controls dynamic-filter
+propagation across the frontend-to-datanode remote query boundary. It is
+separate from `datafusion.optimizer.enable_dynamic_filter_pushdown`, which
+controls DataFusion's native optimizer pushdown. Changing one does not imply
+changing the other.
+
+## Other request controls and limitations
+
+Read preference accepts the generic `read_preference` hint and the dedicated
+`x-greptime-read-preference` HTTP header; when both are supplied, the dedicated
+header wins. HTTP response deadlines use `x-greptime-timeout` (a duration such
+as `5s`). The server default is zero, meaning no response deadline. A header
+value of zero also disables the response deadline; an invalid value falls back
+to the server default. A larger valid header value can currently override the
+server default; no new cap is imposed. This limits the HTTP response, not
+individual statements or response-stream/body processing.
+
+Internal flow/query-reserved keys are not public query options. Internal
+server-marked Flight metadata remains internal. Invalid values for recognized
+query options produce an error. Existing tolerance for unknown initialization
+settings from third-party MySQL/PostgreSQL clients is preserved.

tests-integration/src/grpc/flight.rs

@@ -499,11 +499,41 @@ mod test {
         query_and_expect(db.frontend().as_ref(), sql, expected).await;
 
         create_table_named(&client, "bar").await;
-        let result = client
-            .sql_with_terminal_metrics(
-                "insert into bar select ts, a, `B` from foo",
-                &[("flow.return_region_seq", "true")],
-            )
+        let runtime = common_runtime::global_runtime().clone();
+        let request_handler = GreptimeRequestHandler::new(
+            db.frontend().clone(),
+            None,
+            Some(runtime.clone()),
+            FlightCompression::default(),
+        );
+        let mut internal_server = GrpcServerBuilder::new(GrpcServerConfig::default(), runtime)
+            .database_handler(request_handler.clone())
+            .flight_handler(Arc::new(request_handler))
+            .add_layer(axum::middleware::from_fn(
+                |mut request: axum::extract::Request, next: axum::middleware::Next| async move {
+                    request
+                        .extensions_mut()
+                        .insert(session::context::Channel::Internal);
+                    next.run(request).await
+                },
+            ))
+            .build();
+        internal_server
+            .start("127.0.0.1:0".parse::<SocketAddr>().unwrap())
+            .await
+            .unwrap();
+        let internal_client = Database::new_with_dbname(
+            "greptime-public",
+            Client::with_urls(vec![internal_server.bind_addr().unwrap().to_string()]),
+        );
+        let result = internal_client
+            .flight_request()
+            .with_flow_extensions(&[("flow.return_region_seq", "true")])
+            .query_with_terminal_metrics(QueryRequest {
+                query: Some(Query::Sql(
+                    "insert into bar select ts, a, `B` from foo".to_string(),
+                )),
+            })
             .await
             .unwrap();
         let OutputData::AffectedRows(affected_rows) = result.output.data else {
@@ -569,6 +599,14 @@ mod test {
         let mut grpc_server = GrpcServerBuilder::new(GrpcServerConfig::default(), runtime)
             .database_handler(greptime_request_handler.clone())
             .flight_handler(Arc::new(greptime_request_handler))
+            .add_layer(axum::middleware::from_fn(
+                |mut request: axum::extract::Request, next: axum::middleware::Next| async move {
+                    request
+                        .extensions_mut()
+                        .insert(session::context::Channel::Internal);
+                    next.run(request).await
+                },
+            ))
             .build();
         grpc_server
             .start("127.0.0.1:0".parse::<SocketAddr>().unwrap())
@@ -617,7 +655,11 @@ mod test {
         assert!(metrics.region_watermarks.is_empty());
 
         let result = client
-            .sql_with_terminal_metrics(sql, &[("flow.return_region_seq", "true")])
+            .flight_request()
+            .with_flow_extensions(&[("flow.return_region_seq", "true")])
+            .query_with_terminal_metrics(QueryRequest {
+                query: Some(Query::Sql(sql.to_string())),
+            })
             .await
             .unwrap();
         let terminal_metrics = result.metrics.clone();
@@ -638,7 +680,9 @@ mod test {
         );
 
         let output = client
-            .sql_with_hint(sql, &[("flow.return_region_seq", "true")])
+            .flight_request()
+            .with_flow_extensions(&[("flow.return_region_seq", "true")])
+            .sql(sql)
             .await
             .unwrap();
         let OutputData::Stream(mut stream) = output.data else {
@@ -677,10 +721,13 @@ mod test {
         assert!(result.region_watermark_map().is_none());
 
         let err = client
-            .sql_with_terminal_metrics(
-                "insert into bar select ts, a, `B` from foo",
-                &[("flow.return_region_seq", "not-a-bool")],
-            )
+            .flight_request()
+            .with_flow_extensions(&[("flow.return_region_seq", "not-a-bool")])
+            .query_with_terminal_metrics(QueryRequest {
+                query: Some(Query::Sql(
+                    "insert into bar select ts, a, `B` from foo".to_string(),
+                )),
+            })
             .await
             .unwrap_err();
         let err_msg = format!("{err:?}");
@@ -689,10 +736,13 @@ mod test {
         client.sql("truncate table bar").await.unwrap();
 
         let result = client
-            .sql_with_terminal_metrics(
-                "insert into bar select ts, a, `B` from foo",
-                &[("flow.return_region_seq", "true")],
-            )
+            .flight_request()
+            .with_flow_extensions(&[("flow.return_region_seq", "true")])
+            .query_with_terminal_metrics(QueryRequest {
+                query: Some(Query::Sql(
+                    "insert into bar select ts, a, `B` from foo".to_string(),
+                )),
+            })
             .await
             .unwrap();
         let OutputData::AffectedRows(affected_rows) = result.output.data else {
@@ -705,6 +755,106 @@ mod test {
         );
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_flight_query_options_are_applied_and_request_scoped() {
+        common_telemetry::init_default_ut_logging();
+
+        let (_db, server) = setup_grpc_server(StorageType::File, "flight_query_options").await;
+        let client = Database::new_with_dbname(
+            "greptime-public",
+            Client::with_urls(vec![server.bind_addr().unwrap().to_string()]),
+        );
+
+        let run = |hint: Option<(&'static str, &'static str)>| {
+            let client = client.clone();
+            async move {
+                let mut request = client.flight_request();
+                if let Some((key, value)) = hint {
+                    request = request.with_hints(&[(key, value)]);
+                }
+                request
+                    .sql("SHOW VARIABLES query.parallelism")
+                    .await
+                    .unwrap()
+                    .data
+                    .pretty_print()
+                    .await
+            }
+        };
+        let default_before = run(None).await;
+        let (canonical, legacy) = tokio::join!(
+            run(Some(("query.parallelism", "4"))),
+            run(Some(("query_parallelism", "7"))),
+        );
+        let has_value = |output: &str, expected: &str| {
+            output.lines().any(|line| {
+                line.trim().starts_with('|')
+                    && line.trim().ends_with('|')
+                    && line
+                        .rsplit('|')
+                        .nth(1)
+                        .is_some_and(|cell| cell.trim() == expected)
+            })
+        };
+        assert!(has_value(&canonical, "4"), "{canonical}");
+        assert!(has_value(&legacy, "7"), "{legacy}");
+        let default_after = run(None).await;
+        assert_eq!(
+            default_after, default_before,
+            "request-scoped hint changed the subsequent default result"
+        );
+        for variable in [
+            "query.allow_query_fallback",
+            "query.enable_remote_dynamic_filter_pushdown",
+            "datafusion.optimizer.enable_dynamic_filter_pushdown",
+        ] {
+            let value = client
+                .flight_request()
+                .with_hints(&[(variable, "false")])
+                .sql(format!("SHOW VARIABLES {variable}"))
+                .await
+                .unwrap()
+                .data
+                .pretty_print()
+                .await;
+            assert!(value.contains("false"), "{variable}: {value}");
+        }
+
+        for (key, value) in [
+            ("query.parallelism", "0"),
+            ("query.unknown_option", "true"),
+            ("datafusion.optimizer.repartition_joins", "invalid"),
+            ("datafusion.execution.batch_size", "12"),
+        ] {
+            let err = client
+                .flight_request()
+                .with_hints(&[(key, value)])
+                .sql("SELECT 1")
+                .await
+                .unwrap_err();
+            assert_eq!(
+                err.status_code(),
+                StatusCode::InvalidArguments,
+                "{key}={value}: {err:?}"
+            );
+        }
+        let err = client
+            .flight_request()
+            .with_hints(&[("query.parallelism", "4"), ("query_parallelism", "7")])
+            .sql("SELECT 1")
+            .await
+            .unwrap_err();
+        assert_eq!(err.status_code(), StatusCode::InvalidArguments);
+
+        let err = client
+            .flight_request()
+            .with_hints(&[("flow.return_region_seq", "true")])
+            .sql("SELECT 1")
+            .await
+            .unwrap_err();
+        assert_eq!(err.status_code(), StatusCode::InvalidArguments);
+    }
+
     async fn wait_for_client_health(client: &Client) {
         tokio::time::timeout(Duration::from_secs(10), async {
             loop {
@@ -738,6 +888,14 @@ mod test {
         );
         let mut grpc_server = GrpcServerBuilder::new(GrpcServerConfig::default(), runtime)
             .flight_handler(Arc::new(greptime_request_handler))
+            .add_layer(axum::middleware::from_fn(
+                |mut request: axum::extract::Request, next: axum::middleware::Next| async move {
+                    request
+                        .extensions_mut()
+                        .insert(session::context::Channel::Internal);
+                    next.run(request).await
+                },
+            ))
             .build();
         grpc_server
             .start("127.0.0.1:0".parse::<SocketAddr>().unwrap())

tests-integration/src/tests/remote_dyn_filter_test.rs

@@ -19,7 +19,8 @@ use frontend::instance::Instance;
 use query::datafusion::QUERY_PARALLELISM_HINT;
 use query::options::QUERY_ENABLE_REMOTE_DYNAMIC_FILTER_PUSHDOWN;
 use servers::query_handler::sql::SqlQueryHandler;
-use session::context::QueryContext;
+use session::Session;
+use session::context::{Channel, QueryContext};
 
 use crate::test_util::execute_sql;
 use crate::tests;
@@ -59,6 +60,255 @@ async fn test_remote_dyn_filter_join_e2e() {
     assert_seq_scan_has_dyn_filter(&explain);
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_query_options_session_scope_and_remote_filter_e2e() {
+    common_telemetry::init_default_ut_logging();
+
+    let distributed = tests::create_distributed_instance("test_query_options_session_scope").await;
+    let frontend = distributed.frontend();
+    prepare_remote_dyn_filter_large_tables(&frontend).await;
+
+    let first = Session::new(None, Channel::Mysql, Default::default(), 1);
+    let second = Session::new(None, Channel::Mysql, Default::default(), 2);
+
+    // Capture the second session's initial value before changing the first
+    // session, without assuming a host-specific default.
+    let second_initial = SqlQueryHandler::do_query(
+        frontend.as_ref(),
+        "SHOW VARIABLES query.parallelism",
+        second.new_query_context(),
+    )
+    .await;
+    assert!(
+        second_initial.iter().all(Result::is_ok),
+        "{second_initial:?}"
+    );
+    let second_initial =
+        output_to_pretty_string(second_initial.into_iter().next().unwrap().unwrap()).await;
+
+    // SET is applied on the connection's session state; SHOW VARIABLES reports
+    // the effective value, and a separate connection retains its own value.
+    let set_and_show = SqlQueryHandler::do_query(
+        frontend.as_ref(),
+        "SET query_parallelism = 2; SHOW VARIABLES query.parallelism; SET query.enable_remote_dynamic_filter_pushdown = false; SHOW VARIABLES query.enable_remote_dynamic_filter_pushdown",
+        first.new_query_context(),
+    )
+    .await;
+    assert!(set_and_show.iter().all(Result::is_ok), "{set_and_show:?}");
+    let mut results = set_and_show.into_iter();
+    let _ = results.next().unwrap().unwrap();
+    let values = output_to_pretty_string(results.next().unwrap().unwrap()).await;
+    assert!(
+        values
+            .lines()
+            .any(|line| line.trim().trim_matches('|').trim() == "2"),
+        "{values}"
+    );
+    let _ = results.next().unwrap().unwrap();
+    let remote_filter = output_to_pretty_string(results.next().unwrap().unwrap()).await;
+    assert_contains(&remote_filter, "false");
+
+    let fallback = SqlQueryHandler::do_query(
+        frontend.as_ref(),
+        "SET query_fallback = true; SHOW VARIABLES query.allow_query_fallback; SET ALLOW_QUERY_FALLBACK = false; SHOW VARIABLES query_fallback",
+        first.new_query_context(),
+    )
+    .await;
+    assert!(fallback.iter().all(Result::is_ok), "{fallback:?}");
+    let mut fallback = fallback.into_iter();
+    let _ = fallback.next().unwrap().unwrap();
+    assert_contains(
+        &output_to_pretty_string(fallback.next().unwrap().unwrap()).await,
+        "true",
+    );
+    let _ = fallback.next().unwrap().unwrap();
+    assert_contains(
+        &output_to_pretty_string(fallback.next().unwrap().unwrap()).await,
+        "false",
+    );
+
+    let second_after_first_set = SqlQueryHandler::do_query(
+        frontend.as_ref(),
+        "SHOW VARIABLES query.parallelism",
+        second.new_query_context(),
+    )
+    .await;
+    assert!(
+        second_after_first_set.iter().all(Result::is_ok),
+        "{second_after_first_set:?}"
+    );
+    let second_after_first_set =
+        output_to_pretty_string(second_after_first_set.into_iter().next().unwrap().unwrap()).await;
+    assert_eq!(second_after_first_set, second_initial);
+
+    let independent = SqlQueryHandler::do_query(
+        frontend.as_ref(),
+        "SET query_parallelism = 7; SHOW VARIABLES query.parallelism",
+        second.new_query_context(),
+    )
+    .await;
+    assert!(independent.iter().all(Result::is_ok), "{independent:?}");
+    let mut independent = independent.into_iter();
+    let _ = independent.next().unwrap().unwrap();
+    let independent = output_to_pretty_string(independent.next().unwrap().unwrap()).await;
+    assert!(
+        independent
+            .lines()
+            .any(|line| line.trim().trim_matches('|').trim() == "7"),
+        "{independent}"
+    );
+    let first_after_second_set = SqlQueryHandler::do_query(
+        frontend.as_ref(),
+        "SHOW VARIABLES query.parallelism",
+        first.new_query_context(),
+    )
+    .await;
+    assert!(
+        first_after_second_set.iter().all(Result::is_ok),
+        "{first_after_second_set:?}"
+    );
+    let first_after_second_set =
+        output_to_pretty_string(first_after_second_set.into_iter().next().unwrap().unwrap()).await;
+    assert_eq!(first_after_second_set, values);
+
+    // Model request hints on the query context used by an HTTP request. A
+    // request-scoped value wins even after a statement changes the session.
+    let mut request_context = first.new_query_context().as_ref().clone();
+    request_context.set_extension("query.enable_remote_dynamic_filter_pushdown", "false");
+    let hinted = SqlQueryHandler::do_query(
+        frontend.as_ref(),
+        "SET query.enable_remote_dynamic_filter_pushdown = true; SHOW VARIABLES query.enable_remote_dynamic_filter_pushdown",
+        Arc::new(request_context),
+    )
+    .await;
+    assert!(hinted.iter().all(Result::is_ok), "{hinted:?}");
+    let mut hinted = hinted.into_iter();
+    let _ = hinted.next().unwrap().unwrap();
+    let effective = output_to_pretty_string(hinted.next().unwrap().unwrap()).await;
+    assert_contains(&effective, "false");
+    let no_hint = SqlQueryHandler::do_query(
+        frontend.as_ref(),
+        "SHOW VARIABLES query.enable_remote_dynamic_filter_pushdown",
+        first.new_query_context(),
+    )
+    .await;
+    assert!(no_hint.iter().all(Result::is_ok), "{no_hint:?}");
+    let no_hint = output_to_pretty_string(no_hint.into_iter().next().unwrap().unwrap()).await;
+    assert_contains(&no_hint, "true");
+
+    // Exercise the session setting through a real distributed FE->DN query:
+    // its physical scan plan exposes whether the remote predicate arrived.
+    let join_sql = remote_dyn_filter_large_join_sql();
+    let explain_sql = format!("EXPLAIN ANALYZE VERBOSE {join_sql}");
+    let enabled = SqlQueryHandler::do_query(
+        frontend.as_ref(),
+        &format!("SET query.enable_remote_dynamic_filter_pushdown = true; {explain_sql}"),
+        first.new_query_context(),
+    )
+    .await;
+    assert!(enabled.iter().all(Result::is_ok), "{enabled:?}");
+    let mut enabled = enabled.into_iter();
+    let _ = enabled.next().unwrap().unwrap();
+    let enabled = output_to_pretty_string(enabled.next().unwrap().unwrap()).await;
+    assert_seq_scan_has_dyn_filter(&enabled);
+
+    let disabled = SqlQueryHandler::do_query(
+        frontend.as_ref(),
+        &format!("SET query.enable_remote_dynamic_filter_pushdown = false; {explain_sql}"),
+        first.new_query_context(),
+    )
+    .await;
+    assert!(disabled.iter().all(Result::is_ok), "{disabled:?}");
+    let mut disabled = disabled.into_iter();
+    let _ = disabled.next().unwrap().unwrap();
+    let disabled = output_to_pretty_string(disabled.next().unwrap().unwrap()).await;
+    assert_no_seq_scan_dyn_filter(&disabled);
+
+    let enabled_result = SqlQueryHandler::do_query(
+        frontend.as_ref(),
+        &format!("SET query.enable_remote_dynamic_filter_pushdown = true; {join_sql}"),
+        first.new_query_context(),
+    )
+    .await;
+    let mut enabled_result = enabled_result.into_iter();
+    let _ = enabled_result.next().unwrap().unwrap();
+    let enabled_result = output_to_pretty_string(enabled_result.next().unwrap().unwrap()).await;
+    let disabled_result = SqlQueryHandler::do_query(
+        frontend.as_ref(),
+        &format!("SET query.enable_remote_dynamic_filter_pushdown = false; {join_sql}"),
+        first.new_query_context(),
+    )
+    .await;
+    let mut disabled_result = disabled_result.into_iter();
+    let _ = disabled_result.next().unwrap().unwrap();
+    let disabled_result = output_to_pretty_string(disabled_result.next().unwrap().unwrap()).await;
+    assert_eq!(enabled_result, disabled_result);
+
+    // Native DataFusion pushdown is independent: disabling it keeps the result
+    // while suppressing the scan dynamic-filter evidence despite remote=true.
+    let native_disabled = SqlQueryHandler::do_query(
+        frontend.as_ref(),
+        &format!(
+            "SET query.enable_remote_dynamic_filter_pushdown = true; SET datafusion.optimizer.enable_dynamic_filter_pushdown = false; {explain_sql}"
+        ),
+        first.new_query_context(),
+    )
+    .await;
+    assert!(
+        native_disabled.iter().all(Result::is_ok),
+        "{native_disabled:?}"
+    );
+    let mut native_disabled = native_disabled.into_iter();
+    let _ = native_disabled.next().unwrap().unwrap();
+    let _ = native_disabled.next().unwrap().unwrap();
+    let native_disabled = output_to_pretty_string(native_disabled.next().unwrap().unwrap()).await;
+    assert_no_seq_scan_dyn_filter(&native_disabled);
+    let native_disabled_result = SqlQueryHandler::do_query(
+        frontend.as_ref(),
+        &format!(
+            "SET query.enable_remote_dynamic_filter_pushdown = true; SET datafusion.optimizer.enable_dynamic_filter_pushdown = false; {join_sql}"
+        ),
+        first.new_query_context(),
+    )
+    .await;
+    let mut native_disabled_result = native_disabled_result.into_iter();
+    let _ = native_disabled_result.next().unwrap().unwrap();
+    let _ = native_disabled_result.next().unwrap().unwrap();
+    let native_disabled_result =
+        output_to_pretty_string(native_disabled_result.next().unwrap().unwrap()).await;
+    assert_eq!(native_disabled_result, enabled_result);
+
+    // Parallel execution plus the native join-repartition optimizer switch
+    // changes the physical plan while preserving the query result.
+    let native_explain_sql = format!("EXPLAIN {join_sql}");
+    let plan_with_repartition = execute_with_session_options(
+        frontend.as_ref(),
+        &first,
+        native_explain_sql.as_str(),
+        "true",
+    )
+    .await;
+    let plan_with_repartition = output_to_pretty_string(plan_with_repartition).await;
+    let plan_without_repartition = execute_with_session_options(
+        frontend.as_ref(),
+        &first,
+        native_explain_sql.as_str(),
+        "false",
+    )
+    .await;
+    let plan_without_repartition = output_to_pretty_string(plan_without_repartition).await;
+    assert_contains(&plan_with_repartition, "HashJoinExec: mode=Partitioned");
+    assert_contains(&plan_without_repartition, "HashJoinExec: mode=CollectLeft");
+    let result_with_repartition =
+        execute_with_session_options(frontend.as_ref(), &first, join_sql, "true").await;
+    let result_without_repartition =
+        execute_with_session_options(frontend.as_ref(), &first, join_sql, "false").await;
+    assert_eq!(
+        output_to_pretty_string(result_with_repartition).await,
+        output_to_pretty_string(result_without_repartition).await
+    );
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn test_remote_dyn_filter_left_join_e2e() {
     common_telemetry::init_default_ut_logging();
@@ -198,6 +448,21 @@ async fn test_remote_dyn_filter_large_join_e2e() {
     assert_no_seq_scan_dyn_filter(&explain_without_rdf);
 }
 
+async fn execute_with_session_options(
+    frontend: &Instance,
+    session: &Session,
+    sql: &str,
+    repartition_joins: &str,
+) -> Output {
+    let statement = format!(
+        "SET query_parallelism = 2; SET query.enable_remote_dynamic_filter_pushdown = true; SET datafusion.optimizer.enable_dynamic_filter_pushdown = true; SET datafusion.optimizer.repartition_joins = {repartition_joins}; {sql}"
+    );
+    let outputs =
+        SqlQueryHandler::do_query(frontend, &statement, session.new_query_context()).await;
+    assert!(outputs.iter().all(Result::is_ok), "{outputs:?}");
+    outputs.into_iter().last().unwrap().unwrap()
+}
+
 async fn prepare_remote_dyn_filter_tables(frontend: &Arc<Instance>) {
     execute_sql(
         frontend,

tests-integration/tests/http.rs

@@ -129,6 +129,7 @@ macro_rules! http_tests {
 
                 test_http_auth,
                 test_sql_api,
+                test_http_query_options,
                 test_http_analyze_stream_tql,
                 test_http_sql_slow_query,
                 test_prometheus_promql_api,
@@ -483,6 +484,155 @@ async fn test_sql_skip_wal(distributed: bool) {
     check_http_skip_wal("sql", &cases, distributed).await;
 }
 
+pub async fn test_http_query_options(store_type: StorageType) {
+    common_telemetry::init_default_ut_logging();
+    let (app, mut guard) =
+        setup_test_http_app_with_frontend(store_type, "test_http_query_options").await;
+    let client = TestClient::new(app).await;
+
+    async fn query(client: &TestClient, sql: &str, hints: Option<&str>) -> TestResponse {
+        let mut request = client.post("/v1/sql").form(&[("sql", sql)]);
+        if let Some(hints) = hints {
+            request = request.header("x-greptime-hints", hints);
+        }
+        request.send().await
+    }
+
+    fn shown_value(body: &Value) -> &Value {
+        let output = body["output"].as_array().unwrap().last().unwrap();
+        &output["records"]["rows"][0][0]
+    }
+
+    async fn assert_variable(client: &TestClient, name: &str, expected: &str, hints: Option<&str>) {
+        let sql = format!("SHOW VARIABLES {name}");
+        let response = query(client, &sql, hints).await;
+        assert_eq!(response.status(), StatusCode::OK, "{name}");
+        let body: Value = response.json().await;
+        assert_eq!(shown_value(&body), expected, "{name}: {body}");
+    }
+
+    let baseline_response = query(&client, "SHOW VARIABLES query.parallelism", None).await;
+    assert_eq!(baseline_response.status(), StatusCode::OK);
+    let baseline: Value = baseline_response.json().await;
+    let default_parallelism = shown_value(&baseline).as_str().unwrap().to_string();
+
+    assert_variable(
+        &client,
+        "query.parallelism",
+        "4",
+        Some("query.parallelism=4"),
+    )
+    .await;
+    assert_variable(
+        &client,
+        "query.parallelism",
+        "7",
+        Some("query_parallelism=7"),
+    )
+    .await;
+    for (name, expected, hints) in [
+        (
+            "query.allow_query_fallback",
+            "false",
+            "query_fallback=false",
+        ),
+        (
+            "query.enable_remote_dynamic_filter_pushdown",
+            "false",
+            "query.enable_remote_dynamic_filter_pushdown=false",
+        ),
+        (
+            "datafusion.optimizer.join_reordering",
+            "false",
+            "datafusion.optimizer.join_reordering=false",
+        ),
+        (
+            "datafusion.optimizer.join_reordering",
+            "true",
+            "datafusion.optimizer.join_reordering=true",
+        ),
+    ] {
+        assert_variable(&client, name, expected, Some(hints)).await;
+    }
+
+    let (four, seven) = tokio::join!(
+        query(
+            &client,
+            "SHOW VARIABLES query.parallelism",
+            Some("query.parallelism=4")
+        ),
+        query(
+            &client,
+            "SHOW VARIABLES query.parallelism",
+            Some("query.parallelism=7")
+        ),
+    );
+    for (response, expected) in [(four, "4"), (seven, "7")] {
+        assert_eq!(response.status(), StatusCode::OK);
+        let body: Value = response.json().await;
+        assert_eq!(shown_value(&body), expected);
+    }
+    assert_variable(&client, "query.parallelism", &default_parallelism, None).await;
+
+    let response = query(
+        &client,
+        "SET query.parallelism=3; SHOW VARIABLES query.parallelism",
+        None,
+    )
+    .await;
+    assert_eq!(response.status(), StatusCode::OK);
+    let body: Value = response.json().await;
+    assert_eq!(body["output"].as_array().unwrap().len(), 2);
+    assert_eq!(shown_value(&body), "3");
+
+    let response = query(
+        &client,
+        "SET query.parallelism=3; SHOW VARIABLES query.parallelism",
+        Some("query.parallelism=4"),
+    )
+    .await;
+    assert_eq!(response.status(), StatusCode::OK);
+    let body: Value = response.json().await;
+    assert_eq!(body["output"].as_array().unwrap().len(), 2);
+    assert_eq!(shown_value(&body), "4");
+    assert_variable(&client, "query.parallelism", &default_parallelism, None).await;
+
+    for (headers, expected) in [
+        ("read_preference=leader", "LEADER"),
+        ("query.parallelism=4,read_preference=leader", "LEADER"),
+    ] {
+        assert_variable(&client, "read_preference", expected, Some(headers)).await;
+    }
+    let dedicated_read_preference = client
+        .post("/v1/sql")
+        .form(&[("sql", "SHOW VARIABLES read_preference")])
+        .header("x-greptime-read-preference", "leader")
+        .send()
+        .await;
+    assert_eq!(dedicated_read_preference.status(), StatusCode::OK);
+    let body: Value = dedicated_read_preference.json().await;
+    assert_eq!(shown_value(&body), "LEADER");
+
+    let invalid_read_preference = client
+        .post("/v1/sql")
+        .form(&[("sql", "SHOW VARIABLES read_preference")])
+        .header("x-greptime-read-preference", "follower")
+        .send()
+        .await;
+    assert_eq!(invalid_read_preference.status(), StatusCode::BAD_REQUEST);
+    let error: ErrorResponse = invalid_read_preference.json().await;
+    assert_eq!(error.code(), ErrorCode::InvalidArguments as u32);
+
+    for hints in ["query.parallelism=0", "query.unsupported_option=true"] {
+        let response = query(&client, "SHOW VARIABLES query.parallelism", Some(hints)).await;
+        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+        let error: ErrorResponse = response.json().await;
+        assert_eq!(error.code(), ErrorCode::InvalidArguments as u32);
+    }
+
+    guard.remove_all().await;
+}
+
 pub async fn test_sql_api(store_type: StorageType) {
     let (app, mut guard) = setup_test_http_app_with_frontend(store_type, "sql_api").await;
     let client = TestClient::new(app).await;