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.accepted_entity_authority();
78        let public_admission = match lane {
79            DynamicReadLane::Public => Some(QueryAdmissionPolicy::default_bounded_read()),
80            DynamicReadLane::Trusted => None,
81        };
82        let (payload, _) = self.execute_structural_projection_from_query(
83            query,
84            authority,
85            catalog.snapshot(),
86            public_admission.as_ref(),
87        )?;
88        let (columns, _fixed_scales, rows, row_count) = payload.into_output_components()?;
89
90        Ok(RowProjectionOutput {
91            entity: catalog.snapshot().entity_name().to_string(),
92            columns,
93            rows,
94            row_count,
95        })
96    }
97
98    fn execute_dynamic_grouped_query_against_catalog(
99        &self,
100        request: &DynamicQuery,
101        lane: DynamicReadLane,
102        catalog: AcceptedSchemaCatalogContext,
103    ) -> Result<GroupedQueryOutput, QueryError> {
104        if !request.has_grouping() {
105            return Err(QueryError::intent(
106                IntentError::grouped_terminal_requires_grouped_query(),
107            ));
108        }
109        if request.grouped_execution_limits().is_none() {
110            return Err(QueryReadAdmissionCode::GroupedQueryRequiresLimits.into());
111        }
112        if !request.selected_fields().is_empty() {
113            return Err(QueryError::intent(
114                IntentError::grouped_output_defined_by_group_and_aggregates(),
115            ));
116        }
117        let query = Self::structural_query_from_dynamic_request(request, &catalog)?;
118        let public_admission = match lane {
119            DynamicReadLane::Public => Some(QueryAdmissionPolicy::default_bounded_read()),
120            DynamicReadLane::Trusted => None,
121        };
122
123        self.execute_structural_grouped_from_query(
124            &query,
125            &catalog,
126            public_admission.as_ref(),
127            request.continuation_cursor(),
128        )
129    }
130
131    fn execute_dynamic_query(
132        &self,
133        request: &DynamicQuery,
134        lane: DynamicReadLane,
135    ) -> Result<RowProjectionOutput, QueryError> {
136        let catalog = self
137            .accepted_schema_catalog_context_for_entity_name(Some(request.entity()))
138            .map_err(QueryError::execute)?;
139        self.execute_dynamic_query_against_catalog(request, lane, catalog)
140    }
141
142    /// Execute one ordinary entity-name-driven dynamic read.
143    ///
144    /// The selected accepted plan must satisfy the built-in bounded public-read
145    /// policy before any row is executed.
146    pub fn execute_public_dynamic_query(
147        &self,
148        request: &DynamicQuery,
149    ) -> Result<RowProjectionOutput, QueryError> {
150        self.execute_dynamic_query(request, DynamicReadLane::Public)
151    }
152
153    /// Execute one ordinary entity-name-driven bounded grouped read.
154    pub fn execute_public_dynamic_grouped_query(
155        &self,
156        request: &DynamicQuery,
157    ) -> Result<GroupedQueryOutput, QueryError> {
158        let catalog = self
159            .accepted_schema_catalog_context_for_entity_name(Some(request.entity()))
160            .map_err(QueryError::execute)?;
161        self.execute_dynamic_grouped_query_against_catalog(
162            request,
163            DynamicReadLane::Public,
164            catalog,
165        )
166    }
167
168    /// Execute one typed read through the binding's immutable accepted entity
169    /// identity. `None` means the opaque binding is stale.
170    #[doc(hidden)]
171    pub fn execute_public_dynamic_query_for_typed_binding(
172        &self,
173        binding: &DynamicTypedEntityBinding,
174        request: &DynamicQuery,
175    ) -> Result<Option<RowProjectionOutput>, QueryError> {
176        let Some(catalog) = self
177            .current_typed_entity_binding_catalog(binding)
178            .map_err(QueryError::execute)?
179        else {
180            return Ok(None);
181        };
182        self.execute_dynamic_query_against_catalog(request, DynamicReadLane::Public, catalog)
183            .map(Some)
184    }
185
186    /// Execute one grouped typed read through the binding's immutable accepted
187    /// entity identity. `None` means the opaque binding is stale.
188    #[doc(hidden)]
189    pub fn execute_public_dynamic_grouped_query_for_typed_binding(
190        &self,
191        binding: &DynamicTypedEntityBinding,
192        request: &DynamicQuery,
193    ) -> Result<Option<GroupedQueryOutput>, QueryError> {
194        let Some(catalog) = self
195            .current_typed_entity_binding_catalog(binding)
196            .map_err(QueryError::execute)?
197        else {
198            return Ok(None);
199        };
200        self.execute_dynamic_grouped_query_against_catalog(
201            request,
202            DynamicReadLane::Public,
203            catalog,
204        )
205        .map(Some)
206    }
207
208    /// Execute one trusted entity-name-driven dynamic read.
209    ///
210    /// This uses accepted schema, planner, executor, and projection authority
211    /// only. The request root still supplies finite physical and aggregate
212    /// execution policy; callers separately own authorization.
213    pub fn execute_trusted_dynamic_query(
214        &self,
215        request: &DynamicQuery,
216    ) -> Result<RowProjectionOutput, QueryError> {
217        self.execute_dynamic_query(request, DynamicReadLane::Trusted)
218    }
219
220    /// Execute one trusted entity-name-driven grouped read.
221    ///
222    /// This bypasses ordinary public admission but retains accepted-schema
223    /// planning, explicit grouped limits, cursor validation, and execution.
224    pub fn execute_trusted_dynamic_grouped_query(
225        &self,
226        request: &DynamicQuery,
227    ) -> Result<GroupedQueryOutput, QueryError> {
228        let catalog = self
229            .accepted_schema_catalog_context_for_entity_name(Some(request.entity()))
230            .map_err(QueryError::execute)?;
231        self.execute_dynamic_grouped_query_against_catalog(
232            request,
233            DynamicReadLane::Trusted,
234            catalog,
235        )
236    }
237}
icydb_core/db/session/query/dynamic.rs

icydb_core/db/session/query/
dynamic.rs
