feat(tesseract): Rollup Join support (#9745)

Author: waralexrom

packages/cubejs-schema-compiler/src/compiler/CubeEvaluator.ts

@@ -571,6 +571,22 @@ export class CubeEvaluator extends CubeSymbols {
     }));
   }
 
+  public preAggregationDescriptionByName(cubeName: string, preAggName: string) {
+    const cube = this.cubeFromPath(cubeName);
+    const preAggregations = cube.preAggregations || {};
+
+    const preAgg = preAggregations[preAggName];
+
+    if (!preAgg) {
+      return undefined;
+    }
+
+    return {
+      name: preAggName,
+      ...(preAgg as Record<string, any>)
+    };
+  }
+
   /**
    * Returns pre-aggregations filtered by the specified selector.
    */
@@ -785,6 +801,14 @@ export class CubeEvaluator extends CubeSymbols {
     return { cubeReferencesUsed, pathReferencesUsed, evaluatedSql };
   }
 
+  /**
+   * Evaluates rollup references for retrieving rollupReference used in Tesseract.
+   * This is a temporary solution until Tesseract takes ownership of all pre-aggregations.
+   */
+  public evaluateRollupReferences<T extends ToString | Array<ToString>>(cube: string, rollupReferences: (...args: Array<unknown>) => T) {
+    return this.evaluateReferences(cube, rollupReferences, { originalSorting: true });
+  }
+
   public evaluatePreAggregationReferences(cube: string, aggregation: PreAggregationDefinition): PreAggregationReferences {
     const timeDimensions: Array<PreAggregationTimeDimensionReference> = [];
 

rust/cubesqlplanner/cubesqlplanner/src/cube_bridge/evaluator.rs

@@ -64,4 +64,16 @@ pub trait CubeEvaluator {
         &self,
         cube_name: String,
     ) -> Result<Vec<Rc<dyn PreAggregationDescription>>, CubeError>;
+    #[nbridge(optional)]
+    fn pre_aggregation_description_by_name(
+        &self,
+        cube_name: String,
+        name: String,
+    ) -> Result<Option<Rc<dyn PreAggregationDescription>>, CubeError>;
+    #[nbridge(vec)]
+    fn evaluate_rollup_references(
+        &self,
+        cube: String,
+        sql: Rc<dyn MemberSql>,
+    ) -> Result<Vec<String>, CubeError>;
 }

rust/cubesqlplanner/cubesqlplanner/src/cube_bridge/pre_aggregation_description.rs

@@ -32,4 +32,7 @@ pub trait PreAggregationDescription {
 
     #[nbridge(field, optional)]
     fn time_dimension_reference(&self) -> Result<Option<Rc<dyn MemberSql>>, CubeError>;
+
+    #[nbridge(field, optional)]
+    fn rollup_references(&self) -> Result<Option<Rc<dyn MemberSql>>, CubeError>;
 }

rust/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/compiled_pre_aggregation.rs

@@ -1,14 +1,40 @@
-use crate::cube_bridge::member_sql::MemberSql;
-use crate::cube_bridge::pre_aggregation_description::PreAggregationDescription;
-use crate::planner::query_tools::QueryTools;
-use crate::planner::sql_evaluator::MemberSymbol;
-use cubenativeutils::CubeError;
+use crate::planner::sql_evaluator::{MemberSymbol, SqlCall};
 use std::fmt::Debug;
 use std::rc::Rc;
+
+#[derive(Clone)]
+pub struct PreAggregationJoinItem {
+    pub from: PreAggregationTable,
+    pub to: PreAggregationTable,
+    pub from_members: Vec<Rc<MemberSymbol>>,
+    pub to_members: Vec<Rc<MemberSymbol>>,
+    pub on_sql: Rc<SqlCall>,
+}
+
+#[derive(Clone)]
+pub struct PreAggregationJoin {
+    pub root: PreAggregationTable,
+    pub items: Vec<PreAggregationJoinItem>,
+}
+
+#[derive(Clone)]
+pub struct PreAggregationTable {
+    pub cube_name: String,
+    pub name: String,
+    pub alias: Option<String>,
+}
+
+#[derive(Clone)]
+pub enum PreAggregationSource {
+    Table(PreAggregationTable),
+    Join(PreAggregationJoin),
+}
+
 #[derive(Clone)]
 pub struct CompiledPreAggregation {
     pub cube_name: String,
     pub name: String,
+    pub source: Rc<PreAggregationSource>,
     pub granularity: Option<String>,
     pub external: Option<bool>,
     pub measures: Vec<Rc<MemberSymbol>>,
@@ -34,104 +60,3 @@ impl Debug for CompiledPreAggregation {
             .finish()
     }
 }
-
-impl CompiledPreAggregation {
-    pub fn try_new(
-        query_tools: Rc<QueryTools>,
-        cube_name: &String,
-        description: Rc<dyn PreAggregationDescription>,
-    ) -> Result<Rc<Self>, CubeError> {
-        let static_data = description.static_data();
-        let measures = if let Some(refs) = description.measure_references()? {
-            Self::symbols_from_ref(query_tools.clone(), cube_name, refs, Self::check_is_measure)?
-        } else {
-            Vec::new()
-        };
-        let dimensions = if let Some(refs) = description.dimension_references()? {
-            Self::symbols_from_ref(
-                query_tools.clone(),
-                cube_name,
-                refs,
-                Self::check_is_dimension,
-            )?
-        } else {
-            Vec::new()
-        };
-        let time_dimensions = if let Some(refs) = description.time_dimension_reference()? {
-            let dims = Self::symbols_from_ref(
-                query_tools.clone(),
-                cube_name,
-                refs,
-                Self::check_is_time_dimension,
-            )?;
-            /*             if dims.len() != 1 {
-                return Err(CubeError::user(format!(
-                    "Pre aggregation should contains only one time dimension"
-                )));
-            } */
-            vec![(dims[0].clone(), static_data.granularity.clone())] //TODO remove unwrap
-        } else {
-            Vec::new()
-        };
-        let allow_non_strict_date_range_match = description
-            .static_data()
-            .allow_non_strict_date_range_match
-            .unwrap_or(false);
-        let res = Rc::new(Self {
-            name: static_data.name.clone(),
-            cube_name: cube_name.clone(),
-            granularity: static_data.granularity.clone(),
-            external: static_data.external,
-            measures,
-            dimensions,
-            time_dimensions,
-            allow_non_strict_date_range_match,
-        });
-        Ok(res)
-    }
-
-    fn symbols_from_ref<F: Fn(&MemberSymbol) -> Result<(), CubeError>>(
-        query_tools: Rc<QueryTools>,
-        cube_name: &String,
-        ref_func: Rc<dyn MemberSql>,
-        check_type_fn: F,
-    ) -> Result<Vec<Rc<MemberSymbol>>, CubeError> {
-        let evaluator_compiler_cell = query_tools.evaluator_compiler().clone();
-        let mut evaluator_compiler = evaluator_compiler_cell.borrow_mut();
-        let sql_call = evaluator_compiler.compile_sql_call(cube_name, ref_func)?;
-        let mut res = Vec::new();
-        for symbol in sql_call.get_dependencies().iter() {
-            check_type_fn(&symbol)?;
-            res.push(symbol.clone());
-        }
-        Ok(res)
-    }
-
-    fn check_is_measure(symbol: &MemberSymbol) -> Result<(), CubeError> {
-        symbol
-            .as_measure()
-            .map_err(|_| CubeError::user(format!("Pre-aggregation measure must be a measure")))?;
-        Ok(())
-    }
-
-    fn check_is_dimension(symbol: &MemberSymbol) -> Result<(), CubeError> {
-        symbol.as_dimension().map_err(|_| {
-            CubeError::user(format!("Pre-aggregation dimension must be a dimension"))
-        })?;
-        Ok(())
-    }
-
-    fn check_is_time_dimension(symbol: &MemberSymbol) -> Result<(), CubeError> {
-        let dimension = symbol.as_dimension().map_err(|_| {
-            CubeError::user(format!(
-                "Pre-aggregation time dimension must be a dimension"
-            ))
-        })?;
-        if dimension.dimension_type() != "time" {
-            return Err(CubeError::user(format!(
-                "Pre-aggregation time dimension must be a dimension"
-            )));
-        }
-        Ok(())
-    }
-}

rust/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/mod.rs

@@ -4,10 +4,12 @@ mod measure_matcher;
 mod optimizer;
 mod original_sql_collector;
 mod original_sql_optimizer;
+mod pre_aggregations_compiler;
 
 pub use compiled_pre_aggregation::*;
 use dimension_matcher::*;
 use measure_matcher::*;
 pub use optimizer::*;
 pub use original_sql_collector::*;
 pub use original_sql_optimizer::*;
+pub use pre_aggregations_compiler::*;

rust/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/optimizer.rs

@@ -1,3 +1,4 @@
+use super::PreAggregationsCompiler;
 use super::*;
 use crate::logical_plan::*;
 use crate::plan::FilterItem;
@@ -43,22 +44,9 @@ impl PreAggregationOptimizer {
         let mut cube_names_collector = CubeNamesCollector::new();
         cube_names_collector.collect(&plan)?;
         let cube_names = cube_names_collector.result();
+        let mut compiler = PreAggregationsCompiler::try_new(self.query_tools.clone(), &cube_names)?;
 
-        let mut compiled_pre_aggregations = Vec::new();
-        for cube_name in cube_names.iter() {
-            let pre_aggregations = self
-                .query_tools
-                .cube_evaluator()
-                .pre_aggregations_for_cube_as_array(cube_name.clone())?;
-            for pre_aggregation in pre_aggregations.iter() {
-                let compiled = CompiledPreAggregation::try_new(
-                    self.query_tools.clone(),
-                    cube_name,
-                    pre_aggregation.clone(),
-                )?;
-                compiled_pre_aggregations.push(compiled);
-            }
-        }
+        let compiled_pre_aggregations = compiler.compile_all_pre_aggregations()?;
 
         for pre_aggregation in compiled_pre_aggregations.iter() {
             let new_query = self.try_rewrite_query(plan.clone(), pre_aggregation)?;
@@ -413,51 +401,44 @@ impl PreAggregationOptimizer {
         &mut self,
         pre_aggregation: &Rc<CompiledPreAggregation>,
     ) -> Result<Rc<PreAggregation>, CubeError> {
-        let pre_aggregation_obj = self.query_tools.base_tools().get_pre_aggregation_by_name(
+        /* let pre_aggregation_obj = self.query_tools.base_tools().get_pre_aggregation_by_name(
             pre_aggregation.cube_name.clone(),
             pre_aggregation.name.clone(),
-        )?;
-        if let Some(table_name) = &pre_aggregation_obj.static_data().table_name {
-            let schema = LogicalSchema {
-                time_dimensions: vec![],
-                dimensions: pre_aggregation
-                    .dimensions
-                    .iter()
-                    .cloned()
-                    .chain(
-                        pre_aggregation
-                            .time_dimensions
-                            .iter()
-                            .map(|(d, _)| d.clone()),
-                    )
-                    .collect(),
-                measures: pre_aggregation.measures.to_vec(),
-                multiplied_measures: HashSet::new(),
-            };
-            let pre_aggregation = PreAggregation {
-                name: pre_aggregation.name.clone(),
-                time_dimensions: pre_aggregation.time_dimensions.clone(),
-                dimensions: pre_aggregation.dimensions.clone(),
-                measures: pre_aggregation.measures.clone(),
-                schema: Rc::new(schema),
-                external: pre_aggregation.external.unwrap_or_default(),
-                granularity: pre_aggregation.granularity.clone(),
-                table_name: table_name.clone(),
-                cube_name: pre_aggregation.cube_name.clone(),
-                pre_aggregation_obj,
-            };
-            let result = Rc::new(pre_aggregation);
-            self.used_pre_aggregations.insert(
-                (result.cube_name.clone(), result.name.clone()),
-                result.clone(),
-            );
-            Ok(result)
-        } else {
-            Err(CubeError::internal(format!(
-                "Cannot find pre aggregation object for cube {} and name {}",
-                pre_aggregation.cube_name, pre_aggregation.name
-            )))
-        }
+        )?; */
+        //if let Some(table_name) = &pre_aggregation_obj.static_data().table_name {
+        let schema = LogicalSchema {
+            time_dimensions: vec![],
+            dimensions: pre_aggregation
+                .dimensions
+                .iter()
+                .cloned()
+                .chain(
+                    pre_aggregation
+                        .time_dimensions
+                        .iter()
+                        .map(|(d, _)| d.clone()),
+                )
+                .collect(),
+            measures: pre_aggregation.measures.to_vec(),
+            multiplied_measures: HashSet::new(),
+        };
+        let pre_aggregation = PreAggregation {
+            name: pre_aggregation.name.clone(),
+            time_dimensions: pre_aggregation.time_dimensions.clone(),
+            dimensions: pre_aggregation.dimensions.clone(),
+            measures: pre_aggregation.measures.clone(),
+            schema: Rc::new(schema),
+            external: pre_aggregation.external.unwrap_or_default(),
+            granularity: pre_aggregation.granularity.clone(),
+            source: pre_aggregation.source.clone(),
+            cube_name: pre_aggregation.cube_name.clone(),
+        };
+        let result = Rc::new(pre_aggregation);
+        self.used_pre_aggregations.insert(
+            (result.cube_name.clone(), result.name.clone()),
+            result.clone(),
+        );
+        Ok(result)
     }
 
     fn is_schema_and_filters_match(

rust/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/original_sql_optimizer.rs

@@ -1,3 +1,4 @@
+use super::PreAggregationsCompiler;
 use super::*;
 use crate::logical_plan::*;
 use crate::planner::query_tools::QueryTools;
@@ -352,25 +353,11 @@ impl OriginalSqlOptimizer {
         let res = if let Some(found_pre_aggregation) = self.foud_pre_aggregations.get(cube_name) {
             Some(found_pre_aggregation.clone())
         } else {
-            let pre_aggregations = self
-                .query_tools
-                .cube_evaluator()
-                .pre_aggregations_for_cube_as_array(cube_name.clone())?;
-            if let Some(found_pre_aggregation) = pre_aggregations
-                .iter()
-                .find(|p| p.static_data().pre_aggregation_type == "originalSql")
-            {
-                let compiled = CompiledPreAggregation::try_new(
-                    self.query_tools.clone(),
-                    cube_name,
-                    found_pre_aggregation.clone(),
-                )?;
-                self.foud_pre_aggregations
-                    .insert(cube_name.clone(), compiled.clone());
-                Some(compiled)
-            } else {
-                None
-            }
+            let mut compiler = PreAggregationsCompiler::try_new(
+                self.query_tools.clone(),
+                &vec![cube_name.clone()],
+            )?;
+            compiler.compile_origin_sql_pre_aggregation(&cube_name)?
         };
         Ok(res)
     }