Skip to main content

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