sift_query_sqlx/execution/
aggregate.rs

1use super::*;
2
3impl PgSiftQuery {
4    pub async fn search_list<E>(
5        &self,
6        query: impl QueryInput<E>,
7        scope: &ServerScope<E>,
8    ) -> Result<Vec<E>, PgSearchError>
9    where
10        E: SearchEntity + Send + Unpin + for<'row> FromRow<'row, PgRow>,
11    {
12        self.observe::<E, _>(
13            "search_list",
14            async {
15                let plan =
16                    self.resolve_plan(query, scope, QueryTarget::TypedList, "search_list")?;
17                if plan.force_empty || plan.paging.size == 0 {
18                    return Ok(Vec::new());
19                }
20                let partition = self.extensions.resolve_partition(&plan, "search_list")?;
21                let rendered = prepare(self.render::<E>(
22                    &plan,
23                    "search_list",
24                    "list",
25                    partition.as_deref(),
26                    RenderRequest::List,
27                )?)?;
28                let deadline = self.deadline::<E>();
29                let connection = self.acquire(deadline).await?;
30                let (_, rows) =
31                    fetch_typed::<E>(connection, rendered, deadline, TimeoutStage::ListQuery)
32                        .await?;
33                Ok(rows)
34            },
35            Vec::len,
36        )
37        .await
38    }
39
40    pub async fn search_first<E>(
41        &self,
42        query: impl QueryInput<E>,
43        scope: &ServerScope<E>,
44    ) -> Result<Option<E>, PgSearchError>
45    where
46        E: SearchEntity + Send + Unpin + for<'row> FromRow<'row, PgRow>,
47    {
48        self.observe::<E, _>(
49            "search_first",
50            async {
51                let plan = self.resolve_plan(query, scope, QueryTarget::First, "search_first")?;
52                if plan.force_empty {
53                    return Ok(None);
54                }
55                let partition = self.extensions.resolve_partition(&plan, "search_first")?;
56                let rendered = prepare(self.render::<E>(
57                    &plan,
58                    "search_first",
59                    "list",
60                    partition.as_deref(),
61                    RenderRequest::List,
62                )?)?;
63                let deadline = self.deadline::<E>();
64                let connection = self.acquire(deadline).await?;
65                let (_, mut rows) =
66                    fetch_typed::<E>(connection, rendered, deadline, TimeoutStage::ListQuery)
67                        .await?;
68                Ok(rows.pop())
69            },
70            |row| usize::from(row.is_some()),
71        )
72        .await
73    }
74
75    pub async fn count<E>(
76        &self,
77        query: impl QueryInput<E>,
78        scope: &ServerScope<E>,
79    ) -> Result<i64, PgSearchError>
80    where
81        E: SearchEntity,
82    {
83        self.observe::<E, _>(
84            "count",
85            async {
86                let plan = self.resolve_plan(query, scope, QueryTarget::Count, "count")?;
87                if plan.force_empty {
88                    return Ok(0);
89                }
90                let partition = self.extensions.resolve_partition(&plan, "count")?;
91                let rendered = prepare(self.render::<E>(
92                    &plan,
93                    "count",
94                    "cluster",
95                    partition.as_deref(),
96                    RenderRequest::Cluster(None),
97                )?)?;
98                let deadline = self.deadline::<E>();
99                let connection = self.acquire(deadline).await?;
100                let (_, (count, _)) =
101                    fetch_cluster::<E>(connection, rendered, None, E::META, deadline).await?;
102                Ok(count)
103            },
104            |_| 0,
105        )
106        .await
107    }
108
109    pub async fn sum<E, T, K>(
110        &self,
111        query: impl QueryInput<E>,
112        scope: &ServerScope<E>,
113        field: FieldRef<E, T, K>,
114    ) -> Result<Option<<T as PgSumType>::Output>, PgSearchError>
115    where
116        E: SearchEntity,
117        T: PgSumType,
118        K: SummableKind,
119        <T as PgSumType>::Output: Send + Unpin + for<'row> Decode<'row, Postgres> + Type<Postgres>,
120    {
121        self.observe::<E, _>(
122            "sum",
123            async {
124                let field_id = field.id();
125                let plan = self.resolve_plan(query, scope, QueryTarget::Sum, "sum")?;
126                if plan.force_empty {
127                    return Ok(None);
128                }
129                let partition = self.extensions.resolve_partition(&plan, "sum")?;
130                let rendered = prepare(self.render::<E>(
131                    &plan,
132                    "sum",
133                    "cluster",
134                    partition.as_deref(),
135                    RenderRequest::Sum(field_id),
136                )?)?;
137                let deadline = self.deadline::<E>();
138                let connection = self.acquire(deadline).await?;
139                let (_, value) =
140                    fetch_sum::<<T as PgSumType>::Output>(connection, rendered, deadline).await?;
141                Ok(value)
142            },
143            |_| 0,
144        )
145        .await
146    }
147}
sift_query_sqlx/execution/aggregate.rs

sift_query_sqlx/execution/
aggregate.rs
