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, ExhaustiveQueryPageOutput,
9        ExhaustiveReadError, GroupedQueryOutput, LiveQueryPageOutput, MissingRowPolicy, QueryError,
10        ReadSetRevisionError, ReadSetRevisionProof, ScalarPageWork,
11        codec::{finalize_hash_sha256, new_hash_sha256_prefixed},
12        commit::{cursor_authentication_key, database_incarnation_id},
13        cursor::{
14            CursorPlanError, ScalarOrderTermContract, ScalarPageMode, ScalarPageToken,
15            ScalarPageTokenAuthority, ScalarPageTokenProgress, ScalarPageTokenWindow,
16            decode_optional_cursor_token, encode_cursor,
17        },
18        executor::{
19            CoveringProjectionMetricsRecorder, PageWorkEnvelope,
20            ProjectionMaterializationMetricsRecorder, ScalarContinuationContext,
21            StructuralProjectionRequest, execute_structural_projection_page,
22        },
23        query::{
24            admission::{QueryAdmissionPolicy, QueryAdmissionSummary},
25            expr::{FilterExpr, OrderTerm as FluentOrderTerm},
26            intent::{IntentError, StructuralQuery},
27        },
28        session::AcceptedSchemaCatalogContext,
29    },
30    traits::CanisterKind,
31};
32use icydb_diagnostic_code::{
33    DiagnosticDecodeReason, DiagnosticExecutionBudgetResource, DiagnosticExecutionLane,
34    QueryReadAdmissionCode,
35};
36use sha2::Digest;
37
38#[cfg(not(test))]
39const SCALAR_PAGE_OUTPUT_ROWS: usize = 1_024;
40#[cfg(test)]
41const SCALAR_PAGE_OUTPUT_ROWS: usize = 2;
42
43#[derive(Clone, Copy)]
44enum DynamicReadLane {
45    Public,
46    Trusted,
47}
48
49struct ScalarCursorContract {
50    signature: crate::db::cursor::ContinuationSignature,
51    authority: ScalarPageTokenAuthority,
52    window: ScalarPageTokenWindow,
53    order_terms: Vec<ScalarOrderTermContract>,
54}
55
56impl<C: CanisterKind> DbSession<C> {
57    fn may_select_exact_single_primary_key(
58        request: &DynamicQuery,
59        catalog: &AcceptedSchemaCatalogContext,
60    ) -> bool {
61        let [primary_key] = catalog.accepted_schema_info().primary_key_names() else {
62            return false;
63        };
64        matches!(
65            request.filter_expr(),
66            Some(FilterExpr::Eq { field, .. } | FilterExpr::In { field, .. })
67                if field.eq_ignore_ascii_case(primary_key)
68        )
69    }
70
71    fn exact_primary_key_candidate_bound(
72        prepared_plan: &crate::db::executor::SharedPreparedExecutionPlan,
73    ) -> Option<usize> {
74        let access = &prepared_plan.logical_plan().access;
75        if access.as_by_key_path().is_some() {
76            return Some(1);
77        }
78
79        access.as_by_keys_path().map(<[crate::value::Value]>::len)
80    }
81
82    fn structural_query_from_dynamic_request(
83        request: &DynamicQuery,
84        catalog: &AcceptedSchemaCatalogContext,
85    ) -> Result<StructuralQuery, QueryError> {
86        Self::structural_query_from_dynamic_request_with_page_limit(request, catalog, None, false)
87    }
88
89    fn structural_query_from_dynamic_request_with_page_limit(
90        request: &DynamicQuery,
91        catalog: &AcceptedSchemaCatalogContext,
92        page_limit: Option<u32>,
93        require_total_order: bool,
94    ) -> Result<StructuralQuery, QueryError> {
95        let schema = catalog.accepted_schema_info();
96        let mut query = StructuralQuery::new(MissingRowPolicy::Ignore);
97        if let Some(filter) = request.filter_expr() {
98            query = query.filter_for_schema(schema, filter.clone());
99        }
100        for order in request.order_terms() {
101            query = query.order_term(order.clone());
102        }
103        if require_total_order && request.order_terms().is_empty() {
104            for primary_key in schema.primary_key_names() {
105                query = query.order_term(FluentOrderTerm::asc(primary_key.clone()));
106            }
107        }
108        if !request.selected_fields().is_empty() {
109            query = query.select_fields(request.selected_fields().iter().cloned());
110        }
111        if let Some(limit) = page_limit.or_else(|| request.row_limit()) {
112            query = query.limit(limit);
113        }
114        for field in request.group_fields() {
115            query = query.group_by_with_schema(field, schema)?;
116        }
117        for aggregate in request.aggregates() {
118            query = query.aggregate(aggregate.clone());
119        }
120        if let Some((max_groups, max_group_bytes)) = request.grouped_execution_limits() {
121            if max_groups == 0 || max_group_bytes == 0 {
122                return Err(QueryReadAdmissionCode::GroupedQueryRequiresLimits.into());
123            }
124            query = query.grouped_limits(u64::from(max_groups), u64::from(max_group_bytes));
125        }
126
127        Ok(query)
128    }
129
130    fn scalar_page_cursor_error() -> QueryError {
131        QueryError::from_cursor_plan_error(CursorPlanError::invalid_continuation_cursor_payload(
132            DiagnosticDecodeReason::CursorTokenDecode,
133        ))
134    }
135
136    fn scalar_cursor_contract(
137        request: &DynamicQuery,
138        catalog: &AcceptedSchemaCatalogContext,
139        envelope: PageWorkEnvelope,
140        prepared_plan: &crate::db::executor::SharedPreparedExecutionPlan,
141        mode: ScalarPageMode,
142        proof: Option<&ReadSetRevisionProof>,
143    ) -> Result<ScalarCursorContract, QueryError> {
144        let mut signature = prepared_plan
145            .continuation_signature_for_runtime()
146            .map_err(QueryError::execute)?;
147        match (mode, proof) {
148            (ScalarPageMode::Live, None) => {}
149            (ScalarPageMode::Exhaustive, Some(proof)) => {
150                let mut hasher = new_hash_sha256_prefixed(b"icydb.exhaustive-cursor-proof.v1");
151                hasher.update(signature.into_bytes());
152                hasher.update(proof.signature_bytes());
153                signature = crate::db::cursor::ContinuationSignature::from_bytes(
154                    finalize_hash_sha256(hasher),
155                );
156            }
157            _ => return Err(Self::scalar_page_cursor_error()),
158        }
159        let root_identity = catalog.runtime_root_identity();
160        let (root_fingerprint_method, root_fingerprint) = root_identity.fingerprint();
161        let authority = ScalarPageTokenAuthority::new(
162            database_incarnation_id()
163                .map_err(QueryError::execute)?
164                .to_bytes(),
165            root_identity.accepted_root_revision().get(),
166            root_fingerprint_method,
167            root_fingerprint,
168            catalog.fingerprint(),
169            prepared_plan.authority_ref().entity_tag(),
170        );
171        let window = ScalarPageTokenWindow::new(0, request.row_limit(), envelope.identity());
172        let canonical_order = prepared_plan
173            .logical_plan()
174            .scalar_plan()
175            .order
176            .as_ref()
177            .ok_or_else(Self::scalar_page_cursor_error)?;
178        let order_terms = canonical_order
179            .fields
180            .iter()
181            .map(|term| ScalarOrderTermContract::new(term.rendered_label(), term.direction()))
182            .collect::<Vec<_>>();
183
184        Ok(ScalarCursorContract {
185            signature,
186            authority,
187            window,
188            order_terms,
189        })
190    }
191
192    fn validate_scalar_page_token(
193        token: &ScalarPageToken,
194        mode: ScalarPageMode,
195        signature: crate::db::cursor::ContinuationSignature,
196        authority: ScalarPageTokenAuthority,
197        window: ScalarPageTokenWindow,
198        order_terms: &[ScalarOrderTermContract],
199        entity: &str,
200    ) -> Result<(), QueryError> {
201        if token.signature() != signature {
202            return Err(QueryError::from_cursor_plan_error(
203                CursorPlanError::continuation_cursor_signature_mismatch(
204                    entity,
205                    &signature,
206                    &token.signature(),
207                ),
208            ));
209        }
210        if token.mode() != mode
211            || token.authority() != authority
212            || token.window() != window
213            || token.order_terms() != order_terms
214        {
215            return Err(Self::scalar_page_cursor_error());
216        }
217
218        Ok(())
219    }
220
221    fn execute_dynamic_grouped_query_against_catalog(
222        &self,
223        request: &DynamicQuery,
224        lane: DynamicReadLane,
225        catalog: AcceptedSchemaCatalogContext,
226    ) -> Result<GroupedQueryOutput, QueryError> {
227        if !request.has_grouping() {
228            return Err(QueryError::intent(
229                IntentError::grouped_terminal_requires_grouped_query(),
230            ));
231        }
232        if request.grouped_execution_limits().is_none() {
233            return Err(QueryReadAdmissionCode::GroupedQueryRequiresLimits.into());
234        }
235        if !request.selected_fields().is_empty() {
236            return Err(QueryError::intent(
237                IntentError::grouped_output_defined_by_group_and_aggregates(),
238            ));
239        }
240        let query = Self::structural_query_from_dynamic_request(request, &catalog)?;
241        let public_admission = match lane {
242            DynamicReadLane::Public => Some(QueryAdmissionPolicy::default_bounded_read()),
243            DynamicReadLane::Trusted => None,
244        };
245
246        self.execute_structural_grouped_from_query(
247            &query,
248            &catalog,
249            public_admission.as_ref(),
250            request.continuation_cursor(),
251        )
252    }
253
254    #[expect(
255        clippy::too_many_lines,
256        reason = "live-page orchestration keeps planning, cursor validation, execution, and response proof in one auditable boundary"
257    )]
258    fn execute_scalar_page_against_catalog(
259        &self,
260        request: &DynamicQuery,
261        continuation: Option<&str>,
262        lane: DynamicReadLane,
263        catalog: AcceptedSchemaCatalogContext,
264        mode: ScalarPageMode,
265        supplied_proof: Option<&ReadSetRevisionProof>,
266    ) -> Result<(LiveQueryPageOutput, Option<ReadSetRevisionProof>), ExhaustiveReadError> {
267        if request.has_grouping()
268            || request.grouped_execution_limits().is_some()
269            || request.continuation_cursor().is_some()
270        {
271            return Err(
272                QueryError::intent(IntentError::scalar_terminal_requires_scalar_query()).into(),
273            );
274        }
275
276        let exhaustive_proof = match mode {
277            ScalarPageMode::Live => None,
278            ScalarPageMode::Exhaustive => {
279                if continuation.is_some() && supplied_proof.is_none() {
280                    return Err(ReadSetRevisionError::ResumeProofRequired.into());
281                }
282                let proof = supplied_proof.cloned().map_or_else(
283                    || self.capture_entity_read_set_revision_proof(catalog.identity().store_path()),
284                    Ok,
285                )?;
286                Self::ensure_read_set_contains_store(&proof, catalog.identity().store_path())?;
287                self.verify_read_set_revision_proof(&proof)?;
288                Some(proof)
289            }
290        };
291
292        let envelope = match lane {
293            DynamicReadLane::Public => PageWorkEnvelope::public_scalar(),
294            DynamicReadLane::Trusted => PageWorkEnvelope::default_scalar(),
295        };
296        let page_row_limit = envelope
297            .limit(DiagnosticExecutionBudgetResource::ResultRows)
298            .and_then(|limit| usize::try_from(limit).ok())
299            .unwrap_or(SCALAR_PAGE_OUTPUT_ROWS)
300            .min(SCALAR_PAGE_OUTPUT_ROWS);
301        let decoded_token = decode_optional_cursor_token(continuation)
302            .map_err(QueryError::from_cursor_plan_error)?
303            .map(|bytes| {
304                ScalarPageToken::decode(
305                    bytes.as_slice(),
306                    &cursor_authentication_key().map_err(QueryError::execute)?,
307                )
308                .map_err(|error| {
309                    QueryError::from_cursor_plan_error(CursorPlanError::from_token_wire_error(
310                        error,
311                    ))
312                })
313            })
314            .transpose()?;
315        let prior_rows_emitted = decoded_token
316            .as_ref()
317            .map_or(0, |token| token.progress().rows_emitted());
318        let remaining_limit = request
319            .row_limit()
320            .map(|limit| u64::from(limit).saturating_sub(prior_rows_emitted));
321        let page_output_limit = remaining_limit
322            .unwrap_or(page_row_limit as u64)
323            .min(page_row_limit as u64);
324        let page_output_limit = usize::try_from(page_output_limit).unwrap_or(page_row_limit);
325        let execution_limit = u32::try_from(page_row_limit).unwrap_or(u32::MAX);
326        let execution_lane = match lane {
327            DynamicReadLane::Public => DiagnosticExecutionLane::PublicRead,
328            DynamicReadLane::Trusted => DiagnosticExecutionLane::TrustedRead,
329        };
330        let exact_candidate =
331            decoded_token.is_none() && Self::may_select_exact_single_primary_key(request, &catalog);
332        let initial_plan = if exact_candidate {
333            let query = Self::structural_query_from_dynamic_request(request, &catalog)?;
334            Some(
335                self.structural_projection_prepared_plan_for_accepted_authority(
336                    &query,
337                    catalog.accepted_entity_authority(),
338                    catalog.snapshot(),
339                    execution_lane,
340                )?,
341            )
342        } else {
343            None
344        };
345        let initial_is_exact_exhaustion =
346            initial_plan.as_ref().is_some_and(|(prepared_plan, _, _)| {
347                Self::exact_primary_key_candidate_bound(prepared_plan)
348                    .is_some_and(|bound| bound <= page_output_limit)
349            });
350        let (prepared_plan, projection, _) = if initial_is_exact_exhaustion {
351            initial_plan.ok_or_else(Self::scalar_page_cursor_error)?
352        } else {
353            let query = Self::structural_query_from_dynamic_request_with_page_limit(
354                request,
355                &catalog,
356                Some(execution_limit),
357                true,
358            )?;
359            self.structural_projection_prepared_plan_for_accepted_authority(
360                &query,
361                catalog.accepted_entity_authority(),
362                catalog.snapshot(),
363                execution_lane,
364            )?
365        };
366        if matches!(lane, DynamicReadLane::Public) {
367            let policy = QueryAdmissionPolicy::default_bounded_read();
368            let summary = policy.evaluate(QueryAdmissionSummary::from_plan(
369                policy.lane(),
370                prepared_plan.logical_plan(),
371            ));
372            if let Some(rejection) = summary.rejection() {
373                return Err(QueryError::from(rejection.code()).into());
374            }
375        }
376
377        let exact_initial_exhaustion = initial_is_exact_exhaustion;
378        let cursor_contract = decoded_token
379            .as_ref()
380            .map(|token| {
381                let contract = Self::scalar_cursor_contract(
382                    request,
383                    &catalog,
384                    envelope,
385                    &prepared_plan,
386                    mode,
387                    exhaustive_proof.as_ref(),
388                )?;
389                Self::validate_scalar_page_token(
390                    token,
391                    mode,
392                    contract.signature,
393                    contract.authority,
394                    contract.window,
395                    contract.order_terms.as_slice(),
396                    request.entity(),
397                )?;
398                Ok::<_, QueryError>(contract)
399            })
400            .transpose()?;
401        let deferred_cursor_plan =
402            (!exact_initial_exhaustion && decoded_token.is_none()).then(|| prepared_plan.clone());
403        let continuation_context = decoded_token
404            .as_ref()
405            .and_then(|token| token.progress().last_emitted_logical().cloned())
406            .map_or_else(
407                ScalarContinuationContext::initial,
408                ScalarContinuationContext::resumed,
409            );
410        if decoded_token.is_some() && !continuation_context.has_cursor_boundary() {
411            return Err(Self::scalar_page_cursor_error().into());
412        }
413
414        let value_catalog = prepared_plan
415            .authority_ref()
416            .accepted_schema_info()
417            .map(crate::db::schema::SchemaInfo::value_catalog_handle)
418            .cloned()
419            .ok_or_else(QueryError::invariant)?;
420        let (columns, _fixed_scales) = projection.into_components();
421        let projection_request = StructuralProjectionRequest::new(
422            self.debug,
423            prepared_plan,
424            CoveringProjectionMetricsRecorder::none(),
425            ProjectionMaterializationMetricsRecorder::none(),
426            execution_lane,
427        );
428        let projection_request = if exact_initial_exhaustion {
429            projection_request
430        } else {
431            projection_request
432                .with_continuation(continuation_context)
433                .with_cursor_emission(page_output_limit)
434        };
435        let page = execute_structural_projection_page(&self.db, projection_request)
436            .map_err(QueryError::execute)?;
437        let row_count = page.rows.row_count();
438        let rows = page
439            .rows
440            .into_value_rows()
441            .into_iter()
442            .map(|row| {
443                row.iter()
444                    .map(|value| {
445                        crate::db::schema::output_value_from_runtime(
446                            value_catalog.enum_catalog(),
447                            value,
448                        )
449                        .map_err(|_| QueryError::invariant())
450                    })
451                    .collect::<Result<Vec<_>, _>>()
452            })
453            .collect::<Result<Vec<_>, _>>()?;
454        let rows_emitted = prior_rows_emitted.saturating_add(u64::from(row_count));
455        let total_limit_reached = request
456            .row_limit()
457            .is_some_and(|limit| rows_emitted >= u64::from(limit));
458        let continuation = if page.has_more && !total_limit_reached {
459            let logical_boundary = page
460                .last_emitted_logical
461                .ok_or_else(Self::scalar_page_cursor_error)?;
462            let cursor_contract = if let Some(contract) = cursor_contract {
463                contract
464            } else {
465                let prepared_plan = deferred_cursor_plan
466                    .as_ref()
467                    .ok_or_else(Self::scalar_page_cursor_error)?;
468                Self::scalar_cursor_contract(
469                    request,
470                    &catalog,
471                    envelope,
472                    prepared_plan,
473                    mode,
474                    exhaustive_proof.as_ref(),
475                )?
476            };
477            let token = ScalarPageToken::new(
478                mode,
479                cursor_contract.signature,
480                cursor_contract.authority,
481                cursor_contract.window,
482                cursor_contract.order_terms,
483                ScalarPageTokenProgress::new(Some(logical_boundary), None, None, 0, rows_emitted),
484            );
485            Some(encode_cursor(
486                token
487                    .encode(&cursor_authentication_key().map_err(QueryError::execute)?)
488                    .map_err(|error| {
489                        QueryError::from_cursor_plan_error(CursorPlanError::from_token_wire_error(
490                            error,
491                        ))
492                    })?
493                    .as_slice(),
494            ))
495        } else {
496            None
497        };
498
499        if let Some(proof) = exhaustive_proof.as_ref() {
500            self.verify_read_set_revision_proof(proof)?;
501        }
502
503        Ok((
504            LiveQueryPageOutput {
505                entity: catalog.snapshot().entity_name().to_string(),
506                columns,
507                rows,
508                row_count,
509                continuation,
510                work: ScalarPageWork {
511                    envelope_identity: envelope.identity(),
512                    entries_visited: page.scanned_keys as u64,
513                    result_rows: row_count,
514                },
515            },
516            exhaustive_proof,
517        ))
518    }
519
520    /// Execute one revision-tolerant bounded scalar page.
521    pub fn execute_public_live_page(
522        &self,
523        request: &DynamicQuery,
524        continuation: Option<&str>,
525    ) -> Result<LiveQueryPageOutput, QueryError> {
526        let catalog = self
527            .accepted_schema_catalog_context_for_entity_name(Some(request.entity()))
528            .map_err(QueryError::execute)?;
529        self.execute_scalar_page_against_catalog(
530            request,
531            continuation,
532            DynamicReadLane::Public,
533            catalog,
534            ScalarPageMode::Live,
535            None,
536        )
537        .map(|(page, _)| page)
538        .map_err(Self::live_page_error)
539    }
540
541    /// Execute one live page through a typed binding's immutable accepted
542    /// entity identity. `None` means the opaque binding is stale.
543    #[doc(hidden)]
544    pub fn execute_public_live_page_for_typed_binding(
545        &self,
546        binding: &DynamicTypedEntityBinding,
547        request: &DynamicQuery,
548        continuation: Option<&str>,
549    ) -> Result<Option<LiveQueryPageOutput>, QueryError> {
550        let Some(catalog) = self
551            .current_typed_entity_binding_catalog(binding)
552            .map_err(QueryError::execute)?
553        else {
554            return Ok(None);
555        };
556        self.execute_scalar_page_against_catalog(
557            request,
558            continuation,
559            DynamicReadLane::Public,
560            catalog,
561            ScalarPageMode::Live,
562            None,
563        )
564        .map(|(page, _)| Some(page))
565        .map_err(Self::live_page_error)
566    }
567
568    /// Execute one ordinary entity-name-driven bounded grouped read.
569    pub fn execute_public_dynamic_grouped_query(
570        &self,
571        request: &DynamicQuery,
572    ) -> Result<GroupedQueryOutput, QueryError> {
573        let catalog = self
574            .accepted_schema_catalog_context_for_entity_name(Some(request.entity()))
575            .map_err(QueryError::execute)?;
576        self.execute_dynamic_grouped_query_against_catalog(
577            request,
578            DynamicReadLane::Public,
579            catalog,
580        )
581    }
582
583    /// Execute one grouped typed read through the binding's immutable accepted
584    /// entity identity. `None` means the opaque binding is stale.
585    #[doc(hidden)]
586    pub fn execute_public_dynamic_grouped_query_for_typed_binding(
587        &self,
588        binding: &DynamicTypedEntityBinding,
589        request: &DynamicQuery,
590    ) -> Result<Option<GroupedQueryOutput>, QueryError> {
591        let Some(catalog) = self
592            .current_typed_entity_binding_catalog(binding)
593            .map_err(QueryError::execute)?
594        else {
595            return Ok(None);
596        };
597        self.execute_dynamic_grouped_query_against_catalog(
598            request,
599            DynamicReadLane::Public,
600            catalog,
601        )
602        .map(Some)
603    }
604
605    /// Execute one trusted entity-name-driven grouped read.
606    ///
607    /// This bypasses ordinary public admission but retains accepted-schema
608    /// planning, explicit grouped limits, cursor validation, and execution.
609    pub fn execute_trusted_dynamic_grouped_query(
610        &self,
611        request: &DynamicQuery,
612    ) -> Result<GroupedQueryOutput, QueryError> {
613        let catalog = self
614            .accepted_schema_catalog_context_for_entity_name(Some(request.entity()))
615            .map_err(QueryError::execute)?;
616        self.execute_dynamic_grouped_query_against_catalog(
617            request,
618            DynamicReadLane::Trusted,
619            catalog,
620        )
621    }
622
623    /// Execute one trusted revision-tolerant bounded dynamic page.
624    ///
625    /// Trusted execution bypasses public admission but retains the same
626    /// physical and aggregate request budgets as every other read lane.
627    pub fn execute_trusted_live_page(
628        &self,
629        request: &DynamicQuery,
630        continuation: Option<&str>,
631    ) -> Result<LiveQueryPageOutput, QueryError> {
632        let catalog = self
633            .accepted_schema_catalog_context_for_entity_name(Some(request.entity()))
634            .map_err(QueryError::execute)?;
635        self.execute_scalar_page_against_catalog(
636            request,
637            continuation,
638            DynamicReadLane::Trusted,
639            catalog,
640            ScalarPageMode::Live,
641            None,
642        )
643        .map(|(page, _)| page)
644        .map_err(Self::live_page_error)
645    }
646
647    /// Execute one revision-strict bounded dynamic page.
648    pub fn execute_public_exhaustive_page(
649        &self,
650        request: &DynamicQuery,
651        continuation: Option<&str>,
652        proof: Option<&ReadSetRevisionProof>,
653    ) -> Result<ExhaustiveQueryPageOutput, ExhaustiveReadError> {
654        let catalog =
655            self.accepted_schema_catalog_context_for_entity_name(Some(request.entity()))?;
656        let (page, proof) = self.execute_scalar_page_against_catalog(
657            request,
658            continuation,
659            DynamicReadLane::Public,
660            catalog,
661            ScalarPageMode::Exhaustive,
662            proof,
663        )?;
664        let proof = proof.ok_or(ReadSetRevisionError::NonCanonical)?;
665        Ok(ExhaustiveQueryPageOutput::from_live_page(page, proof))
666    }
667
668    /// Execute one exhaustive page through a typed binding's accepted identity.
669    #[doc(hidden)]
670    pub fn execute_public_exhaustive_page_for_typed_binding(
671        &self,
672        binding: &DynamicTypedEntityBinding,
673        request: &DynamicQuery,
674        continuation: Option<&str>,
675        proof: Option<&ReadSetRevisionProof>,
676    ) -> Result<Option<ExhaustiveQueryPageOutput>, ExhaustiveReadError> {
677        let Some(catalog) = self.current_typed_entity_binding_catalog(binding)? else {
678            return Ok(None);
679        };
680        let (page, proof) = self.execute_scalar_page_against_catalog(
681            request,
682            continuation,
683            DynamicReadLane::Public,
684            catalog,
685            ScalarPageMode::Exhaustive,
686            proof,
687        )?;
688        let proof = proof.ok_or(ReadSetRevisionError::NonCanonical)?;
689        Ok(Some(ExhaustiveQueryPageOutput::from_live_page(page, proof)))
690    }
691
692    /// Execute one trusted revision-strict bounded dynamic page.
693    pub fn execute_trusted_exhaustive_page(
694        &self,
695        request: &DynamicQuery,
696        continuation: Option<&str>,
697        proof: Option<&ReadSetRevisionProof>,
698    ) -> Result<ExhaustiveQueryPageOutput, ExhaustiveReadError> {
699        let catalog =
700            self.accepted_schema_catalog_context_for_entity_name(Some(request.entity()))?;
701        let (page, proof) = self.execute_scalar_page_against_catalog(
702            request,
703            continuation,
704            DynamicReadLane::Trusted,
705            catalog,
706            ScalarPageMode::Exhaustive,
707            proof,
708        )?;
709        let proof = proof.ok_or(ReadSetRevisionError::NonCanonical)?;
710        Ok(ExhaustiveQueryPageOutput::from_live_page(page, proof))
711    }
712
713    fn live_page_error(error: ExhaustiveReadError) -> QueryError {
714        match error {
715            ExhaustiveReadError::Query(error) => error,
716            ExhaustiveReadError::Revision(_) => QueryError::invariant(),
717        }
718    }
719}
icydb_core/db/session/query/dynamic.rs

icydb_core/db/session/query/
dynamic.rs
