sift_query_sqlx/renderer/
mod.rs

1use std::fmt::Write;
2use std::sync::Arc;
3
4use sift_query_algorithm::{FNV1A_64_OFFSET_BASIS, StableHasher64};
5use sift_query_core::{
6    BucketRequest, FieldId, FieldKind, PagingMode, QueryPlan, QueryTarget, ScalarValue,
7    SearchEntity, SearchOptions, TypedList, ValueType,
8};
9
10use crate::RenderError;
11
12mod cursor;
13pub(crate) mod predicate;
14mod types;
15mod window;
16
17use cursor::render_cursor;
18use predicate::render_arena;
19use types::{render_aggregate, supports_aggregate};
20
21#[derive(Debug, Clone, PartialEq)]
22pub enum PgBindValue {
23    Scalar(ScalarValue),
24    Array(TypedList),
25}
26
27#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
28#[repr(transparent)]
29pub struct QueryShapeId(pub u64);
30
31#[derive(Debug, Clone, PartialEq)]
32pub struct RenderedPgQuery {
33    pub sql: Arc<str>,
34    pub binds: Vec<PgBindValue>,
35    pub shape_id: QueryShapeId,
36}
37
38#[derive(Debug, Clone, Default)]
39pub struct PgRenderer;
40
41impl PgRenderer {
42    /// Renders a bounded time or numeric bucket aggregation using only a
43    /// server-declared bucket request.
44    pub fn render_bucket<E: SearchEntity>(
45        &self,
46        plan: &QueryPlan<E>,
47        request: &BucketRequest<E>,
48    ) -> Result<RenderedPgQuery, RenderError> {
49        let field_id = request.field_id();
50        plan.validate_field_access([field_id])
51            .map_err(|error| RenderError::InvalidCondition(error.to_string()))?;
52        let field = plan
53            .meta
54            .field(field_id)
55            .ok_or(RenderError::UnknownField(field_id.0))?;
56        if !field.selectable || field.kind == FieldKind::FilterOnly {
57            return Err(RenderError::InvalidCondition(
58                "bucket field must be selectable".to_owned(),
59            ));
60        }
61        let mut state = RenderState::default();
62        let mut sql = sql_buffer(plan);
63        match request {
64            BucketRequest::Time(request) => {
65                if !is_time_value_type(field.value_type) {
66                    return Err(RenderError::InvalidCondition(
67                        "time buckets require a date/time field".to_owned(),
68                    ));
69                }
70                let unit = request.unit().sql_name();
71                let timezone = request.timezone();
72                let null_label = state.bind(ScalarValue::Text("null".to_owned()));
73                write!(
74                    sql,
75                    "SELECT CASE WHEN {field} IS NULL THEN {null_label} ELSE (date_trunc('{unit}', {field} AT TIME ZONE '{timezone}'))::text END AS \"sq_bucket\", count(*)::bigint AS \"sq_count\" FROM {from}",
76                    field = field.column,
77                    null_label = null_label,
78                    unit = unit,
79                    timezone = timezone,
80                    from = plan.meta.from
81                )
82                .expect("writing to String cannot fail");
83            }
84            BucketRequest::Numeric(request) => {
85                if !matches!(
86                    field.value_type,
87                    ValueType::I32 | ValueType::I64 | ValueType::F32 | ValueType::F64
88                ) || request
89                    .boundaries()
90                    .iter()
91                    .any(|value| value.value_type() != field.value_type)
92                {
93                    return Err(RenderError::InvalidCondition(
94                        "numeric buckets require matching numeric boundaries".to_owned(),
95                    ));
96                }
97                sql.push_str("SELECT CASE WHEN ");
98                sql.push_str(field.column);
99                sql.push_str(" IS NULL THEN ");
100                sql.push_str(&state.bind(ScalarValue::Text("null".to_owned())));
101                for (index, boundary) in request.boundaries().iter().enumerate() {
102                    write!(
103                        sql,
104                        " WHEN {} < {} THEN {}",
105                        field.column,
106                        state.bind(boundary.clone()),
107                        state.bind(ScalarValue::Text(sift_query_core::numeric_bucket_label(
108                            request.boundaries(),
109                            index
110                        )))
111                    )
112                    .expect("writing to String cannot fail");
113                }
114                write!(
115                    sql,
116                    " ELSE {} END AS \"sq_bucket\", count(*)::bigint AS \"sq_count\" FROM {}",
117                    state.bind(ScalarValue::Text(sift_query_core::numeric_bucket_label(
118                        request.boundaries(),
119                        request.boundaries().len()
120                    ))),
121                    plan.meta.from
122                )
123                .expect("writing to String cannot fail");
124            }
125        }
126        render_filters(plan, &mut state, &mut sql, false)?;
127        match request {
128            BucketRequest::Time(_) => sql.push_str(" GROUP BY sq_bucket ORDER BY sq_bucket ASC"),
129            BucketRequest::Numeric(_) => {
130                write!(
131                    sql,
132                    " GROUP BY sq_bucket ORDER BY min({}) ASC NULLS LAST",
133                    field.column
134                )
135                .expect("writing to String cannot fail");
136            }
137        }
138        Ok(state.finish(sql, plan.shape_fingerprint_seed()))
139    }
140
141    /// Renders a bounded, server-declared facet query over a row-phase field.
142    pub fn render_facet<E: SearchEntity>(
143        &self,
144        plan: &QueryPlan<E>,
145        field_id: FieldId,
146        limit: u32,
147        include_null: bool,
148    ) -> Result<RenderedPgQuery, RenderError> {
149        self.render_facet_with_options(plan, field_id, limit, include_null, false)
150    }
151
152    pub fn render_facet_with_options<E: SearchEntity>(
153        &self,
154        plan: &QueryPlan<E>,
155        field_id: FieldId,
156        limit: u32,
157        include_null: bool,
158        exclude_self_filter: bool,
159    ) -> Result<RenderedPgQuery, RenderError> {
160        let field = plan
161            .meta
162            .field(field_id)
163            .ok_or(RenderError::UnknownField(field_id.0))?;
164        if plan.meta.is_grouped() {
165            return Err(RenderError::InvalidCondition(
166                "facets require a row-phase entity".to_owned(),
167            ));
168        }
169        let mut state = RenderState::default();
170        let mut sql = sql_buffer(plan);
171        if plan.distinct {
172            write!(
173                sql,
174                "SELECT sq_facet AS \"sq_key\", count(*)::bigint AS \"sq_count\", sum(count(*)) OVER ()::bigint AS \"sq_total\" FROM (SELECT DISTINCT {}, {} AS sq_facet FROM {}",
175                render_projection(plan, false)?,
176                field.column,
177                plan.meta.from
178            )
179            .expect("writing to String cannot fail");
180            let filter_start = sql.len();
181            render_filters_excluding(
182                plan,
183                &mut state,
184                &mut sql,
185                false,
186                exclude_self_filter.then_some(field_id),
187            )?;
188            if !include_null {
189                if sql.len() > filter_start {
190                    sql.push_str(" AND ");
191                } else {
192                    sql.push_str(" WHERE ");
193                }
194                write!(sql, "{} IS NOT NULL", field.column).expect("writing to String cannot fail");
195            }
196            sql.push_str(
197                ") sq_facet_source GROUP BY sq_facet ORDER BY count(*) DESC, sq_facet ASC LIMIT ",
198            );
199            // The LIMIT placeholder is appended through the normal bind state
200            // so the statement remains fully parameterized.
201            let placeholder = state.bind(ScalarValue::I64(i64::from(limit)));
202            sql.push_str(&placeholder);
203            return Ok(state.finish(sql, plan.shape_fingerprint_seed()));
204        }
205        write!(
206            sql,
207            "SELECT {} AS \"sq_key\", count(*)::bigint AS \"sq_count\", sum(count(*)) OVER ()::bigint AS \"sq_total\" FROM {}",
208            field.column, plan.meta.from
209        )
210        .expect("writing to String cannot fail");
211        let filter_start = sql.len();
212        render_filters_excluding(
213            plan,
214            &mut state,
215            &mut sql,
216            false,
217            exclude_self_filter.then_some(field_id),
218        )?;
219        if !include_null {
220            if sql.len() == filter_start {
221                sql.push_str(" WHERE ");
222            } else {
223                sql.push_str(" AND ");
224            }
225            write!(sql, "{} IS NOT NULL", field.column).expect("writing to String cannot fail");
226        }
227        write!(
228            sql,
229            " GROUP BY {} ORDER BY count(*) DESC, {} ASC LIMIT {}",
230            field.column,
231            field.column,
232            state.bind(ScalarValue::I64(i64::from(limit)))
233        )
234        .expect("writing to String cannot fail");
235        Ok(state.finish(sql, plan.shape_fingerprint_seed()))
236    }
237
238    pub fn render_list<E: SearchEntity>(
239        &self,
240        plan: &QueryPlan<E>,
241    ) -> Result<RenderedPgQuery, RenderError> {
242        if !matches!(
243            plan.target,
244            QueryTarget::TypedPage
245                | QueryTarget::MapPage
246                | QueryTarget::TypedList
247                | QueryTarget::First
248                | QueryTarget::StaticPage
249        ) {
250            return Err(RenderError::WrongTarget);
251        }
252        let mut state = RenderState::default();
253        let mut sql = sql_buffer(plan);
254        sql.push_str("SELECT ");
255        if plan.distinct {
256            sql.push_str("DISTINCT ");
257        }
258        sql.push_str(&render_projection(plan, true)?);
259        write!(sql, " FROM {}", plan.meta.from).expect("writing to String cannot fail");
260        render_filters(plan, &mut state, &mut sql, true)?;
261        render_group_by(plan, &mut sql);
262        render_having(plan, &mut state, &mut sql)?;
263        render_order(plan, &mut state, &mut sql)?;
264        let limit = state.bind(ScalarValue::I64(i64::from(plan.paging.size)));
265        write!(sql, " LIMIT {limit}").expect("writing to String cannot fail");
266        if !matches!(plan.paging.mode, PagingMode::Cursor(_)) {
267            let offset = state.bind(ScalarValue::I64(
268                i64::try_from(plan.paging.offset)
269                    .map_err(|_| RenderError::InvalidCondition("offset overflow".to_owned()))?,
270            ));
271            write!(sql, " OFFSET {offset}").expect("writing to String cannot fail");
272        }
273        Ok(state.finish(sql, plan.shape_fingerprint_seed()))
274    }
275
276    pub fn render_count<E: SearchEntity>(
277        &self,
278        plan: &QueryPlan<E>,
279    ) -> Result<RenderedPgQuery, RenderError> {
280        self.render_cluster(plan, None)
281    }
282
283    pub fn render_cluster<E: SearchEntity>(
284        &self,
285        plan: &QueryPlan<E>,
286        options: Option<&SearchOptions<E>>,
287    ) -> Result<RenderedPgQuery, RenderError> {
288        let requests = options.map_or(&[][..], SearchOptions::requests);
289        for request in requests {
290            plan.validate_field_access([request.field])
291                .map_err(|error| RenderError::InvalidCondition(error.to_string()))?;
292            if !supports_aggregate(request.function, request.input_type) {
293                return Err(RenderError::InvalidCondition(
294                    "summary field type does not support the PostgreSQL aggregate".to_owned(),
295                ));
296            }
297        }
298        let needs_subquery = plan.meta.is_grouped() || plan.distinct;
299        let mut state = RenderState::default();
300        let mut sql = sql_buffer(plan);
301        if needs_subquery {
302            sql.push_str("SELECT count(*) AS \"sq_count\"");
303            for (index, request) in requests.iter().enumerate() {
304                write!(
305                    sql,
306                    ", {} AS \"sq_{index}\"",
307                    render_aggregate(request.function, &format!("sq_{index}"))
308                )
309                .expect("writing to String cannot fail");
310            }
311            sql.push_str(" FROM (SELECT ");
312            if plan.distinct {
313                sql.push_str("DISTINCT ");
314                sql.push_str(&render_projection(plan, false)?);
315                for (index, request) in requests.iter().enumerate() {
316                    if !plan.projection.contains(&request.field) {
317                        return Err(RenderError::InvalidCondition(
318                            "DISTINCT summaries require the summarized field in the final projection"
319                                .to_owned(),
320                        ));
321                    }
322                    let field = plan
323                        .meta
324                        .field(request.field)
325                        .ok_or(RenderError::UnknownField(request.field.0))?;
326                    write!(sql, ", {} AS \"sq_{index}\"", field.column)
327                        .expect("writing to String cannot fail");
328                }
329            } else if requests.is_empty() {
330                sql.push('1');
331            } else {
332                for (index, request) in requests.iter().enumerate() {
333                    if index > 0 {
334                        sql.push_str(", ");
335                    }
336                    let field = plan
337                        .meta
338                        .field(request.field)
339                        .ok_or(RenderError::UnknownField(request.field.0))?;
340                    write!(sql, "{} AS \"sq_{index}\"", field.column)
341                        .expect("writing to String cannot fail");
342                }
343            }
344            write!(sql, " FROM {}", plan.meta.from).expect("writing to String cannot fail");
345            render_filters(plan, &mut state, &mut sql, false)?;
346            render_group_by(plan, &mut sql);
347            render_having(plan, &mut state, &mut sql)?;
348            sql.push_str(") sq_cluster");
349        } else {
350            sql.push_str("SELECT count(*) AS \"sq_count\"");
351            for (index, request) in requests.iter().enumerate() {
352                let field = plan
353                    .meta
354                    .field(request.field)
355                    .ok_or(RenderError::UnknownField(request.field.0))?;
356                write!(
357                    sql,
358                    ", {} AS \"sq_{index}\"",
359                    render_aggregate(request.function, field.column)
360                )
361                .expect("writing to String cannot fail");
362            }
363            write!(sql, " FROM {}", plan.meta.from).expect("writing to String cannot fail");
364            render_filters(plan, &mut state, &mut sql, false)?;
365        }
366        Ok(state.finish(sql, plan.shape_fingerprint_seed()))
367    }
368
369    pub fn render_sum<E: SearchEntity>(
370        &self,
371        plan: &QueryPlan<E>,
372        field_id: FieldId,
373    ) -> Result<RenderedPgQuery, RenderError> {
374        let field = plan
375            .meta
376            .field(field_id)
377            .ok_or(RenderError::UnknownField(field_id.0))?;
378        plan.validate_field_access([field_id])
379            .map_err(|error| RenderError::InvalidCondition(error.to_string()))?;
380        let mut state = RenderState::default();
381        let mut sql = sql_buffer(plan);
382        if plan.distinct {
383            let alias = field
384                .result_alias
385                .ok_or_else(|| RenderError::MissingAlias(field.public_name.to_owned()))?;
386            if !plan.projection.contains(&field_id) {
387                return Err(RenderError::InvalidCondition(
388                    "SUM over DISTINCT entities requires the field in the final projection"
389                        .to_owned(),
390                ));
391            }
392            write!(
393                sql,
394                "SELECT sum({}) AS \"sq_sum\" FROM (SELECT DISTINCT {} FROM {}",
395                quote_ident(alias),
396                render_projection(plan, true)?,
397                plan.meta.from
398            )
399            .expect("writing to String cannot fail");
400            render_filters(plan, &mut state, &mut sql, false)?;
401            render_group_by(plan, &mut sql);
402            render_having(plan, &mut state, &mut sql)?;
403            sql.push_str(") sq_sum_source");
404        } else if plan.meta.is_grouped() {
405            sql.push_str("SELECT sum(\"sq_value\") AS \"sq_sum\" FROM (SELECT ");
406            write!(
407                sql,
408                "{} AS \"sq_value\" FROM {}",
409                field.column, plan.meta.from
410            )
411            .expect("writing to String cannot fail");
412            render_filters(plan, &mut state, &mut sql, false)?;
413            render_group_by(plan, &mut sql);
414            render_having(plan, &mut state, &mut sql)?;
415            sql.push_str(") sq_sum_source");
416        } else {
417            write!(
418                sql,
419                "SELECT sum({}) AS \"sq_sum\" FROM {}",
420                field.column, plan.meta.from
421            )
422            .expect("writing to String cannot fail");
423            render_filters(plan, &mut state, &mut sql, false)?;
424        }
425        Ok(state.finish(sql, plan.shape_fingerprint_seed()))
426    }
427}
428
429mod helpers;
430
431use helpers::{
432    RenderState, is_time_value_type, quote_ident, render_filters, render_filters_excluding,
433    render_group_by, render_having, render_order, render_projection, sql_buffer,
434};
435
436pub(crate) fn hash_shape(schema_fingerprint: u64, sql: &str) -> u64 {
437    let mut hash = StableHasher64::new(schema_fingerprint ^ FNV1A_64_OFFSET_BASIS);
438    hash.write(sql);
439    hash.finish()
440}
441#[cfg(test)]
442mod tests;
sift_query_sqlx/renderer/mod.rs

sift_query_sqlx/renderer/
mod.rs
