rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-09-21 04:35:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(frontend): remove gRPC DDL panics for DropView and non-timestamp time index (#8739)

Browse Source 
* fix(frontend): remove gRPC DDL panics for DropView and non-timestamp time index

Direct gRPC DDL bypasses the SQL parser, so two client-controlled DDL
payloads could panic a request handler:

- QX-152: DdlExpr::DropView hit todo!() (instance/grpc.rs:247-248).
  Wire it to the real drop-view implementation (drop_view was
  pub(crate); widened to pub) so a DropView DDL returns a structured
  error (e.g. TableNotFound) instead of panicking.
- QX-153: a CreateTableExpr whose time_index column is not a timestamp
  reached Schema::new's unwrap (ddl.rs:2346 -> schema.rs:114-119).
  create_table_info now uses Schema::try_new with ConvertSchemaSnafu
  context (InvalidArguments), and the direct gRPC CreateTable arm
  validates the request via validate_create_expr (which now also checks
  the time-index column type is a timestamp) before any catalog work.

SQL/HTTP paths were already protected by the parser; unchanged.
Tests: qx_152_drop_view_via_grpc_ddl_returns_error_not_panic,
qx_153_create_table_with_non_timestamp_time_index_via_grpc_returns_error
(asserts InvalidArguments), test_create_table_info_rejects_non_timestamp_time_index.

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

* test(frontend): add gRPC DDL happy-path coverage for DropView and CreateTable

Per review: the initial tests only asserted error paths. Add:
- drop_if_exists=true on a missing view succeeds (no error)
- dropping an existing view via gRPC DDL succeeds end-to-end
- a valid CreateTableExpr with a timestamp time index still succeeds
  (guards validate_create_expr against rejecting good requests)
- qx_152 test now asserts the TableNotFound status instead of is_err()

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

---------

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-08-07 06:48:15 +00:00
committed by GitHub
parent fbbe5af767
commit 97ca104129
4 changed files with 596 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/frontend/src/instance.rs
+523 -4
View File
@@ -1691,7 +1691,7 @@ mod tests {
use common_meta::cache::LayeredCacheRegistryBuilder;
use common_meta::kv_backend::memory::MemoryKvBackend;
use common_meta::procedure_executor::{ExecutorContext, ProcedureExecutor};
use common_meta::rpc::ddl::{SubmitDdlTaskRequest, SubmitDdlTaskResponse};
use common_meta::rpc::ddl::{DdlTask, SubmitDdlTaskRequest, SubmitDdlTaskResponse};
use common_meta::rpc::procedure::{
MigrateRegionRequest, MigrateRegionResponse, ProcedureStateResponse,
};
@@ -1722,7 +1722,9 @@ mod tests {
};
use store_api::storage::ScanRequest;
use strfmt::Format;
use table::metadata::{FilterPushDownType, TableInfo, TableInfoBuilder, TableMetaBuilder};
use table::metadata::{
FilterPushDownType, TableInfo, TableInfoBuilder, TableMetaBuilder, TableType,
};
use table::table_name::TableName;
use table::test_util::{EmptyTable, MemTable};
use table::{Table, TableRef};
@@ -2307,6 +2309,139 @@ mod tests {
}
}
/// A test [`ProcedureExecutor`] that completes create/drop DDL tasks against the
/// in-memory catalog, mimicking what the meta DDL procedures do in production.
/// This allows happy-path DDL requests (create/drop table/view) to be exercised
/// end to end through the gRPC ingress.
struct MockProcedureExecutor {
catalog_manager: Arc<catalog::memory::MemoryCatalogManager>,
next_table_id: std::sync::atomic::AtomicU32,
submitted: std::sync::Mutex<Vec<DdlTask>>,
}
impl MockProcedureExecutor {
fn new(catalog_manager: Arc<catalog::memory::MemoryCatalogManager>) -> Self {
Self {
catalog_manager,
next_table_id: std::sync::atomic::AtomicU32::new(1026),
submitted: std::sync::Mutex::new(Vec::new()),
}
}
}
#[async_trait::async_trait]
impl ProcedureExecutor for MockProcedureExecutor {
async fn submit_ddl_task(
&self,
_ctx: &ExecutorContext,
request: SubmitDdlTaskRequest,
) -> common_meta::error::Result<SubmitDdlTaskResponse> {
self.submitted.lock().unwrap().push(request.task.clone());
match request.task {
DdlTask::CreateTable(task) => {
let table_id = self
.next_table_id
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let mut table_info = task.table_info;
table_info.ident.table_id = table_id;
self.catalog_manager
.register_table_sync(catalog::RegisterTableRequest {
catalog: table_info.catalog_name.clone(),
schema: table_info.schema_name.clone(),
table_name: table_info.name.clone(),
table_id,
table: table::dist_table::DistTable::table(Arc::new(table_info)),
})
.map_err(BoxedError::new)
.context(common_meta::error::ExternalSnafu)?;
Ok(SubmitDdlTaskResponse {
key: Vec::new(),
table_ids: vec![table_id],
})
}
DdlTask::CreateView(task) => {
let view_id = self
.next_table_id
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let mut view_info = task.view_info;
view_info.ident.table_id = view_id;
self.catalog_manager
.register_table_sync(catalog::RegisterTableRequest {
catalog: task.create_view.catalog_name.clone(),
schema: task.create_view.schema_name.clone(),
table_name: task.create_view.view_name.clone(),
table_id: view_id,
table: table::dist_table::DistTable::table(Arc::new(view_info)),
})
.map_err(BoxedError::new)
.context(common_meta::error::ExternalSnafu)?;
Ok(SubmitDdlTaskResponse {
key: Vec::new(),
table_ids: vec![view_id],
})
}
DdlTask::DropView(task) => {
self.catalog_manager
.deregister_table_sync(catalog::DeregisterTableRequest {
catalog: task.catalog.clone(),
schema: task.schema.clone(),
table_name: task.view.clone(),
})
.map_err(BoxedError::new)
.context(common_meta::error::ExternalSnafu)?;
Ok(SubmitDdlTaskResponse::default())
}
other => common_meta::error::UnsupportedSnafu {
operation: format!("mock submit_ddl_task: {other:?}"),
}
.fail(),
}
}
async fn migrate_region(
&self,
_ctx: &ExecutorContext,
_request: MigrateRegionRequest,
) -> common_meta::error::Result<MigrateRegionResponse> {
common_meta::error::UnsupportedSnafu {
operation: "migrate_region",
}
.fail()
}
async fn reconcile(
&self,
_ctx: &ExecutorContext,
_request: ReconcileRequest,
) -> common_meta::error::Result<ReconcileResponse> {
common_meta::error::UnsupportedSnafu {
operation: "reconcile",
}
.fail()
}
async fn query_procedure_state(
&self,
_ctx: &ExecutorContext,
_pid: &str,
) -> common_meta::error::Result<ProcedureStateResponse> {
common_meta::error::UnsupportedSnafu {
operation: "query_procedure_state",
}
.fail()
}
async fn list_procedures(
&self,
_ctx: &ExecutorContext,
) -> common_meta::error::Result<ProcedureDetailResponse> {
common_meta::error::UnsupportedSnafu {
operation: "list_procedures",
}
.fail()
}
}
fn test_cache_registry(
kv_backend: common_meta::kv_backend::KvBackendRef,
) -> TestResult<common_meta::cache::LayeredCacheRegistryRef> {
@@ -2485,10 +2620,29 @@ mod tests {
target_table: TableRef,
plugins: Plugins,
metric_names_table: Option<TableRef>,
) -> TestResult<Instance> {
let catalog_manager = catalog::memory::MemoryCatalogManager::new_with_table(source_table);
test_instance_with_catalog_manager(
catalog_manager,
target_table,
plugins,
metric_names_table,
Arc::new(NoopProcedureExecutor),
)
.await
}
/// Builds a test frontend `Instance` over the given (already source-registered)
/// catalog manager, completing DDL tasks through `procedure_executor`.
async fn test_instance_with_catalog_manager(
catalog_manager: Arc<catalog::memory::MemoryCatalogManager>,
target_table: TableRef,
plugins: Plugins,
metric_names_table: Option<TableRef>,
procedure_executor: ProcedureExecutorRef,
) -> TestResult<Instance> {
let kv_backend = Arc::new(MemoryKvBackend::new());
let process_manager = Arc::new(ProcessManager::new("test-frontend".to_string(), None));
let catalog_manager = catalog::memory::MemoryCatalogManager::new_with_table(source_table);
let target_table_name = "target";
catalog_manager
.register_table_sync(catalog::RegisterTableRequest {
@@ -2529,7 +2683,7 @@ mod tests {
cache_registry,
catalog_manager,
Arc::new(client::client_manager::NodeClients::default()),
Arc::new(NoopProcedureExecutor),
procedure_executor,
process_manager,
)
.with_plugin(plugins)
@@ -3277,4 +3431,369 @@ mod tests {
assert_eq!(result.is_ok(), is_ok);
}
}
/// A `DropView` DDL sent through the direct gRPC ingress must return an error
/// (e.g. table not found) instead of panicking on `todo!()`.
#[tokio::test]
async fn qx_152_drop_view_via_grpc_ddl_returns_error_not_panic() -> TestResult<()> {
let instance =
test_instance_with_tables(test_table(1024, "source")?, test_table(1025, "target")?)
.await?;
let request = api::v1::greptime_request::Request::Ddl(api::v1::DdlRequest {
expr: Some(api::v1::ddl_request::Expr::DropView(
api::v1::DropViewExpr {
catalog_name: String::new(),
schema_name: String::new(),
view_name: "non_existent_view".to_string(),
view_id: None,
drop_if_exists: false,
},
)),
});
let result = servers::query_handler::grpc::GrpcQueryHandler::do_query(
&instance,
request,
QueryContext::arc(),
)
.await;
let err = match result {
Ok(_) => panic!("DropView DDL request must return an error instead of panicking"),
Err(err) => err,
};
assert_eq!(
err.status_code(),
StatusCode::TableNotFound,
"dropping a non-existent view without IF EXISTS must report TableNotFound, got {err}"
);
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`.
#[tokio::test]
async fn qx_152_drop_view_if_exists_missing_view_via_grpc_ddl_succeeds() -> TestResult<()> {
let catalog_manager =
catalog::memory::MemoryCatalogManager::new_with_table(test_table(1024, "source")?);
let procedure_executor = Arc::new(MockProcedureExecutor::new(catalog_manager.clone()));
let instance = test_instance_with_catalog_manager(
catalog_manager,
test_table(1025, "target")?,
Plugins::new(),
None,
procedure_executor.clone() as ProcedureExecutorRef,
)
.await?;
let request = api::v1::greptime_request::Request::Ddl(api::v1::DdlRequest {
expr: Some(api::v1::ddl_request::Expr::DropView(
api::v1::DropViewExpr {
catalog_name: String::new(),
schema_name: String::new(),
view_name: "non_existent_view".to_string(),
view_id: None,
drop_if_exists: true,
},
)),
});
let result = servers::query_handler::grpc::GrpcQueryHandler::do_query(
&instance,
request,
QueryContext::arc(),
)
.await;
let output = match result {
Ok(output) => output,
Err(err) => {
panic!("DROP VIEW IF EXISTS on a missing view must succeed, got error: {err}")
}
};
assert!(
matches!(output.data, OutputData::AffectedRows(0)),
"DROP VIEW IF EXISTS on a missing view must report 0 affected rows"
);
assert!(
procedure_executor.submitted.lock().unwrap().is_empty(),
"DROP VIEW IF EXISTS on a missing view must not submit a DDL task"
);
Ok(())
}
/// A `CREATE VIEW` followed by `DROP VIEW` through the direct gRPC ingress must
/// succeed end to end: the view is registered in the catalog and then removed.
#[tokio::test]
async fn qx_152_drop_existing_view_via_grpc_ddl_succeeds() -> TestResult<()> {
let catalog_manager =
catalog::memory::MemoryCatalogManager::new_with_table(test_table(1024, "source")?);
let procedure_executor = Arc::new(MockProcedureExecutor::new(catalog_manager.clone()));
let instance = test_instance_with_catalog_manager(
catalog_manager,
test_table(1025, "target")?,
Plugins::new(),
None,
procedure_executor.clone() as ProcedureExecutorRef,
)
.await?;
// The default "greptime.public" schema must be visible to the kv-backed table
// metadata manager for `CREATE VIEW`/`CREATE TABLE` to pass validation.
instance
.table_metadata_manager()
.schema_manager()
.create(
common_meta::key::schema_name::SchemaNameKey::new("greptime", "public"),
None,
true,
)
.await
.unwrap();
let create_view_request = api::v1::greptime_request::Request::Ddl(api::v1::DdlRequest {
expr: Some(api::v1::ddl_request::Expr::CreateView(
api::v1::CreateViewExpr {
catalog_name: String::new(),
schema_name: String::new(),
view_name: "my_view".to_string(),
logical_plan: vec![1, 2, 3],
create_if_not_exists: false,
or_replace: false,
table_names: vec![],
columns: vec![],
plan_columns: vec![],
definition: "CREATE VIEW my_view AS SELECT * FROM source".to_string(),
},
)),
});
let output = match servers::query_handler::grpc::GrpcQueryHandler::do_query(
&instance,
create_view_request,
QueryContext::arc(),
)
.await
{
Ok(output) => output,
Err(err) => panic!("CREATE VIEW via gRPC DDL must succeed, got error: {err}"),
};
assert!(
matches!(output.data, OutputData::AffectedRows(0)),
"CREATE VIEW via gRPC DDL must report 0 affected rows"
);
// The view is registered in the catalog as a view.
let view = instance
.catalog_manager()
.table("greptime", "public", "my_view", None)
.await
.unwrap()
.expect("view should exist after CREATE VIEW");
assert_eq!(view.table_info().table_type, TableType::View);
let drop_view_request = api::v1::greptime_request::Request::Ddl(api::v1::DdlRequest {
expr: Some(api::v1::ddl_request::Expr::DropView(
api::v1::DropViewExpr {
catalog_name: String::new(),
schema_name: String::new(),
view_name: "my_view".to_string(),
view_id: None,
drop_if_exists: false,
},
)),
});
let output = match servers::query_handler::grpc::GrpcQueryHandler::do_query(
&instance,
drop_view_request,
QueryContext::arc(),
)
.await
{
Ok(output) => output,
Err(err) => panic!("DROP VIEW via gRPC DDL must succeed, got error: {err}"),
};
assert!(
matches!(output.data, OutputData::AffectedRows(0)),
"DROP VIEW via gRPC DDL must report 0 affected rows"
);
// The view is gone after the drop.
assert!(
instance
.catalog_manager()
.table("greptime", "public", "my_view", None)
.await
.unwrap()
.is_none(),
"view should be removed after DROP VIEW"
);
let submitted = procedure_executor.submitted.lock().unwrap();
assert_eq!(
submitted.len(),
2,
"expected create and drop view tasks, got {submitted:?}"
);
assert!(matches!(&submitted[0], DdlTask::CreateView(_)));
assert!(matches!(&submitted[1], DdlTask::DropView(_)));
Ok(())
}
/// A direct gRPC `CreateTable` whose time index column is not a timestamp must
/// be rejected with `InvalidArguments` instead of panicking while building the schema.
#[tokio::test]
async fn qx_153_create_table_with_non_timestamp_time_index_via_grpc_returns_error()
-> TestResult<()> {
let instance =
test_instance_with_tables(test_table(1024, "source")?, test_table(1025, "target")?)
.await?;
let request = api::v1::greptime_request::Request::Ddl(api::v1::DdlRequest {
expr: Some(api::v1::ddl_request::Expr::CreateTable(
api::v1::CreateTableExpr {
catalog_name: String::new(),
schema_name: String::new(),
table_name: "demo".to_string(),
desc: String::new(),
column_defs: vec![api::v1::ColumnDef {
name: "host".to_string(),
data_type: api::v1::ColumnDataType::String as i32,
is_nullable: true,
default_constraint: vec![],
semantic_type: 0,
comment: String::new(),
datatype_extension: None,
options: None,
}],
time_index: "host".to_string(),
primary_keys: vec![],
create_if_not_exists: false,
table_options: HashMap::new(),
table_id: None,
engine: "mito".to_string(),
},
)),
});
let result = servers::query_handler::grpc::GrpcQueryHandler::do_query(
&instance,
request,
QueryContext::arc(),
)
.await;
let err = match result {
Ok(_) => panic!("CreateTable with a non-timestamp time index must be rejected"),
Err(err) => err,
};
assert_eq!(err.status_code(), StatusCode::InvalidArguments, "{err}");
Ok(())
}
/// A valid `CREATE TABLE` (timestamp time index) through the direct gRPC ingress
/// must succeed, guarding that the `validate_create_expr` ingress check doesn't
/// accidentally reject good requests.
#[tokio::test]
async fn qx_153_create_table_with_timestamp_time_index_via_grpc_succeeds() -> TestResult<()> {
let catalog_manager =
catalog::memory::MemoryCatalogManager::new_with_table(test_table(1024, "source")?);
let procedure_executor = Arc::new(MockProcedureExecutor::new(catalog_manager.clone()));
let instance = test_instance_with_catalog_manager(
catalog_manager,
test_table(1025, "target")?,
Plugins::new(),
None,
procedure_executor.clone() as ProcedureExecutorRef,
)
.await?;
// The default "greptime.public" schema must be visible to the kv-backed table
// metadata manager for `CREATE TABLE` to pass validation.
instance
.table_metadata_manager()
.schema_manager()
.create(
common_meta::key::schema_name::SchemaNameKey::new("greptime", "public"),
None,
true,
)
.await
.unwrap();
let request = api::v1::greptime_request::Request::Ddl(api::v1::DdlRequest {
expr: Some(api::v1::ddl_request::Expr::CreateTable(
api::v1::CreateTableExpr {
catalog_name: String::new(),
schema_name: String::new(),
table_name: "demo".to_string(),
desc: String::new(),
column_defs: vec![
api::v1::ColumnDef {
name: "host".to_string(),
data_type: api::v1::ColumnDataType::String as i32,
is_nullable: true,
default_constraint: vec![],
semantic_type: 0,
comment: String::new(),
datatype_extension: None,
options: None,
},
api::v1::ColumnDef {
name: "ts".to_string(),
data_type: api::v1::ColumnDataType::TimestampMillisecond as i32,
is_nullable: true,
default_constraint: vec![],
semantic_type: 0,
comment: String::new(),
datatype_extension: None,
options: None,
},
],
time_index: "ts".to_string(),
primary_keys: vec![],
create_if_not_exists: false,
table_options: HashMap::new(),
table_id: None,
engine: "mito".to_string(),
},
)),
});
let output = match servers::query_handler::grpc::GrpcQueryHandler::do_query(
&instance,
request,
QueryContext::arc(),
)
.await
{
Ok(output) => output,
Err(err) => panic!("CREATE TABLE via gRPC DDL must succeed, got error: {err}"),
};
assert!(
matches!(output.data, OutputData::AffectedRows(0)),
"CREATE TABLE via gRPC DDL must report 0 affected rows"
);
// The table is registered in the catalog.
let table = instance
.catalog_manager()
.table("greptime", "public", "demo", None)
.await
.unwrap()
.expect("table should exist after CREATE TABLE");
assert_eq!(table.table_info().table_type, TableType::Base);
let submitted = procedure_executor.submitted.lock().unwrap();
assert_eq!(
submitted.len(),
1,
"expected one create table task, got {submitted:?}"
);
assert!(matches!(&submitted[0], DdlTask::CreateTable(_)));
Ok(())
}
}
src/frontend/src/instance/grpc.rs
+13 -2
View File
@@ -176,6 +176,9 @@ impl GrpcQueryHandler for Instance {
match expr {
DdlExpr::CreateTable(mut expr) => {
// Direct gRPC DDL bypasses the SQL parser, so validate the
// request here (e.g. the time index must be a timestamp).
operator::expr_helper::validate_create_expr(&expr)?;
let _ = self
.statement_executor
.create_table_inner(&mut expr, None, ctx.clone())
@@ -244,8 +247,16 @@ impl GrpcQueryHandler for Instance {
Output::new_with_affected_rows(0)
}
DdlExpr::DropView(_) => {
todo!("implemented in the following PR")
DdlExpr::DropView(expr) => {
self.statement_executor
.drop_view(
expr.catalog_name,
expr.schema_name,
expr.view_name,
expr.drop_if_exists,
ctx.clone(),
)
.await?
}
DdlExpr::CommentOn(expr) => {
self.statement_executor
src/operator/src/expr_helper.rs
+26 -5
View File
@@ -36,6 +36,7 @@ use common_error::ext::BoxedError;
use common_grpc_expr::util::ColumnExpr;
use common_time::Timezone;
use datafusion::sql::planner::object_name_to_table_reference;
use datatypes::prelude::ConcreteDataType;
use datatypes::schema::{
COLUMN_FULLTEXT_OPT_KEY_ANALYZER, COLUMN_FULLTEXT_OPT_KEY_BACKEND,
COLUMN_FULLTEXT_OPT_KEY_CASE_SENSITIVE, COLUMN_FULLTEXT_OPT_KEY_FALSE_POSITIVE_RATE,
@@ -442,14 +443,34 @@ pub fn validate_create_expr(create: &CreateTableExpr) -> Result<()> {
}
// verify time_index exists
let _ = column_to_indices
.get(&create.time_index)
.with_context(|| InvalidSqlSnafu {
let time_index_idx =
column_to_indices
.get(&create.time_index)
.with_context(|| InvalidSqlSnafu {
err_msg: format!(
"column name `{}` is not found in column list",
create.time_index
),
})?;
// verify time_index is a timestamp column
let time_index_column = &create.column_defs[*time_index_idx];
let data_type = ConcreteDataType::from(
ColumnDataTypeWrapper::try_new(
time_index_column.data_type,
time_index_column.datatype_extension.clone(),
)
.context(ColumnDataTypeSnafu)?,
);
ensure!(
data_type.is_timestamp(),
InvalidSqlSnafu {
err_msg: format!(
"column name `{}` is not found in column list",
"column `{}` is not a timestamp type, it can't be used as time index",
create.time_index
),
})?;
}
);
// verify primary_key exists
for pk in &create.primary_keys {
src/operator/src/statement/ddl.rs
+34 -2
View File
@@ -1213,7 +1213,7 @@ impl StatementExecutor {
/// Drop a view
#[tracing::instrument(skip_all)]
pub(crate) async fn drop_view(
pub async fn drop_view(
&self,
catalog: String,
schema: String,
@@ -2343,7 +2343,7 @@ pub fn create_table_info(
}
let next_column_id = column_schemas.len() as u32;
let schema = Arc::new(Schema::new(column_schemas));
let schema = Arc::new(Schema::try_new(column_schemas).context(ConvertSchemaSnafu)?);
let primary_key_indices = create_table
.primary_keys
@@ -3443,6 +3443,38 @@ WITH ('repartition.column.hint' = ' host ')",
);
}
#[test]
fn test_create_table_info_rejects_non_timestamp_time_index() {
let expr = CreateTableExpr {
catalog_name: "greptime".to_string(),
schema_name: "public".to_string(),
table_name: "demo".to_string(),
desc: String::new(),
column_defs: vec![api::v1::ColumnDef {
name: "host".to_string(),
data_type: api::v1::ColumnDataType::String as i32,
is_nullable: true,
default_constraint: vec![],
semantic_type: 0,
comment: String::new(),
datatype_extension: None,
options: None,
}],
time_index: "host".to_string(),
primary_keys: vec![],
create_if_not_exists: false,
table_options: HashMap::new(),
table_id: None,
engine: "mito".to_string(),
};
let err = create_table_info(&expr, vec![]).unwrap_err();
assert_eq!(
common_error::ext::ErrorExt::status_code(&err),
common_error::status_code::StatusCode::InvalidArguments
);
}
#[test]
fn test_json2_requires_append_mode() {
let cases = [