Skip to content

Commit 8503379

Browse files
waralexromwaralexrom
committed
in work
1 parent 58734ba commit 8503379

File tree

17 files changed

+325
-90
lines changed

17 files changed

+325
-90
lines changed

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -734,6 +734,7 @@ export class BaseQuery {
734734
offset = offset || 'end';
735735
return this.timeDimensions.map(
736736
d => [d, (dateFrom, dateTo, dateField, dimensionDateFrom, dimensionDateTo, isFromStartToEnd) => {
737+
console.log("!!!!!! IsFromStartToEnd: ", isFromStartToEnd, " --");
737738
// dateFrom based window
738739
const conditions = [];
739740
if (trailingInterval !== 'unbounded') {
@@ -1408,7 +1409,6 @@ export class BaseQuery {
14081409
) || undefined
14091410
);
14101411
const baseQueryAlias = this.cubeAlias('base');
1411-
console.log("!!! date join contdition: ", dateJoinCondition);
14121412
const dateJoinConditionSql =
14131413
dateJoinCondition.map(
14141414
([d, f]) => f(
@@ -3263,7 +3263,7 @@ export class BaseQuery {
32633263
group_by_exprs: '{{ group_by | map(attribute=\'index\') | join(\', \') }}',
32643264
join: '{{ join_type }} JOIN {{ source }} ON {{ condition }}',
32653265
cte: '{{ alias }} AS ({{ query | indent(2, true) }})',
3266-
time_seria_select: 'SELECT date_from::timestamp AS "date_from",\n' +
3266+
time_series_select: 'SELECT date_from::timestamp AS "date_from",\n' +
32673267
'date_to::timestamp AS "date_to" \n' +
32683268
'FROM(\n' +
32693269
' VALUES ' +

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -23,9 +23,9 @@ pub struct TimeShiftReference {
2323

2424
#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)]
2525
pub struct RollingWindow {
26-
trailing: Option<String>,
27-
leading: Option<String>,
28-
offset: Option<String>,
26+
pub trailing: Option<String>,
27+
pub leading: Option<String>,
28+
pub offset: Option<String>,
2929
}
3030

3131
#[derive(Serialize, Deserialize, Debug)]

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

Lines changed: 114 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1,76 +1,139 @@
1-
use super::{time_seria, Schema, SingleAliasedSource};
1+
use super::{time_series, Schema, SingleAliasedSource};
22
use crate::planner::sql_templates::PlanSqlTemplates;
33
use crate::planner::{BaseJoinCondition, BaseMember, VisitorContext};
44
use cubenativeutils::CubeError;
55

66
use std::rc::Rc;
77

88
pub struct RollingWindowJoinCondition {
9-
tailing_interval: Option<String>,
9+
data_source: String,
10+
time_series_source: String,
11+
trailing_interval: Option<String>,
1012
leading_interval: Option<String>,
1113
offset: String,
12-
is_from_start_to_end: bool,
13-
time_dimension: Vec<Rc<BaseMember>>,
14+
time_dimension: Rc<dyn BaseMember>,
1415
}
1516

1617
impl RollingWindowJoinCondition {
1718
pub fn new(
19+
data_source: String,
20+
time_series_source: String,
1821
trailing_interval: Option<String>,
1922
leading_interval: Option<String>,
2023
offset: String,
21-
is_from_start_to_end: bool,
22-
dimensions: Vec<Rc<BaseMember>>,
24+
time_dimension: Rc<dyn BaseMember>,
2325
) -> Self {
2426
Self {
25-
tailing_interval,
27+
data_source,
28+
time_series_source,
29+
trailing_interval,
2630
leading_interval,
2731
offset,
28-
is_from_start_to_end,
2932
time_dimension,
3033
}
3134
}
3235

36+
/*
37+
*
38+
offset = offset || 'end';
39+
return this.timeDimensions.map(
40+
d => [d, (dateFrom, dateTo, dateField, dimensionDateFrom, dimensionDateTo, isFromStartToEnd) => {
41+
// dateFrom based window
42+
const conditions = [];
43+
if (trailingInterval !== 'unbounded') {
44+
const startDate = isFromStartToEnd || offset === 'start' ? dateFrom : dateTo;
45+
const trailingStart = trailingInterval ? this.subtractInterval(startDate, trailingInterval) : startDate;
46+
const sign = offset === 'start' ? '>=' : '>';
47+
conditions.push(`${dateField} ${sign} ${trailingStart}`);
48+
}
49+
if (leadingInterval !== 'unbounded') {
50+
const endDate = isFromStartToEnd || offset === 'end' ? dateTo : dateFrom;
51+
const leadingEnd = leadingInterval ? this.addInterval(endDate, leadingInterval) : endDate;
52+
const sign = offset === 'end' ? '<=' : '<';
53+
conditions.push(`${dateField} ${sign} ${leadingEnd}`);
54+
}
55+
return conditions.length ? conditions.join(' AND ') : '1 = 1';
56+
}]
57+
);
58+
*/
3359
pub fn to_sql(
3460
&self,
3561
templates: &PlanSqlTemplates,
3662
context: Rc<VisitorContext>,
3763
schema: Rc<Schema>,
3864
) -> Result<String, CubeError> {
39-
let result = if self.dimensions.is_empty() {
40-
format!("1 = 1")
65+
let mut conditions = vec![];
66+
/* let date_column_alias = if let Some(column) = schema.find_column_for_member(&self.time_dimension.full_name(), &None) {
67+
templates.column_reference(&source, &column.alias.clone())
4168
} else {
42-
let conditions = vec![];
43-
self.dimensions
44-
.iter()
45-
.map(|dim| -> Result<String, CubeError> {
46-
if let Some(trailing_interval) = self.trailing_interval {
47-
if tailing_interval == "unbounded" {
48-
let seria_column = "date_from",
49-
}
50-
}
69+
dimension.to_sql(context.clone(), schema.clone())
70+
} */
71+
let date_column_alias =
72+
self.resolve_time_column_alias(templates, context.clone(), schema.clone())?;
73+
if let Some(trailing_interval) = &self.trailing_interval {
74+
if trailing_interval != "unbounded" {
75+
let start_date = if self.offset == "start" {
76+
templates
77+
.column_reference(&Some(self.time_series_source.clone()), "date_from")?
78+
} else {
79+
templates.column_reference(&Some(self.time_series_source.clone()), "date_to")?
80+
};
5181

82+
let trailing_start = if let Some(trailing_interval) = &self.trailing_interval {
83+
format!("{start_date} - interval '{trailing_interval}'")
84+
} else {
85+
start_date
86+
};
5287

53-
})
54-
.collect::<Result<Vec<_>, _>>()?
55-
.join(" AND ")
88+
let sign = if self.offset == "start" { ">=" } else { ">" };
89+
90+
conditions.push(format!("{date_column_alias} {sign} {trailing_start}"));
91+
}
92+
}
93+
94+
if let Some(leading_interval) = &self.trailing_interval {
95+
if leading_interval != "unbounded" {
96+
let end_date = if self.offset == "end" {
97+
templates.column_reference(&Some(self.time_series_source.clone()), "date_to")?
98+
} else {
99+
templates
100+
.column_reference(&Some(self.time_series_source.clone()), "date_from")?
101+
};
102+
103+
let leading_end = if let Some(leading_interval) = &self.leading_interval {
104+
format!("{end_date} + interval '{leading_interval}'")
105+
} else {
106+
end_date
107+
};
108+
109+
let sign = if self.offset == "end" { "<=" } else { "<" };
110+
111+
conditions.push(format!("{date_column_alias} {sign} {leading_end}"));
112+
}
113+
}
114+
115+
let result = if conditions.is_empty() {
116+
templates.always_true()?
117+
} else {
118+
conditions.join(" AND ")
56119
};
57120
Ok(result)
58121
}
59122

60-
fn resolve_member_alias(
123+
fn resolve_time_column_alias(
61124
&self,
62125
templates: &PlanSqlTemplates,
63126
context: Rc<VisitorContext>,
64-
source: &String,
65-
dimension: &Rc<dyn BaseMember>,
66127
schema: Rc<Schema>,
67128
) -> Result<String, CubeError> {
68-
let schema = schema.extract_source_schema(source);
69-
let source = Some(source.clone());
70-
if let Some(column) = schema.find_column_for_member(&dimension.full_name(), &source) {
129+
let schema = schema.extract_source_schema(&self.data_source);
130+
let source = Some(self.data_source.clone());
131+
if let Some(column) =
132+
schema.find_column_for_member(&self.time_dimension.full_name(), &source)
133+
{
71134
templates.column_reference(&source, &column.alias.clone())
72135
} else {
73-
dimension.to_sql(context.clone(), schema.clone())
136+
self.time_dimension.to_sql(context.clone(), schema.clone())
74137
}
75138
}
76139
}
@@ -162,6 +225,7 @@ impl DimensionJoinCondition {
162225
pub enum JoinCondition {
163226
DimensionJoinCondition(DimensionJoinCondition),
164227
BaseJoinCondition(Rc<dyn BaseJoinCondition>),
228+
RollingWindowJoinCondition(RollingWindowJoinCondition),
165229
}
166230

167231
impl JoinCondition {
@@ -179,6 +243,24 @@ impl JoinCondition {
179243
))
180244
}
181245

246+
pub fn new_rolling_join(
247+
data_source: String,
248+
time_series_source: String,
249+
trailing_interval: Option<String>,
250+
leading_interval: Option<String>,
251+
offset: String,
252+
time_dimension: Rc<dyn BaseMember>,
253+
) -> Self {
254+
Self::RollingWindowJoinCondition(RollingWindowJoinCondition::new(
255+
data_source,
256+
time_series_source,
257+
trailing_interval,
258+
leading_interval,
259+
offset,
260+
time_dimension,
261+
))
262+
}
263+
182264
pub fn new_base_join(base: Rc<dyn BaseJoinCondition>) -> Self {
183265
Self::BaseJoinCondition(base)
184266
}
@@ -192,6 +274,9 @@ impl JoinCondition {
192274
match &self {
193275
JoinCondition::DimensionJoinCondition(cond) => cond.to_sql(templates, context, schema),
194276
JoinCondition::BaseJoinCondition(cond) => cond.to_sql(context, schema),
277+
JoinCondition::RollingWindowJoinCondition(cond) => {
278+
cond.to_sql(templates, context, schema)
279+
}
195280
}
196281
}
197282
}

‎rust/cubesqlplanner/cubesqlplanner/src/plan/mod.rs‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -8,19 +8,18 @@ pub mod order;
88
pub mod query_plan;
99
pub mod schema;
1010
pub mod select;
11-
pub mod time_seria;
11+
pub mod time_series;
1212
pub mod union;
1313

1414
pub use builder::{JoinBuilder, SelectBuilder};
1515
pub use cte::Cte;
1616
pub use expression::{Expr, MemberExpression};
1717
pub use filter::{Filter, FilterGroup, FilterItem};
1818
pub use from::{From, FromSource, SingleAliasedSource, SingleSource};
19-
pub use join::{Join, JoinCondition, JoinItem};
19+
pub use join::{Join, JoinCondition, JoinItem, RollingWindowJoinCondition};
2020
pub use order::OrderBy;
2121
pub use query_plan::QueryPlan;
2222
pub use schema::{Schema, SchemaColumn, SchemaCube};
2323
pub use select::{AliasedExpr, Select};
24-
pub use time_seria::TimeSeria;
24+
pub use time_series::TimeSeries;
2525
pub use union::Union;
26-

‎rust/cubesqlplanner/cubesqlplanner/src/plan/query_plan.rs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,27 +1,27 @@
1-
use super::{Schema, Select, TimeSeria, Union};
1+
use super::{Schema, Select, TimeSeries, Union};
22
use crate::planner::sql_templates::PlanSqlTemplates;
33
use cubenativeutils::CubeError;
44
use std::rc::Rc;
55

66
pub enum QueryPlan {
77
Select(Rc<Select>),
88
Union(Rc<Union>),
9-
TimeSeria(Rc<TimeSeria>),
9+
TimeSeries(Rc<TimeSeries>),
1010
}
1111

1212
impl QueryPlan {
1313
pub fn make_schema(&self, self_alias: Option<String>) -> Schema {
1414
match self {
1515
QueryPlan::Select(select) => select.make_schema(self_alias),
1616
QueryPlan::Union(union) => union.make_schema(self_alias),
17-
QueryPlan::TimeSeria(seria) => seria.make_schema(self_alias),
17+
QueryPlan::TimeSeries(series) => series.make_schema(self_alias),
1818
}
1919
}
2020
pub fn to_sql(&self, templates: &PlanSqlTemplates) -> Result<String, CubeError> {
2121
match self {
2222
QueryPlan::Select(s) => s.to_sql(templates),
2323
QueryPlan::Union(u) => u.to_sql(templates),
24-
QueryPlan::TimeSeria(seria) => seria.to_sql(templates),
24+
QueryPlan::TimeSeries(series) => series.to_sql(templates),
2525
}
2626
}
2727
}

‎rust/cubesqlplanner/cubesqlplanner/src/plan/time_seria.rs‎ renamed to ‎rust/cubesqlplanner/cubesqlplanner/src/plan/time_series.rs‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,25 +3,26 @@ use crate::planner::sql_templates::PlanSqlTemplates;
33
use cubenativeutils::CubeError;
44
use std::rc::Rc;
55

6-
pub struct TimeSeria {
6+
pub struct TimeSeries {
77
pub time_dimension_name: String,
88
pub from_date: Option<String>,
99
pub to_date: Option<String>,
1010
pub seria: Vec<Vec<String>>,
1111
}
1212

13-
impl TimeSeria {
13+
impl TimeSeries {
1414
pub fn make_schema(&self, self_alias: Option<String>) -> Schema {
15-
let column = SchemaColumn::new(
15+
/* let column = SchemaColumn::new(
1616
self_alias,
1717
format!("from_date"),
1818
self.time_dimension_name.clone(),
1919
);
20-
Schema::new(vec![column], vec![])
20+
Schema::new(vec![column], vec![]) */
21+
Schema::empty()
2122
}
2223

2324
pub fn to_sql(&self, templates: &PlanSqlTemplates) -> Result<String, CubeError> {
24-
templates.time_seria_select(
25+
templates.time_series_select(
2526
self.from_date.clone(),
2627
self.to_date.clone(),
2728
self.seria.clone(),

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

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
use crate::cube_bridge::measure_definition::{MeasureDefinition, TimeShiftReference};
22
use crate::planner::sql_evaluator::EvaluationNode;
3+
use crate::planner::BaseMember;
34
use crate::planner::BaseTimeDimension;
45
use cubenativeutils::CubeError;
56
use lazy_static::lazy_static;
@@ -61,15 +62,23 @@ impl MultiStageTimeShift {
6162
#[derive(Clone)]
6263
pub enum MultiStageLeafMemberType {
6364
Measure,
64-
TimeSeria(Rc<BaseTimeDimension>),
65+
TimeSeries(Rc<BaseTimeDimension>),
66+
}
67+
68+
#[derive(Clone)]
69+
pub struct RollingWindowDescription {
70+
pub time_dimension: Rc<dyn BaseMember>,
71+
pub trailing: Option<String>,
72+
pub leading: Option<String>,
73+
pub offset: String,
6574
}
6675

6776
#[derive(Clone)]
6877
pub enum MultiStageInodeMemberType {
6978
Rank,
7079
Aggregate,
7180
Calculate,
72-
RollingWindow,
81+
RollingWindow(RollingWindowDescription),
7382
}
7483

7584
#[derive(Clone)]

0 commit comments

Comments
 (0)