sift_query_sqlx/execution/
mod.rs

1//! PostgreSQL execution configuration and query entry points.
2use std::sync::Arc;
3use std::time::Duration;
4
5use indexmap::IndexMap;
6use sift_query_core::{
7    AggregateFn, ConfigError, FieldRef, GroupScope, ParamError, ParamParser, QueryBuilder,
8    QueryInput, QueryPlan, QueryPolicy, QueryTarget, RawParams, RowValue, SearchEntity,
9    SearchOptions, ServerScope, SummableKind, ValueType,
10};
11use sqlx::postgres::{PgPool, PgRow};
12use sqlx::{Decode, FromRow, Postgres, Type};
13use tokio::sync::Semaphore;
14use tokio::time::{Instant, timeout_at};
15
16use crate::cache::RenderRequest;
17use crate::execution::io::{fetch_cluster, fetch_maps, fetch_sum, fetch_typed, prepare};
18use crate::{
19    CountStrategy, ExecutionStrategy, PartitionResolver, PgRenderer, PgSearchError, PgSumType,
20    RenderedPgQuery, ResultFilter, ShapeCacheStats, SqlInterceptor, TimeoutStage,
21};
22
23mod aggregate;
24mod config;
25pub(crate) mod cursor_page;
26pub(crate) mod external;
27pub(crate) mod io;
28mod prepare;
29pub(crate) mod projection;
30mod search;
31pub(crate) mod strategy;
32pub(crate) mod stream;
33pub(crate) mod window;
34
35pub type RowMap = IndexMap<String, RowValue>;
36
37#[derive(Debug, Clone, PartialEq)]
38#[cfg_attr(feature = "serde", derive(serde::Serialize))]
39pub struct Page<T> {
40    pub data: Vec<T>,
41    pub total_count: i64,
42    pub summaries: Vec<SummaryValue>,
43    pub next_cursor: Option<String>,
44}
45
46impl<T> Page<T> {
47    /// Creates an empty page with the same metadata contract as a successful
48    /// zero-row query.
49    #[must_use]
50    pub fn empty() -> Self {
51        Self {
52            data: Vec::new(),
53            total_count: 0,
54            summaries: Vec::new(),
55            next_cursor: None,
56        }
57    }
58}
59
60/// A cursor page fetched without a count or cluster query.
61#[derive(Debug, Clone, PartialEq)]
62#[cfg_attr(feature = "serde", derive(serde::Serialize))]
63pub struct CursorPage<T> {
64    pub data: Vec<T>,
65    pub next_cursor: Option<String>,
66    pub has_more: bool,
67}
68
69impl<T> CursorPage<T> {
70    /// Creates an empty cursor page. An empty page never has a continuation.
71    #[must_use]
72    pub fn empty() -> Self {
73        Self {
74            data: Vec::new(),
75            next_cursor: None,
76            has_more: false,
77        }
78    }
79}
80
81#[derive(Debug, Clone, PartialEq)]
82#[cfg_attr(feature = "serde", derive(serde::Serialize))]
83pub struct SummaryValue {
84    pub field: &'static str,
85    pub function: AggregateFn,
86    pub value_type: ValueType,
87    pub value: Option<RowValue>,
88}
89
90#[derive(Debug, Clone, Copy, PartialEq, Eq)]
91pub struct PgExecutionPolicy {
92    pub timeout: Duration,
93    pub strategy: ExecutionStrategy,
94    pub count_strategy: CountStrategy,
95    pub max_parallel_queries: usize,
96    /// Number of SQL templates retained by the optional shape cache. Zero disables it.
97    pub shape_cache_capacity: usize,
98}
99
100impl Default for PgExecutionPolicy {
101    fn default() -> Self {
102        Self {
103            timeout: Duration::from_secs(5),
104            strategy: ExecutionStrategy::SequentialBestEffort,
105            count_strategy: CountStrategy::Separate,
106            max_parallel_queries: 16,
107            shape_cache_capacity: 0,
108        }
109    }
110}
111
112#[derive(Debug, Clone)]
113pub struct PgSiftQuery {
114    pool: PgPool,
115    pub(crate) policy: QueryPolicy,
116    pub(crate) execution: PgExecutionPolicy,
117    pub(crate) renderer: PgRenderer,
118    pub(crate) parallel_gate: Arc<Semaphore>,
119    pub(crate) scan_gate: Arc<Semaphore>,
120    pub(crate) scan_wait_gate: Arc<Semaphore>,
121    shape_cache: Option<Arc<crate::QueryShapeCache>>,
122    pub(crate) extensions: crate::enrichment::PgExtensions,
123    pub(crate) observers: crate::observe::Observers,
124}
sift_query_sqlx/execution/mod.rs

sift_query_sqlx/execution/
mod.rs
