fix(tesseract): Fix rolling window external pre-aggregation

Author: waralexrom

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

@@ -330,11 +330,11 @@ export class BaseQuery {
      */
     this.customSubQueryJoins = this.options.subqueryJoins ?? [];
     this.useNativeSqlPlanner = this.options.useNativeSqlPlanner ?? getEnv('nativeSqlPlanner');
-    this.canUseNativeSqlPlannerPreAggregation = false;
-    if (this.useNativeSqlPlanner && !this.neverUseSqlPlannerPreaggregation()) {
+    this.canUseNativeSqlPlannerPreAggregation = true;
+/*     if (this.useNativeSqlPlanner && !this.neverUseSqlPlannerPreaggregation()) {
       const hasMultiStageMeasures = this.fullKeyQueryAggregateMeasures({ hasMultipliedForPreAggregation: true }).multiStageMembers.length > 0;
       this.canUseNativeSqlPlannerPreAggregation = hasMultiStageMeasures;
-    }
+    } */
     this.queryLevelJoinHints = this.options.joinHints ?? [];
     this.prebuildJoin();
 
@@ -775,6 +775,13 @@ export class BaseQuery {
     );
   }
 
+  driverTools(external) {
+    if (external && this.options.disableExternalPreAggregations && this.externalQueryClass) {
+      return this.externalQuery();
+    }
+    return this;
+  }
+
   buildSqlAndParamsRust(exportAnnotatedSql) {
     const order = this.options.order && R.pipe(
       R.map((hash) => ((!hash || !hash.id) ? null : hash)),

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

@@ -4,6 +4,7 @@ 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,
@@ -16,6 +17,7 @@ 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,

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

@@ -0,0 +1,45 @@
+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,
+};
+use cubenativeutils::wrappers::NativeContextHolder;
+use cubenativeutils::wrappers::NativeObjectHandle;
+use cubenativeutils::CubeError;
+use std::any::Any;
+use std::rc::Rc;
+
+#[nativebridge::native_bridge]
+pub trait DriverTools {
+    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 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 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 interval_and_minimal_time_unit(&self, interval: String) -> Result<Vec<String>, CubeError>;
+    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>;
+}

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

@@ -28,3 +28,4 @@ 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/physical_plan_builder/builder.rs

@@ -41,15 +41,14 @@ impl PhysicalPlanBuilderContext {
 
 pub struct PhysicalPlanBuilder {
     query_tools: Rc<QueryTools>,
-    _plan_sql_templates: PlanSqlTemplates,
+    plan_sql_templates: PlanSqlTemplates,
 }
 
 impl PhysicalPlanBuilder {
-    pub fn new(query_tools: Rc<QueryTools>) -> Self {
-        let plan_sql_templates = query_tools.plan_sql_templates();
+    pub fn new(query_tools: Rc<QueryTools>, plan_sql_templates: PlanSqlTemplates) -> Self {
         Self {
             query_tools,
-            _plan_sql_templates: plan_sql_templates,
+            plan_sql_templates,
         }
     }
 
@@ -970,7 +969,7 @@ impl PhysicalPlanBuilder {
             ));
         };
 
-        let templates = self.query_tools.plan_sql_templates();
+        let templates = self.query_tools.plan_sql_templates(false)?;
 
         let ts_date_range = if templates.supports_generated_time_series()
             && granularity_obj.is_predefined_granularity()

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

@@ -52,9 +52,7 @@ impl RegularRollingWindowJoinCondition {
             };
 
             let trailing_start = if let Some(trailing_interval) = &self.trailing_interval {
-                templates
-                    .base_tools()
-                    .subtract_interval(start_date, trailing_interval.clone())?
+                templates.subtract_interval(start_date, trailing_interval.clone())?
             } else {
                 start_date
             };
@@ -72,9 +70,7 @@ impl RegularRollingWindowJoinCondition {
             };
 
             let leading_end = if let Some(leading_interval) = &self.leading_interval {
-                templates
-                    .base_tools()
-                    .add_interval(end_date, leading_interval.clone())?
+                templates.add_interval(end_date, leading_interval.clone())?
             } else {
                 end_date
             };

rust/cubesqlplanner/cubesqlplanner/src/plan/time_series.rs

@@ -44,7 +44,6 @@ impl TimeSeries {
             && self.granularity.is_predefined_granularity()
         {
             let interval_description = templates
-                .base_tools()
                 .interval_and_minimal_time_unit(self.granularity.granularity_interval().clone())?;
             if interval_description.len() != 2 {
                 return Err(CubeError::internal(

rust/cubesqlplanner/cubesqlplanner/src/planner/base_query.rs

@@ -78,14 +78,14 @@ impl<IT: InnerTypes> BaseQuery<IT> {
     }
 
     fn build_sql_and_params_impl(&self) -> Result<NativeObjectHandle<IT>, CubeError> {
-        let templates = self.query_tools.plan_sql_templates();
+        let templates = self.query_tools.plan_sql_templates(false)?;
         let query_planner = QueryPlanner::new(self.request.clone(), self.query_tools.clone());
         let logical_plan = query_planner.plan()?;
 
         let (optimized_plan, used_pre_aggregations) =
             self.try_pre_aggregations(logical_plan.clone())?;
 
-        let physical_plan_builder = PhysicalPlanBuilder::new(self.query_tools.clone());
+        let physical_plan_builder = PhysicalPlanBuilder::new(self.query_tools.clone(), templates.clone());
         let original_sql_pre_aggregations = if !self.request.is_pre_aggregation_query() {
             OriginalSqlCollector::new(self.query_tools.clone()).collect(&optimized_plan)?
         } else {