feat(cubestore): add stale-while-revalidate timeout for SQL query cache (cube-js#10822)

* feat(cubestore): add stale-while-revalidate timeout for SQL query cache

Add CUBESTORE_QUERY_CACHE_STALE_WHILE_REVALIDATE env config param that
enables stale-while-revalidate behavior for CubeStore's SQL result cache.

When enabled (set to a number of seconds, e.g. 30), the cache maintains
a secondary lookup keyed only by SQL query text + inline tables (ignoring
partition/chunk IDs). If a query arrives and the exact cache key (which
includes current partition/chunk snapshot) misses, but a stale entry
exists and was created within the configured timeout, the stale result is
served immediately while a background task refreshes the cache with fresh
data.

This is disabled by default (0 = disabled). The feature reduces query
latency during compaction or partition changes when the underlying data
hasn't materially changed.

Changes:
- config/mod.rs: Add query_cache_stale_while_revalidate_secs to ConfigObj
  trait and ConfigObjImpl, parsed from CUBESTORE_QUERY_CACHE_STALE_WHILE_REVALIDATE
- sql/cache.rs: Add stale_cache (Moka cache keyed by SqlQueueCacheKey),
  stale-while-revalidate lookup in get(), background refresh via tokio::spawn
- app_metrics.rs: Add DATA_QUERIES_CACHE_STALE_HIT counter metric
- Unit test for stale-while-revalidate behavior

Co-authored-by: Pavel Tiunov <pavel.tiunov@gmail.com>

* refactor(cubestore): use queue cache for background stale-while-revalidate refresh

The background refresh spawned by stale-while-revalidate now integrates
with the queue cache dedup mechanism:

- Extract try_register_background() which atomically checks if a query is
  already in flight in the queue cache and registers a new entry if not.
- On stale hit, try_register_background() is called first. If another
  execution is already in flight (from a previous stale hit or a
  concurrent foreground request), we skip spawning entirely.
- The background spawn registers a watch::Sender in the queue cache, so
  any concurrent foreground get() for the same SQL that doesn't hit the
  stale cache will join as a waiter instead of starting a duplicate
  execution.
- Queue cache entry cleanup happens lazily: sender is dropped when the
  background task completes, closing the channel. The next get() for that
  key detects the closed channel via has_changed().is_err() and pops it.

Co-authored-by: Pavel Tiunov <pavel.tiunov@gmail.com>

* refactor(cubestore): reuse queue cache flow for background stale-while-revalidate

Extract get_inner() with is_background_refresh flag from get(). The
background spawn now calls get_inner(true) which skips the stale cache
check but reuses the entire queue cache dedup flow — no separate
implementation for the background path.

- get() is a thin wrapper: delegates to get_inner(false)
- get_inner(is_background_refresh=true) skips stale check, goes
  straight to queue cache logic for natural dedup
- get() now takes self: &Arc<Self> so the spawn can clone the Arc
- get_inner() takes self: Arc<Self> and returns Pin<Box<dyn Future>>
  to satisfy Send + 'static bounds for tokio::spawn
- Removed try_register_background() and all duplicated cache update
  logic from previous approach

Co-authored-by: Pavel Tiunov <pavel.tiunov@gmail.com>

* fix(cubestore): fix stale-while-revalidate test to simulate partition change

The test now invalidates the exact result cache before the second get()
call to simulate a partition/chunk change that would make the exact cache
key miss while the stale entry (keyed by SQL only) still matches.

Co-authored-by: Pavel Tiunov <pavel.tiunov@gmail.com>

* fix(cubestore): address review feedback on stale-while-revalidate cache

- Add weigher to stale_cache so max_capacity is in bytes, not entry count
- Fix clippy: use Duration::from_secs as method reference instead of closure
- Add test: background refresh failure preserves stale entry for next request
- Add test: stale timeout expiry falls through to fresh execution

Co-authored-by: Pavel Tiunov <pavel.tiunov@gmail.com>

* feat(cubestore): add size/weight metrics for stale cache

Add cs.sql.query.data.cache.stale.size and cs.sql.query.data.cache.stale.weight
gauge metrics mirroring the existing result cache metrics. Reported alongside
result cache metrics on cache mutations and zeroed on drop.

Co-authored-by: Pavel Tiunov <pavel.tiunov@gmail.com>

---------

Co-authored-by: Cursor Agent <cursoragent@cursor.com>

Author: paveltiunov

rust/cubestore/cubestore/src/app_metrics.rs

@@ -13,10 +13,16 @@ pub static WORKER_POOL_ERROR: Counter = metrics::counter("cs.worker_pool.errors"
 /// Incoming SQL queries that do data reads.
 pub static DATA_QUERIES: Counter = metrics::counter("cs.sql.query.data");
 pub static DATA_QUERIES_CACHE_HIT: Counter = metrics::counter("cs.sql.query.data.cache.hit");
+pub static DATA_QUERIES_CACHE_STALE_HIT: Counter =
+    metrics::counter("cs.sql.query.data.cache.stale_hit");
 // Approximate number of entries in this cache.
 pub static DATA_QUERIES_CACHE_SIZE: Gauge = metrics::gauge("cs.sql.query.data.cache.size");
 // Approximate total weighted size of entries in this cache.
 pub static DATA_QUERIES_CACHE_WEIGHT: Gauge = metrics::gauge("cs.sql.query.data.cache.weight");
+pub static DATA_QUERIES_STALE_CACHE_SIZE: Gauge =
+    metrics::gauge("cs.sql.query.data.cache.stale.size");
+pub static DATA_QUERIES_STALE_CACHE_WEIGHT: Gauge =
+    metrics::gauge("cs.sql.query.data.cache.stale.weight");
 pub static DATA_QUERY_TIME_MS: Histogram = metrics::histogram("cs.sql.query.data.ms");
 pub static DATA_QUERY_LOGICAL_PLAN_TOTAL_CREATION_TIME_US: Histogram =
     metrics::histogram("cs.sql.query.data.planning.logical_plan.total_creation.us");

rust/cubestore/cubestore/src/config/mod.rs

@@ -510,6 +510,8 @@ pub trait ConfigObj: DIService {
 
     fn query_cache_time_to_idle_secs(&self) -> Option<u64>;
 
+    fn query_cache_stale_while_revalidate_secs(&self) -> Option<u64>;
+
     fn metadata_cache_max_capacity_bytes(&self) -> u64;
 
     fn metadata_cache_time_to_idle_secs(&self) -> u64;
@@ -643,6 +645,7 @@ pub struct ConfigObjImpl {
     pub query_cache_max_capacity_bytes: u64,
     pub query_queue_cache_max_capacity: u64,
     pub query_cache_time_to_idle_secs: Option<u64>,
+    pub query_cache_stale_while_revalidate_secs: Option<u64>,
     pub metadata_cache_max_capacity_bytes: u64,
     pub metadata_cache_time_to_idle_secs: u64,
     pub stream_replay_check_interval_secs: u64,
@@ -956,6 +959,9 @@ impl ConfigObj for ConfigObjImpl {
     fn query_cache_time_to_idle_secs(&self) -> Option<u64> {
         self.query_cache_time_to_idle_secs
     }
+    fn query_cache_stale_while_revalidate_secs(&self) -> Option<u64> {
+        self.query_cache_stale_while_revalidate_secs
+    }
 
     fn metadata_cache_max_capacity_bytes(&self) -> u64 {
         self.metadata_cache_max_capacity_bytes
@@ -1233,6 +1239,8 @@ impl Config {
 
     pub fn default() -> Config {
         let query_timeout = env_parse("CUBESTORE_QUERY_TIMEOUT", 120);
+        let query_cache_stale_while_revalidate_secs: u64 =
+            env_parse("CUBESTORE_QUERY_CACHE_STALE_WHILE_REVALIDATE", 0);
         let query_cache_time_to_idle_secs = env_parse(
             "CUBESTORE_QUERY_CACHE_TIME_TO_IDLE",
             // 1 hour
@@ -1528,6 +1536,13 @@ impl Config {
                 } else {
                     Some(query_cache_time_to_idle_secs)
                 },
+                query_cache_stale_while_revalidate_secs: if query_cache_stale_while_revalidate_secs
+                    == 0
+                {
+                    None
+                } else {
+                    Some(query_cache_stale_while_revalidate_secs)
+                },
                 metadata_cache_max_capacity_bytes: env_parse(
                     "CUBESTORE_METADATA_CACHE_MAX_CAPACITY_BYTES",
                     0,
@@ -1740,6 +1755,7 @@ impl Config {
                 query_cache_max_capacity_bytes: 512 << 20,
                 query_queue_cache_max_capacity: 10000,
                 query_cache_time_to_idle_secs: Some(600),
+                query_cache_stale_while_revalidate_secs: None,
                 metadata_cache_max_capacity_bytes: 0,
                 metadata_cache_time_to_idle_secs: 1_000,
                 meta_store_log_upload_interval: 30,
@@ -2393,6 +2409,7 @@ impl Config {
             self.config_obj.query_cache_max_capacity_bytes(),
             self.config_obj.query_cache_time_to_idle_secs(),
             self.config_obj.query_queue_cache_max_capacity(),
+            self.config_obj.query_cache_stale_while_revalidate_secs(),
         ));
 
         let query_cache_to_move = query_cache.clone();

rust/cubestore/cubestore/src/queryplanner/mod.rs

@@ -1063,7 +1063,7 @@ pub mod tests {
             Arc::new(test_utils::MetaStoreMock {}),
             Arc::new(test_utils::CacheStoreMock {}),
             &vec![],
-            Arc::new(SqlResultCache::new(1 << 20, None, 10000)),
+            Arc::new(SqlResultCache::new(1 << 20, None, 10000, None)),
             Arc::new(SessionContext::new().state()),
         )
     }