diff --git a/docs/rfcs/2026-09-11-metric-export-import.md b/docs/rfcs/2026-09-11-metric-export-import.md index 3d2c93e4695..8ddc31f40b0 100644 --- a/docs/rfcs/2026-09-11-metric-export-import.md +++ b/docs/rfcs/2026-09-11-metric-export-import.md @@ -390,14 +390,34 @@ validation, schema-option inheritance and target metadata resolution. Bound the request size and the importer's DDL buffering separately. The production CLI needs an authenticated server operation that checks CREATE -authorization for every member and reuses the ordinary DDL path. Transport, -capability discovery and request limits must be agreed before accepting this -interface. Initially bound a batch to 128 statements and 1 MiB of SQL, with an -explicit oversized-statement error; separately bound the importer's parsed DDL -buffer. A target without batch-DDL capability uses ordinary DDL if it supports +authorization for every member and reuses the ordinary DDL path. Bound the +importer's parsed DDL buffer separately from the server request limits below. +A target without batch-DDL capability uses ordinary DDL if it supports the snapshot's data layout. Do not retry an ambiguously completed batch through another path. +The server operation is `POST /v1/ddl/logical-tables` with an +`application/x-www-form-urlencoded` body containing `sql` and optional `db`. +`db` uses the same catalog/schema interpretation and defaults as `/v1/sql`. +Authenticated `GET /v1/capabilities` advertises `metric_batch_ddl: 1` when the +SQL handler supports the operation, independently of packed import/export. +The form body is limited to 4 MiB before decoding, or the configured HTTP body +limit when that limit is stricter. Decoded SQL is limited to 1 MiB and 1–128 +statements. Only explicit Metric logical CREATE TABLE statements in one +catalog/schema with one literal `on_physical_table` value are accepted; +duplicate normalized target names are rejected. + +All members pass SQL interception, authorization and ordinary CREATE preparation +before one logical-table batch procedure is submitted. Success uses the normal +GreptimeDB v1 response with one `affectedrows: 0` output per input statement in +order; this is CREATE's output, not a count of newly created tables. A failure +returns one error without a partial success list. Validation failures before +submission have no DDL side effects. Once submitted, the existing procedure may +have applied some or all changes even if the HTTP request times out or disconnects; +there is no cross-table rollback guarantee. `IF NOT EXISTS` keeps ordinary CREATE +semantics. The operation does not regroup requests or retry through single-table +CREATE. + ## Packed data COPY The CLI retains one `COPY DATABASE FROM` operation per `(chunk, schema)` task. diff --git a/src/frontend/AGENTS.md b/src/frontend/AGENTS.md index e16cbfc9872..bab27676655 100644 --- a/src/frontend/AGENTS.md +++ b/src/frontend/AGENTS.md @@ -39,6 +39,9 @@ remote datanodes via `operator`/`client`. - **SQL query** (`instance.rs`): `do_query_inner` handles parsing, interceptors, permission checks, timeout/cancellation, and delegates planning/execution to `StatementExecutor`. Distributed scans enter through `region_query.rs`. +- **Logical-table batch DDL** (`instance.rs`): dedicated SQL handler → bounded + parse/interception → all-member permissions → operator CREATE preparation → + one logical-table batch procedure. Keep pre-submission validation side-effect free. - **Insert** (`instance/grpc.rs`): `handle_inserts` / `handle_row_inserts` → `check_permission` → `operator`'s `Inserter` (schema validation, optional auto-create, partition routing, meter admission) → local `RegionServer` diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs index 6e531214127..d3e6f2addfc 100644 --- a/src/frontend/src/instance.rs +++ b/src/frontend/src/instance.rs @@ -782,6 +782,98 @@ impl Instance { query_interceptor.post_execute(Output { data, meta }, query_ctx) } + async fn batch_create_logical_tables( + &self, + query: &str, + query_ctx: QueryContextRef, + ) -> Result> { + use servers::query_handler::sql::{ + MAX_LOGICAL_TABLE_DDL_BYTES, MAX_LOGICAL_TABLE_DDL_STATEMENTS, + }; + ensure!(!self.is_suspended(), error::SuspendedSnafu); + ensure!( + query.len() <= MAX_LOGICAL_TABLE_DDL_BYTES, + InvalidSqlSnafu { + err_msg: "logical-table DDL exceeds 1 MiB" + } + ); + let interceptor = self.plugins.get::>(); + let interceptor = interceptor.as_ref(); + let query = interceptor.pre_parsing(query, query_ctx.clone())?; + ensure!( + query.len() <= MAX_LOGICAL_TABLE_DDL_BYTES, + InvalidSqlSnafu { + err_msg: "logical-table DDL exceeds 1 MiB" + } + ); + let statements = parse_stmt(query.as_ref(), query_ctx.sql_dialect()).map_err(|_| { + InvalidSqlSnafu { + err_msg: "invalid logical-table DDL", + } + .build() + })?; + ensure!( + !statements.is_empty() && statements.len() <= MAX_LOGICAL_TABLE_DDL_STATEMENTS, + InvalidSqlSnafu { + err_msg: "logical-table batch requires 1 to 128 statements" + } + ); + let original_statements = interceptor.map(|_| statements.clone()); + let statements = interceptor.post_parsing(statements, query_ctx.clone())?; + ensure!( + !statements.is_empty() && statements.len() <= MAX_LOGICAL_TABLE_DDL_STATEMENTS, + InvalidSqlSnafu { + err_msg: "logical-table batch requires 1 to 128 statements" + } + ); + let measure_rewritten_sql = || { + let mut remaining = MAX_LOGICAL_TABLE_DDL_BYTES - (statements.len() - 1); + statements.iter().all(|statement| { + matches!(statement, Statement::CreateTable(create) if create.consume_sql_size_budget(&mut remaining)) + }) + }; + ensure!( + original_statements + .as_ref() + .is_none_or(|original| original == &statements) + || measure_rewritten_sql(), + InvalidSqlSnafu { + err_msg: "invalid or oversized rewritten logical-table DDL" + } + ); + for statement in &statements { + ensure!( + matches!(statement, Statement::CreateTable(_)), + InvalidSqlSnafu { + err_msg: "batch requires explicit Metric logical CREATE TABLE statements" + } + ); + interceptor.pre_execute(Some(statement), None, query_ctx.clone())?; + } + for statement in &statements { + self.check_sql_permission(statement, &query_ctx).await?; + check_permission(self.plugins.clone(), statement, &query_ctx)?; + } + let count = statements.len(); + let creates = statements + .into_iter() + .filter_map(|statement| { + if let Statement::CreateTable(create) = statement { + Some(create) + } else { + None + } + }) + .collect(); + self.statement_executor + .batch_create_logical_tables(creates, query_ctx.clone()) + .await + .context(TableOperationSnafu)?; + (0..count) + .map(|_| interceptor.post_execute(Output::new_with_affected_rows(0), query_ctx.clone())) + .collect() + } + #[tracing::instrument(skip_all, name = "SqlQueryHandler::do_query")] async fn do_query_inner(&self, query: &str, query_ctx: QueryContextRef) -> Vec> { if self.is_suspended() { @@ -1113,6 +1205,21 @@ impl Instance { #[async_trait] impl SqlQueryHandler for Instance { + fn supports_metric_batch_ddl(&self) -> bool { + true + } + + async fn create_logical_tables( + &self, + query: &str, + query_ctx: QueryContextRef, + ) -> server_error::Result> { + self.batch_create_logical_tables(query, query_ctx) + .await + .map_err(BoxedError::new) + .context(ExecuteQuerySnafu) + } + async fn do_query( &self, query: &str, @@ -4145,6 +4252,214 @@ mod tests { Ok(()) } + #[tokio::test] + async fn logical_ddl_validates_and_authorizes_rewritten_batch_before_submission() + -> TestResult<()> { + struct Rewrite(&'static str); + impl SqlQueryInterceptor for Rewrite { + type Error = Error; + fn pre_parsing<'a>( + &self, + query: &'a str, + _: QueryContextRef, + ) -> Result> { + Ok(if self.0 == "sql_bytes" { + std::borrow::Cow::Owned(format!("{query}{}", " ".repeat(1024 * 1024 + 1))) + } else { + std::borrow::Cow::Borrowed(query) + }) + } + fn pre_execute( + &self, + statement: Option<&Statement>, + _: Option<&LogicalPlan>, + _: QueryContextRef, + ) -> Result<()> { + ensure!( + !(self.0 == "pre_execute" + && matches!(statement, + Some(Statement::CreateTable(create)) if create.name.to_string() == "allowed2")), + InvalidSqlSnafu { + err_msg: "second member rejected by interceptor" + } + ); + Ok(()) + } + fn post_parsing( + &self, + mut statements: Vec, + _: QueryContextRef, + ) -> Result> { + match self.0 { + "target" => { + statements[1] = parse_one_sql( + "CREATE TABLE denied (ts TIMESTAMP TIME INDEX) ENGINE=metric WITH (on_physical_table='phy')", + ) + } + "count" => statements = (0..129).map(|i| parse_one_sql(&format!( + "CREATE TABLE allowed_{i} (ts TIMESTAMP TIME INDEX) ENGINE=metric WITH (on_physical_table='phy')" + ))).collect(), + "type" => statements[1] = parse_one_sql("SELECT 1"), + "secret_bytes" | "array_bytes" | "small_option" => { + if let Statement::CreateTable(create) = &mut statements[1] { + let value = "x".repeat(if self.0 == "small_option" { 10 } else { 2 * 1024 * 1024 }); + if self.0 == "array_bytes" { + create.options.insert_options("non_scalar", vec![value.as_str()].into()); + } else { + create.options.insert("secret_access_key".to_string(), value); + } + } + } + "partition_shape" => { + if let Statement::CreateTable(create) = &mut statements[1] { + create.partitions = Some(sql::statements::create::Partitions { + column_list: vec![], + exprs: vec![sqlparser::ast::Expr::Identifier(sqlparser::ast::Ident::new("hidden"))], + }); + } + } + "vector_hidden" => { + statements[1] = parse_one_sql("CREATE TABLE allowed2 (ts TIMESTAMP TIME INDEX, v VECTOR(3)) ENGINE=metric WITH (on_physical_table='phy')"); + if let Statement::CreateTable(create) = &mut statements[1] { + create.columns[1].column_def.options.push(sqlparser::ast::ColumnOptionDef { + name: None, + option: sqlparser::ast::ColumnOption::Comment("private-value".repeat(200_000)), + }); + } + } + mode if mode.starts_with("fulltext_") || mode.starts_with("skipping_") + || mode.starts_with("inverted_") || mode.starts_with("vector_") => { + if let Statement::CreateTable(create) = &mut statements[1] { + let (kind, value_type) = mode.split_once('_').unwrap(); + let value = "x".repeat(if value_type == "small" { 10 } else { 2 * 1024 * 1024 }); + let extensions = &mut create.columns[1].extensions; + let options = match kind { + "fulltext" => &mut extensions.fulltext_index_options, + "skipping" => &mut extensions.skipping_index_options, + "inverted" => &mut extensions.inverted_index_options, + _ => &mut extensions.vector_options, + }.get_or_insert_default(); + if kind == "vector" { + options.insert("dim".to_string(), "3".to_string()); + } + if value_type == "array" { + options.insert_options("non_scalar", vec![value.as_str()].into()); + } else { + options.insert("secret_access_key".to_string(), value); + } + } + } + "physical_null" | "physical_array" => { + let value = if self.0 == "physical_null" { "NULL" } else { "['x']" }; + statements[1] = parse_one_sql(&format!( + "CREATE TABLE allowed2 (ts TIMESTAMP TIME INDEX) ENGINE=metric WITH (on_physical_table='phy', physical_metric_table={value})" + )); + } + "ast_bytes" => { + if let Statement::CreateTable(create) = &mut statements[1] { + create.name = + ObjectName(vec![sqlparser::ast::ObjectNamePart::Identifier( + sqlparser::ast::Ident::new("t".repeat(1024 * 1024)), + )]); + } + } + _ => {} + } + Ok(statements) + } + } + for mode in [ + "none", + "target", + "count", + "type", + "sql_bytes", + "ast_bytes", + "secret_bytes", + "array_bytes", + "small_option", + "physical_null", + "physical_array", + "fulltext_secret", + "fulltext_array", + "fulltext_small", + "skipping_secret", + "skipping_array", + "skipping_small", + "inverted_secret", + "inverted_array", + "inverted_small", + "vector_secret", + "vector_array", + "vector_small", + "vector_hidden", + "partition_shape", + "pre_execute", + "cross_catalog", + "valid", + ] { + let catalog = + catalog::memory::MemoryCatalogManager::new_with_table(test_table(1024, "source")?); + let executor = Arc::new(MockProcedureExecutor::new(catalog.clone())); + let plugins = Plugins::new(); + plugins.insert::(Arc::new(RejectUnresolvedPermissionChecker)); + plugins.insert::>(Arc::new(Rewrite(mode))); + plugins.insert(QueryOptions { + disallow_cross_catalog_query: mode == "cross_catalog", + }); + let instance = test_instance_with_catalog_manager( + catalog, + test_table(1025, "target")?, + plugins, + None, + executor.clone(), + ) + .await?; + let create = |name| { + format!( + "CREATE TABLE {name} (ts TIMESTAMP TIME INDEX, host STRING) ENGINE=metric WITH (on_physical_table='phy')" + ) + }; + let sql = format!( + "{};{}", + create("allowed"), + create(match mode { + "none" => "denied", + "cross_catalog" => "other.public.allowed2", + _ => "allowed2", + }) + ); + let result = + SqlQueryHandler::create_logical_tables(&instance, &sql, QueryContext::arc()).await; + let submits = matches!(mode, "valid" | "small_option") + || (mode.ends_with("_small") && !mode.starts_with("vector_")); + if matches!(mode, "none" | "target") { + let err = result.unwrap_err(); + assert_eq!( + err.status_code(), + StatusCode::PermissionDenied, + "{mode}: {err}" + ); + } else { + assert_eq!( + result.unwrap_err().status_code(), + match mode { + _ if submits => StatusCode::Unsupported, + "cross_catalog" => StatusCode::AccessDenied, + _ => StatusCode::InvalidArguments, + }, + "{mode}" + ); + } + assert_eq!( + executor.submitted.lock().unwrap().len(), + usize::from(submits), + "{mode}" + ); + } + Ok(()) + } + /// `DROP VIEW IF EXISTS` on a missing view through the direct gRPC ingress must /// succeed with 0 affected rows (no error, no DDL task submitted), instead of /// returning `TableNotFound`. diff --git a/src/operator/src/statement/ddl.rs b/src/operator/src/statement/ddl.rs index 2e3c88af463..fa9da9f322e 100644 --- a/src/operator/src/statement/ddl.rs +++ b/src/operator/src/statement/ddl.rs @@ -93,7 +93,9 @@ use sql::statements::create::{ }; use sql::statements::statement::Statement; use sqlparser::ast::{Expr, Ident, UnaryOperator, Value as ParserValue}; -use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME}; +use store_api::metric_engine_consts::{ + LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY, +}; use store_api::mito_engine_options::APPEND_MODE_KEY; use substrait::{DFLogicalSubstraitConvertor, SubstraitPlan}; use table::TableRef; @@ -341,7 +343,22 @@ impl StatementExecutor { #[tracing::instrument(skip_all)] pub async fn create_table(&self, stmt: CreateTable, ctx: QueryContextRef) -> Result { - let (catalog, schema, _table) = table_idents_to_full_name(&stmt.name, &ctx) + let mut create_expr = self.prepare_create_table(&stmt, &ctx).await?; + self.create_table_inner( + &mut create_expr, + stmt.partitions, + ctx, + TriggerReason::Manual, + ) + .await + } + + async fn prepare_create_table( + &self, + stmt: &CreateTable, + ctx: &QueryContextRef, + ) -> Result { + let (catalog, schema, _table) = table_idents_to_full_name(&stmt.name, ctx) .map_err(BoxedError::new) .context(error::ExternalSnafu)?; @@ -356,7 +373,7 @@ impl StatementExecutor { .context(TableMetadataManagerSnafu)? .map(|v| v.into_inner()); - let create_expr = &mut expr_helper::create_to_expr(&stmt, &ctx)?; + let mut create_expr = expr_helper::create_to_expr(stmt, ctx)?; // TTL and compaction options are handled separately; ingestion quota is database-only. if let Some(schema_options) = schema_options { for (key, value) in schema_options.extra_options.iter() { @@ -370,8 +387,72 @@ impl StatementExecutor { } } - self.create_table_inner(create_expr, stmt.partitions, ctx, TriggerReason::Manual) - .await + Ok(create_expr) + } + + /// Prepares every explicit logical CREATE before submitting one batch procedure. + pub async fn batch_create_logical_tables( + &self, + statements: Vec, + ctx: QueryContextRef, + ) -> Result<()> { + let mut expressions = Vec::with_capacity(statements.len()); + let mut targets = std::collections::HashSet::new(); + let mut group = None; + for statement in statements { + ensure!( + statement.engine == METRIC_ENGINE_NAME + && statement.options.get(LOGICAL_TABLE_METADATA_KEY).is_some() + && statement + .options + .value(PHYSICAL_TABLE_METADATA_KEY) + .is_none(), + InvalidSqlSnafu { + err_msg: "batch requires explicit Metric logical CREATE TABLE statements" + } + ); + let expr = self.prepare_create_table(&statement, &ctx).await?; + let key = ( + expr.catalog_name.clone(), + expr.schema_name.clone(), + expr.table_options.get(LOGICAL_TABLE_METADATA_KEY).cloned(), + ); + ensure!( + group.as_ref().is_none_or(|group| group == &key), + InvalidSqlSnafu { + err_msg: "batch must share catalog, schema and literal on_physical_table" + } + ); + group = Some(key); + ensure!( + targets.insert(expr.table_name.clone()), + InvalidSqlSnafu { + err_msg: "duplicate target in logical-table batch" + } + ); + self.validate_logical_create(&expr, statement.partitions.as_ref(), &ctx) + .await?; + expressions.push(expr); + } + self.create_logical_tables(&expressions, ctx, TriggerReason::Manual) + .await?; + Ok(()) + } + + async fn validate_logical_create( + &self, + expr: &CreateTableExpr, + partitions: Option<&Partitions>, + ctx: &QueryContextRef, + ) -> Result<()> { + ensure_table_definition_writable(&expr.schema_name, &expr.table_name)?; + if let Some(partitions) = partitions + && !partitions.exprs.is_empty() + { + self.validate_logical_table_partition_rule(expr, partitions, ctx) + .await?; + } + Ok(()) } #[tracing::instrument(skip_all)] @@ -472,12 +553,8 @@ impl StatementExecutor { .table_options .contains_key(LOGICAL_TABLE_METADATA_KEY) { - if let Some(partitions) = partitions.as_ref() - && !partitions.exprs.is_empty() - { - self.validate_logical_table_partition_rule(create_table, partitions, &query_ctx) - .await?; - } + self.validate_logical_create(create_table, partitions.as_ref(), &query_ctx) + .await?; // Create logical tables self.create_logical_tables( std::slice::from_ref(create_table), diff --git a/src/servers/src/error.rs b/src/servers/src/error.rs index af6b2499e7f..6e558c8234b 100644 --- a/src/servers/src/error.rs +++ b/src/servers/src/error.rs @@ -200,6 +200,9 @@ pub enum Error { location: Location, }, + #[snafu(display("Unsupported operation: {}", operation))] + UnsupportedOperation { operation: String }, + #[snafu(display("Invalid query: {}", reason))] InvalidQuery { reason: String, @@ -767,7 +770,7 @@ impl ErrorExt for Error { | InvalidPromRemoteReadQueryResult { .. } | OtlpMetricModeIncompatible { .. } => StatusCode::IllegalState, - UnsupportedDataType { .. } => StatusCode::Unsupported, + UnsupportedDataType { .. } | UnsupportedOperation { .. } => StatusCode::Unsupported, #[cfg(not(windows))] UpdateJemallocMetrics { .. } => StatusCode::Internal, @@ -941,7 +944,7 @@ impl From for Error { } } -fn log_error_if_necessary(error: &Error) { +pub(crate) fn log_error_if_necessary(error: &Error) { if error.status_code().should_log_error() { error!(error; "Failed to handle HTTP request "); } else { diff --git a/src/servers/src/http.rs b/src/servers/src/http.rs index 34f214bae26..4555b85ac69 100644 --- a/src/servers/src/http.rs +++ b/src/servers/src/http.rs @@ -681,7 +681,12 @@ impl HttpServerBuilder { } pub fn with_sql_handler(self, sql_handler: ServerSqlQueryHandlerRef) -> Self { - let sql_router = HttpServer::route_sql(ApiState { sql_handler }); + let batch_body_limit = if self.options.body_limit.0 == 0 { + 4 * 1024 * 1024 + } else { + (self.options.body_limit.0 as usize).min(4 * 1024 * 1024) + }; + let sql_router = HttpServer::route_sql(ApiState { sql_handler }, batch_body_limit); Self { router: self @@ -1390,16 +1395,24 @@ impl HttpServer { .with_state(log_state) } - fn route_sql(api_state: ApiState) -> Router { + fn route_sql(api_state: ApiState, batch_body_limit: usize) -> Router { Router::new() .route( "/capabilities", - routing::get(|| async { - axum::Json( - serde_json::json!({"metric_packed_import": 1, "metric_packed_export": 1}), - ) + routing::get(|State(state): State| async move { + let mut capabilities = + serde_json::json!({"metric_packed_import": 1, "metric_packed_export": 1}); + if state.sql_handler.supports_metric_batch_ddl() { + capabilities["metric_batch_ddl"] = serde_json::json!(1); + } + axum::Json(capabilities) }), ) + .route( + "/ddl/logical-tables", + routing::post(handler::create_logical_tables) + .layer(DefaultBodyLimit::max(batch_body_limit)), + ) .route( "/sql", routing::get(handler::sql).post(handler::sql).layer( @@ -1939,6 +1952,15 @@ mod test { client.get("/v1/capabilities").send().await.status(), StatusCode::UNAUTHORIZED ); + assert_eq!( + client + .post("/v1/ddl/logical-tables") + .form(&[("sql", "SELECT 1")]) + .send() + .await + .status(), + StatusCode::UNAUTHORIZED + ); let response = client .get("/v1/capabilities") .header("Authorization", "Basic dXNlcjpwYXNzd29yZA==") @@ -1952,6 +1974,84 @@ mod test { } } + #[tokio::test] + async fn logical_ddl_form_limits_and_unsupported_handler() { + use axum::body::{Body, to_bytes}; + use tower::ServiceExt; + + async fn post_form(app: Router, body: String) -> (StatusCode, serde_json::Value) { + let response = app + .oneshot( + Request::builder() + .method(Method::POST) + .uri("/v1/ddl/logical-tables") + .header("Content-Type", "application/x-www-form-urlencoded") + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let body = to_bytes(response.into_body(), 4096).await.unwrap(); + (status, serde_json::from_slice(&body).unwrap()) + } + + for limit in [0, 64] { + let (tx, _rx) = mpsc::channel(1); + let server = HttpServerBuilder::new(HttpOptions { + body_limit: ReadableSize(limit), + ..Default::default() + }) + .with_sql_handler(Arc::new(DummyInstance { _tx: tx })) + .build(); + let app = server.build(server.make_app()).unwrap(); + let client = TestClient::new(app.clone()).await; + assert_eq!( + client.get("/v1/ddl/logical-tables").send().await.status(), + StatusCode::METHOD_NOT_ALLOWED + ); + let small = client + .post("/v1/ddl/logical-tables") + .form(&[("sql", "CREATE TABLE t")]) + .send() + .await; + assert_eq!( + small.json::().await["code"], + common_error::status_code::StatusCode::Unsupported as u32 + ); + let (_, body) = post_form( + app.clone(), + if limit == 0 { + format!("sql={}", "%20".repeat(1024 * 1024)) + } else { + "sql=".to_string() + &" ".repeat(65) + }, + ) + .await; + let code = body["code"].clone(); + assert_eq!( + code, + if limit == 0 { + common_error::status_code::StatusCode::Unsupported as u32 + } else { + common_error::status_code::StatusCode::InvalidArguments as u32 + } + ); + for body in [ + format!("sql={}", "x".repeat(1024 * 1024 + 1)), + format!("sql={}", "%C3%A9".repeat(512 * 1024 + 1)), + format!("sql=CREATE+TABLE+t&padding={}", "x".repeat(4 * 1024 * 1024)), + ] { + let (status, body) = post_form(app.clone(), body).await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!( + body["code"], + common_error::status_code::StatusCode::InvalidArguments as u32 + ); + } + } + } + #[tokio::test] pub async fn test_extra_router_respects_api_surface() { // Downstream routes added through `with_extra_router()` (e.g. enterprise diff --git a/src/servers/src/http/handler.rs b/src/servers/src/http/handler.rs index bd31ee35ace..8cbfc25ff07 100644 --- a/src/servers/src/http/handler.rs +++ b/src/servers/src/http/handler.rs @@ -44,7 +44,10 @@ use sql::parser::{ParseOptions, ParserContext}; use sql::statements::statement::Statement; use tokio::sync::{Notify, watch}; -use crate::error::{CollectRecordbatchSnafu, FailedToParseQuerySnafu, InvalidQuerySnafu, Result}; +use crate::error::{ + CollectRecordbatchSnafu, FailedToParseQuerySnafu, InvalidQuerySnafu, Result, + log_error_if_necessary, +}; use crate::http::header::collect_plan_metrics; use crate::http::result::arrow_result::ArrowResponse; use crate::http::result::csv_result::CsvResponse; @@ -229,6 +232,83 @@ pub async fn sql( resp.with_execution_time(start.elapsed().as_millis() as u64) } +/// Form parameters for the bounded logical-table DDL operation. +#[derive(Deserialize)] +pub struct LogicalTableDdl { + sql: Option, + db: Option, +} + +/// Creates one batch of Metric logical tables using the normal v1 response. +pub async fn create_logical_tables( + State(state): State, + Extension(mut query_ctx): Extension, + form: std::result::Result, FormRejection>, +) -> HttpResponse { + let start = Instant::now(); + let form = match form { + Ok(Form(form)) => form, + Err(_) => { + return HttpResponse::Error(ErrorResponse::from_error_message( + StatusCode::InvalidArguments, + "invalid or oversized form body".to_string(), + )); + } + }; + let Some(sql) = form.sql else { + return HttpResponse::Error(ErrorResponse::from_error_message( + StatusCode::InvalidArguments, + "sql parameter is required.".to_string(), + )); + }; + if sql.len() > crate::query_handler::sql::MAX_LOGICAL_TABLE_DDL_BYTES { + return HttpResponse::Error(ErrorResponse::from_error_message( + StatusCode::InvalidArguments, + "logical-table DDL exceeds 1 MiB".to_string(), + )); + } + if let Some(db) = form.db { + let (catalog, schema) = parse_catalog_and_schema_from_db_string(&db); + query_ctx.set_current_catalog(&catalog); + query_ctx.set_current_schema(&schema); + } + query_ctx.set_channel(Channel::HttpSql); + let query_ctx = Arc::new(query_ctx); + if let Some((status, msg)) = validate_schema(state.sql_handler.clone(), query_ctx.clone()).await + { + return HttpResponse::Error(ErrorResponse::from_error_message(status, msg)); + } + let outputs = match state + .sql_handler + .create_logical_tables(&sql, query_ctx) + .await + { + Ok(outputs) + if outputs + .iter() + .all(|output| matches!(output.data, OutputData::AffectedRows(0))) => + { + outputs.into_iter().map(Ok).collect() + } + other => { + let status = match other { + Err(error) => { + log_error_if_necessary(&error); + error.status_code() + } + Ok(_) => StatusCode::Internal, + }; + return HttpResponse::Error( + ErrorResponse::from_error_message(status, "logical-table batch failed".to_string()) + .with_execution_time(start.elapsed().as_millis() as u64), + ); + } + }; + GreptimedbV1Response::from_output(outputs) + .await + .with_execution_time(start.elapsed().as_millis() as u64) +} + /// Limits each statement's response before collecting batches and converting rows. fn limit_output_rows(output: Output, limit: usize) -> Result { let mut remaining = limit; @@ -994,13 +1074,16 @@ mod tests { use std::sync::atomic::AtomicUsize; use async_trait::async_trait; + use common_error::ext::{BoxedError, PlainError}; use common_query::OutputMeta; + use common_telemetry::tracing::instrument::WithSubscriber; use datafusion_expr::LogicalPlan; use datatypes::prelude::ConcreteDataType; use datatypes::schema::{ColumnSchema, Schema}; use datatypes::vectors::{BinaryVector, UInt32Vector, VectorRef}; use futures::stream; use query::query_engine::DescribeResult; + use snafu::IntoError; use super::*; use crate::query_handler::sql::SqlQueryHandler; @@ -1015,6 +1098,12 @@ mod tests { std::mem::take(&mut *self.outputs.lock().unwrap()) } + async fn create_logical_tables(&self, _: &str, _: QueryContextRef) -> Result> { + std::mem::take(&mut *self.outputs.lock().unwrap()) + .into_iter() + .collect() + } + async fn do_analyze_stream_query(&self, _: &str, _: QueryContextRef) -> Result { unimplemented!() } @@ -1068,6 +1157,61 @@ mod tests { .await } + #[tokio::test] + async fn test_logical_ddl_logs_diagnostics_without_exposing_sql_or_response_details() { + let sql = "CREATE TABLE logs (ts TIMESTAMP TIME INDEX) ENGINE=metric WITH (on_physical_table='request-only-value')"; + for (status, level) in [ + (StatusCode::InvalidArguments, "WARN"), + (StatusCode::Internal, "ERROR"), + ] { + let logs = tempfile::NamedTempFile::new().unwrap(); + let writer = logs.reopen().unwrap(); + let subscriber = common_telemetry::tracing_subscriber::fmt() + .without_time() + .with_ansi(false) + .with_writer(move || writer.try_clone().unwrap()) + .finish(); + let error = crate::error::ExecuteQuerySnafu.into_error(BoxedError::new( + PlainError::new("sensitive-option-value".to_string(), status), + )); + let response = create_logical_tables( + State(ApiState { + sql_handler: Arc::new(TestSqlQueryHandler { + outputs: Mutex::new(vec![Err(error)]), + }), + }), + Extension(QueryContext::with_db_name(None)), + Ok(Form(LogicalTableDdl { + sql: Some(sql.to_string()), + db: None, + })), + ) + .with_subscriber(subscriber) + .await + .into_response(); + let diagnostics = std::fs::read_to_string(logs.path()).unwrap(); + assert!(diagnostics.contains(level), "{diagnostics}"); + assert!( + diagnostics.contains("Failed to execute query"), + "{diagnostics}" + ); + assert!( + diagnostics.contains("sensitive-option-value"), + "{diagnostics}" + ); + assert!(!diagnostics.contains(sql), "{diagnostics}"); + assert!(!diagnostics.contains("request-only-value"), "{diagnostics}"); + assert!(!format!("{:?}", response.headers()).contains("sensitive-option-value")); + let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let body: Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(body["code"], status as u32); + assert_eq!(body["error"], "logical-table batch failed"); + assert!(body.get("output").is_none()); + } + } + fn number_batches() -> RecordBatches { let schema = Arc::new(Schema::new(vec![ColumnSchema::new( "number", diff --git a/src/servers/src/query_handler/sql.rs b/src/servers/src/query_handler/sql.rs index 9bf99a44441..1b65ff002db 100644 --- a/src/servers/src/query_handler/sql.rs +++ b/src/servers/src/query_handler/sql.rs @@ -26,8 +26,30 @@ use crate::error::Result; pub type ServerSqlQueryHandlerRef = Arc; +/// Maximum decoded SQL size accepted by the logical-table batch operation. +pub const MAX_LOGICAL_TABLE_DDL_BYTES: usize = 1024 * 1024; +/// Maximum number of CREATE statements in one logical-table batch. +pub const MAX_LOGICAL_TABLE_DDL_STATEMENTS: usize = 128; + #[async_trait] pub trait SqlQueryHandler { + /// Whether this handler implements the bounded logical-table DDL operation. + fn supports_metric_batch_ddl(&self) -> bool { + false + } + + /// Validates and authorizes the whole batch before submitting one procedure. + async fn create_logical_tables( + &self, + _query: &str, + _query_ctx: QueryContextRef, + ) -> Result> { + crate::error::UnsupportedOperationSnafu { + operation: "metric batch DDL", + } + .fail() + } + async fn do_query(&self, query: &str, query_ctx: QueryContextRef) -> Vec>; /// Executes an HTTP analyze-stream query. diff --git a/src/sql/src/statements/create.rs b/src/sql/src/statements/create.rs index 2cb78431969..725743ee679 100644 --- a/src/sql/src/statements/create.rs +++ b/src/sql/src/statements/create.rs @@ -13,7 +13,7 @@ // limitations under the License. use std::collections::HashMap; -use std::fmt::{Display, Formatter}; +use std::fmt::{Display, Formatter, Write}; use common_catalog::consts::FILE_ENGINE; use datatypes::json::{JSON2_DEFAULT_MAX_AUTO_EXPANDED_PATHS, JsonSettings}; @@ -68,6 +68,36 @@ fn format_table_constraint(constraints: &[TableConstraint]) -> String { constraints.iter().map(|c| format_indent!(c)).join(LINE_SEP) } +fn write_list( + f: &mut dyn Write, + values: &[T], + separator: &str, + prefix: &str, +) -> std::fmt::Result { + for (index, value) in values.iter().enumerate() { + if index > 0 { + f.write_str(separator)?; + } + write!(f, "{prefix}{value}")?; + } + Ok(()) +} + +// sqlparser's Custom Display joins all modifiers into an intermediate String. +fn write_data_type(f: &mut dyn Write, data_type: &DataType) -> std::fmt::Result { + if let DataType::Custom(name, modifiers) = data_type { + write!(f, "{name}")?; + if !modifiers.is_empty() { + f.write_char('(')?; + write_list(f, modifiers, COMMA_SEP, "")?; + f.write_char(')')?; + } + Ok(()) + } else { + write!(f, "{data_type}") + } +} + /// Table constraint for create table statement. #[derive(Debug, PartialEq, Eq, Clone, Visit, VisitMut, Serialize)] pub enum TableConstraint { @@ -81,7 +111,9 @@ impl Display for TableConstraint { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { match self { TableConstraint::PrimaryKey { columns } => { - write!(f, "PRIMARY KEY ({})", format_list_comma!(columns)) + f.write_str("PRIMARY KEY (")?; + write_list(f, columns, COMMA_SEP, "")?; + f.write_str(")") } TableConstraint::TimeIndex { column } => { write!(f, "TIME INDEX ({})", column) @@ -143,12 +175,34 @@ pub struct Json2Options { impl Display for Json2Options { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - let mut options = Vec::with_capacity(self.type_hints.len() + 1); + f.write_str("(\n ")?; if let Some(max) = self.max_auto_expanded_paths { - options.push(format!("max_auto_expanded_paths = {max}")); + write!(f, "max_auto_expanded_paths = {max}")?; } - options.extend(self.type_hints.iter().map(format_json_type_hint)); - write!(f, "(\n {}\n )", options.iter().join(",\n ")) + for (index, hint) in self.type_hints.iter().enumerate() { + if index > 0 || self.max_auto_expanded_paths.is_some() { + f.write_str(",\n ")?; + } + for (index, segment) in hint.path.iter().enumerate() { + if index > 0 { + f.write_char('.')?; + } + f.write_char('"')?; + for part in segment.split_inclusive('"') { + f.write_str(part)?; + if part.ends_with('"') { + f.write_char('"')?; + } + } + f.write_char('"')?; + } + f.write_char(' ')?; + write_data_type(f, &hint.data_type)?; + if hint.inverted_index { + f.write_str(" INVERTED INDEX")?; + } + } + f.write_str("\n )") } } @@ -203,14 +257,20 @@ impl Column { impl Display for Column { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - if let Some(vector_options) = &self.extensions.vector_options + self.write_sql(f, false) + } +} + +impl Column { + fn write_sql(&self, f: &mut dyn Write, complete: bool) -> std::fmt::Result { + if !complete + && let Some(vector_options) = &self.extensions.vector_options && let Some(dim) = vector_options.get(VECTOR_OPT_DIM) { - write!(f, "{} VECTOR({})", self.column_def.name, dim)?; - return Ok(()); + return write!(f, "{} VECTOR({})", self.column_def.name, dim); } - - write!(f, "{} {}", self.column_def.name, self.column_def.data_type)?; + write!(f, "{} ", self.column_def.name)?; + write_data_type(f, &self.column_def.data_type)?; if let Some(options) = &self.extensions.json2_options { write!(f, "{options}")?; } @@ -218,30 +278,18 @@ impl Display for Column { write!(f, " {option}")?; } - if let Some(fulltext_options) = &self.extensions.fulltext_index_options { - if !fulltext_options.is_empty() { - let options = fulltext_options.kv_pairs(); - write!(f, " FULLTEXT INDEX WITH({})", format_list_comma!(options))?; - } else { - write!(f, " FULLTEXT INDEX")?; - } - } - - if let Some(skipping_index_options) = &self.extensions.skipping_index_options { - if !skipping_index_options.is_empty() { - let options = skipping_index_options.kv_pairs(); - write!(f, " SKIPPING INDEX WITH({})", format_list_comma!(options))?; - } else { - write!(f, " SKIPPING INDEX")?; - } - } - - if let Some(inverted_index_options) = &self.extensions.inverted_index_options { - if !inverted_index_options.is_empty() { - let options = inverted_index_options.kv_pairs(); - write!(f, " INVERTED INDEX WITH({})", format_list_comma!(options))?; - } else { - write!(f, " INVERTED INDEX")?; + for (kind, options) in [ + ("FULLTEXT", &self.extensions.fulltext_index_options), + ("SKIPPING", &self.extensions.skipping_index_options), + ("INVERTED", &self.extensions.inverted_index_options), + ] { + if let Some(options) = options { + write!(f, " {kind} INDEX")?; + if !options.is_empty() { + f.write_str(" WITH(")?; + options.write_sql(f, COMMA_SEP, "", complete)?; + f.write_str(")")?; + } } } @@ -317,24 +365,6 @@ fn json_type_hint_sql_data_type(data_type: &ConcreteDataType) -> Result String { - let path = hint - .path - .iter() - .map(|segment| format_json_path_segment(segment)) - .join("."); - let inverted_index = if hint.inverted_index { - " INVERTED INDEX" - } else { - "" - }; - format!("{} {}{}", path, hint.data_type, inverted_index) -} - -fn format_json_path_segment(segment: &str) -> String { - format!("\"{}\"", segment.replace('"', "\"\"")) -} - /// Partition on columns or values. /// /// - `column_list` is the list of columns in `PARTITION ON COLUMNS` clause. @@ -359,12 +389,11 @@ impl Partitions { impl Display for Partitions { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { if !self.column_list.is_empty() { - write!( - f, - "PARTITION ON COLUMNS ({}) (\n{}\n)", - format_list_comma!(self.column_list), - format_list_indent!(self.exprs), - )?; + f.write_str("PARTITION ON COLUMNS (")?; + write_list(f, &self.column_list, COMMA_SEP, "")?; + f.write_str(") (\n")?; + write_list(f, &self.exprs, LINE_SEP, " ")?; + f.write_str("\n)")?; } Ok(()) } @@ -372,6 +401,53 @@ impl Display for Partitions { impl Display for CreateTable { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + self.write_sql(f, false) + } +} + +impl CreateTable { + /// Debits the SQL size of a rewritten AST, including unredacted and non-scalar + /// options. Returns false on exhaustion or a shape hidden by normal Display. + /// No SQL or secret value is returned. Delegated sqlparser formatters retain + /// their own allocation behavior; this is not a general AST memory limit. + pub fn consume_sql_size_budget(&self, remaining: &mut usize) -> bool { + struct Budget<'a>(&'a mut usize); + impl Write for Budget<'_> { + fn write_str(&mut self, value: &str) -> std::fmt::Result { + *self.0 = self.0.checked_sub(value.len()).ok_or(std::fmt::Error)?; + Ok(()) + } + } + if self + .partitions + .as_ref() + .is_some_and(|p| p.column_list.is_empty() && !p.exprs.is_empty()) + { + return false; + } + for column in &self.columns { + if let Some(options) = &column.extensions.vector_options { + let DataType::Custom(name, tokens) = &column.column_def.data_type else { + return false; + }; + if options.len() != 1 + || name.0.len() != 1 + || tokens.len() != 1 + || !name.0[0] + .as_ident() + .is_some_and(|name| name.value.eq_ignore_ascii_case("VECTOR")) + || !tokens[0].parse::().ok().is_some_and(|dim| { + options.get(VECTOR_OPT_DIM) == Some(dim.to_string().as_str()) + }) + { + return false; + } + } + } + self.write_sql(&mut Budget(remaining), true).is_ok() + } + + fn write_sql(&self, f: &mut dyn Write, complete: bool) -> std::fmt::Result { write!(f, "CREATE ")?; if self.engine == FILE_ENGINE { write!(f, "EXTERNAL ")?; @@ -381,16 +457,25 @@ impl Display for CreateTable { write!(f, "IF NOT EXISTS ")?; } writeln!(f, "{} (", &self.name)?; - writeln!(f, "{},", format_list_indent!(self.columns))?; - writeln!(f, "{}", format_table_constraint(&self.constraints))?; + for (index, column) in self.columns.iter().enumerate() { + if index > 0 { + f.write_str(LINE_SEP)?; + } + f.write_str(" ")?; + column.write_sql(f, complete)?; + } + f.write_str(",\n")?; + write_list(f, &self.constraints, LINE_SEP, " ")?; + f.write_char('\n')?; writeln!(f, ")")?; if let Some(partitions) = &self.partitions { writeln!(f, "{partitions}")?; } writeln!(f, "ENGINE={}", &self.engine)?; if !self.options.is_empty() { - let options = self.options.kv_pairs(); - write!(f, "WITH(\n{}\n)", format_list_indent!(options))?; + f.write_str("WITH(\n")?; + self.options.write_sql(f, LINE_SEP, " ", complete)?; + f.write_str("\n)")?; } Ok(()) } @@ -623,6 +708,120 @@ mod tests { use crate::parser::{ParseOptions, ParserContext}; use crate::statements::statement::Statement; + #[test] + fn test_rewritten_create_sql_budget() { + let parse = |sql| { + let mut statements = ParserContext::create_with_dialect( + sql, + &GreptimeDbDialect {}, + ParseOptions::default(), + ) + .unwrap(); + let Statement::CreateTable(create) = statements.remove(0) else { + panic!("CREATE expected") + }; + create + }; + let create = parse( + "CREATE TABLE demo (ts TIMESTAMP TIME INDEX, host STRING) ENGINE=metric WITH (on_physical_table='phy')", + ); + let size = create.to_string().len(); + let mut remaining = size; + assert!(create.consume_sql_size_budget(&mut remaining)); + assert_eq!(remaining, 0); + assert!(!create.consume_sql_size_budget(&mut (size - 1))); + + for location in ["table", "fulltext", "skipping", "inverted"] { + for value_type in ["secret", "array", "struct"] { + for length in [10, 8192] { + let mut create = create.clone(); + let extensions = &mut create.columns[1].extensions; + let map = match location { + "table" => &mut create.options, + "fulltext" => extensions.fulltext_index_options.get_or_insert_default(), + "skipping" => extensions.skipping_index_options.get_or_insert_default(), + _ => extensions.inverted_index_options.get_or_insert_default(), + }; + let value = "private-value".repeat(length); + match value_type { + "secret" => map.insert("secret_access_key".into(), value.clone()), + "array" => map.insert_options("payload", vec![value.as_str()].into()), + _ => map.insert_options( + "payload", + crate::util::OptionValue::try_new(Expr::Struct { + values: vec![Expr::Value( + sqlparser::ast::Value::SingleQuotedString(value.clone()).into(), + )], + fields: vec![], + }) + .unwrap(), + ), + } + assert!(!create.to_string().contains(&value)); + assert_eq!( + create.consume_sql_size_budget(&mut 4096), + length == 10, + "{location}/{value_type}" + ); + } + } + } + let mut vector = parse("CREATE TABLE demo (ts TIMESTAMP TIME INDEX, v VECTOR(3))"); + vector.columns[1].column_def.options.push(ColumnOptionDef { + name: None, + option: sqlparser::ast::ColumnOption::Comment("private-value".repeat(8192)), + }); + assert!(!vector.to_string().contains("private-value")); + assert!(!vector.consume_sql_size_budget(&mut 4096)); + vector.columns[1].column_def.options.clear(); + assert!(vector.consume_sql_size_budget(&mut 4096)); + if let DataType::Custom(_, tokens) = &mut vector.columns[1].column_def.data_type { + tokens[0] = format!("{}3", "0".repeat(8192)); + } + assert!(!vector.consume_sql_size_budget(&mut 4096)); + vector.columns[1] + .extensions + .vector_options + .as_mut() + .unwrap() + .insert("extra".into(), "x".into()); + assert!(!vector.consume_sql_size_budget(&mut 4096)); + + let mut create = create; + create.columns[1].extensions.json2_options = Some(Json2Options { + max_auto_expanded_paths: Some(1), + type_hints: vec![JsonTypeHint { + path: vec!["x".repeat(8192)], + data_type: DataType::Text, + inverted_index: true, + }], + }); + assert!(!create.consume_sql_size_budget(&mut 4096)); + create.columns[1].extensions.json2_options = None; + create.partitions = Some(Partitions { + column_list: vec![], + exprs: vec![Expr::Identifier(Ident::new("hidden"))], + }); + assert!(!create.consume_sql_size_budget(&mut 4096)); + } + + #[test] + fn test_json2_streamed_quote_escaping() { + let path = "a\\\"\"b"; + let options = Json2Options { + max_auto_expanded_paths: None, + type_hints: vec![JsonTypeHint { + path: vec![path.into()], + data_type: DataType::Text, + inverted_index: false, + }], + }; + assert_eq!( + options.to_string(), + format!("(\n \"{}\" TEXT\n )", path.replace('"', "\"\"")) + ); + } + #[test] fn test_display_create_table() { let sql = r"create table if not exists demo( diff --git a/src/sql/src/statements/option_map.rs b/src/sql/src/statements/option_map.rs index d562a61ce04..5077088b67a 100644 --- a/src/sql/src/statements/option_map.rs +++ b/src/sql/src/statements/option_map.rs @@ -139,26 +139,44 @@ impl OptionMap { } pub fn kv_pairs(&self) -> Vec { - let mut result = Vec::with_capacity(self.options.len() + self.secrets.len()); - for (k, v) in self - .options - .iter() - .filter_map(|(k, v)| v.as_string().map(|v| (k, v))) - { - if k.contains(".") { - result.push(format!("'{k}' = '{}'", v.escape_debug())); - } else { - result.push(format!("{k} = '{}'", v.escape_debug())); + self.sql_pairs(false).map(|pair| pair.to_string()).collect() + } + + /// Streams the same pairs as `kv_pairs`, or all actual values for size validation. + pub(crate) fn write_sql( + &self, + f: &mut dyn std::fmt::Write, + separator: &str, + prefix: &str, + complete: bool, + ) -> std::fmt::Result { + for (index, pair) in self.sql_pairs(complete).enumerate() { + if index > 0 { + f.write_str(separator)?; } + write!(f, "{prefix}{pair}")?; } - for (k, _) in self.secrets.iter() { - if k.contains(".") { - result.push(format!("'{k}' = '******'")); + Ok(()) + } + + fn sql_pairs(&self, complete: bool) -> impl Iterator> { + let options = self.options.iter().filter_map(move |(key, value)| { + let value = match value.as_string() { + Some(value) => Either::Right(value), + None if complete => Either::Left(value), + None => return None, + }; + Some(SqlPair { key, value }) + }); + let secrets = self.secrets.iter().map(move |(key, value)| SqlPair { + key, + value: Either::Right(if complete { + value.expose_secret() } else { - result.push(format!("{k} = '******'")); - } - } - result + "******" + }), + }); + options.chain(secrets) } pub fn entries(&self) -> impl Iterator)> { @@ -174,6 +192,25 @@ impl OptionMap { } } +struct SqlPair<'a> { + key: &'a str, + value: Either<&'a OptionValue, &'a str>, +} + +impl std::fmt::Display for SqlPair<'_> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + if self.key.contains('.') { + write!(f, "'{}'", self.key)?; + } else { + f.write_str(self.key)?; + } + match self.value { + Either::Left(value) => write!(f, " = {value}"), + Either::Right(value) => write!(f, " = '{}'", value.escape_debug()), + } + } +} + impl> From for OptionMap { fn from(value: I) -> Self { let mut result = OptionMap::default(); diff --git a/src/sql/src/util.rs b/src/sql/src/util.rs index c0c237cb0c0..a33c2576bbe 100644 --- a/src/sql/src/util.rs +++ b/src/sql/src/util.rs @@ -16,7 +16,6 @@ use std::collections::HashSet; use std::fmt::{Display, Formatter}; use std::ops::ControlFlow; -use itertools::Itertools; use promql_parser::label::{METRIC_NAME, MatchOp}; use promql_parser::parser::{ Expr as PromExpr, MatrixSelector as PromMatrixSelector, VectorSelector as PromVectorSelector, @@ -161,12 +160,20 @@ impl Display for OptionValue { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { if let Some(s) = self.as_string() { write!(f, "'{s}'") - } else if let Some(s) = self.as_list() { - write!( - f, - "[{}]", - s.into_iter().map(|x| format!("'{x}'")).join(", ") - ) + } else if let Expr::Array(array) = &self.0 + && array + .elem + .iter() + .all(|expr| Self::expr_as_string(expr).is_some()) + { + f.write_str("[")?; + for (index, expr) in array.elem.iter().enumerate() { + if index > 0 { + f.write_str(", ")?; + } + write!(f, "'{}'", Self::expr_as_string(expr).unwrap())?; + } + f.write_str("]") } else { write!(f, "'{}'", self.0) } @@ -667,6 +674,7 @@ fn extract_tables_from_sql_query(query: &sqlparser::ast::Query, names: &mut Hash #[cfg(test)] mod tests { + use itertools::Itertools; use sqlparser::tokenizer::Token; use super::*; diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index eff6a8685f2..54a4e2d67b8 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -517,6 +517,459 @@ async fn test_sql_skip_wal(distributed: bool) { check_http_skip_wal("sql", &cases, distributed).await; } +fn logical_ddl(name: &str) -> String { + format!( + "CREATE TABLE IF NOT EXISTS {name} (ts TIMESTAMP TIME INDEX, host STRING PRIMARY KEY, val DOUBLE) ENGINE=metric WITH (on_physical_table='phy.dot')" + ) +} + +#[tokio::test(flavor = "multi_thread")] +async fn logical_ddl_http_standalone() { + check_logical_ddl_http(false).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn logical_ddl_http_distributed() { + check_logical_ddl_http(true).await; +} + +#[tokio::test(flavor = "multi_thread")] +#[ignore = "manual logical DDL throughput measurement"] +async fn logical_ddl_http_benchmark() { + common_telemetry::init_default_ut_logging(); + let count: usize = std::env::var("GT_LOGICAL_DDL_BENCH_TABLES") + .unwrap_or_else(|_| "1000".to_string()) + .parse() + .unwrap(); + for (name, endpoint, batch_size) in [ + ("single", "/v1/sql", 1), + ("multi_sql", "/v1/sql", 128), + ("batch", "/v1/ddl/logical-tables", 128), + ] { + let mut instance = + MockInstanceImpl::Standalone( + tests_integration::standalone::GreptimeDbStandaloneBuilder::new(&format!( + "logical_ddl_bench_{name}" + )) + .with_event_recorder_options( + tests_integration::test_util::test_event_recorder_options(), + ) + .build() + .await, + ); + let server = HttpServerBuilder::new(HttpOptions::default()) + .with_sql_handler(instance.frontend()) + .build(); + let client = TestClient::new(server.build(server.make_app()).unwrap()).await; + let response = execute_sql(&client, "CREATE TABLE \"phy.dot\" (ts TIMESTAMP TIME INDEX, host STRING PRIMARY KEY, val DOUBLE) ENGINE=metric WITH (physical_metric_table='')").await; + assert!(response.status().is_success(), "{}", response.text().await); + let statements = (0..count) + .map(|i| logical_ddl(&format!("bench_{i}"))) + .collect::>(); + let start = std::time::Instant::now(); + for chunk in statements.chunks(batch_size) { + let response = client + .post(endpoint) + .form(&[("sql", chunk.join(";"))]) + .send() + .await; + assert!(response.status().is_success(), "{}", response.text().await); + assert_eq!( + response.json::().await["output"] + .as_array() + .unwrap() + .len(), + chunk.len() + ); + } + let elapsed = start.elapsed(); + let requests = count.div_ceil(batch_size); + let procedures = if name == "batch" { requests } else { count }; + wait_for_event_data(&client, + "SELECT count(DISTINCT procedure_id), count(DISTINCT table_name) FROM greptime_private.events WHERE type='create_logical_tables' AND json_get_string(procedure_trigger, 'type')='Submitted' AND table_name LIKE 'bench_%'", + &format!("[[{procedures},{count}]]")).await; + common_telemetry::info!( + "Logical DDL: mode={name}, tables={count}, concurrency=1, HTTP requests={requests}, procedures={procedures}, elapsed={elapsed:?}" + ); + instance.shutdown().await; + } +} + +async fn check_logical_ddl_http(distributed: bool) { + common_telemetry::init_default_ut_logging(); + let events = tests_integration::test_util::test_event_recorder_options(); + let mut instance = if distributed { + MockInstanceImpl::Distributed( + tests_integration::cluster::GreptimeDbClusterBuilder::new("batch_logical_ddl") + .await + .with_event_recorder_options(events) + .with_datanodes(3) + .build(true) + .await, + ) + } else { + MockInstanceImpl::Standalone( + tests_integration::standalone::GreptimeDbStandaloneBuilder::new("batch_logical_ddl") + .with_event_recorder_options(events) + .build() + .await, + ) + }; + let frontend = instance.frontend(); + let server = HttpServerBuilder::new(HttpOptions::default()) + .with_sql_handler(frontend.clone()) + .build(); + let client = TestClient::new(server.build(server.make_app()).unwrap()).await; + assert_eq!( + client + .get("/v1/capabilities") + .send() + .await + .json::() + .await["metric_batch_ddl"], + 1 + ); + let response = execute_sql(&client, "CREATE TABLE \"phy.dot\" (ts TIMESTAMP TIME INDEX, host STRING PRIMARY KEY, val DOUBLE) ENGINE=metric WITH (physical_metric_table='')").await; + assert!(response.status().is_success(), "{}", response.text().await); + + for sql in [ + String::new(), + format!("{}; SELECT 1", logical_ddl("rejected")), + format!("{}; CREATE VIEW v AS SELECT 1", logical_ddl("rejected")), + format!( + "{}; CREATE TABLE copy LIKE batch_0", + logical_ddl("rejected") + ), + format!("{}; CREATE TABLE copy AS SELECT 1", logical_ddl("rejected")), + logical_ddl("rejected").replace("WITH (", "WITH (physical_metric_table='', "), + format!( + "{}; CREATE TABLE ordinary (ts TIMESTAMP TIME INDEX)", + logical_ddl("rejected") + ), + format!( + "{};{}", + logical_ddl("rejected"), + logical_ddl("public.rejected") + ), + format!("{};{}", logical_ddl("rejected"), logical_ddl("other.t")), + format!( + "{};{}", + logical_ddl("rejected"), + logical_ddl("other_physical").replace("phy.dot", "other") + ), + (0..129) + .map(|i| logical_ddl(&format!("rejected_{i}"))) + .collect::>() + .join(";"), + ] { + let response = client + .post("/v1/ddl/logical-tables") + .form(&[("sql", sql)]) + .send() + .await; + assert!(!response.status().is_success(), "{}", response.text().await); + } + validate_data( + "batch_rejected", + &client, + "SELECT count(*) FROM information_schema.tables WHERE table_name LIKE 'rejected%'", + "[[0]]", + ) + .await; + + let sql = (0..128) + .map(|i| logical_ddl(&format!("batch_{i}"))) + .collect::>() + .join(";"); + let response = client + .post("/v1/ddl/logical-tables") + .form(&[("sql", sql)]) + .send() + .await; + assert!(response.status().is_success(), "{}", response.text().await); + assert_eq!( + response.json::().await["output"], + json!(vec![json!({"affectedrows": 0}); 128]) + ); + wait_for_event_data(&client, + "SELECT count(DISTINCT procedure_id), count(DISTINCT table_name) FROM greptime_private.events WHERE type='create_logical_tables' AND json_get_string(procedure_trigger, 'type')='Submitted' AND table_name LIKE 'batch_%'", + "[[1,128]]").await; + + let physical_id = frontend + .catalog_manager() + .table("greptime", "public", "phy.dot", None) + .await + .unwrap() + .unwrap() + .table_info() + .table_id(); + let mut ids = std::collections::HashSet::new(); + for i in 0..128 { + let table = frontend + .catalog_manager() + .table("greptime", "public", &format!("batch_{i}"), None) + .await + .unwrap() + .unwrap(); + let id = table.table_info().table_id(); + assert!(ids.insert(id)); + assert_ne!(id, physical_id); + let bound_physical_id = frontend + .table_metadata_manager() + .table_route_manager() + .get_physical_table_id(id) + .await + .unwrap(); + assert_eq!(bound_physical_id, physical_id); + assert_eq!( + table + .table_info() + .meta + .options + .extra_options + .get("on_physical_table") + .map(String::as_str), + Some("phy.dot") + ); + assert_eq!(table.schema().column_schemas().len(), 3); + } + let sql = logical_ddl("one_at_byte_limit"); + let padded = format!("{sql}{}", " ".repeat(1024 * 1024 - sql.len())); + let response = client + .post("/v1/ddl/logical-tables") + .form(&[("sql", padded)]) + .send() + .await; + assert!(response.status().is_success(), "{}", response.text().await); + + let response = execute_sql(&client, "CREATE DATABASE inherited WITH ('skip_wal'='true', ttl='1h', 'compaction.type'='twcs'); CREATE TABLE inherited.phy (ts TIMESTAMP TIME INDEX, host STRING PRIMARY KEY, val DOUBLE) ENGINE=metric WITH (physical_metric_table='')").await; + assert!(response.status().is_success(), "{}", response.text().await); + let inherited = logical_ddl("inherited.ordinary").replace("phy.dot", "phy"); + let response = execute_sql(&client, &inherited).await; + assert!(response.status().is_success(), "{}", response.text().await); + let sql = format!( + "{};{}", + logical_ddl("batched").replace("phy.dot", "phy"), + logical_ddl("explicit") + .replace("phy.dot", "phy") + .replace("WITH (", "WITH (skip_wal='false', ") + ); + let response = client + .post("/v1/ddl/logical-tables") + .form(&[("sql", sql.as_str()), ("db", "greptime-inherited")]) + .send() + .await; + assert!(response.status().is_success(), "{}", response.text().await); + let catalog = frontend.catalog_manager(); + let ordinary = catalog + .table("greptime", "inherited", "ordinary", None) + .await + .unwrap() + .unwrap(); + let batched = catalog + .table("greptime", "inherited", "batched", None) + .await + .unwrap() + .unwrap(); + let explicit = catalog + .table("greptime", "inherited", "explicit", None) + .await + .unwrap() + .unwrap(); + assert_eq!( + ordinary.table_info().meta.options, + batched.table_info().meta.options + ); + assert!(batched.table_info().meta.options.skip_wal); + assert!(!explicit.table_info().meta.options.skip_wal); + assert!( + !batched + .table_info() + .meta + .options + .extra_options + .contains_key("compaction.type") + ); + + let response = execute_sql(&client, "CREATE TABLE partitioned (ts TIMESTAMP TIME INDEX, host STRING PRIMARY KEY, val DOUBLE) PARTITION ON COLUMNS (host) (host < 'm', host >= 'm') ENGINE=metric WITH (physical_metric_table='')").await; + assert!(response.status().is_success(), "{}", response.text().await); + let partitioned = logical_ddl("bad_partition") + .replace("phy.dot", "partitioned") + .replace( + "ENGINE=", + "PARTITION ON COLUMNS (host) (host < 'n', host >= 'n') ENGINE=", + ); + let response = execute_sql(&client, &partitioned).await; + assert!(!response.status().is_success()); + let sql = format!( + "{};{partitioned}", + logical_ddl("before_bad_partition").replace("phy.dot", "partitioned") + ); + let response = client + .post("/v1/ddl/logical-tables") + .form(&[("sql", sql)]) + .send() + .await; + assert!(!response.status().is_success()); + assert!( + catalog + .table("greptime", "public", "before_bad_partition", None) + .await + .unwrap() + .is_none() + ); + + let matching = partitioned + .replace("bad_partition", "matching_partition") + .replace("'n'", "'m'"); + let response = execute_sql(&client, &matching).await; + assert!(response.status().is_success(), "{}", response.text().await); + let sql = format!( + "{};{}", + matching.replace("matching_partition", "batch_matching_partition"), + logical_ddl("batch_inherited_partition").replace("phy.dot", "partitioned") + ); + let response = client + .post("/v1/ddl/logical-tables") + .form(&[("sql", sql)]) + .send() + .await; + assert!(response.status().is_success(), "{}", response.text().await); + + let original_id = frontend + .catalog_manager() + .table("greptime", "public", "batch_0", None) + .await + .unwrap() + .unwrap() + .table_info() + .table_id(); + let sql = format!("{};{}", logical_ddl("batch_0"), logical_ddl("batch_new")); + for _ in 0..2 { + let response = client + .post("/v1/ddl/logical-tables") + .form(&[("sql", &sql)]) + .send() + .await; + assert!(response.status().is_success(), "{}", response.text().await); + } + let old = frontend + .catalog_manager() + .table("greptime", "public", "batch_0", None) + .await + .unwrap() + .unwrap(); + assert_eq!(original_id, old.table_info().table_id()); + let response = execute_sql( + &client, + "INSERT INTO batch_0 (ts, host, val) VALUES (1, 'a', 2.0); INSERT INTO batch_new (ts, host, val) VALUES (2, 'b', 3.0)", + ) + .await; + assert!(response.status().is_success(), "{}", response.text().await); + validate_data( + "batch_values", + &client, + "SELECT val FROM batch_0 UNION ALL SELECT val FROM batch_new ORDER BY val", + "[[2.0],[3.0]]", + ) + .await; + let ctx = session::context::QueryContext::arc(); + let provider = + auth::static_user_provider_from_option("static_user_provider:cmd:readonly:ro=password") + .unwrap(); + let user = auth::UserProvider::authenticate( + &provider, + auth::Identity::UserId("readonly", None), + auth::Password::PlainText("password".to_string().into()), + ) + .await + .unwrap(); + ctx.set_current_user(user); + frontend + .plugins() + .get_or_insert::(auth::DefaultPermissionChecker::arc); + let result = servers::query_handler::sql::SqlQueryHandler::create_logical_tables( + frontend.as_ref(), + &logical_ddl("read_only_denied"), + ctx, + ) + .await; + assert!(result.is_err()); + assert!( + frontend + .catalog_manager() + .table("greptime", "public", "read_only_denied", None) + .await + .unwrap() + .is_none() + ); + + struct FailAfterExecution(&'static str); + impl servers::interceptor::SqlQueryInterceptor for FailAfterExecution { + type Error = frontend::error::Error; + fn post_execute( + &self, + _: common_query::Output, + _: QueryContextRef, + ) -> frontend::error::Result { + use common_query::Output; + match self.0 { + "rows" => Ok(Output::new_with_affected_rows(1)), + "records" => Ok(Output::new_with_record_batches( + common_recordbatch::RecordBatches::empty(), + )), + "stream" => Ok(Output::new_with_stream(Box::pin( + common_recordbatch::RecordBatchStreamWrapper::new( + Arc::new(datatypes::schema::Schema::new(vec![])), + futures::stream::iter([Err::( + common_recordbatch::error::CreateRecordBatchesSnafu { + reason: "sensitive-option-value", + } + .build(), + )]), + ), + ))), + _ => frontend::error::InvalidSqlSnafu { + err_msg: "sensitive-option-value", + } + .fail(), + } + } + } + frontend + .plugins() + .insert::>(Arc::new( + FailAfterExecution("error"), + )); + for mode in ["error", "rows", "records", "stream"] { + frontend + .plugins() + .map_mut::, _, _>( + |plugin| *plugin.unwrap() = Arc::new(FailAfterExecution(mode)), + ); + let name = format!("created_before_error_{mode}"); + let response = client + .post("/v1/ddl/logical-tables") + .form(&[("sql", logical_ddl(&name))]) + .send() + .await; + assert!(!response.status().is_success(), "{mode}"); + assert!(!format!("{:?}", response.headers()).contains("sensitive-option-value")); + let body = response.json::().await; + assert!(body.get("output").is_none()); + assert_eq!(body["error"], "logical-table batch failed"); + assert!( + frontend + .catalog_manager() + .table("greptime", "public", &name, None) + .await + .unwrap() + .is_some() + ); + } + instance.shutdown().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;