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        .with_distinct_output_offset(usize::try_from(prior_rows_emitted).unwrap_or(usize::MAX));
429        let projection_request = if exact_initial_exhaustion {
430            projection_request
431        } else {
432            projection_request
433                .with_continuation(continuation_context)
434                .with_cursor_emission(page_output_limit)
435        };
436        let page = execute_structural_projection_page(&self.db, projection_request)
437            .map_err(QueryError::execute)?;
438        let row_count = page.rows.row_count();
439        let rows = page
440            .rows
441            .into_value_rows()
442            .into_iter()
443            .map(|row| {
444                row.iter()
445                    .map(|value| {
446                        crate::db::schema::output_value_from_runtime(
447                            value_catalog.enum_catalog(),
448                            value,
449                        )
450                        .map_err(|_| QueryError::invariant())
451                    })
452                    .collect::<Result<Vec<_>, _>>()
453            })
454            .collect::<Result<Vec<_>, _>>()?;
455        let rows_emitted = prior_rows_emitted.saturating_add(u64::from(row_count));
456        let total_limit_reached = request
457            .row_limit()
458            .is_some_and(|limit| rows_emitted >= u64::from(limit));
459        let continuation = if page.has_more && !total_limit_reached {
460            let logical_boundary = page
461                .last_emitted_logical
462                .ok_or_else(Self::scalar_page_cursor_error)?;
463            let cursor_contract = if let Some(contract) = cursor_contract {
464                contract
465            } else {
466                let prepared_plan = deferred_cursor_plan
467                    .as_ref()
468                    .ok_or_else(Self::scalar_page_cursor_error)?;
469                Self::scalar_cursor_contract(
470                    request,
471                    &catalog,
472                    envelope,
473                    prepared_plan,
474                    mode,
475                    exhaustive_proof.as_ref(),
476                )?
477            };
478            let token = ScalarPageToken::new(
479                mode,
480                cursor_contract.signature,
481                cursor_contract.authority,
482                cursor_contract.window,
483                cursor_contract.order_terms,
484                ScalarPageTokenProgress::new(Some(logical_boundary), None, None, 0, rows_emitted),
485            );
486            Some(encode_cursor(
487                token
488                    .encode(&cursor_authentication_key().map_err(QueryError::execute)?)
489                    .map_err(|error| {
490                        QueryError::from_cursor_plan_error(CursorPlanError::from_token_wire_error(
491                            error,
492                        ))
493                    })?
494                    .as_slice(),
495            ))
496        } else {
497            None
498        };
499
500        if let Some(proof) = exhaustive_proof.as_ref() {
501            self.verify_read_set_revision_proof(proof)?;
502        }
503
504        Ok((
505            LiveQueryPageOutput {
506                entity: catalog.snapshot().entity_name().to_string(),
507                columns,
508                rows,
509                row_count,
510                continuation,
511                work: ScalarPageWork {
512                    envelope_identity: envelope.identity(),
513                    entries_visited: page.scanned_keys as u64,
514                    result_rows: row_count,
515                },
516            },
517            exhaustive_proof,
518        ))
519    }
520
521    /// Execute one revision-tolerant bounded scalar page.
522    pub fn execute_public_live_page(
523        &self,
524        request: &DynamicQuery,
525        continuation: Option<&str>,
526    ) -> Result<LiveQueryPageOutput, QueryError> {
527        let catalog = self
528            .accepted_schema_catalog_context_for_entity_name(Some(request.entity()))
529            .map_err(QueryError::execute)?;
530        self.execute_scalar_page_against_catalog(
531            request,
532            continuation,
533            DynamicReadLane::Public,
534            catalog,
535            ScalarPageMode::Live,
536            None,
537        )
538        .map(|(page, _)| page)
539        .map_err(Self::live_page_error)
540    }
541
542    /// Execute one live page through a typed binding's immutable accepted
543    /// entity identity. `None` means the opaque binding is stale.
544    #[doc(hidden)]
545    pub fn execute_public_live_page_for_typed_binding(
546        &self,
547        binding: &DynamicTypedEntityBinding,
548        request: &DynamicQuery,
549        continuation: Option<&str>,
550    ) -> Result<Option<LiveQueryPageOutput>, QueryError> {
551        let Some(catalog) = self
552            .current_typed_entity_binding_catalog(binding)
553            .map_err(QueryError::execute)?
554        else {
555            return Ok(None);
556        };
557        self.execute_scalar_page_against_catalog(
558            request,
559            continuation,
560            DynamicReadLane::Public,
561            catalog,
562            ScalarPageMode::Live,
563            None,
564        )
565        .map(|(page, _)| Some(page))
566        .map_err(Self::live_page_error)
567    }
568
569    /// Execute one ordinary entity-name-driven bounded grouped read.
570    pub fn execute_public_dynamic_grouped_query(
571        &self,
572        request: &DynamicQuery,
573    ) -> Result<GroupedQueryOutput, QueryError> {
574        let catalog = self
575            .accepted_schema_catalog_context_for_entity_name(Some(request.entity()))
576            .map_err(QueryError::execute)?;
577        self.execute_dynamic_grouped_query_against_catalog(
578            request,
579            DynamicReadLane::Public,
580            catalog,
581        )
582    }
583
584    /// Execute one grouped typed read through the binding's immutable accepted
585    /// entity identity. `None` means the opaque binding is stale.
586    #[doc(hidden)]
587    pub fn execute_public_dynamic_grouped_query_for_typed_binding(
588        &self,
589        binding: &DynamicTypedEntityBinding,
590        request: &DynamicQuery,
591    ) -> Result<Option<GroupedQueryOutput>, QueryError> {
592        let Some(catalog) = self
593            .current_typed_entity_binding_catalog(binding)
594            .map_err(QueryError::execute)?
595        else {
596            return Ok(None);
597        };
598        self.execute_dynamic_grouped_query_against_catalog(
599            request,
600            DynamicReadLane::Public,
601            catalog,
602        )
603        .map(Some)
604    }
605
606    /// Execute one trusted entity-name-driven grouped read.
607    ///
608    /// This bypasses ordinary public admission but retains accepted-schema
609    /// planning, explicit grouped limits, cursor validation, and execution.
610    pub fn execute_trusted_dynamic_grouped_query(
611        &self,
612        request: &DynamicQuery,
613    ) -> Result<GroupedQueryOutput, QueryError> {
614        let catalog = self
615            .accepted_schema_catalog_context_for_entity_name(Some(request.entity()))
616            .map_err(QueryError::execute)?;
617        self.execute_dynamic_grouped_query_against_catalog(
618            request,
619            DynamicReadLane::Trusted,
620            catalog,
621        )
622    }
623
624    /// Execute one trusted revision-tolerant bounded dynamic page.
625    ///
626    /// Trusted execution bypasses public admission but retains the same
627    /// physical and aggregate request budgets as every other read lane.
628    pub fn execute_trusted_live_page(
629        &self,
630        request: &DynamicQuery,
631        continuation: Option<&str>,
632    ) -> Result<LiveQueryPageOutput, QueryError> {
633        let catalog = self
634            .accepted_schema_catalog_context_for_entity_name(Some(request.entity()))
635            .map_err(QueryError::execute)?;
636        self.execute_scalar_page_against_catalog(
637            request,
638            continuation,
639            DynamicReadLane::Trusted,
640            catalog,
641            ScalarPageMode::Live,
642            None,
643        )
644        .map(|(page, _)| page)
645        .map_err(Self::live_page_error)
646    }
647
648    /// Execute one revision-strict bounded dynamic page.
649    pub fn execute_public_exhaustive_page(
650        &self,
651        request: &DynamicQuery,
652        continuation: Option<&str>,
653        proof: Option<&ReadSetRevisionProof>,
654    ) -> Result<ExhaustiveQueryPageOutput, ExhaustiveReadError> {
655        let catalog =
656            self.accepted_schema_catalog_context_for_entity_name(Some(request.entity()))?;
657        let (page, proof) = self.execute_scalar_page_against_catalog(
658            request,
659            continuation,
660            DynamicReadLane::Public,
661            catalog,
662            ScalarPageMode::Exhaustive,
663            proof,
664        )?;
665        let proof = proof.ok_or(ReadSetRevisionError::NonCanonical)?;
666        Ok(ExhaustiveQueryPageOutput::from_live_page(page, proof))
667    }
668
669    /// Execute one exhaustive page through a typed binding's accepted identity.
670    #[doc(hidden)]
671    pub fn execute_public_exhaustive_page_for_typed_binding(
672        &self,
673        binding: &DynamicTypedEntityBinding,
674        request: &DynamicQuery,
675        continuation: Option<&str>,
676        proof: Option<&ReadSetRevisionProof>,
677    ) -> Result<Option<ExhaustiveQueryPageOutput>, ExhaustiveReadError> {
678        let Some(catalog) = self.current_typed_entity_binding_catalog(binding)? else {
679            return Ok(None);
680        };
681        let (page, proof) = self.execute_scalar_page_against_catalog(
682            request,
683            continuation,
684            DynamicReadLane::Public,
685            catalog,
686            ScalarPageMode::Exhaustive,
687            proof,
688        )?;
689        let proof = proof.ok_or(ReadSetRevisionError::NonCanonical)?;
690        Ok(Some(ExhaustiveQueryPageOutput::from_live_page(page, proof)))
691    }
692
693    /// Execute one trusted revision-strict bounded dynamic page.
694    pub fn execute_trusted_exhaustive_page(
695        &self,
696        request: &DynamicQuery,
697        continuation: Option<&str>,
698        proof: Option<&ReadSetRevisionProof>,
699    ) -> Result<ExhaustiveQueryPageOutput, ExhaustiveReadError> {
700        let catalog =
701            self.accepted_schema_catalog_context_for_entity_name(Some(request.entity()))?;
702        let (page, proof) = self.execute_scalar_page_against_catalog(
703            request,
704            continuation,
705            DynamicReadLane::Trusted,
706            catalog,
707            ScalarPageMode::Exhaustive,
708            proof,
709        )?;
710        let proof = proof.ok_or(ReadSetRevisionError::NonCanonical)?;
711        Ok(ExhaustiveQueryPageOutput::from_live_page(page, proof))
712    }
713
714    fn live_page_error(error: ExhaustiveReadError) -> QueryError {
715        match error {
716            ExhaustiveReadError::Query(error) => error,
717            ExhaustiveReadError::Revision(_) => QueryError::invariant(),
718        }
719    }
720}
icydb_core/db/session/query/dynamic.rs

icydb_core/db/session/query/
dynamic.rs
