feat(tesseract): Support of rolling window (including to_date) pre-aggregations

Author: waralexrom

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

@@ -331,10 +331,11 @@ export class BaseQuery {
     this.customSubQueryJoins = this.options.subqueryJoins ?? [];
     this.useNativeSqlPlanner = this.options.useNativeSqlPlanner ?? getEnv('nativeSqlPlanner');
     this.canUseNativeSqlPlannerPreAggregation = true;
-/*     if (this.useNativeSqlPlanner && !this.neverUseSqlPlannerPreaggregation()) {
-      const hasMultiStageMeasures = this.fullKeyQueryAggregateMeasures({ hasMultipliedForPreAggregation: true }).multiStageMembers.length > 0;
-      this.canUseNativeSqlPlannerPreAggregation = hasMultiStageMeasures;
-    } */
+    if (this.useNativeSqlPlanner && !this.neverUseSqlPlannerPreaggregation()) {
+      const fullAggregateMeasures = this.fullKeyQueryAggregateMeasures({ hasMultipliedForPreAggregation: true });
+
+      this.canUseNativeSqlPlannerPreAggregation = fullAggregateMeasures.multiStageMembers.length > 0 || fullAggregateMeasures.cumulativeMeasures.length > 0;
+    }
     this.queryLevelJoinHints = this.options.joinHints ?? [];
     this.prebuildJoin();
 
@@ -776,7 +777,7 @@ export class BaseQuery {
   }
 
   driverTools(external) {
-    if (external && this.options.disableExternalPreAggregations && this.externalQueryClass) {
+    if (external && !this.options.disableExternalPreAggregations && this.externalQueryClass) {
       return this.externalQuery();
     }
     return this;

packages/cubejs-schema-compiler/src/adapter/CubeStoreQuery.ts

@@ -334,4 +334,13 @@ export class CubeStoreQuery extends BaseQuery {
       }
     );
   }
+
+  public sqlTemplates() {
+    const templates = super.sqlTemplates();
+    templates.statements.time_series_select = '{% for time_item in seria  %}' +
+    'select to_timestamp(\'{{ time_item[0] }}\') date_from, to_timestamp(\'{{ time_item[1] }}\') date_to \n' +
+    '{% if not loop.last %} UNION ALL\n{% endif %}' +
+    '{% endfor %}';
+    return templates;
+  }
 }

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

@@ -1353,7 +1353,7 @@ SELECT 1 AS revenue,  cast('2024-01-01' AS timestamp) as time UNION ALL
     { visitors__created_at_week: '2017-01-09T00:00:00.000Z', visitors__revenue_rolling_three_day: '900' }
   ]));
 
-  it('rolling count', async () => runQueryTest({
+  it('rolling count 11', async () => runQueryTest({
     measures: [
       'visitors.countRolling'
     ],

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

@@ -1,10 +1,10 @@
 use super::base_query_options::FilterItem;
+use super::driver_tools::{DriverTools, NativeDriverTools};
 use super::filter_group::{FilterGroup, NativeFilterGroup};
 use super::filter_params::{FilterParams, NativeFilterParams};
 use super::pre_aggregation_obj::{NativePreAggregationObj, PreAggregationObj};
 use super::security_context::{NativeSecurityContext, SecurityContext};
 use super::sql_templates_render::{NativeSqlTemplatesRender, SqlTemplatesRender};
-use super::driver_tools::{NativeDriverTools, DriverTools};
 use super::sql_utils::{NativeSqlUtils, SqlUtils};
 use cubenativeutils::wrappers::serializer::{
     NativeDeserialize, NativeDeserializer, NativeSerialize,
@@ -18,12 +18,6 @@ use std::rc::Rc;
 #[nativebridge::native_bridge]
 pub trait BaseTools {
     fn driver_tools(&self, external: bool) -> Result<Rc<dyn DriverTools>, CubeError>;
-    fn convert_tz(&self, field: String) -> Result<String, CubeError>;
-    fn time_grouped_column(
-        &self,
-        granularity: String,
-        dimension: String,
-    ) -> Result<String, CubeError>;
     fn sql_templates(&self) -> Result<Rc<dyn SqlTemplatesRender>, CubeError>;
     fn security_context_for_rust(&self) -> Result<Rc<dyn SecurityContext>, CubeError>;
     fn sql_utils_for_rust(&self) -> Result<Rc<dyn SqlUtils>, CubeError>;
@@ -35,10 +29,6 @@ pub trait BaseTools {
         &self,
         used_filters: Option<Vec<FilterItem>>,
     ) -> Result<Rc<dyn FilterGroup>, CubeError>;
-    fn timestamp_precision(&self) -> Result<u32, CubeError>;
-    fn time_stamp_cast(&self, field: String) -> Result<String, CubeError>; //TODO move to templates
-    fn date_time_cast(&self, field: String) -> Result<String, CubeError>; //TODO move to templates
-    fn in_db_time_zone(&self, date: String) -> Result<String, CubeError>;
     fn generate_time_series(
         &self,
         granularity: String,
@@ -51,23 +41,8 @@ pub trait BaseTools {
         origin: String,
     ) -> Result<Vec<Vec<String>>, CubeError>;
     fn get_allocated_params(&self) -> Result<Vec<String>, CubeError>;
-    fn subtract_interval(&self, date: String, interval: String) -> Result<String, CubeError>;
-    fn add_interval(&self, date: String, interval: String) -> Result<String, CubeError>;
-    fn add_timestamp_interval(&self, date: String, interval: String) -> Result<String, CubeError>;
     fn all_cube_members(&self, path: String) -> Result<Vec<String>, CubeError>;
     fn interval_and_minimal_time_unit(&self, interval: 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>;
-    fn date_bin(
-        &self,
-        interval: String,
-        source: String,
-        origin: String,
-    ) -> Result<String, CubeError>;
-
     fn get_pre_aggregation_by_name(
         &self,
         cube_name: String,

rust/cubesqlplanner/cubesqlplanner/src/cube_bridge/driver_tools.rs

@@ -1,10 +1,4 @@
-use super::base_query_options::FilterItem;
-use super::filter_group::{FilterGroup, NativeFilterGroup};
-use super::filter_params::{FilterParams, NativeFilterParams};
-use super::pre_aggregation_obj::{NativePreAggregationObj, PreAggregationObj};
-use super::security_context::{NativeSecurityContext, SecurityContext};
 use super::sql_templates_render::{NativeSqlTemplatesRender, SqlTemplatesRender};
-use super::sql_utils::{NativeSqlUtils, SqlUtils};
 use cubenativeutils::wrappers::serializer::{
     NativeDeserialize, NativeDeserializer, NativeSerialize,
 };

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

@@ -6,6 +6,7 @@ pub mod case_item;
 pub mod case_label;
 pub mod cube_definition;
 pub mod dimension_definition;
+pub mod driver_tools;
 pub mod evaluator;
 pub mod filter_group;
 pub mod filter_params;
@@ -28,4 +29,3 @@ pub mod segment_definition;
 pub mod sql_templates_render;
 pub mod sql_utils;
 pub mod struct_with_sql_member;
-pub mod driver_tools;

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

@@ -1,5 +1,4 @@
 use super::*;
-use crate::cube_bridge::pre_aggregation_obj::PreAggregationObj;
 use crate::logical_plan::*;
 use crate::plan::FilterItem;
 use crate::planner::query_tools::QueryTools;
@@ -29,7 +28,7 @@ impl MatchState {
 
 pub struct PreAggregationOptimizer {
     query_tools: Rc<QueryTools>,
-    used_pre_aggregations: HashMap<(String, String), Rc<dyn PreAggregationObj>>,
+    used_pre_aggregations: HashMap<(String, String), Rc<PreAggregation>>,
 }
 
 impl PreAggregationOptimizer {
@@ -71,7 +70,7 @@ impl PreAggregationOptimizer {
         Ok(None)
     }
 
-    pub fn get_used_pre_aggregations(&self) -> Vec<Rc<dyn PreAggregationObj>> {
+    pub fn get_used_pre_aggregations(&self) -> Vec<Rc<PreAggregation>> {
         self.used_pre_aggregations.values().cloned().collect()
     }
 
@@ -445,15 +444,14 @@ impl PreAggregationOptimizer {
                 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(
-                (
-                    pre_aggregation.cube_name.clone(),
-                    pre_aggregation.name.clone(),
-                ),
-                pre_aggregation_obj.clone(),
+                (result.cube_name.clone(), result.name.clone()),
+                result.clone(),
             );
-            Ok(Rc::new(pre_aggregation))
+            Ok(result)
         } else {
             Err(CubeError::internal(format!(
                 "Cannot find pre aggregation object for cube {} and name {}",

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

@@ -1,4 +1,5 @@
 use super::*;
+use crate::cube_bridge::pre_aggregation_obj::PreAggregationObj;
 use crate::planner::sql_evaluator::MemberSymbol;
 use itertools::Itertools;
 use std::rc::Rc;
@@ -13,6 +14,7 @@ pub struct PreAggregation {
     pub granularity: Option<String>,
     pub table_name: String,
     pub cube_name: String,
+    pub pre_aggregation_obj: Rc<dyn PreAggregationObj>,
 }
 
 impl PrettyPrint for PreAggregation {

rust/cubesqlplanner/cubesqlplanner/src/physical_plan_builder/builder.rs

@@ -101,6 +101,7 @@ impl PhysicalPlanBuilder {
     ) -> Result<Rc<Select>, CubeError> {
         let mut render_references = HashMap::new();
         let mut measure_references = HashMap::new();
+        let mut dimensions_references = HashMap::new();
         let mut context_factory = context.make_sql_nodes_factory();
         let from = match &logical_plan.source {
             SimpleQuerySource::LogicalJoin(join) => self.process_logical_join(
@@ -113,8 +114,8 @@ impl PhysicalPlanBuilder {
                 let res = self.process_pre_aggregation(
                     pre_aggregation,
                     context,
-                    &mut render_references,
                     &mut measure_references,
+                    &mut dimensions_references,
                 )?;
                 for member in logical_plan.schema.time_dimensions.iter() {
                     context_factory.add_dimensions_with_ignored_timezone(member.full_name());
@@ -127,6 +128,7 @@ impl PhysicalPlanBuilder {
         let mut select_builder = SelectBuilder::new(from);
         context_factory.set_ungrouped(logical_plan.ungrouped);
         context_factory.set_pre_aggregation_measures_references(measure_references);
+        context_factory.set_pre_aggregation_dimensions_references(dimensions_references);
 
         let mut group_by = Vec::new();
         for member in logical_plan.schema.dimensions.iter() {
@@ -184,8 +186,8 @@ impl PhysicalPlanBuilder {
         &self,
         pre_aggregation: &Rc<PreAggregation>,
         _context: &PhysicalPlanBuilderContext,
-        render_references: &mut HashMap<String, QualifiedColumnName>,
         measure_references: &mut HashMap<String, QualifiedColumnName>,
+        dimensions_references: &mut HashMap<String, QualifiedColumnName>,
     ) -> Result<Rc<From>, CubeError> {
         let mut pre_aggregation_schema = Schema::empty();
         let pre_aggregation_alias = PlanSqlTemplates::memeber_alias_name(
@@ -200,7 +202,7 @@ impl PhysicalPlanBuilder {
                 &dim.alias_suffix(),
                 self.query_tools.clone(),
             )?;
-            render_references.insert(
+            dimensions_references.insert(
                 dim.full_name(),
                 QualifiedColumnName::new(Some(pre_aggregation_alias.clone()), alias.clone()),
             );
@@ -213,16 +215,10 @@ impl PhysicalPlanBuilder {
                 granularity,
                 self.query_tools.clone(),
             )?;
-            render_references.insert(
+            dimensions_references.insert(
                 dim.full_name(),
                 QualifiedColumnName::new(Some(pre_aggregation_alias.clone()), alias.clone()),
             );
-            if let Some(granularity) = &granularity {
-                render_references.insert(
-                    format!("{}_{}", dim.full_name(), granularity),
-                    QualifiedColumnName::new(Some(pre_aggregation_alias.clone()), alias.clone()),
-                );
-            }
             pre_aggregation_schema.add_column(SchemaColumn::new(alias, Some(dim.full_name())));
         }
         for meas in pre_aggregation.measures.iter() {
@@ -969,9 +965,7 @@ impl PhysicalPlanBuilder {
             ));
         };
 
-        let templates = self.query_tools.plan_sql_templates(false)?;
-
-        let ts_date_range = if templates.supports_generated_time_series()
+        let ts_date_range = if self.plan_sql_templates.supports_generated_time_series()
             && granularity_obj.is_predefined_granularity()
         {
             if let Some(date_range) = time_dimension_symbol

rust/cubesqlplanner/cubesqlplanner/src/plan/join.rs

@@ -117,7 +117,7 @@ pub struct ToDateRollingWindowJoinCondition {
     time_series_source: String,
     granularity: String,
     time_dimension: Expr,
-    query_tools: Rc<QueryTools>,
+    _query_tools: Rc<QueryTools>,
 }
 
 impl ToDateRollingWindowJoinCondition {
@@ -131,7 +131,7 @@ impl ToDateRollingWindowJoinCondition {
             time_series_source,
             granularity,
             time_dimension,
-            query_tools,
+            _query_tools: query_tools,
         }
     }
 
@@ -146,10 +146,7 @@ impl ToDateRollingWindowJoinCondition {
             templates.column_reference(&Some(self.time_series_source.clone()), "date_to")?;
         let date_to =
             templates.column_reference(&Some(self.time_series_source.clone()), "date_from")?;
-        let grouped_from = self
-            .query_tools
-            .base_tools()
-            .time_grouped_column(self.granularity.clone(), date_from)?;
+        let grouped_from = templates.time_grouped_column(self.granularity.clone(), date_from)?;
         let result = format!("{date_column} >= {grouped_from} and {date_column} <= {date_to}");
         Ok(result)
     }