feat(query): add RAP definition cache (#19757)

* feat(query): add RAP definition cache

Add a query-node local row access policy definition cache keyed by (tenant, policy_id) to avoid repeated meta-store RPCs during bind.

The cache uses DashMap<(Tenant, u64), Arc<OnceCell<_>>> so cache hits stay cheap and concurrent misses on the same key share one in-flight load. Errors are not cached.

Also add sqllogictest coverage for attach/detach/re-attach cycles and a periodic background cleanup task to reclaim orphaned entries left on remote nodes after DROP ROW ACCESS POLICY.

* chore(test): remove redundant RAP definition cache tests

The RAP definition cache is keyed by (tenant, policy_id) where policy_id
is a monotonic meta-store sequence number. Cache correctness relies on:

1. Immutability: a given policy_id always maps to the same definition,
   so a cached entry can never become stale while the policy exists.
2. Invalidation on drop: the DROP ROW ACCESS POLICY interpreter calls
   invalidate(tenant, policy_id), removing the entry on the node that
   executes the DDL.
3. Bounded staleness on remote nodes: a background task clears the
   entire cache every 5 minutes, so remote nodes converge within one
   cleanup interval after a policy is dropped.

Given these properties, the removed tests add no unique coverage:

- The inline unit test only verified DashMap::remove semantics.
- The sqllogictest (05_0016) duplicated attach/detach/re-attach cycles
  already covered by 05_0004_ddl_security_policy.test, and its
  consecutive-query approach cannot observably confirm a cache hit
  (unlike 05_0011 which uses EXPLAIN to verify result cache behavior).

Author: TCeason

Cargo.lock

@@ -9,11 +9,14 @@ edition = { workspace = true }
 
 [dependencies]
 async-trait = { workspace = true }
+dashmap = { workspace = true }
 databend-common-base = { workspace = true }
 databend-common-exception = { workspace = true }
 databend-common-meta-app = { workspace = true }
 databend-common-meta-store = { workspace = true }
 databend-meta-client = { workspace = true }
+log = { workspace = true }
+tokio = { workspace = true }
 
 [build-dependencies]
 

src/query/ee_features/row_access_policy/Cargo.toml

@@ -12,8 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod row_access_policy_cache;
 pub mod row_access_policy_handler;
 
+pub use row_access_policy_cache::RowAccessPolicyCacheManager;
 pub use row_access_policy_handler::RowAccessPolicyHandler;
 pub use row_access_policy_handler::RowAccessPolicyHandlerWrapper;
 pub use row_access_policy_handler::get_row_access_policy_handler;

src/query/ee_features/row_access_policy/src/lib.rs

@@ -0,0 +1,141 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use dashmap::DashMap;
+use databend_common_base::base::GlobalInstance;
+use databend_common_exception::Result;
+use databend_common_meta_app::row_access_policy::RowAccessPolicyMeta;
+use databend_common_meta_app::tenant::Tenant;
+use databend_common_meta_store::MetaStore;
+use databend_meta_client::types::SeqV;
+use log::debug;
+use log::info;
+use tokio::sync::OnceCell;
+
+use crate::row_access_policy_handler::get_row_access_policy_handler;
+
+/// Background cleanup interval for orphaned cache entries.
+/// Entries for dropped policies on remote nodes are never accessed again,
+/// so periodic full clear is the simplest way to bound memory growth.
+const RAP_CACHE_CLEANUP_INTERVAL: Duration = Duration::from_secs(300);
+
+/// Cache key: (tenant, policy_id).
+type CacheKey = (Tenant, u64);
+
+/// Each entry is an `Arc<OnceCell<...>>` so that concurrent misses on the same
+/// key only trigger one meta-store RPC (singleflight).
+///
+/// `OnceCell::get_or_try_init` will NOT persist an `Err` — if the first caller
+/// fails, the cell stays empty and the next caller retries.
+type CacheEntry = Arc<OnceCell<SeqV<RowAccessPolicyMeta>>>;
+
+pub struct RowAccessPolicyCacheManager {
+    cache: DashMap<CacheKey, CacheEntry>,
+}
+
+impl RowAccessPolicyCacheManager {
+    fn new() -> Self {
+        Self {
+            cache: DashMap::new(),
+        }
+    }
+
+    pub fn init() -> Result<()> {
+        let manager = Arc::new(Self::new());
+        GlobalInstance::set(manager.clone());
+
+        // Background task to reclaim orphaned entries.
+        // On remote nodes, entries for dropped policies are never looked up
+        // again (policy_id is monotonic), so periodic clear is the only way
+        // to reclaim them.
+        databend_common_base::runtime::spawn(async move {
+            loop {
+                tokio::time::sleep(RAP_CACHE_CLEANUP_INTERVAL).await;
+                let count = manager.cache.len();
+                if count > 0 {
+                    manager.cache.clear();
+                    debug!(
+                        "row_access_policy_cache: cleared {} orphaned/stale entries",
+                        count
+                    );
+                }
+            }
+        });
+
+        Ok(())
+    }
+
+    pub fn instance() -> Arc<RowAccessPolicyCacheManager> {
+        GlobalInstance::get()
+    }
+
+    /// Synchronous fast path — returns `Some` only when the definition is
+    /// already cached.  The binder calls this first to avoid `block_on`
+    /// overhead on cache hits.
+    pub fn try_get(&self, tenant: &Tenant, policy_id: u64) -> Option<SeqV<RowAccessPolicyMeta>> {
+        let key = (tenant.clone(), policy_id);
+        self.cache.get(&key).and_then(|cell| cell.get().cloned())
+    }
+
+    /// Invalidate one cached definition when the underlying RAP is dropped.
+    pub fn invalidate(&self, tenant: &Tenant, policy_id: u64) {
+        let key = (tenant.clone(), policy_id);
+        self.cache.remove(&key);
+    }
+
+    /// Async path — used on cache miss.  Guarantees singleflight: concurrent
+    /// callers for the same `(tenant, policy_id)` share one in-flight RPC.
+    /// Errors are never cached.
+    pub async fn get_or_load(
+        &self,
+        meta_api: Arc<MetaStore>,
+        tenant: &Tenant,
+        policy_id: u64,
+    ) -> Result<SeqV<RowAccessPolicyMeta>> {
+        let key = (tenant.clone(), policy_id);
+
+        // Get-or-insert the OnceCell for this key.
+        let cell = self
+            .cache
+            .entry(key)
+            .or_insert_with(|| Arc::new(OnceCell::new()))
+            .value()
+            .clone();
+
+        let result = cell
+            .get_or_try_init(|| {
+                let meta_api = meta_api.clone();
+                let tenant = tenant.clone();
+                async move {
+                    let handler = get_row_access_policy_handler();
+                    let res = handler
+                        .get_row_access_policy_by_id(meta_api, &tenant, policy_id)
+                        .await;
+                    if res.is_ok() {
+                        info!(
+                            "row_access_policy_cache: loaded policy_id={} from meta store",
+                            policy_id
+                        );
+                    }
+                    res
+                }
+            })
+            .await?;
+
+        Ok(result.clone())
+    }
+}

src/query/ee_features/row_access_policy/src/row_access_policy_cache.rs

@@ -41,6 +41,7 @@ use databend_common_users::RoleCacheManager;
 use databend_common_users::UserApiProvider;
 use databend_common_users::builtin::BuiltIn;
 use databend_enterprise_resources_management::DummyResourcesManagement;
+use databend_enterprise_row_access_policy_feature::RowAccessPolicyCacheManager;
 use databend_meta_runtime::DatabendRuntime;
 use databend_storages_common_cache::CacheManager;
 use databend_storages_common_cache::TempDirManager;
@@ -164,6 +165,7 @@ impl GlobalServices {
             .await?;
         }
         RoleCacheManager::init()?;
+        RowAccessPolicyCacheManager::init()?;
 
         DataOperator::init(&config.storage, config.spill.storage_params.clone()).await?;
         ShareTableConfig::init(

src/query/service/src/global_services.rs

@@ -23,6 +23,7 @@ use databend_common_meta_app::principal::OwnershipObject;
 use databend_common_sql::plans::DropRowAccessPolicyPlan;
 use databend_common_users::RoleCacheManager;
 use databend_common_users::UserApiProvider;
+use databend_enterprise_row_access_policy_feature::RowAccessPolicyCacheManager;
 use databend_enterprise_row_access_policy_feature::get_row_access_policy_handler;
 
 use crate::interpreters::Interpreter;
@@ -80,6 +81,7 @@ impl Interpreter for DropRowAccessPolicyInterpreter {
         }
 
         if let Some(policy_id) = policy_id {
+            RowAccessPolicyCacheManager::instance().invalidate(&tenant, policy_id);
             let role_api = UserApiProvider::instance().role_api(&tenant);
             role_api
                 .revoke_ownership(&OwnershipObject::RowAccessPolicy { policy_id })

src/query/service/src/interpreters/interpreter_row_access_policy_drop.rs

@@ -63,7 +63,7 @@ use databend_common_meta_app::tenant::Tenant;
 use databend_common_storage::StageFileInfo;
 use databend_common_storage::StageFilesInfo;
 use databend_common_users::UserApiProvider;
-use databend_enterprise_row_access_policy_feature::get_row_access_policy_handler;
+use databend_enterprise_row_access_policy_feature::RowAccessPolicyCacheManager;
 use databend_meta_client::types::MetaId;
 use databend_storages_common_table_meta::table::ChangeType;
 use log::debug;
@@ -502,8 +502,6 @@ impl Binder {
     ) -> Result<SExpr> {
         LicenseManagerSwitch::instance()
             .check_enterprise_enabled(self.ctx.get_license_key(), Feature::RowAccessPolicy)?;
-        let meta_api = UserApiProvider::instance().get_meta_store_client();
-        let handler = get_row_access_policy_handler();
         // Collect arguments in policy.columns_ids order (matches the policy parameter list).
         // Previously this iterated `fields` (schema order) with a contains-filter, which
         // silently reordered arguments when USING column order differed from schema order.
@@ -521,18 +519,32 @@ impl Binder {
             })
             .collect();
         let policy = policy.policy_id;
+        let tenant = self.ctx.get_tenant();
+        let cache = RowAccessPolicyCacheManager::instance();
         let start = std::time::Instant::now();
-        let res = databend_common_base::runtime::block_on(handler.get_row_access_policy_by_id(
-            meta_api,
-            &self.ctx.get_tenant(),
-            policy,
-        ))?;
-        let fetch_elapsed = start.elapsed();
-        info!(
-            "row_access_policy: policy_id={}, fetch_ms={:.3}",
-            policy,
-            fetch_elapsed.as_secs_f64() * 1000.0,
-        );
+        // Fast sync path: avoid block_on overhead on cache hits.
+        let res = match cache.try_get(&tenant, policy) {
+            Some(cached) => {
+                debug!(
+                    "row_access_policy: policy_id={}, cache_hit, elapsed_ms={:.3}",
+                    policy,
+                    start.elapsed().as_secs_f64() * 1000.0,
+                );
+                cached
+            }
+            None => {
+                let meta_api = UserApiProvider::instance().get_meta_store_client();
+                let loaded = databend_common_base::runtime::block_on(
+                    cache.get_or_load(meta_api, &tenant, policy),
+                )?;
+                info!(
+                    "row_access_policy: policy_id={}, cache_miss, fetch_ms={:.3}",
+                    policy,
+                    start.elapsed().as_secs_f64() * 1000.0,
+                );
+                loaded
+            }
+        };
         let body = res.data.body;
         let settings = self.ctx.get_settings();
         let sql_dialect = settings.get_sql_dialect()?;