use chrono::NaiveDate;
use super::{AggregationSqlGenerator, Result};
use crate::{
compiler::{
aggregation::{AggregateExpression, AggregationPlan, GroupByExpression},
fact_table::{FactTableMetadata, PartialPeriodConfig, TemporalGrain},
},
db::where_clause::WhereClause,
runtime::partial_period::BranchPlan,
};
#[derive(Debug, Clone)]
pub struct PartialPeriodSql {
pub sql: String,
pub params: Vec<serde_json::Value>,
}
impl AggregationSqlGenerator {
pub fn generate_partial_period(
&self,
plan: &AggregationPlan,
config: &PartialPeriodConfig,
branch_plan: &BranchPlan,
extra_where: Option<&WhereClause>,
) -> Result<PartialPeriodSql> {
let mut params: Vec<serde_json::Value> = Vec::new();
let mut branches: Vec<String> = Vec::new();
if let Some((gte, lt)) = branch_plan.partial_leading {
let sql = self.build_branch(
plan,
&config.fine_grain_view,
&config.time_grain_column,
Some(config.time_grain_trunc),
gte,
lt,
extra_where,
&plan.metadata,
&mut params,
)?;
branches.push(sql);
}
if let Some((gte, lt)) = branch_plan.complete_middle {
let sql = self.build_branch(
plan,
&plan.request.table_name,
&config.time_grain_column,
None, gte,
lt,
extra_where,
&plan.metadata,
&mut params,
)?;
branches.push(sql);
}
let (gte, lt) = branch_plan.current_period;
let sql = self.build_branch(
plan,
&config.fine_grain_view,
&config.time_grain_column,
Some(config.time_grain_trunc),
gte,
lt,
extra_where,
&plan.metadata,
&mut params,
)?;
branches.push(sql);
let union_sql = branches.join("\nUNION ALL\n");
Ok(PartialPeriodSql {
sql: union_sql,
params,
})
}
#[allow(clippy::too_many_arguments)] fn build_branch(
&self,
plan: &AggregationPlan,
view_name: &str,
time_col: &str,
trunc_grain: Option<TemporalGrain>,
date_gte: NaiveDate,
date_lt: NaiveDate,
extra_where: Option<&WhereClause>,
metadata: &FactTableMetadata,
params: &mut Vec<serde_json::Value>,
) -> Result<String> {
let select_sql = self.build_branch_select(
&plan.group_by_expressions,
&plan.aggregate_expressions,
time_col,
trunc_grain,
)?;
let from_sql = format!("FROM {view_name}");
let where_sql =
self.build_branch_where(time_col, date_gte, date_lt, extra_where, metadata, params)?;
let group_sql = if !plan.group_by_expressions.is_empty() {
self.build_branch_group_by(&plan.group_by_expressions, time_col, trunc_grain)?
} else {
String::new()
};
let mut parts: Vec<&str> = vec![&select_sql, &from_sql, &where_sql, &group_sql];
parts.retain(|s| !s.is_empty());
Ok(parts.join("\n"))
}
fn build_branch_select(
&self,
group_by_expressions: &[GroupByExpression],
aggregate_expressions: &[AggregateExpression],
time_col: &str,
trunc_grain: Option<TemporalGrain>,
) -> Result<String> {
let mut columns = Vec::new();
for expr in group_by_expressions {
let alias = group_by_alias(expr);
let sql = self.branch_group_by_expr_sql(expr, time_col, trunc_grain)?;
columns.push(format!("{sql} AS {alias}"));
}
for expr in aggregate_expressions {
let column = self.aggregate_expression_to_sql(expr)?;
let alias = aggregate_alias(expr);
columns.push(format!("{column} AS {alias}"));
}
Ok(format!("SELECT\n {}", columns.join(",\n ")))
}
fn build_branch_where(
&self,
time_col: &str,
date_gte: NaiveDate,
date_lt: NaiveDate,
extra_where: Option<&WhereClause>,
metadata: &FactTableMetadata,
params: &mut Vec<serde_json::Value>,
) -> Result<String> {
let gte_param =
self.emit_value_param(&serde_json::Value::String(date_gte.to_string()), params);
let lt_param =
self.emit_value_param(&serde_json::Value::String(date_lt.to_string()), params);
let quoted_col = self.quote_identifier(time_col);
let mut where_parts = vec![format!(
"{quoted_col} >= {gte_param} AND {quoted_col} < {lt_param}"
)];
if let Some(wc) = extra_where {
let extra_sql = self.where_clause_to_sql_parameterized(wc, metadata, params)?;
if !extra_sql.is_empty() {
where_parts.push(extra_sql);
}
}
Ok(format!("WHERE {}", where_parts.join(" AND ")))
}
fn build_branch_group_by(
&self,
group_by_expressions: &[GroupByExpression],
time_col: &str,
trunc_grain: Option<TemporalGrain>,
) -> Result<String> {
let mut exprs = Vec::new();
for expr in group_by_expressions {
exprs.push(self.branch_group_by_expr_sql(expr, time_col, trunc_grain)?);
}
Ok(format!("GROUP BY {}", exprs.join(", ")))
}
fn branch_group_by_expr_sql(
&self,
expr: &GroupByExpression,
time_col: &str,
trunc_grain: Option<TemporalGrain>,
) -> Result<String> {
match (expr, trunc_grain) {
(GroupByExpression::TemporalBucket { column, .. }, Some(grain))
if column == time_col =>
{
Ok(self.temporal_bucket_sql(column, grain.to_temporal_bucket()))
},
(GroupByExpression::NativeColumn { column, .. }, Some(grain)) if column == time_col => {
Ok(self.temporal_bucket_sql(column, grain.to_temporal_bucket()))
},
(
GroupByExpression::TemporalBucket { column, .. }
| GroupByExpression::NativeColumn { column, .. },
None,
) if column == time_col => Ok(self.quote_identifier(column)),
_ => self.group_by_expression_to_sql(expr),
}
}
}
fn group_by_alias(expr: &GroupByExpression) -> &str {
match expr {
GroupByExpression::JsonbPath { alias, .. }
| GroupByExpression::TemporalBucket { alias, .. }
| GroupByExpression::CalendarPath { alias, .. }
| GroupByExpression::NativeColumn { alias, .. } => alias,
}
}
fn aggregate_alias(expr: &AggregateExpression) -> &str {
match expr {
AggregateExpression::Count { alias }
| AggregateExpression::CountDistinct { alias, .. }
| AggregateExpression::MeasureAggregate { alias, .. }
| AggregateExpression::AdvancedAggregate { alias, .. }
| AggregateExpression::BoolAggregate { alias, .. } => alias,
}
}