From 9d9039e07b305339e023d3311eef2da35e4ac063 Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Fri, 9 Oct 2026 16:10:55 +0800 Subject: [PATCH] feat: validate public query hints across protocols Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/servers/src/grpc/flight.rs | 168 ++++++++++++++++++-- src/servers/src/grpc/greptime_handler.rs | 57 ++++++- src/servers/src/hint_headers.rs | 161 ++++++++++++++++++- src/servers/src/http/hints.rs | 39 +++-- src/servers/src/http/read_preference.rs | 30 ++-- src/servers/tests/http/http_handler_test.rs | 7 + 6 files changed, 418 insertions(+), 44 deletions(-) diff --git a/src/servers/src/grpc/flight.rs b/src/servers/src/grpc/flight.rs index 874e98b7326..6a1c1a5d45e 100644 --- a/src/servers/src/grpc/flight.rs +++ b/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, ) -> TonicResult>> { - 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::() .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> { @@ -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, + } + + #[async_trait] + impl crate::query_handler::grpc::GrpcQueryHandler for FlightQueryHandler { + async fn do_query( + &self, + _query: greptime_request::Request, + _ctx: QueryContextRef, + ) -> Result { + 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> + Send>> { + Box::pin(stream::empty()) + } + } + + fn flight_handler(calls: Arc) -> 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(); diff --git a/src/servers/src/grpc/greptime_handler.rs b/src/servers/src/grpc/greptime_handler.rs index ad0a693ac59..e4210698035 100644 --- a/src/servers/src/grpc/greptime_handler.rs +++ b/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, +) -> Result { + 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, ) -> Result { 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(); diff --git a/src/servers/src/hint_headers.rs b/src/servers/src/hint_headers.rs index 3d13dd9bf5e..e782405ef9c 100644 --- a/src/servers/src/hint_headers.rs +++ b/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(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(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(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> { + use std::collections::HashMap; + + let mut normalized = Vec::with_capacity(hints.len()); + let mut seen = HashMap::::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::().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::().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; diff --git a/src/servers/src/http/hints.rs b/src/servers/src/http/hints.rs index 09e4879dd50..d070bd26d5c 100644 --- a/src/servers/src/http/hints.rs +++ b/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, 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::() { - 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::() { + 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")); } diff --git a/src/servers/src/http/read_preference.rs b/src/servers/src/http/read_preference.rs index 04e9771b890..e2f1c637c1d 100644 --- a/src/servers/src/http/read_preference.rs +++ b/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, 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::() { - 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::().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::() { + common_telemetry::debug!("Setting read preference to {}", read_preference); + query_ctx.set_read_preference(read_preference); + } } next.run(request).await } diff --git a/src/servers/tests/http/http_handler_test.rs b/src/servers/tests/http/http_handler_test.rs index 2c8064727a6..6a4e560c108 100644 --- a/src/servers/tests/http/http_handler_test.rs +++ b/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),