cratestack_sqlx/query/read/
aggregate_column.rs

1//! `aggregate.sum/avg/min/max(col)` — single-column scalar aggregates
2//! with filter + read policy. Caller picks the decode type at the
3//! call site since PG's `SUM(int)` returns i64, `AVG(int)` returns
4//! f64/Decimal, etc.
5
6use cratestack_core::{CratestackContext, CratestackError};
7use cratestack_sql::{IntoColumnName, ReadSource};
8
9use crate::query::support::{ReadPolicyKind, push_scoped_conditions};
10use crate::{FilterExpr, SqlxRuntime, sqlx};
11
12use super::aggregate::AggregateOp;
13
14#[derive(Clone)]
15pub struct AggregateColumn<'a, M: 'static, PK: 'static> {
16    runtime: &'a SqlxRuntime,
17    descriptor: &'static dyn ReadSource<M, PK>,
18    op: AggregateOp,
19    column: &'static str,
20    filters: Vec<FilterExpr>,
21}
22
23impl<'a, M: 'static, PK: 'static> AggregateColumn<'a, M, PK> {
24    pub(super) fn new<C: IntoColumnName>(
25        runtime: &'a SqlxRuntime,
26        descriptor: &'static dyn ReadSource<M, PK>,
27        op: AggregateOp,
28        column: C,
29    ) -> Self {
30        Self {
31            runtime,
32            descriptor,
33            op,
34            column: column.into_column_name(),
35            filters: Vec::new(),
36        }
37    }
38
39    pub fn where_(mut self, filter: crate::Filter) -> Self {
40        self.filters.push(FilterExpr::from(filter));
41        self
42    }
43
44    pub fn where_expr(mut self, filter: FilterExpr) -> Self {
45        self.filters.push(filter);
46        self
47    }
48
49    pub fn where_any(mut self, filters: impl IntoIterator<Item = FilterExpr>) -> Self {
50        self.filters.push(FilterExpr::any(filters));
51        self
52    }
53
54    pub fn where_optional<F>(mut self, filter: Option<F>) -> Self
55    where
56        F: Into<FilterExpr>,
57    {
58        if let Some(filter) = filter {
59            self.filters.push(filter.into());
60        }
61        self
62    }
63
64    fn build_query(&self, ctx: &CratestackContext) -> sqlx::QueryBuilder<sqlx::Postgres> {
65        let mut query = sqlx::QueryBuilder::<sqlx::Postgres>::new("SELECT ");
66        query
67            .push(self.op.function_name())
68            .push("(")
69            .push(self.column)
70            .push(") FROM ")
71            .push(self.descriptor.table_name());
72        push_scoped_conditions(
73            &mut query,
74            self.descriptor,
75            &self.filters,
76            None::<(&'static str, i64)>,
77            ctx,
78            ReadPolicyKind::List,
79        );
80        query
81    }
82
83    pub async fn run<T>(self, ctx: &CratestackContext) -> Result<Option<T>, CratestackError>
84    where
85        T: Send + Unpin + for<'r> sqlx::Decode<'r, sqlx::Postgres> + sqlx::Type<sqlx::Postgres>,
86    {
87        // Inside an `@isolation` procedure: run on its transaction
88        // (docs/design/procedure-isolation.md §4).
89        if let Some(bound) = self.runtime.bound() {
90            return crate::bound::in_bound_savepoint!(bound, |sp| self.run_in_tx(sp, ctx));
91        }
92        let mut query = self.build_query(ctx);
93        let value: (Option<T>,) = query
94            .build_query_as::<(Option<T>,)>()
95            .fetch_one(self.runtime.pool())
96            .await
97            .map_err(crate::error::cratestack_error_from_sqlx)?;
98        Ok(value.0)
99    }
100
101    pub async fn run_in_tx<'tx, T>(
102        self,
103        tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
104        ctx: &CratestackContext,
105    ) -> Result<Option<T>, CratestackError>
106    where
107        T: Send + Unpin + for<'r> sqlx::Decode<'r, sqlx::Postgres> + sqlx::Type<sqlx::Postgres>,
108    {
109        let mut query = self.build_query(ctx);
110        let value: (Option<T>,) = query
111            .build_query_as::<(Option<T>,)>()
112            .fetch_one(&mut **tx)
113            .await
114            .map_err(crate::error::cratestack_error_from_sqlx)?;
115        Ok(value.0)
116    }
117}
cratestack_sqlx/query/read/aggregate_column.rs

cratestack_sqlx/query/read/
aggregate_column.rs
