8 tests remaining

Author: waralexrom

packages/cubejs-schema-compiler/test/integration/postgres/sql-generation.test.ts

@@ -945,8 +945,6 @@ describe('SQL Generation', () => {
       timezone: 'America/Los_Angeles'
     });
 
-    console.log(query.buildSqlAndParams());
-
     expect(query.buildSqlAndParams()[0]).toMatch(/HLL_COUNT\.MERGE/);
     expect(query.buildSqlAndParams()[0]).toMatch(/HLL_COUNT\.INIT/);
   });
@@ -971,7 +969,7 @@ describe('SQL Generation', () => {
 
     console.log(query.buildSqlAndParams());
 
-    expect(query.buildSqlAndParams()[0]).toMatch(/OFFSET (\d) LIMIT (\d)/);
+    expect(query.buildSqlAndParams()[0]).toMatch(/OFFSET (\d)\s+LIMIT (\d)/);
   });
 
   it('calculated join', async () => {
@@ -2535,6 +2533,9 @@ describe('SQL Generation', () => {
 
       const sqlBuild = query.buildSqlAndParams();
 
+      console.log(sqlBuild[0]);
+      console.log(sqlBuild[1]);
+
       expect(sqlBuild[0].includes('America/Los_Angeles')).toEqual(true);
       expect(sqlBuild[1][0]).toEqual(granularityTest.from);
       expect(sqlBuild[1][1]).toEqual(granularityTest.to);

rust/cubesqlplanner/cubesqlplanner/src/cube_bridge/base_tools.rs

@@ -47,4 +47,9 @@ pub trait BaseTools {
     ) -> Result<Vec<Vec<String>>, CubeError>;
     fn get_allocated_params(&self) -> Result<Vec<String>, CubeError>;
     fn all_cube_members(&self, path: String) -> Result<Vec<String>, CubeError>;
+    //===== TODO Move to templates
+    fn hll_init(&self, sql: String) -> Result<String, CubeError>;
+    fn hll_merge(&self, sql: String) -> Result<String, CubeError>;
+    fn hll_cardinality_merge(&self, sql: String) -> Result<String, CubeError>;
+    fn count_distinct_approx(&self, sql: String) -> Result<String, CubeError>;
 }

rust/cubesqlplanner/cubesqlplanner/src/planner/base_measure.rs

@@ -200,7 +200,7 @@ impl BaseMeasure {
 
     pub fn can_used_as_addictive_in_multplied(&self) -> Result<bool, CubeError> {
         let measure_type = self.measure_type();
-        let res = if measure_type == "countDistinct" {
+        let res = if measure_type == "countDistinct" || measure_type == "countDistinctApprox" {
             true
         } else if measure_type == "count" {
             if let Some(definition) = &self.definition {

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

@@ -186,18 +186,21 @@ pub struct MultiStageMember {
     member_type: MultiStageMemberType,
     evaluation_node: Rc<MemberSymbol>,
     is_ungrupped: bool,
+    has_aggregates_on_top: bool,
 }
 
 impl MultiStageMember {
     pub fn new(
         member_type: MultiStageMemberType,
         evaluation_node: Rc<MemberSymbol>,
         is_ungrupped: bool,
+        has_aggregates_on_top: bool,
     ) -> Rc<Self> {
         Rc::new(Self {
             member_type,
             evaluation_node,
             is_ungrupped,
+            has_aggregates_on_top,
         })
     }
 
@@ -216,4 +219,8 @@ impl MultiStageMember {
     pub fn is_ungrupped(&self) -> bool {
         self.is_ungrupped
     }
+
+    pub fn has_aggregates_on_top(&self) -> bool {
+        self.has_aggregates_on_top
+    }
 }

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

@@ -384,6 +384,9 @@ impl MultiStageMemberQueryPlanner {
 
         let mut node_factory = SqlNodesFactory::new();
         node_factory.set_time_shifts(self.description.state().time_shifts().clone());
+        if self.description.member().has_aggregates_on_top() {
+            node_factory.set_count_approx_as_state(true);
+        }
 
         if cte_query_properties
             .full_key_aggregate_measures()?

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

@@ -173,6 +173,7 @@ impl MultiStageQueryPlanner {
                         )),
                         time_dimension.member_evaluator(),
                         true,
+                        false,
                     ),
                     state.clone(),
                     vec![],
@@ -196,6 +197,7 @@ impl MultiStageQueryPlanner {
                 MultiStageMemberType::Leaf(MultiStageLeafMemberType::Measure),
                 member,
                 self.query_properties.ungrouped(),
+                true,
             ),
             state,
             vec![],
@@ -336,6 +338,7 @@ impl MultiStageQueryPlanner {
                         MultiStageMemberType::Inode(inode_member),
                         member,
                         self.query_properties.ungrouped(),
+                        false,
                     ),
                     state.clone(),
                     input,
@@ -386,6 +389,7 @@ impl MultiStageQueryPlanner {
                     MultiStageMemberType::Leaf(MultiStageLeafMemberType::Measure),
                     member.clone(),
                     self.query_properties.ungrouped(),
+                    false,
                 ),
                 state.clone(),
                 vec![],
@@ -435,6 +439,7 @@ impl MultiStageQueryPlanner {
                     MultiStageMemberType::Inode(multi_stage_member),
                     member,
                     is_ungrupped,
+                    false,
                 ),
                 state.clone(),
                 input,

rust/cubesqlplanner/cubesqlplanner/src/planner/sql_evaluator/sql_nodes/factory.rs

@@ -13,6 +13,7 @@ pub struct SqlNodesFactory {
     time_shifts: HashMap<String, String>,
     ungrouped: bool,
     ungrouped_measure: bool,
+    count_approx_as_state: bool,
     render_references: HashMap<String, QualifiedColumnName>,
     rendered_as_multiplied_measures: HashSet<String>,
     ungrouped_measure_references: HashMap<String, QualifiedColumnName>,
@@ -28,6 +29,7 @@ impl SqlNodesFactory {
             time_shifts: HashMap::new(),
             ungrouped: false,
             ungrouped_measure: false,
+            count_approx_as_state: false,
             render_references: HashMap::new(),
             ungrouped_measure_references: HashMap::new(),
             cube_name_references: HashMap::new(),
@@ -78,6 +80,10 @@ impl SqlNodesFactory {
         self.rolling_window = value;
     }
 
+    pub fn set_count_approx_as_state(&mut self, value: bool) {
+        self.count_approx_as_state = value;
+    }
+
     pub fn set_ungrouped_measure_references(
         &mut self,
         value: HashMap<String, QualifiedColumnName>,
@@ -161,7 +167,11 @@ impl SqlNodesFactory {
         } else if self.rolling_window {
             RollingWindowNode::new(input)
         } else {
-            FinalMeasureSqlNode::new(input, self.rendered_as_multiplied_measures.clone())
+            FinalMeasureSqlNode::new(
+                input,
+                self.rendered_as_multiplied_measures.clone(),
+                self.count_approx_as_state,
+            )
         }
     }
 

rust/cubesqlplanner/cubesqlplanner/src/planner/sql_evaluator/sql_nodes/final_measure.rs

@@ -12,16 +12,19 @@ use std::rc::Rc;
 pub struct FinalMeasureSqlNode {
     input: Rc<dyn SqlNode>,
     rendered_as_multiplied_measures: HashSet<String>,
+    count_approx_as_state: bool,
 }
 
 impl FinalMeasureSqlNode {
     pub fn new(
         input: Rc<dyn SqlNode>,
         rendered_as_multiplied_measures: HashSet<String>,
+        count_approx_as_state: bool,
     ) -> Rc<Self> {
         Rc::new(Self {
             input,
             rendered_as_multiplied_measures,
+            count_approx_as_state,
         })
     }
 
@@ -60,6 +63,12 @@ impl SqlNode for FinalMeasureSqlNode {
 
                 if ev.is_calculated() {
                     input
+                } else if ev.measure_type() == "countDistinctApprox" {
+                    if self.count_approx_as_state {
+                        query_tools.base_tools().hll_init(input)?
+                    } else {
+                        query_tools.base_tools().count_distinct_approx(input)?
+                    }
                 } else if self.is_count_disttinct(ev) {
                     templates.count_distinct(&input)?
                 } else {

rust/cubesqlplanner/cubesqlplanner/src/planner/sql_evaluator/sql_nodes/rolling_window.rs

@@ -39,16 +39,20 @@ impl SqlNode for RollingWindowNode {
                     templates,
                 )?;
                 if m.is_cumulative() {
-                    let aggregate_function = if m.measure_type() == "sum"
-                        || m.measure_type() == "count"
-                        || m.measure_type() == "runningTotal"
-                    {
-                        "sum"
+                    if m.measure_type() == "countDistinctApprox" {
+                        query_tools.base_tools().hll_cardinality_merge(input)?
                     } else {
-                        m.measure_type()
-                    };
+                        let aggregate_function = if m.measure_type() == "sum"
+                            || m.measure_type() == "count"
+                            || m.measure_type() == "runningTotal"
+                        {
+                            "sum"
+                        } else {
+                            m.measure_type()
+                        };
 
-                    format!("{}({})", aggregate_function, input)
+                        format!("{}({})", aggregate_function, input)
+                    }
                 } else {
                     input
                 }