1use 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 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 #[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 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 #[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 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 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 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 #[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 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}