fix: report query depth instead of failing or crashing on it

rust/cube/cubesqlplanner/cubesqlplanner/src/planner/planners/multi_stage/depth_guard.rs

@@ -0,0 +1,101 @@
+//! Depth guard for chained multi-stage members.
+//!
+//! Every multi-stage member on a dependency path becomes its own CTE, and `MultiStageQueryPlanner`
+//! plans them by descending from a member into its children. Planning runs synchronously on the
+//! caller's thread, so a long enough chain exhausts that thread's stack and the process aborts —
+//! a crash no error handler sees, on either side of the native boundary.
+//!
+//! The guard turns that into a reportable error. It measures only multi-stage members that are
+//! planned as a stage: calculation chains recurse as well, but cost a fraction of a stage per
+//! level, and a reference -- a view member proxying one, say -- inherits `multi_stage` from what
+//! it resolves to while planning collapses it into that member and opens no stage of its own.
+
+use crate::planner::symbols::MemberSymbol;
+use cubenativeutils::CubeError;
+use std::collections::{HashMap, HashSet};
+use std::rc::Rc;
+use std::sync::OnceLock;
+
+/// Multi-stage members a single dependency path may carry.
+///
+/// Far above any hand-written model — real ones chain a handful of stages — and below the depth
+/// at which planning runs out of stack, which a release build reaches around a few hundred
+/// minimal stages, sooner for heavy ones and sooner still on a smaller caller stack.
+const DEFAULT_MAX_MULTI_STAGE_DEPTH: usize = 32;
+
+fn max_multi_stage_depth() -> usize {
+    static MAX_DEPTH: OnceLock<usize> = OnceLock::new();
+    *MAX_DEPTH.get_or_init(|| match std::env::var("CUBEJS_MAX_MULTI_STAGE_DEPTH") {
+        // A malformed value falls back to the default rather than refusing to plan: this is a
+        // safety valve, and a typo in it must not take queries down.
+        Ok(value) => match value.parse::<usize>() {
+            Ok(0) | Err(_) => DEFAULT_MAX_MULTI_STAGE_DEPTH,
+            Ok(depth) => depth,
+        },
+        Err(_) => DEFAULT_MAX_MULTI_STAGE_DEPTH,
+    })
+}
+
+pub fn check_multi_stage_depth(roots: &[Rc<MemberSymbol>]) -> Result<(), CubeError> {
+    let limit = max_multi_stage_depth();
+    for root in roots {
+        let depth = multi_stage_depth(root);
+        if depth > limit {
+            return Err(CubeError::user(format!(
+                "Member '{}' chains {} multi-stage members deep, against a limit of {}. Each \
+                 one is planned as a separate stage, and this many cannot be planned. Collapse \
+                 the intermediate stages into fewer members, or raise \
+                 CUBEJS_MAX_MULTI_STAGE_DEPTH.",
+                root.full_name(),
+                depth,
+                limit
+            )));
+        }
+    }
+    Ok(())
+}
+
+/// Multi-stage members on the longest dependency path starting at `root`, `root` included.
+///
+/// Walks an explicit stack and memoizes per symbol identity: the graphs this guards against are
+/// exactly the ones a recursive walk could not survive, and a member reachable by many paths
+/// must not be re-expanded per path.
+fn multi_stage_depth(root: &Rc<MemberSymbol>) -> usize {
+    let mut depth_below: HashMap<*const MemberSymbol, usize> = HashMap::new();
+    let mut on_path: HashSet<*const MemberSymbol> = HashSet::new();
+    // Addresses are the memo keys, so every symbol one stands for has to outlive the walk;
+    // otherwise a freed symbol's address could come back as a different one.
+    let mut measured: Vec<Rc<MemberSymbol>> = Vec::new();
+    let mut pending = vec![(root.clone(), false)];
+
+    while let Some((symbol, dependencies_visited)) = pending.pop() {
+        let key = Rc::as_ptr(&symbol);
+        if dependencies_visited {
+            let below = symbol
+                .get_dependencies()
+                .iter()
+                .filter_map(|dependency| depth_below.get(&Rc::as_ptr(dependency)))
+                .copied()
+                .max()
+                .unwrap_or(0);
+            let opens_a_stage = symbol.is_multi_stage() && !symbol.is_reference();
+            depth_below.insert(key, below + usize::from(opens_a_stage));
+            on_path.remove(&key);
+            measured.push(symbol);
+            continue;
+        }
+        if depth_below.contains_key(&key) || !on_path.insert(key) {
+            // Already measured, or a cycle: a cyclic model does not terminate in planning
+            // either, and this walk must not be the thing that hangs on it.
+            continue;
+        }
+        let dependencies = symbol.get_dependencies();
+        pending.push((symbol, true));
+        pending.extend(dependencies.into_iter().map(|d| (d, false)));
+    }
+
+    depth_below
+        .get(&Rc::as_ptr(root))
+        .copied()
+        .unwrap_or_default()
+}

rust/cube/cubesqlplanner/cubesqlplanner/src/planner/planners/multi_stage/mod.rs

@@ -1,10 +1,12 @@
+mod depth_guard;
 mod member;
 mod member_query_planner;
 mod multi_stage_query_planner;
 mod planning_scope;
 mod query_description;
 mod time_shift_state;
 
+pub use depth_guard::check_multi_stage_depth;
 pub use member::*;
 pub use member_query_planner::MultiStageMemberQueryPlanner;
 pub use multi_stage_query_planner::MultiStageQueryPlanner;

rust/cube/cubesqlplanner/cubesqlplanner/src/planner/top_level_planner.rs

@@ -1,4 +1,4 @@
-use super::planners::multi_stage::PlanningScope;
+use super::planners::multi_stage::{check_multi_stage_depth, PlanningScope};
 use super::planners::QueryPlanner;
 use super::state::State;
 use super::QueryProperties;
@@ -31,6 +31,8 @@ impl TopLevelPlanner {
     }
 
     pub fn plan(&self) -> Result<(String, Vec<PreAggregationUsage>), CubeError> {
+        check_multi_stage_depth(&self.request.all_used_symbols()?)?;
+
         let query_planner = QueryPlanner::new(self.request.clone(), self.query_tools.clone());
         let mut scope = PlanningScope::new();
         let query = query_planner.plan(&mut scope)?;

rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/cubestore/snapshots/cubesqlplanner__tests__integration__cubestore__switch_rolling__switch_rolling_four_measures.snap

@@ -0,0 +1,11 @@
+---
+source: cubesqlplanner/cubesqlplanner/src/tests/integration/cubestore/switch_rolling.rs
+expression: normalize(result)
+---
+sales__category|sales__created_at_month|sales__rolling_amount|sales__prev_rolling_amount|sales__rolling_amount_change|sales__rolling_amount_growth
+books|2024-04-01 00:00:00|900.0000000000|157.0000000000|743.0000000000|4.7324840764
+books|2024-05-01 00:00:00|1202.0000000000|350.0000000000|852.0000000000|2.4342857143
+books|2024-06-01 00:00:00|1502.0000000000|600.0000000000|902.0000000000|1.5033333333
+toys|2024-04-01 00:00:00|90.0000000000|18.0000000000|72.0000000000|4.0000000000
+toys|2024-05-01 00:00:00|121.0000000000|35.0000000000|86.0000000000|2.4571428571
+toys|2024-06-01 00:00:00|151.0000000000|60.0000000000|91.0000000000|1.5166666667

rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/cubestore/switch_rolling.rs

@@ -6,8 +6,6 @@
 //! Requires `--features integration-cubestore` and a `cubestored` binary;
 //! without them both paths return `None` and only matching is asserted.
 //!
-//! One shape is `#[ignore]`d for a CubeStore limitation unrelated to the
-//! calc-group grain — see the note on it; run it with `--ignored` to reproduce.
 //! Two defects these tests were originally written against are fixed: the
 //! rolling-rewrite schema widening in #11410 and the aggregating-index decimal
 //! cast in #11413.
@@ -398,17 +396,7 @@ async fn test_growth_case_entrypoint() {
 }
 
 /// Four switch entrypoints at once — the dashboard shape, and the deepest
-/// FullKeyAggregate plan this model produces. CubeStore cannot decode a
-/// serialized plan this deep:
-///
-/// ```text
-/// Error during planning: Error decoding expr as protobuf: failed to decode
-/// Protobuf message: ... recursion limit reached
-/// ```
-///
-/// Unrelated to the calc-group grain — the plan is valid and the raw half of
-/// this test returns the expected rows.
-#[ignore = "CubeStore cannot decode a serialized plan this deep (protobuf recursion limit)"]
+/// FullKeyAggregate plan this model produces.
 #[tokio::test(flavor = "multi_thread")]
 async fn test_four_entrypoints_in_one_query() {
     run_both(

rust/cube/cubesqlplanner/cubesqlplanner/src/tests/mod.rs

@@ -14,6 +14,7 @@ mod filter_params_time_shifts;
 mod join_hints_collector;
 mod measure_symbol;
 mod member_expressions_on_views;
+mod multi_stage_depth;
 mod no_query_tools_leak;
 mod positional_params;
 mod string_measures;

rust/cube/cubesqlplanner/cubesqlplanner/src/tests/multi_stage_depth.rs

@@ -0,0 +1,169 @@
+use crate::test_fixtures::cube_bridge::MockSchema;
+use crate::test_fixtures::test_utils::TestContext;
+use indoc::indoc;
+
+/// The documented default of `CUBEJS_MAX_MULTI_STAGE_DEPTH`, spelled out so that changing the
+/// default has to come with a decision about these cases.
+const DEFAULT_LIMIT: usize = 32;
+
+const CUBE_HEADER: &str = indoc! {r#"
+    cubes:
+        - name: orders
+          sql: "SELECT * FROM ms_orders"
+          dimensions:
+              - name: id
+                type: number
+                sql: id
+                primary_key: true
+              - name: category
+                type: string
+                sql: category
+          measures:
+              - name: amount
+                type: sum
+                sql: amount
+"#};
+
+/// `stages` measures, each computing over the previous one as its own multi-stage stage, so the
+/// deepest member carries a chain of `stages` + 1 multi-stage members.
+fn chained_stages_schema(stages: usize) -> String {
+    let mut yaml = String::from(CUBE_HEADER);
+    yaml.push_str(concat!(
+        "          - name: stage_0\n",
+        "            type: number\n",
+        "            sql: \"{CUBE.amount}\"\n",
+        "            multi_stage: true\n",
+        "            add_group_by:\n",
+        "                - orders.category\n",
+    ));
+    for stage in 1..stages {
+        yaml.push_str(&format!(
+            "          - name: stage_{stage}\n            type: number\n            sql: \"{{CUBE.stage_{previous}}} + 1\"\n            multi_stage: true\n            add_group_by:\n                - orders.category\n",
+            stage = stage,
+            previous = stage - 1
+        ));
+    }
+    yaml
+}
+
+/// The same chain length built from plain calculated measures, which are not stages.
+fn chained_plain_schema(members: usize) -> String {
+    let mut yaml = String::from(CUBE_HEADER);
+    yaml.push_str(concat!(
+        "          - name: plain_0\n",
+        "            type: number\n",
+        "            sql: \"{CUBE.amount}\"\n",
+    ));
+    for member in 1..members {
+        yaml.push_str(&format!(
+            "          - name: plain_{member}\n            type: number\n            sql: \"{{CUBE.plain_{previous}}} + 1\"\n",
+            member = member,
+            previous = member - 1
+        ));
+    }
+    yaml
+}
+
+/// The same chain, plus a view re-exporting its deepest measure. A view member is a proxy that
+/// inherits `multi_stage` from what it resolves to and adds a dependency level of its own.
+fn chained_stages_view_schema(stages: usize) -> String {
+    let mut yaml = chained_stages_schema(stages);
+    yaml.push_str(&format!(
+        "views:\n    - name: orders_view\n      cubes:\n          - join_path: orders\n            includes:\n                - category\n                - stage_{}\n",
+        stages - 1
+    ));
+    yaml
+}
+
+fn query_for(measure: &str) -> String {
+    format!(
+        indoc! {r#"
+            measures:
+              - orders.{}
+            dimensions:
+              - orders.category
+        "#},
+        measure
+    )
+}
+
+fn build(yaml: &str, measure: &str) -> Result<String, cubenativeutils::CubeError> {
+    let schema = MockSchema::from_yaml(yaml).unwrap();
+    TestContext::new(schema)
+        .unwrap()
+        .build_sql(&query_for(measure))
+}
+
+fn build_on_view(yaml: &str, measure: &str) -> Result<String, cubenativeutils::CubeError> {
+    let schema = MockSchema::from_yaml(yaml).unwrap();
+    TestContext::new(schema).unwrap().build_sql(&format!(
+        "measures:\n  - orders_view.{}\ndimensions:\n  - orders_view.category\n",
+        measure
+    ))
+}
+
+#[test]
+fn plans_a_chain_up_to_the_depth_limit() {
+    let stages = DEFAULT_LIMIT;
+    let sql = build(
+        &chained_stages_schema(stages),
+        &format!("stage_{}", stages - 1),
+    )
+    .unwrap();
+    assert_eq!(sql.matches(" AS (").count(), stages + 1);
+}
+
+#[test]
+fn chain_past_the_depth_limit_names_depth() {
+    let stages = DEFAULT_LIMIT + 8;
+    let err = build(
+        &chained_stages_schema(stages),
+        &format!("stage_{}", stages - 1),
+    )
+    .map(|_| ())
+    .expect_err("a chain past the limit must be refused, not planned");
+
+    let message = err.to_string();
+    assert!(
+        message.contains(&format!(
+            "chains {} multi-stage members deep, against a limit of {}",
+            stages, DEFAULT_LIMIT
+        )),
+        "message must name the depth reached and the budget, got: {}",
+        message
+    );
+    assert!(
+        message.contains(&format!("orders.stage_{}", stages - 1)),
+        "message must name the member the chain hangs from, got: {}",
+        message
+    );
+    assert!(
+        message.contains("CUBEJS_MAX_MULTI_STAGE_DEPTH"),
+        "message must name the knob that raises the budget, got: {}",
+        message
+    );
+}
+
+/// Only stages count. Reference and calculation chains recurse as well, but cost a fraction of
+/// a stage per level, and models built over views reach depths a stage budget would refuse.
+#[test]
+fn a_plain_member_chain_is_not_a_multi_stage_chain() {
+    let members = DEFAULT_LIMIT + 8;
+    build(
+        &chained_plain_schema(members),
+        &format!("plain_{}", members - 1),
+    )
+    .unwrap();
+}
+
+/// A proxy is collapsed before planning and becomes no stage of its own, so the same chain must
+/// not be refused merely because a view re-exports it.
+#[test]
+fn a_view_proxy_is_not_an_extra_stage() {
+    let stages = DEFAULT_LIMIT;
+    build_on_view(
+        &chained_stages_view_schema(stages),
+        &format!("stage_{}", stages - 1),
+    )
+    .unwrap();
+}

rust/cubestore/cubestore/Cargo.toml

@@ -110,7 +110,7 @@ deepsize = "0.2.0"
 anyhow = "1.0"
 arc-swap = "1.7.1"
 object_store = "0.11.1"
-prost = "0.13.1"
+prost = { version = "0.13.1", features = ["no-recursion-limit"] }
 comfy-table = "7.2.2"
 
 [target.'cfg(target_os = "linux")'.dependencies]

rust/cubestore/cubestore/src/bin/cubestored.rs

@@ -1,4 +1,4 @@
-use cubestore::config::{validate_config, Config, CubeServices};
+use cubestore::config::{env_parse_lenient, validate_config, Config, CubeServices};
 use cubestore::http::status::serve_status_probes;
 use cubestore::telemetry::{init_agent_sender, track_event};
 use cubestore::util::logger::init_cube_logger;
@@ -82,6 +82,16 @@ fn main() {
     if let Ok(var) = std::env::var("CUBESTORE_EVENT_LOOP_MAX_BLOCKING_THREADS") {
         tokio_builder.max_blocking_threads(var.parse().unwrap());
     }
+    // Parsing, planning and plan serialization all recurse once per level of query nesting on
+    // this runtime's threads, so the depth a query may reach is bounded by their stack. Tokio
+    // would otherwise leave it at the 2 MiB platform default, which holds only a few dozen
+    // levels of nested subqueries. Select workers size theirs through
+    // CUBESTORE_SELECT_WORKER_STACK_SIZE, and deserializing a plan has to fit that one too --
+    // see DEFAULT_MAX_QUERY_PLAN_DEPTH.
+    tokio_builder.thread_stack_size(env_parse_lenient(
+        "CUBESTORE_MAIN_STACK_SIZE",
+        8 * 1024 * 1024,
+    ));
     let runtime = tokio_builder.build().unwrap();
     runtime.block_on(async move {
         init_agent_sender().await;

rust/cubestore/cubestore/src/lib.rs

@@ -211,7 +211,17 @@ impl From<std::io::Error> for CubeError {
 
 impl From<ParserError> for CubeError {
     fn from(v: ParserError) -> Self {
-        CubeError::from_error(format!("{:?}", v))
+        match v {
+            // This variant carries no message of its own, so name what ran out and what the
+            // budget was. It stays a user error: the query is the thing that has to change.
+            ParserError::RecursionLimitExceeded => CubeError::user(format!(
+                "Query is nested too deeply to parse: it exceeds the {} levels of nested \
+                 expressions, subqueries and parenthesised groups this node accepts. Flatten \
+                 the query, or raise CUBESTORE_SQL_PARSER_RECURSION_LIMIT.",
+                crate::sql::parser::sql_parser_recursion_limit()
+            )),
+            v => CubeError::from_error(format!("{:?}", v)),
+        }
     }
 }