From bf7eb5e200be81e077dcdfa8c1c5db6e343a63b5 Mon Sep 17 00:00:00 2001 From: camilesing Date: Wed, 3 Dec 2025 12:48:40 +0800 Subject: [PATCH 1/2] feat: add explicit FLUSH PRIVILEGES to refresh role cache for query node --- src/meta/app/src/tenant/tenant.rs | 2 +- src/query/ast/src/ast/statements/system_action.rs | 2 ++ src/query/ast/src/parser/statement.rs | 10 ++++++++-- src/query/ast/src/parser/token.rs | 2 ++ .../src/interpreters/interpreter_system_action.rs | 5 +++++ .../src/servers/flight/v1/actions/kill_query.rs | 2 +- src/query/service/src/servers/flight/v1/actions/mod.rs | 6 ++++-- .../src/servers/flight/v1/actions/set_priority.rs | 2 +- .../src/servers/flight/v1/actions/system_action.rs | 2 +- .../src/servers/flight/v1/actions/truncate_table.rs | 2 +- src/query/sql/src/planner/binder/system.rs | 6 ++++++ src/query/sql/src/planner/plans/system.rs | 3 +++ tests/nox/java_client/prepare.py | 5 +---- .../suites/base/20+_others/20_0017_system_action.test | 3 +++ 14 files changed, 39 insertions(+), 13 deletions(-) diff --git a/src/meta/app/src/tenant/tenant.rs b/src/meta/app/src/tenant/tenant.rs index df13818e5fe61..2d8e6a4dd2013 100644 --- a/src/meta/app/src/tenant/tenant.rs +++ b/src/meta/app/src/tenant/tenant.rs @@ -24,7 +24,7 @@ use crate::app_error::TenantIsEmpty; /// Tenant is not stored directly in meta-store. /// /// It is just a type for use on the client side. -#[derive(Clone, Debug, PartialEq, Eq, Hash)] +#[derive(Clone, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)] pub struct Tenant { // TODO: consider using NonEmptyString? pub tenant: String, diff --git a/src/query/ast/src/ast/statements/system_action.rs b/src/query/ast/src/ast/statements/system_action.rs index fbe3a89759f5e..aaa33a8bb90cf 100644 --- a/src/query/ast/src/ast/statements/system_action.rs +++ b/src/query/ast/src/ast/statements/system_action.rs @@ -32,6 +32,7 @@ impl Display for SystemStmt { #[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut)] pub enum SystemAction { Backtrace(bool), + FlushPrivileges, } impl Display for SystemAction { @@ -41,6 +42,7 @@ impl Display for SystemAction { true => write!(f, "ENABLE EXCEPTION_BACKTRACE"), false => write!(f, "DISABLE EXCEPTION_BACKTRACE"), }, + SystemAction::FlushPrivileges => write!(f, "FLUSH PRIVILEGES"), } } } diff --git a/src/query/ast/src/parser/statement.rs b/src/query/ast/src/parser/statement.rs index 85570a1c6881f..869d7b008ac3c 100644 --- a/src/query/ast/src/parser/statement.rs +++ b/src/query/ast/src/parser/statement.rs @@ -5054,15 +5054,21 @@ pub fn priority(i: Input) -> IResult { } pub fn action(i: Input) -> IResult { - let mut backtrace = parser_fn(map( + let backtrace = parser_fn(map( rule! { #switch ~ EXCEPTION_BACKTRACE }, |(switch, _)| SystemAction::Backtrace(switch), )); + let flush_privileges = parser_fn(map( + rule! { + FLUSH ~ PRIVILEGES + }, + |_| SystemAction::FlushPrivileges, + )); // add other system action type here rule!( - #backtrace + #backtrace | #flush_privileges ) .parse(i) } diff --git a/src/query/ast/src/parser/token.rs b/src/query/ast/src/parser/token.rs index 4b120493368df..086d1c10f5e4f 100644 --- a/src/query/ast/src/parser/token.rs +++ b/src/query/ast/src/parser/token.rs @@ -683,6 +683,8 @@ pub enum TokenKind { FORMAT_NAME, #[token("FORMATS", ignore(ascii_case))] FORMATS, + #[token("FLUSH", ignore(ascii_case))] + FLUSH, #[token("FRAGMENTS", ignore(ascii_case))] FRAGMENTS, #[token("FRIDAY", ignore(ascii_case))] diff --git a/src/query/service/src/interpreters/interpreter_system_action.rs b/src/query/service/src/interpreters/interpreter_system_action.rs index e0b2b192a3a70..125c04a14255b 100644 --- a/src/query/service/src/interpreters/interpreter_system_action.rs +++ b/src/query/service/src/interpreters/interpreter_system_action.rs @@ -20,6 +20,7 @@ use databend_common_exception::set_backtrace; use databend_common_exception::Result; use databend_common_sql::plans::SystemAction; use databend_common_sql::plans::SystemPlan; +use databend_common_users::RoleCacheManager; use crate::clusters::ClusterHelper; use crate::clusters::FlightParams; @@ -90,6 +91,10 @@ impl Interpreter for SystemActionInterpreter { SystemAction::Backtrace(switch) => { set_backtrace(switch); } + SystemAction::FlushPrivileges => { + let tenant = self.ctx.get_tenant(); + RoleCacheManager::instance().force_reload(&tenant).await?; + } } Ok(PipelineBuildResult::create()) } diff --git a/src/query/service/src/servers/flight/v1/actions/kill_query.rs b/src/query/service/src/servers/flight/v1/actions/kill_query.rs index 592bf7173a2fd..ec01f5dc12f28 100644 --- a/src/query/service/src/servers/flight/v1/actions/kill_query.rs +++ b/src/query/service/src/servers/flight/v1/actions/kill_query.rs @@ -24,7 +24,7 @@ use crate::servers::flight::v1::actions::create_session; pub static KILL_QUERY: &str = "/actions/kill_query"; pub async fn kill_query(plan: KillPlan) -> Result { - let session = create_session()?; + let session = create_session(None)?; let version = GlobalConfig::version(); let query_context = session.create_query_context(version).await?; let interpreter = KillInterpreter::from_flight(query_context, plan)?; diff --git a/src/query/service/src/servers/flight/v1/actions/mod.rs b/src/query/service/src/servers/flight/v1/actions/mod.rs index f5f244b92f168..a3d633e8f35f6 100644 --- a/src/query/service/src/servers/flight/v1/actions/mod.rs +++ b/src/query/service/src/servers/flight/v1/actions/mod.rs @@ -27,6 +27,7 @@ use std::sync::Arc; use databend_common_catalog::session_type::SessionType; use databend_common_config::GlobalConfig; use databend_common_exception::Result; +use databend_common_meta_app::tenant::Tenant; use databend_common_settings::Settings; pub use flight_actions::flight_actions; pub use flight_actions::FlightActions; @@ -43,9 +44,10 @@ pub use truncate_table::TRUNCATE_TABLE; use crate::sessions::Session; use crate::sessions::SessionManager; -pub(crate) fn create_session() -> Result> { +pub(crate) fn create_session(tenant: Option) -> Result> { let config = GlobalConfig::instance(); - let settings = Settings::create(config.query.tenant_id.clone()); + let tenant_id = tenant.unwrap_or(config.query.tenant_id.clone()); + let settings = Settings::create(tenant_id.clone()); match SessionManager::instance().create_with_settings(SessionType::FlightRPC, settings, None) { Err(cause) => Err(cause), Ok(session) => Ok(Arc::new(session)), diff --git a/src/query/service/src/servers/flight/v1/actions/set_priority.rs b/src/query/service/src/servers/flight/v1/actions/set_priority.rs index 6767a02904f3a..9e5bd534a43e3 100644 --- a/src/query/service/src/servers/flight/v1/actions/set_priority.rs +++ b/src/query/service/src/servers/flight/v1/actions/set_priority.rs @@ -24,7 +24,7 @@ use crate::servers::flight::v1::actions::create_session; pub static SET_PRIORITY: &str = "/actions/set_priority"; pub async fn set_priority(plan: SetPriorityPlan) -> Result { - let session = create_session()?; + let session = create_session(None)?; let version = GlobalConfig::version(); let query_context = session.create_query_context(version).await?; let interpreter = SetPriorityInterpreter::from_flight(query_context, plan)?; diff --git a/src/query/service/src/servers/flight/v1/actions/system_action.rs b/src/query/service/src/servers/flight/v1/actions/system_action.rs index a234793eff00c..3b13a10a29ba3 100644 --- a/src/query/service/src/servers/flight/v1/actions/system_action.rs +++ b/src/query/service/src/servers/flight/v1/actions/system_action.rs @@ -23,7 +23,7 @@ use crate::servers::flight::v1::actions::create_session; pub static SYSTEM_ACTION: &str = "/actions/system_action"; pub async fn system_action(plan: SystemPlan) -> Result<()> { - let session = create_session()?; + let session = create_session(Some(plan.clone().tenant))?; let version = GlobalConfig::version(); let query_context = session.create_query_context(version).await?; let interpreter = SystemActionInterpreter::from_flight(query_context, plan)?; diff --git a/src/query/service/src/servers/flight/v1/actions/truncate_table.rs b/src/query/service/src/servers/flight/v1/actions/truncate_table.rs index e36b81d246d9d..cc03070c774ca 100644 --- a/src/query/service/src/servers/flight/v1/actions/truncate_table.rs +++ b/src/query/service/src/servers/flight/v1/actions/truncate_table.rs @@ -23,7 +23,7 @@ use crate::servers::flight::v1::actions::create_session; pub static TRUNCATE_TABLE: &str = "/actions/truncate_table"; pub async fn truncate_table(plan: TruncateTablePlan) -> Result<()> { - let session = create_session()?; + let session = create_session(None)?; let version = GlobalConfig::version(); let query_context = session.create_query_context(version).await?; let interpreter = TruncateTableInterpreter::from_flight(query_context, plan)?; diff --git a/src/query/sql/src/planner/binder/system.rs b/src/query/sql/src/planner/binder/system.rs index d811441bde23e..a4e18545b4552 100644 --- a/src/query/sql/src/planner/binder/system.rs +++ b/src/query/sql/src/planner/binder/system.rs @@ -25,9 +25,15 @@ impl Binder { #[async_backtrace::framed] pub(super) async fn bind_system(&mut self, stmt: &SystemStmt) -> Result { let SystemStmt { action } = stmt; + let tenant = self.ctx.get_tenant(); match action { AstSystemAction::Backtrace(switch) => Ok(Plan::System(Box::new(SystemPlan { action: SystemAction::Backtrace(*switch), + tenant, + }))), + AstSystemAction::FlushPrivileges => Ok(Plan::System(Box::new(SystemPlan { + action: SystemAction::FlushPrivileges, + tenant, }))), } } diff --git a/src/query/sql/src/planner/plans/system.rs b/src/query/sql/src/planner/plans/system.rs index 2c43342cea0f4..be0e8b45b839b 100644 --- a/src/query/sql/src/planner/plans/system.rs +++ b/src/query/sql/src/planner/plans/system.rs @@ -12,15 +12,18 @@ // See the License for the specific language governing permissions and // limitations under the License. +use databend_common_meta_app::tenant::Tenant; use serde::Deserialize; use serde::Serialize; #[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq)] pub struct SystemPlan { pub action: SystemAction, + pub tenant: Tenant, } #[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)] pub enum SystemAction { Backtrace(bool), + FlushPrivileges, } diff --git a/tests/nox/java_client/prepare.py b/tests/nox/java_client/prepare.py index 4c44b767b0c1c..ea6d935b6b9e3 100644 --- a/tests/nox/java_client/prepare.py +++ b/tests/nox/java_client/prepare.py @@ -57,10 +57,7 @@ def create_user(): "CREATE USER databend IDENTIFIED BY 'databend' with default_role='account_admin'" ) exec("GRANT ROLE account_admin TO USER databend") - # need for cluster to sync the GRANT op - time.sleep(16) - for p in [8001, 8002, 8003]: - exec("SHOW GRANTS FOR USER databend", port=p) + exec("SYSTEM FLUSH PRIVILEGES") def download_testng(): diff --git a/tests/sqllogictests/suites/base/20+_others/20_0017_system_action.test b/tests/sqllogictests/suites/base/20+_others/20_0017_system_action.test index b0fea51f0ae2a..c96ada33a1418 100644 --- a/tests/sqllogictests/suites/base/20+_others/20_0017_system_action.test +++ b/tests/sqllogictests/suites/base/20+_others/20_0017_system_action.test @@ -3,3 +3,6 @@ SYSTEM ENABLE EXCEPTION_BACKTRACE; statement ok SYSTEM DISABLE EXCEPTION_BACKTRACE; + +statement ok +SYSTEM FLUSH PRIVILEGES; \ No newline at end of file From 198a47ecb379b9c7f2ac2893ec38ef8844ea7147 Mon Sep 17 00:00:00 2001 From: TCeason Date: Tue, 9 Dec 2025 09:59:25 +0800 Subject: [PATCH 2/2] - SystemPlan is only produced by the SQL binder and that binder always runs in a QueryContext whose tenant is fixed to GlobalConfig::instance().query.tenant_id unless the server is in management mode. When management mode is enabled, ManagementModeAccess blocks Plan::System altogether so no user query can ever build a system action with a different tenant. As a result, the tenant field inside SystemPlan was guaranteed to equal the global config tenant and provided no extra routing information anywhere else in the stack. - All flight actions (system_action, kill_query, set_priority, truncate_table) run inside a query server whose GlobalConfig already defines a single tenant. There is no supported deployment where a single query process serves multiple tenants concurrently over flight RPC. Therefore, the create_session helper in src/query/service/src/servers/flight/v1/actions/mod.rs can safely derive the tenant from GlobalConfig every time; the optional parameter was never used to switch context to another tenant. - Because both ends (planner and flight handler) collapse to the same static tenant, the tenant field in SystemPlan and the Option parameter to create_session were redundant wiring that couldn't change behavior. Removing them simplifies the code without affecting any observable semantics and makes it clear that system/flight actions always run under the server's configured tenant. --- src/meta/app/src/tenant/tenant.rs | 2 +- .../service/src/servers/flight/v1/actions/kill_query.rs | 2 +- src/query/service/src/servers/flight/v1/actions/mod.rs | 7 +++---- .../service/src/servers/flight/v1/actions/set_priority.rs | 2 +- .../service/src/servers/flight/v1/actions/system_action.rs | 2 +- .../src/servers/flight/v1/actions/truncate_table.rs | 2 +- src/query/sql/src/planner/binder/system.rs | 3 --- src/query/sql/src/planner/plans/system.rs | 2 -- 8 files changed, 8 insertions(+), 14 deletions(-) diff --git a/src/meta/app/src/tenant/tenant.rs b/src/meta/app/src/tenant/tenant.rs index 2d8e6a4dd2013..df13818e5fe61 100644 --- a/src/meta/app/src/tenant/tenant.rs +++ b/src/meta/app/src/tenant/tenant.rs @@ -24,7 +24,7 @@ use crate::app_error::TenantIsEmpty; /// Tenant is not stored directly in meta-store. /// /// It is just a type for use on the client side. -#[derive(Clone, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)] +#[derive(Clone, Debug, PartialEq, Eq, Hash)] pub struct Tenant { // TODO: consider using NonEmptyString? pub tenant: String, diff --git a/src/query/service/src/servers/flight/v1/actions/kill_query.rs b/src/query/service/src/servers/flight/v1/actions/kill_query.rs index ec01f5dc12f28..592bf7173a2fd 100644 --- a/src/query/service/src/servers/flight/v1/actions/kill_query.rs +++ b/src/query/service/src/servers/flight/v1/actions/kill_query.rs @@ -24,7 +24,7 @@ use crate::servers::flight::v1::actions::create_session; pub static KILL_QUERY: &str = "/actions/kill_query"; pub async fn kill_query(plan: KillPlan) -> Result { - let session = create_session(None)?; + let session = create_session()?; let version = GlobalConfig::version(); let query_context = session.create_query_context(version).await?; let interpreter = KillInterpreter::from_flight(query_context, plan)?; diff --git a/src/query/service/src/servers/flight/v1/actions/mod.rs b/src/query/service/src/servers/flight/v1/actions/mod.rs index a3d633e8f35f6..97f8db0d67f47 100644 --- a/src/query/service/src/servers/flight/v1/actions/mod.rs +++ b/src/query/service/src/servers/flight/v1/actions/mod.rs @@ -27,7 +27,6 @@ use std::sync::Arc; use databend_common_catalog::session_type::SessionType; use databend_common_config::GlobalConfig; use databend_common_exception::Result; -use databend_common_meta_app::tenant::Tenant; use databend_common_settings::Settings; pub use flight_actions::flight_actions; pub use flight_actions::FlightActions; @@ -44,10 +43,10 @@ pub use truncate_table::TRUNCATE_TABLE; use crate::sessions::Session; use crate::sessions::SessionManager; -pub(crate) fn create_session(tenant: Option) -> Result> { +pub(crate) fn create_session() -> Result> { let config = GlobalConfig::instance(); - let tenant_id = tenant.unwrap_or(config.query.tenant_id.clone()); - let settings = Settings::create(tenant_id.clone()); + let tenant_id = config.query.tenant_id.clone(); + let settings = Settings::create(tenant_id); match SessionManager::instance().create_with_settings(SessionType::FlightRPC, settings, None) { Err(cause) => Err(cause), Ok(session) => Ok(Arc::new(session)), diff --git a/src/query/service/src/servers/flight/v1/actions/set_priority.rs b/src/query/service/src/servers/flight/v1/actions/set_priority.rs index 9e5bd534a43e3..6767a02904f3a 100644 --- a/src/query/service/src/servers/flight/v1/actions/set_priority.rs +++ b/src/query/service/src/servers/flight/v1/actions/set_priority.rs @@ -24,7 +24,7 @@ use crate::servers::flight::v1::actions::create_session; pub static SET_PRIORITY: &str = "/actions/set_priority"; pub async fn set_priority(plan: SetPriorityPlan) -> Result { - let session = create_session(None)?; + let session = create_session()?; let version = GlobalConfig::version(); let query_context = session.create_query_context(version).await?; let interpreter = SetPriorityInterpreter::from_flight(query_context, plan)?; diff --git a/src/query/service/src/servers/flight/v1/actions/system_action.rs b/src/query/service/src/servers/flight/v1/actions/system_action.rs index 3b13a10a29ba3..a234793eff00c 100644 --- a/src/query/service/src/servers/flight/v1/actions/system_action.rs +++ b/src/query/service/src/servers/flight/v1/actions/system_action.rs @@ -23,7 +23,7 @@ use crate::servers::flight::v1::actions::create_session; pub static SYSTEM_ACTION: &str = "/actions/system_action"; pub async fn system_action(plan: SystemPlan) -> Result<()> { - let session = create_session(Some(plan.clone().tenant))?; + let session = create_session()?; let version = GlobalConfig::version(); let query_context = session.create_query_context(version).await?; let interpreter = SystemActionInterpreter::from_flight(query_context, plan)?; diff --git a/src/query/service/src/servers/flight/v1/actions/truncate_table.rs b/src/query/service/src/servers/flight/v1/actions/truncate_table.rs index cc03070c774ca..e36b81d246d9d 100644 --- a/src/query/service/src/servers/flight/v1/actions/truncate_table.rs +++ b/src/query/service/src/servers/flight/v1/actions/truncate_table.rs @@ -23,7 +23,7 @@ use crate::servers::flight::v1::actions::create_session; pub static TRUNCATE_TABLE: &str = "/actions/truncate_table"; pub async fn truncate_table(plan: TruncateTablePlan) -> Result<()> { - let session = create_session(None)?; + let session = create_session()?; let version = GlobalConfig::version(); let query_context = session.create_query_context(version).await?; let interpreter = TruncateTableInterpreter::from_flight(query_context, plan)?; diff --git a/src/query/sql/src/planner/binder/system.rs b/src/query/sql/src/planner/binder/system.rs index a4e18545b4552..aa9d0107b784f 100644 --- a/src/query/sql/src/planner/binder/system.rs +++ b/src/query/sql/src/planner/binder/system.rs @@ -25,15 +25,12 @@ impl Binder { #[async_backtrace::framed] pub(super) async fn bind_system(&mut self, stmt: &SystemStmt) -> Result { let SystemStmt { action } = stmt; - let tenant = self.ctx.get_tenant(); match action { AstSystemAction::Backtrace(switch) => Ok(Plan::System(Box::new(SystemPlan { action: SystemAction::Backtrace(*switch), - tenant, }))), AstSystemAction::FlushPrivileges => Ok(Plan::System(Box::new(SystemPlan { action: SystemAction::FlushPrivileges, - tenant, }))), } } diff --git a/src/query/sql/src/planner/plans/system.rs b/src/query/sql/src/planner/plans/system.rs index be0e8b45b839b..e9901059af00e 100644 --- a/src/query/sql/src/planner/plans/system.rs +++ b/src/query/sql/src/planner/plans/system.rs @@ -12,14 +12,12 @@ // See the License for the specific language governing permissions and // limitations under the License. -use databend_common_meta_app::tenant::Tenant; use serde::Deserialize; use serde::Serialize; #[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Eq)] pub struct SystemPlan { pub action: SystemAction, - pub tenant: Tenant, } #[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]