mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-21 19:32:34 +00:00
feat: add bounded Metric logical-table batch DDL (#9427)
* feat: add bounded Metric logical-table batch DDL Signed-off-by: jeremyhi <fengjiachun@gmail.com> * fix: enforce logical batch DDL interceptor boundaries Signed-off-by: jeremyhi <fengjiachun@gmail.com> * fix: bound rewritten DDL size through shared SQL formatting Signed-off-by: jeremyhi <fengjiachun@gmail.com> * fix: preserve logical DDL error diagnostics Signed-off-by: jeremyhi <fengjiachun@gmail.com> * test: verify logical DDL error logging Signed-off-by: jeremyhi <fengjiachun@gmail.com> --------- Signed-off-by: jeremyhi <fengjiachun@gmail.com>
This commit is contained in:
jeremyhi
@@ -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.
|
||||
|
||||
@@ -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`
|
||||
|
||||
@@ -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<Vec<Output>> {
|
||||
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::<SqlQueryInterceptorRef<Error>>();
|
||||
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<Result<Output>> {
|
||||
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<Vec<Output>> {
|
||||
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<std::borrow::Cow<'a, str>> {
|
||||
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<Statement>,
|
||||
_: QueryContextRef,
|
||||
) -> Result<Vec<Statement>> {
|
||||
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::<PermissionCheckerRef>(Arc::new(RejectUnresolvedPermissionChecker));
|
||||
plugins.insert::<SqlQueryInterceptorRef<Error>>(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`.
|
||||
|
||||
@@ -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<TableRef> {
|
||||
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<CreateTableExpr> {
|
||||
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<CreateTable>,
|
||||
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),
|
||||
|
||||
@@ -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<std::io::Error> 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 {
|
||||
|
||||
+106
-6
@@ -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<S>(api_state: ApiState) -> Router<S> {
|
||||
fn route_sql<S>(api_state: ApiState, batch_body_limit: usize) -> Router<S> {
|
||||
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<ApiState>| 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::<serde_json::Value>().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
|
||||
|
||||
@@ -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<String>,
|
||||
db: Option<String>,
|
||||
}
|
||||
|
||||
/// Creates one batch of Metric logical tables using the normal v1 response.
|
||||
pub async fn create_logical_tables(
|
||||
State(state): State<ApiState>,
|
||||
Extension(mut query_ctx): Extension<QueryContext>,
|
||||
form: std::result::Result<Form<LogicalTableDdl>, 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<Output> {
|
||||
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<Vec<Output>> {
|
||||
std::mem::take(&mut *self.outputs.lock().unwrap())
|
||||
.into_iter()
|
||||
.collect()
|
||||
}
|
||||
|
||||
async fn do_analyze_stream_query(&self, _: &str, _: QueryContextRef) -> Result<Output> {
|
||||
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",
|
||||
|
||||
@@ -26,8 +26,30 @@ use crate::error::Result;
|
||||
|
||||
pub type ServerSqlQueryHandlerRef = Arc<dyn SqlQueryHandler + Send + Sync>;
|
||||
|
||||
/// 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<Vec<Output>> {
|
||||
crate::error::UnsupportedOperationSnafu {
|
||||
operation: "metric batch DDL",
|
||||
}
|
||||
.fail()
|
||||
}
|
||||
|
||||
async fn do_query(&self, query: &str, query_ctx: QueryContextRef) -> Vec<Result<Output>>;
|
||||
|
||||
/// Executes an HTTP analyze-stream query.
|
||||
|
||||
@@ -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<T: Display>(
|
||||
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<DataType
|
||||
Ok(sql_type)
|
||||
}
|
||||
|
||||
fn format_json_type_hint(hint: &JsonTypeHint) -> 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::<u32>().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(
|
||||
|
||||
@@ -139,26 +139,44 @@ impl OptionMap {
|
||||
}
|
||||
|
||||
pub fn kv_pairs(&self) -> Vec<String> {
|
||||
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<Item = SqlPair<'_>> {
|
||||
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<Item = (&str, Either<&OptionValue, &str>)> {
|
||||
@@ -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<I: IntoIterator<Item = (String, String)>> From<I> for OptionMap {
|
||||
fn from(value: I) -> Self {
|
||||
let mut result = OptionMap::default();
|
||||
|
||||
+15
-7
@@ -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::*;
|
||||
|
||||
@@ -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::<Vec<_>>();
|
||||
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::<Value>().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::<Value>()
|
||||
.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::<Vec<_>>()
|
||||
.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::<Vec<_>>()
|
||||
.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::<Value>().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::PermissionCheckerRef, _>(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<common_query::Output> {
|
||||
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::RecordBatch, _>(
|
||||
common_recordbatch::error::CreateRecordBatchesSnafu {
|
||||
reason: "sensitive-option-value",
|
||||
}
|
||||
.build(),
|
||||
)]),
|
||||
),
|
||||
))),
|
||||
_ => frontend::error::InvalidSqlSnafu {
|
||||
err_msg: "sensitive-option-value",
|
||||
}
|
||||
.fail(),
|
||||
}
|
||||
}
|
||||
}
|
||||
frontend
|
||||
.plugins()
|
||||
.insert::<servers::interceptor::SqlQueryInterceptorRef<frontend::error::Error>>(Arc::new(
|
||||
FailAfterExecution("error"),
|
||||
));
|
||||
for mode in ["error", "rows", "records", "stream"] {
|
||||
frontend
|
||||
.plugins()
|
||||
.map_mut::<servers::interceptor::SqlQueryInterceptorRef<frontend::error::Error>, _, _>(
|
||||
|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::<Value>().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;
|
||||
|
||||
Reference in New Issue
Block a user