[pull] master from cube-js:master

rust/cubesql/cubesql/src/compile/engine/df/scan.rs

@@ -11,7 +11,7 @@ use crate::{
 };
 use async_trait::async_trait;
 use chrono::{Datelike, NaiveDate};
-use cubeclient::models::{V1LoadRequestQuery, V1LoadResponse};
+use cubeclient::models::{V1LoadRequestQuery, V1LoadResponse, V1LoadResult};
 pub use datafusion::{
     arrow::{
         array::{
@@ -1194,9 +1194,15 @@ pub fn convert_transport_response(
     response
         .results
         .into_iter()
-        .map(|r| {
-            let mut response = JsonValueObject::new(r.data.clone());
-            let updated_schema = if let Some(last_refresh_time) = r.last_refresh_time.clone() {
+        .map(|result| {
+            let V1LoadResult {
+                data,
+                last_refresh_time,
+                ..
+            } = result;
+
+            let mut response = JsonValueObject::new(data);
+            let updated_schema = if let Some(last_refresh_time) = last_refresh_time {
                 let mut metadata = schema.metadata().clone();
                 metadata.insert("lastRefreshTime".to_string(), last_refresh_time);
                 Arc::new(Schema::new_with_metadata(

rust/cubesqlplanner/cubesqlplanner/Cargo.toml

@@ -22,6 +22,7 @@ lazy_static = "1.4.0"
 regex = "1.3.9"
 typed-builder = "0.21.2"
 indoc = "2.0.7"
+indexmap = "2.12"
 
 
 [features]

rust/cubesqlplanner/cubesqlplanner/src/planner/planners/multi_stage/multi_stage_query_planner.rs

@@ -21,6 +21,7 @@ use crate::planner::sql_evaluator::MemberSymbol;
 use crate::planner::GranularityHelper;
 use crate::planner::QueryProperties;
 use cubenativeutils::CubeError;
+use indexmap::IndexMap;
 use itertools::Itertools;
 use std::collections::HashSet;
 use std::rc::Rc;
@@ -238,7 +239,6 @@ impl MultiStageQueryPlanner {
         Ok(())
     }
 
-    //TODO refactor needed
     fn try_make_childs_for_case_switch(
         &self,
         case: &CaseSwitchDefinition,
@@ -248,67 +248,72 @@ impl MultiStageQueryPlanner {
         resolved_multi_stage_dimensions: &mut HashSet<String>,
         cte_state: &mut CteState,
     ) -> Result<bool, CubeError> {
-        if let CaseSwitchItem::Member(switch_member) = &case.switch {
-            let mut processed_deps = HashSet::new();
-            for itm in &case.items {
-                let mut state = new_state.clone_state();
-                let filter = BaseFilter::try_new(
-                    self.query_tools.clone(),
-                    switch_member.clone(),
-                    FilterType::Dimension,
-                    FilterOperator::Equal,
-                    Some(vec![Some(itm.value.clone())]),
-                )?;
-                state.add_dimension_filter(FilterItem::Item(filter));
-                let state = Rc::new(state);
-
-                for dep in itm.sql.get_dependencies() {
-                    let dep = dep.resolve_reference_chain();
-                    if !processed_deps.contains(&dep.full_name()) {
-                        processed_deps.insert(dep.full_name());
-                        result.push(self.make_queries_descriptions(
-                            dep,
-                            state.clone(),
-                            descriptions,
-                            resolved_multi_stage_dimensions,
-                            cte_state,
-                        )?);
+        let CaseSwitchItem::Member(switch_member) = &case.switch else {
+            return Ok(false);
+        };
+
+        // Collect, per dependency, the union of switch values that need it.
+        // `None` marks an unrestricted (open ELSE) entry: such a dependency
+        // must be processed without a prefilter on switch_member, since the
+        // outer CASE will dispatch by value at row level.
+        let mut deps: IndexMap<String, (Rc<MemberSymbol>, Option<Vec<String>>)> = IndexMap::new();
+
+        let mut record = |dep: Rc<MemberSymbol>, branch_values: Option<Vec<String>>| {
+            let dep = dep.resolve_reference_chain();
+            let entry = deps
+                .entry(dep.full_name())
+                .or_insert_with(|| (dep.clone(), Some(Vec::new())));
+            match (&mut entry.1, branch_values) {
+                (None, _) => {} // already unrestricted
+                (slot @ Some(_), None) => *slot = None,
+                (Some(values), Some(branch)) => {
+                    for v in branch {
+                        if !values.contains(&v) {
+                            values.push(v);
+                        }
                     }
                 }
             }
-            if let Some(else_sql) = &case.else_sql {
-                let mut state = new_state.clone_state();
-                if let Some(else_values) = case.get_else_values() {
-                    if !else_values.is_empty() {
-                        let filter = BaseFilter::try_new(
-                            self.query_tools.clone(),
-                            switch_member.clone(),
-                            FilterType::Dimension,
-                            FilterOperator::Equal,
-                            Some(else_values.into_iter().map(|v| Some(v)).collect_vec()),
-                        )?;
-                        state.add_dimension_filter(FilterItem::Item(filter));
-                    }
-                }
-                let state = Rc::new(state);
-                for dep in else_sql.get_dependencies() {
-                    let dep = dep.resolve_reference_chain();
-                    if !processed_deps.contains(&dep.full_name()) {
-                        processed_deps.insert(dep.full_name());
-                        result.push(self.make_queries_descriptions(
-                            dep,
-                            state.clone(),
-                            descriptions,
-                            resolved_multi_stage_dimensions,
-                            cte_state,
-                        )?);
-                    }
+        };
+
+        for itm in &case.items {
+            for dep in itm.sql.get_dependencies() {
+                record(dep, Some(vec![itm.value.clone()]));
+            }
+        }
+
+        if let Some(else_sql) = &case.else_sql {
+            let else_values = case.get_else_values();
+            for dep in else_sql.get_dependencies() {
+                record(dep.clone(), else_values.clone());
+            }
+        }
+
+        for (_, (dep, values)) in deps {
+            let mut state = new_state.clone_state();
+            if let Some(values) = values {
+                if !values.is_empty() {
+                    let filter = BaseFilter::try_new(
+                        self.query_tools.clone(),
+                        switch_member.clone(),
+                        FilterType::Dimension,
+                        FilterOperator::Equal,
+                        Some(values.into_iter().map(Some).collect_vec()),
+                    )?;
+                    state.add_dimension_filter(FilterItem::Item(filter));
                 }
             }
-            Ok(true)
-        } else {
-            Ok(false)
+            let state = Rc::new(state);
+            result.push(self.make_queries_descriptions(
+                dep,
+                state,
+                descriptions,
+                resolved_multi_stage_dimensions,
+                cte_state,
+            )?);
         }
+
+        Ok(true)
     }
 
     fn make_queries_descriptions(

rust/cubesqlplanner/cubesqlplanner/src/test_fixtures/schemas/yaml_files/common/integration_multi_stage.yaml

@@ -223,7 +223,7 @@ cubes:
                   dir: desc
 
           - name: status_weighted_amount
-            type: number
+            type: sum
             multi_stage: true
             case:
                 switch: "{CUBE.status}"

rust/cubesqlplanner/cubesqlplanner/src/tests/integration/multi_stage/case_switch.rs

@@ -9,17 +9,6 @@ fn create_context() -> TestContext {
 
 const SEED: &str = "integration_multi_stage_tables.sql";
 
-// BUG: Case/switch multi-stage measures generate incorrect SQL.
-// The leaf CTE is filtered by the first WHEN value (e.g. status = 'completed')
-// even though no such filter was specified in the query.
-// Expected: each WHEN value should produce a separate leaf CTE with its own
-// filter, then the final CTE combines results via CASE expression
-// (cross-join pattern, similar to calc_groups test).
-// Actual: only one leaf CTE with first WHEN value as filter, so only
-// 'completed' rows are processed. Result shows completed totals (300, 500, 600)
-// instead of weighted sums across all statuses (375, 600, 750).
-
-#[ignore]
 #[tokio::test(flavor = "multi_thread")]
 async fn test_case_switch_measure() {
     let ctx = create_context();
@@ -42,7 +31,6 @@ async fn test_case_switch_measure() {
     }
 }
 
-#[ignore]
 #[tokio::test(flavor = "multi_thread")]
 async fn test_case_switch_with_dimension() {
     let ctx = create_context();
@@ -60,6 +48,27 @@ async fn test_case_switch_with_dimension() {
               - "2024-03-31"
         order:
           - id: orders.status
+          - id: orders.created_at
+    "#};
+
+    ctx.build_sql(query).unwrap();
+
+    if let Some(result) = ctx.try_execute_pg(query, SEED).await {
+        insta::assert_snapshot!(result);
+    }
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_case_switch_dimension_only() {
+    let ctx = create_context();
+
+    let query = indoc! {r#"
+        measures:
+          - orders.status_weighted_amount
+        dimensions:
+          - orders.status
+        order:
+          - id: orders.status
     "#};
 
     ctx.build_sql(query).unwrap();
@@ -69,7 +78,6 @@ async fn test_case_switch_with_dimension() {
     }
 }
 
-#[ignore]
 #[tokio::test(flavor = "multi_thread")]
 async fn test_case_switch_with_filter() {
     let ctx = create_context();

rust/cubesqlplanner/cubesqlplanner/src/tests/integration/multi_stage/snapshots/cubesqlplanner__tests__integration__multi_stage__case_switch__case_switch_dimension_only.snap

@@ -0,0 +1,9 @@
+---
+source: cubesqlplanner/src/tests/integration/multi_stage/case_switch.rs
+expression: result
+---
+orders__status | orders__status_weighted_amount
+---------------+-------------------------------
+cancelled      | 0.00                          
+completed      | 1400.00                       
+pending        | 325.000

rust/cubesqlplanner/cubesqlplanner/src/tests/integration/multi_stage/snapshots/cubesqlplanner__tests__integration__multi_stage__case_switch__case_switch_measure.snap

@@ -0,0 +1,9 @@
+---
+source: cubesqlplanner/src/tests/integration/multi_stage/case_switch.rs
+expression: result
+---
+orders__created_at_month | orders__status_weighted_amount
+-------------------------+-------------------------------
+2024-01-01 00:00:00      | 375.000                       
+2024-02-01 00:00:00      | 600.000                       
+2024-03-01 00:00:00      | 750.000

rust/cubesqlplanner/cubesqlplanner/src/tests/integration/multi_stage/snapshots/cubesqlplanner__tests__integration__multi_stage__case_switch__case_switch_with_dimension.snap

@@ -0,0 +1,15 @@
+---
+source: cubesqlplanner/src/tests/integration/multi_stage/case_switch.rs
+expression: result
+---
+orders__status | orders__created_at_month | orders__status_weighted_amount
+---------------+--------------------------+-------------------------------
+cancelled      | 2024-01-01 00:00:00      | 0.00                          
+cancelled      | 2024-02-01 00:00:00      | 0.00                          
+cancelled      | 2024-03-01 00:00:00      | 0.00                          
+completed      | 2024-01-01 00:00:00      | 300.00                        
+completed      | 2024-02-01 00:00:00      | 500.00                        
+completed      | 2024-03-01 00:00:00      | 600.00                        
+pending        | 2024-01-01 00:00:00      | 75.000                        
+pending        | 2024-02-01 00:00:00      | 100.000                       
+pending        | 2024-03-01 00:00:00      | 150.000