feat(sqlplanner): Base multi staging support (#8832)

Author: waralexrom

packages/cubejs-schema-compiler/src/adapter/BaseQuery.js

@@ -614,7 +614,11 @@ export class BaseQuery {
       joinRoot: this.join.root,
       joinGraph: this.joinGraph,
       cubeEvaluator: this.cubeEvaluator,
+      order: this.options.order,
       filters: this.options.filters,
+      limit: this.options.limit ? this.options.limit.toString() : null,
+      rowLimit: this.options.rowLimit ? this.options.rowLimit.toString() : null,
+      offset: this.options.offset ? this.options.offset.toString() : null,
       baseTools: this,
 
     };
@@ -1030,6 +1034,7 @@ export class BaseQuery {
 
     const multipliedMeasures = measuresToRender(true, false)(measureToHierarchy);
     const regularMeasures = measuresToRender(false, false)(measureToHierarchy);
+
     const cumulativeMeasures =
       R.pipe(
         R.map(multiplied => R.xprod([multiplied], measuresToRender(multiplied, true)(measureToHierarchy))),
@@ -1518,7 +1523,7 @@ export class BaseQuery {
       if (m.expressionName && !collectedMeasures.length && !m.isMemberExpression) {
         throw new UserError(`Subquery dimension ${m.expressionName} should reference at least one measure`);
       }
-      if (!collectedMeasures.length && m.isMemberExpression && m.query.allCubeNames.length > 1 && m.measureSql() === "COUNT(*)") {
+      if (!collectedMeasures.length && m.isMemberExpression && m.query.allCubeNames.length > 1 && m.measureSql() === 'COUNT(*)') {
         const cubeName = m.expressionCubeName ? `\`${m.expressionCubeName}\` ` : '';
         throw new UserError(`The query contains \`COUNT(*)\` expression but cube/view ${cubeName}is missing \`count\` measure`);
       }
@@ -2438,7 +2443,6 @@ export class BaseQuery {
       fn,
       context
     );
-
     return context.joinHints;
   }
 

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

@@ -97,7 +97,7 @@ describe('SQL Generation', () => {
             type: 'prior',
           }]
         },
-        cagr_1d: {
+        cagr_day: {
           multi_stage: true,
           sql: \`ROUND(100 * \${revenue} / NULLIF(\${revenue_day_ago}, 0))\`,
           type: 'number',
@@ -601,7 +601,7 @@ describe('SQL Generation', () => {
     await compiler.compile();
     const query = new PostgresQuery({ joinGraph, cubeEvaluator, compiler }, q);
 
-    console.log(query.buildSqlAndParams());
+    // console.log(query.buildSqlAndParams());
 
     const res = await dbRunner.testQuery(query.buildSqlAndParams());
     console.log(JSON.stringify(res));
@@ -836,7 +836,7 @@ describe('SQL Generation', () => {
     measures: [
       'visitors.revenue',
       'visitors.revenue_day_ago',
-      'visitors.cagr_1d'
+      'visitors.cagr_day'
     ],
     timeDimensions: [{
       dimension: 'visitors.created_at',
@@ -848,8 +848,8 @@ describe('SQL Generation', () => {
     }],
     timezone: 'America/Los_Angeles'
   }, [
-    { visitors__created_at_day: '2017-01-05T00:00:00.000Z', visitors__cagr_1d: '150', visitors__revenue: '300', visitors__revenue_day_ago: '200' },
-    { visitors__created_at_day: '2017-01-06T00:00:00.000Z', visitors__cagr_1d: '300', visitors__revenue: '900', visitors__revenue_day_ago: '300' }
+    { visitors__created_at_day: '2017-01-05T00:00:00.000Z', visitors__cagr_day: '150', visitors__revenue: '300', visitors__revenue_day_ago: '200' },
+    { visitors__created_at_day: '2017-01-06T00:00:00.000Z', visitors__cagr_day: '300', visitors__revenue: '900', visitors__revenue_day_ago: '300' }
   ]));
 
   it('sql utils', async () => runQueryTest({

rust/cubesqlplanner/cubesqlplanner/src/cube_bridge/base_query_options.rs

@@ -29,6 +29,18 @@ pub struct FilterItem {
     pub values: Option<Vec<Option<String>>>,
 }
 
+#[derive(Serialize, Deserialize, Debug)]
+pub struct OrderByItem {
+    pub id: String,
+    pub desc: Option<bool>,
+}
+
+impl OrderByItem {
+    pub fn is_desc(&self) -> bool {
+        self.desc.unwrap_or(false)
+    }
+}
+
 impl FilterItem {
     pub fn member(&self) -> Option<&String> {
         self.member.as_ref().or(self.dimension.as_ref())
@@ -43,9 +55,11 @@ pub struct BaseQueryOptionsStatic {
     pub time_dimensions: Option<Vec<TimeDimension>>,
     pub timezone: Option<String>,
     pub filters: Option<Vec<FilterItem>>,
-    #[serde(rename = "joinRoot")]
-    pub join_root: Option<String>, //TODO temporaty. join graph should be rewrited in rust or taked
-                                   //from Js CubeCompiller
+    pub order: Option<Vec<OrderByItem>>,
+    pub limit: Option<String>,
+    #[serde(rename = "rowLimit")]
+    pub row_limit: Option<String>,
+    pub offset: Option<String>,
 }
 
 #[nativebridge::native_bridge(BaseQueryOptionsStatic)]

rust/cubesqlplanner/cubesqlplanner/src/cube_bridge/dimension_definition.rs

@@ -13,7 +13,10 @@ use std::rc::Rc;
 pub struct DimenstionDefinitionStatic {
     #[serde(rename = "type")]
     pub dimension_type: String,
+    #[serde(rename = "ownedByCube")]
     pub owned_by_cube: Option<bool>,
+    #[serde(rename = "multiStage")]
+    pub multi_stage: Option<bool>,
 }
 
 #[nativebridge::native_bridge(DimenstionDefinitionStatic)]

rust/cubesqlplanner/cubesqlplanner/src/cube_bridge/measure_definition.rs

@@ -1,5 +1,6 @@
 use super::cube_definition::{CubeDefinition, NativeCubeDefinition};
 use super::measure_filter::{MeasureFiltersVec, NativeMeasureFiltersVec};
+use super::member_order_by::{MemberOrderByVec, NativeMemberOrderByVec};
 use super::memeber_sql::{MemberSql, NativeMemberSql};
 use cubenativeutils::wrappers::serializer::{
     NativeDeserialize, NativeDeserializer, NativeSerialize,
@@ -11,11 +12,31 @@ use serde::{Deserialize, Serialize};
 use std::any::Any;
 use std::rc::Rc;
 
+#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)]
+pub struct TimeShiftReference {
+    pub interval: String,
+    #[serde(rename = "type")]
+    pub shift_type: Option<String>,
+    #[serde(rename = "timeDimension")]
+    pub time_dimension: String,
+}
+
 #[derive(Serialize, Deserialize, Debug)]
 pub struct MeasureDefinitionStatic {
     #[serde(rename = "type")]
     pub measure_type: String,
+    #[serde(rename = "ownedByCube")]
     pub owned_by_cube: Option<bool>,
+    #[serde(rename = "multiStage")]
+    pub multi_stage: Option<bool>,
+    #[serde(rename = "reduceByReferences")]
+    pub reduce_by_references: Option<Vec<String>>,
+    #[serde(rename = "addGroupByReferences")]
+    pub add_group_by_references: Option<Vec<String>>,
+    #[serde(rename = "groupByReferences")]
+    pub group_by_references: Option<Vec<String>>,
+    #[serde(rename = "timeShiftReferences")]
+    pub time_shift_references: Option<Vec<TimeShiftReference>>,
 }
 
 #[nativebridge::native_bridge(MeasureDefinitionStatic)]
@@ -29,4 +50,8 @@ pub trait MeasureDefinition {
     #[optional]
     #[field]
     fn filters(&self) -> Result<Option<Rc<dyn MeasureFiltersVec>>, CubeError>;
+
+    #[optional]
+    #[field]
+    fn order_by(&self) -> Result<Option<Rc<dyn MemberOrderByVec>>, CubeError>;
 }

rust/cubesqlplanner/cubesqlplanner/src/cube_bridge/member_order_by.rs

@@ -0,0 +1,57 @@
+use super::memeber_sql::{MemberSql, NativeMemberSql};
+use cubenativeutils::wrappers::object::NativeArray;
+use cubenativeutils::wrappers::serializer::{
+    NativeDeserialize, NativeDeserializer, NativeSerialize,
+};
+use cubenativeutils::wrappers::NativeContextHolder;
+use cubenativeutils::wrappers::NativeObjectHandle;
+use cubenativeutils::CubeError;
+use std::any::Any;
+use std::marker::PhantomData;
+use std::rc::Rc;
+
+#[nativebridge::native_bridge]
+pub trait MemberOrderBy {
+    #[field]
+    fn sql(&self) -> Result<Rc<dyn MemberSql>, CubeError>;
+    #[field]
+    fn dir(&self) -> Result<String, CubeError>;
+}
+
+pub trait MemberOrderByVec {
+    fn items(&self) -> &Vec<Rc<dyn MemberOrderBy>>;
+}
+
+pub struct NativeMemberOrderByVec<IT: InnerTypes> {
+    items: Vec<Rc<dyn MemberOrderBy>>,
+    phantom: PhantomData<IT>,
+}
+
+impl<IT: InnerTypes> NativeMemberOrderByVec<IT> {
+    pub fn try_new(native_items: NativeObjectHandle<IT>) -> Result<Self, CubeError> {
+        let items = native_items
+            .into_array()?
+            .to_vec()?
+            .into_iter()
+            .map(|v| -> Result<Rc<dyn MemberOrderBy>, CubeError> {
+                Ok(Rc::new(NativeMemberOrderBy::from_native(v)?))
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+        Ok(Self {
+            items,
+            phantom: PhantomData::default(),
+        })
+    }
+}
+
+impl<IT: InnerTypes> MemberOrderByVec for NativeMemberOrderByVec<IT> {
+    fn items(&self) -> &Vec<Rc<dyn MemberOrderBy>> {
+        &self.items
+    }
+}
+
+impl<IT: InnerTypes> NativeDeserialize<IT> for NativeMemberOrderByVec<IT> {
+    fn from_native(v: NativeObjectHandle<IT>) -> Result<Self, CubeError> {
+        Self::try_new(v)
+    }
+}

rust/cubesqlplanner/cubesqlplanner/src/cube_bridge/mod.rs

@@ -12,6 +12,7 @@ pub mod join_item_definition;
 pub mod measure_definition;
 pub mod measure_filter;
 pub mod member_definition;
+pub mod member_order_by;
 pub mod memeber_sql;
 pub mod security_context;
 pub mod sql_templates_render;

rust/cubesqlplanner/cubesqlplanner/src/plan/expression.rs

@@ -2,18 +2,25 @@ use crate::planner::{BaseMember, VisitorContext};
 use cubenativeutils::CubeError;
 use std::rc::Rc;
 
+#[derive(Clone)]
 pub enum Expr {
     Field(Rc<dyn BaseMember>),
-    Reference(String, String),
+    Reference(Option<String>, String),
+    Asterix,
 }
 
 impl Expr {
     pub fn to_sql(&self, context: Rc<VisitorContext>) -> Result<String, CubeError> {
         match self {
             Expr::Field(field) => field.to_sql(context),
             Expr::Reference(cube_alias, field_alias) => {
-                Ok(format!("{}.{}", cube_alias, field_alias))
+                if let Some(cube_alias) = cube_alias {
+                    Ok(format!("{}.{}", cube_alias, field_alias))
+                } else {
+                    Ok(field_alias.clone())
+                }
             }
+            Expr::Asterix => Ok("*".to_string()),
         }
     }
 }

rust/cubesqlplanner/cubesqlplanner/src/plan/filter.rs

@@ -4,14 +4,16 @@ use cubenativeutils::CubeError;
 use std::fmt;
 use std::rc::Rc;
 
+#[derive(Clone)]
 pub enum FilterGroupOperator {
     Or,
     And,
 }
 
+#[derive(Clone)]
 pub struct FilterGroup {
-    operator: FilterGroupOperator,
-    items: Vec<FilterItem>,
+    pub operator: FilterGroupOperator,
+    pub items: Vec<FilterItem>,
 }
 
 impl FilterGroup {
@@ -49,7 +51,11 @@ impl FilterItem {
                     .iter()
                     .map(|itm| itm.to_sql(context.clone()))
                     .collect::<Result<Vec<_>, _>>()?;
-                format!("({})", items_sql.join(&operator))
+                if items_sql.is_empty() {
+                    format!("( 1 = 1 )")
+                } else {
+                    format!("({})", items_sql.join(&operator))
+                }
             }
             FilterItem::Item(item) => {
                 let sql = item.to_sql(context.clone())?;

rust/cubesqlplanner/cubesqlplanner/src/plan/from.rs

@@ -1,5 +1,4 @@
-use super::Join;
-use super::QueryPlan;
+use super::{Join, QueryPlan, Subquery};
 use crate::planner::{BaseCube, VisitorContext};
 use cubenativeutils::CubeError;
 use std::rc::Rc;
@@ -9,7 +8,8 @@ pub enum FromSource {
     Empty,
     Cube(Rc<BaseCube>),
     Join(Rc<Join>),
-    Subquery(Rc<QueryPlan>, String),
+    Subquery(Subquery),
+    TableReference(String, Option<String>),
 }
 
 #[derive(Clone)]
@@ -26,12 +26,16 @@ impl From {
         Self::new(FromSource::Cube(cube))
     }
 
+    pub fn new_from_table_reference(reference: String, alias: Option<String>) -> Self {
+        Self::new(FromSource::TableReference(reference, alias))
+    }
+
     pub fn new_from_join(join: Rc<Join>) -> Self {
         Self::new(FromSource::Join(join))
     }
 
     pub fn new_from_subquery(plan: Rc<QueryPlan>, alias: String) -> Self {
-        Self::new(FromSource::Subquery(plan, alias))
+        Self::new(FromSource::Subquery(Subquery::new(plan, alias)))
     }
 
     pub fn to_sql(&self, context: Rc<VisitorContext>) -> Result<String, CubeError> {
@@ -44,8 +48,13 @@ impl From {
             FromSource::Join(j) => {
                 format!("{}", j.to_sql(context.clone())?)
             }
-            FromSource::Subquery(s, alias) => {
-                format!("({}) AS {}", s.to_sql()?, alias)
+            FromSource::Subquery(s) => s.to_sql()?,
+            FromSource::TableReference(r, alias) => {
+                if let Some(alias) = alias {
+                    format!(" {} as {} ", r, alias)
+                } else {
+                    format!(" {} ", r)
+                }
             }
         };
         Ok(sql)