icydb_core/db/session/query/
dynamic.rs

1//! Module: db::session::query::dynamic
2//! Responsibility: lower and execute public dynamic reads against accepted schema.
3//! Does not own: query planning, accepted schema construction, or row projection.
4//! Boundary: entity-name requests converge on the shared structural read lane.
5
6use crate::{
7    db::{
8        DbSession, DynamicQuery, DynamicTypedEntityBinding, GroupedQueryOutput, MissingRowPolicy,
9        QueryError, RowProjectionOutput,
10        query::{
11            admission::QueryAdmissionPolicy,
12            intent::{IntentError, StructuralQuery},
13        },
14        session::AcceptedSchemaCatalogContext,
15    },
16    traits::CanisterKind,
17};
18use icydb_diagnostic_code::QueryReadAdmissionCode;
19
20#[derive(Clone, Copy)]
21enum DynamicReadLane {
22    Public,
23    Trusted,
24}
25
26impl<C: CanisterKind> DbSession<C> {
27    fn structural_query_from_dynamic_request(
28        request: &DynamicQuery,
29        catalog: &AcceptedSchemaCatalogContext,
30    ) -> Result<StructuralQuery, QueryError> {
31        let schema = catalog.accepted_schema_info();
32        let mut query = StructuralQuery::new(MissingRowPolicy::Ignore);
33        if let Some(filter) = request.filter_expr() {
34            query = query.filter_for_schema(&schema, filter.clone());
35        }
36        for order in request.order_terms() {
37            query = query.order_term(order.clone());
38        }
39        if !request.selected_fields().is_empty() {
40            query = query.select_fields(request.selected_fields().iter().cloned());
41        }
42        if let Some(limit) = request.row_limit() {
43            query = query.limit(limit);
44        }
45        for field in request.group_fields() {
46            query = query.group_by_with_schema(field, &schema)?;
47        }
48        for aggregate in request.aggregates() {
49            query = query.aggregate(aggregate.clone());
50        }
51        if let Some((max_groups, max_group_bytes)) = request.grouped_execution_limits() {
52            if max_groups == 0 || max_group_bytes == 0 {
53                return Err(QueryReadAdmissionCode::GroupedQueryRequiresLimits.into());
54            }
55            query = query.grouped_limits(u64::from(max_groups), u64::from(max_group_bytes));
56        }
57
58        Ok(query)
59    }
60
61    fn execute_dynamic_query_against_catalog(
62        &self,
63        request: &DynamicQuery,
64        lane: DynamicReadLane,
65        catalog: AcceptedSchemaCatalogContext,
66    ) -> Result<RowProjectionOutput, QueryError> {
67        if request.has_grouping()
68            || request.grouped_execution_limits().is_some()
69            || request.continuation_cursor().is_some()
70        {
71            return Err(QueryError::intent(
72                IntentError::scalar_terminal_requires_scalar_query(),
73            ));
74        }
75        let query = Self::structural_query_from_dynamic_request(request, &catalog)?;
76
77        let authority = catalog
78            .accepted_entity_authority()
79            .map_err(QueryError::execute)?;
80        let public_admission = match lane {
81            DynamicReadLane::Public => Some(QueryAdmissionPolicy::default_bounded_read()),
82            DynamicReadLane::Trusted => None,
83        };
84        let (payload, _) = self.execute_structural_projection_from_query(
85            query,
86            authority,
87            catalog.snapshot(),
88            public_admission.as_ref(),
89        )?;
90        let (columns, _fixed_scales, rows, row_count) = payload.into_output_components()?;
91
92        Ok(RowProjectionOutput {
93            entity: catalog.snapshot().entity_name().to_string(),
94            columns,
95            rows,
96            row_count,
97        })
98    }
99
100    fn execute_dynamic_grouped_query_against_catalog(
101        &self,
102        request: &DynamicQuery,
103        lane: DynamicReadLane,
104        catalog: AcceptedSchemaCatalogContext,
105    ) -> Result<GroupedQueryOutput, QueryError> {
106        if !request.has_grouping() {
107            return Err(QueryError::intent(
108                IntentError::grouped_terminal_requires_grouped_query(),
109            ));
110        }
111        if request.grouped_execution_limits().is_none() {
112            return Err(QueryReadAdmissionCode::GroupedQueryRequiresLimits.into());
113        }
114        if !request.selected_fields().is_empty() {
115            return Err(QueryError::intent(
116                IntentError::grouped_output_defined_by_group_and_aggregates(),
117            ));
118        }
119        let query = Self::structural_query_from_dynamic_request(request, &catalog)?;
120        let public_admission = match lane {
121            DynamicReadLane::Public => Some(QueryAdmissionPolicy::default_bounded_read()),
122            DynamicReadLane::Trusted => None,
123        };
124
125        self.execute_structural_grouped_from_query(
126            &query,
127            &catalog,
128            public_admission.as_ref(),
129            request.continuation_cursor(),
130        )
131    }
132
133    fn execute_dynamic_query(
134        &self,
135        request: &DynamicQuery,
136        lane: DynamicReadLane,
137    ) -> Result<RowProjectionOutput, QueryError> {
138        let catalog = self
139            .accepted_schema_catalog_context_for_entity_name(Some(request.entity()))
140            .map_err(QueryError::execute)?;
141        self.execute_dynamic_query_against_catalog(request, lane, catalog)
142    }
143
144    /// Execute one ordinary entity-name-driven dynamic read.
145    ///
146    /// The selected accepted plan must satisfy the built-in bounded public-read
147    /// policy before any row is executed.
148    pub fn execute_public_dynamic_query(
149        &self,
150        request: &DynamicQuery,
151    ) -> Result<RowProjectionOutput, QueryError> {
152        self.execute_dynamic_query(request, DynamicReadLane::Public)
153    }
154
155    /// Execute one ordinary entity-name-driven bounded grouped read.
156    pub fn execute_public_dynamic_grouped_query(
157        &self,
158        request: &DynamicQuery,
159    ) -> Result<GroupedQueryOutput, QueryError> {
160        let catalog = self
161            .accepted_schema_catalog_context_for_entity_name(Some(request.entity()))
162            .map_err(QueryError::execute)?;
163        self.execute_dynamic_grouped_query_against_catalog(
164            request,
165            DynamicReadLane::Public,
166            catalog,
167        )
168    }
169
170    /// Execute one typed read through the binding's immutable accepted entity
171    /// identity. `None` means the opaque binding is stale.
172    #[doc(hidden)]
173    pub fn execute_public_dynamic_query_for_typed_binding(
174        &self,
175        binding: &DynamicTypedEntityBinding,
176        request: &DynamicQuery,
177    ) -> Result<Option<RowProjectionOutput>, QueryError> {
178        let Some(catalog) = self
179            .current_typed_entity_binding_catalog(binding)
180            .map_err(QueryError::execute)?
181        else {
182            return Ok(None);
183        };
184        self.execute_dynamic_query_against_catalog(request, DynamicReadLane::Public, catalog)
185            .map(Some)
186    }
187
188    /// Execute one grouped typed read through the binding's immutable accepted
189    /// entity identity. `None` means the opaque binding is stale.
190    #[doc(hidden)]
191    pub fn execute_public_dynamic_grouped_query_for_typed_binding(
192        &self,
193        binding: &DynamicTypedEntityBinding,
194        request: &DynamicQuery,
195    ) -> Result<Option<GroupedQueryOutput>, QueryError> {
196        let Some(catalog) = self
197            .current_typed_entity_binding_catalog(binding)
198            .map_err(QueryError::execute)?
199        else {
200            return Ok(None);
201        };
202        self.execute_dynamic_grouped_query_against_catalog(
203            request,
204            DynamicReadLane::Public,
205            catalog,
206        )
207        .map(Some)
208    }
209
210    /// Execute one trusted entity-name-driven dynamic read.
211    ///
212    /// This uses accepted schema, planner, executor, and projection authority
213    /// only. Like trusted SQL reads, callers own authorization and resource
214    /// policy.
215    pub fn execute_trusted_dynamic_query(
216        &self,
217        request: &DynamicQuery,
218    ) -> Result<RowProjectionOutput, QueryError> {
219        self.execute_dynamic_query(request, DynamicReadLane::Trusted)
220    }
221
222    /// Execute one trusted entity-name-driven grouped read.
223    ///
224    /// This bypasses ordinary public admission but retains accepted-schema
225    /// planning, explicit grouped limits, cursor validation, and execution.
226    pub fn execute_trusted_dynamic_grouped_query(
227        &self,
228        request: &DynamicQuery,
229    ) -> Result<GroupedQueryOutput, QueryError> {
230        let catalog = self
231            .accepted_schema_catalog_context_for_entity_name(Some(request.entity()))
232            .map_err(QueryError::execute)?;
233        self.execute_dynamic_grouped_query_against_catalog(
234            request,
235            DynamicReadLane::Trusted,
236            catalog,
237        )
238    }
239}
icydb_core/db/session/query/dynamic.rs

icydb_core/db/session/query/
dynamic.rs
