Skip to main content

icydb_core/db/query/fluent/load/
terminals.rs

1//! Module: query::fluent::load::terminals
2//! Responsibility: fluent load terminal APIs and terminal-plan explanation entrypoints.
3//! Does not own: planner semantic validation or executor runtime routing decisions.
4//! Boundary: delegates to session planning/execution and returns typed query results.
5//!
6//! Terminal Execution Model
7//!
8//! Fluent terminals are concrete descriptors with a 1:1 mapping to session
9//! execution and explain entrypoints. This module may orchestrate descriptor
10//! construction, non-paged gating, and public output shaping, but it must not
11//! carry terminal-kind enums, transport output enums, or match-based execution
12//! dispatch. Adding a new terminal means adding a new descriptor type and one
13//! direct `TerminalStrategyDriver` implementation for that descriptor.
14
15use std::num::NonZeroU32;
16
17#[cfg(feature = "diagnostics")]
18use crate::db::FluentTerminalExecutionAttribution;
19use crate::{
20    db::{
21        DbSession, PersistedRow, Query,
22        query::{
23            admission::QueryAdmissionPolicy,
24            api::ResponseCardinalityExt,
25            builder::{
26                AvgBySlotTerminal, AvgDistinctBySlotTerminal, CountDistinctBySlotTerminal,
27                CountRowsTerminal, DistinctValuesBySlotTerminal, ExistsRowsTerminal,
28                FirstIdTerminal, FirstValueBySlotTerminal, LastIdTerminal, LastValueBySlotTerminal,
29                MaxIdBySlotTerminal, MaxIdTerminal, MedianIdBySlotTerminal, MinIdBySlotTerminal,
30                MinIdTerminal, MinMaxIdBySlotTerminal, NthIdBySlotTerminal, SumBySlotTerminal,
31                SumDistinctBySlotTerminal, ValueProjectionExpr, ValuesBySlotTerminal,
32                ValuesBySlotWithIdsTerminal,
33            },
34            explain::{ExplainAggregateTerminalPlan, ExplainExecutionNodeDescriptor},
35            fluent::load::{FluentLoadQuery, LoadQueryResult},
36            intent::{IntentError, QueryError},
37            plan::AggregateKind,
38            read_intent::{
39                COMPLETE_SMALL_EXECUTION_LIMIT, COMPLETE_SMALL_MAX_ROWS,
40                PUBLIC_PAGE_MAX_RESPONSE_BYTES,
41            },
42        },
43        response::EntityResponse,
44    },
45    traits::EntityValue,
46    types::{Decimal, Id},
47    value::{OutputValue, Value},
48};
49
50type MinMaxByIds<E> = Option<(Id<E>, Id<E>)>;
51
52///
53/// TerminalStrategyDriver
54///
55/// TerminalStrategyDriver is the fluent terminal wiring adapter between a
56/// query-owned strategy object and the session-owned execution/explain
57/// boundary. Implementations are deliberately thin: they only choose the
58/// matching `DbSession` method for an existing strategy type.
59///
60
61trait TerminalStrategyDriver<E: PersistedRow + EntityValue> {
62    type Output;
63    type ExplainOutput;
64
65    fn execute(
66        self,
67        session: &DbSession<E::Canister>,
68        query: &Query<E>,
69    ) -> Result<Self::Output, QueryError>;
70
71    fn explain(
72        &self,
73        session: &DbSession<E::Canister>,
74        query: &Query<E>,
75    ) -> Result<Self::ExplainOutput, QueryError>;
76}
77
78// Define one aggregate-style terminal driver implementation. The macro keeps
79// terminal descriptors 1:1 with session methods while removing repeated explain
80// wiring that is identical for every aggregate terminal.
81macro_rules! impl_aggregate_terminal_driver {
82    ($terminal:ty, $output:ty, $execute:ident) => {
83        impl<E> TerminalStrategyDriver<E> for $terminal
84        where
85            E: PersistedRow + EntityValue,
86        {
87            type Output = $output;
88            type ExplainOutput = ExplainAggregateTerminalPlan;
89
90            fn execute(
91                self,
92                session: &DbSession<E::Canister>,
93                query: &Query<E>,
94            ) -> Result<Self::Output, QueryError> {
95                session.$execute(query, self)
96            }
97
98            fn explain(
99                &self,
100                session: &DbSession<E::Canister>,
101                query: &Query<E>,
102            ) -> Result<Self::ExplainOutput, QueryError> {
103                session.explain_query_prepared_aggregate_terminal_with_visible_indexes(query, self)
104            }
105        }
106    };
107}
108
109// Define one projection-style terminal driver implementation. Projection
110// terminals share the execution/explain shape but return execution descriptors
111// rather than aggregate-terminal plans.
112macro_rules! impl_projection_terminal_driver {
113    ($terminal:ty, $output:ty, $execute:ident) => {
114        impl<E> TerminalStrategyDriver<E> for $terminal
115        where
116            E: PersistedRow + EntityValue,
117        {
118            type Output = $output;
119            type ExplainOutput = ExplainExecutionNodeDescriptor;
120
121            fn execute(
122                self,
123                session: &DbSession<E::Canister>,
124                query: &Query<E>,
125            ) -> Result<Self::Output, QueryError> {
126                session.$execute(query, self)
127            }
128
129            fn explain(
130                &self,
131                session: &DbSession<E::Canister>,
132                query: &Query<E>,
133            ) -> Result<Self::ExplainOutput, QueryError> {
134                session.explain_query_prepared_projection_terminal_with_visible_indexes(query, self)
135            }
136        }
137    };
138}
139
140impl_aggregate_terminal_driver!(CountRowsTerminal, u32, execute_fluent_count_rows_terminal);
141impl_aggregate_terminal_driver!(
142    ExistsRowsTerminal,
143    bool,
144    execute_fluent_exists_rows_terminal
145);
146impl_aggregate_terminal_driver!(MinIdTerminal, Option<Id<E>>, execute_fluent_min_id_terminal);
147impl_aggregate_terminal_driver!(MaxIdTerminal, Option<Id<E>>, execute_fluent_max_id_terminal);
148impl_aggregate_terminal_driver!(
149    MinIdBySlotTerminal,
150    Option<Id<E>>,
151    execute_fluent_min_id_by_slot
152);
153impl_aggregate_terminal_driver!(
154    MaxIdBySlotTerminal,
155    Option<Id<E>>,
156    execute_fluent_max_id_by_slot
157);
158impl_aggregate_terminal_driver!(
159    SumBySlotTerminal,
160    Option<Decimal>,
161    execute_fluent_sum_by_slot
162);
163impl_aggregate_terminal_driver!(
164    SumDistinctBySlotTerminal,
165    Option<Decimal>,
166    execute_fluent_sum_distinct_by_slot
167);
168impl_aggregate_terminal_driver!(
169    AvgBySlotTerminal,
170    Option<Decimal>,
171    execute_fluent_avg_by_slot
172);
173impl_aggregate_terminal_driver!(
174    AvgDistinctBySlotTerminal,
175    Option<Decimal>,
176    execute_fluent_avg_distinct_by_slot
177);
178impl_aggregate_terminal_driver!(
179    FirstIdTerminal,
180    Option<Id<E>>,
181    execute_fluent_first_id_terminal
182);
183impl_aggregate_terminal_driver!(
184    LastIdTerminal,
185    Option<Id<E>>,
186    execute_fluent_last_id_terminal
187);
188impl_aggregate_terminal_driver!(
189    NthIdBySlotTerminal,
190    Option<Id<E>>,
191    execute_fluent_nth_id_by_slot
192);
193impl_aggregate_terminal_driver!(
194    MedianIdBySlotTerminal,
195    Option<Id<E>>,
196    execute_fluent_median_id_by_slot
197);
198impl_aggregate_terminal_driver!(
199    MinMaxIdBySlotTerminal,
200    MinMaxByIds<E>,
201    execute_fluent_min_max_id_by_slot
202);
203
204impl_projection_terminal_driver!(
205    ValuesBySlotTerminal,
206    Vec<Value>,
207    execute_fluent_values_by_slot
208);
209impl_projection_terminal_driver!(
210    DistinctValuesBySlotTerminal,
211    Vec<Value>,
212    execute_fluent_distinct_values_by_slot
213);
214impl_projection_terminal_driver!(
215    CountDistinctBySlotTerminal,
216    u32,
217    execute_fluent_count_distinct_by_slot
218);
219impl_projection_terminal_driver!(
220    ValuesBySlotWithIdsTerminal,
221    Vec<(Id<E>, Value)>,
222    execute_fluent_values_by_with_ids_slot
223);
224impl_projection_terminal_driver!(
225    FirstValueBySlotTerminal,
226    Option<Value>,
227    execute_fluent_first_value_by_slot
228);
229impl_projection_terminal_driver!(
230    LastValueBySlotTerminal,
231    Option<Value>,
232    execute_fluent_last_value_by_slot
233);
234
235// Convert one runtime projection value into the public output boundary type.
236fn output(value: Value) -> OutputValue {
237    OutputValue::from(value)
238}
239
240// Convert one ordered runtime projection vector into the public output form.
241fn output_values(values: Vec<Value>) -> Vec<OutputValue> {
242    values.into_iter().map(output).collect()
243}
244
245// Convert one ordered runtime `(id, value)` projection vector into the public output form.
246fn output_values_with_ids<E: PersistedRow>(
247    values: Vec<(Id<E>, Value)>,
248) -> Vec<(Id<E>, OutputValue)> {
249    values
250        .into_iter()
251        .map(|(id, value)| (id, output(value)))
252        .collect()
253}
254
255fn non_zero_u32(value: u32) -> Result<NonZeroU32, QueryError> {
256    NonZeroU32::new(value).ok_or_else(QueryError::invariant)
257}
258
259impl<E> FluentLoadQuery<'_, E>
260where
261    E: PersistedRow,
262{
263    // ------------------------------------------------------------------
264    // Execution (single semantic boundary)
265    // ------------------------------------------------------------------
266
267    /// Execute this query using the session's policy settings.
268    pub fn execute(&self) -> Result<LoadQueryResult<E>, QueryError>
269    where
270        E: EntityValue,
271    {
272        self.with_admitted_non_paged(DbSession::execute_query_result)
273    }
274
275    /// Execute this query without the default bounded read-admission gate.
276    ///
277    /// This is for trusted maintenance/admin code that has its own caller
278    /// authorization and resource policy. Application-facing reads should use
279    /// `execute`.
280    pub fn execute_trusted(&self) -> Result<LoadQueryResult<E>, QueryError>
281    where
282        E: EntityValue,
283    {
284        self.with_non_paged(DbSession::execute_query_result)
285    }
286
287    /// Execute this query through the scalar rows-only session boundary.
288    pub fn execute_rows(&self) -> Result<EntityResponse<E>, QueryError>
289    where
290        E: EntityValue,
291    {
292        self.with_admitted_non_paged(DbSession::execute_scalar_query_rows)
293    }
294
295    /// Execute scalar rows without the default bounded read-admission gate.
296    ///
297    /// This is for trusted maintenance/admin code that has its own caller
298    /// authorization and resource policy. Application-facing reads should use
299    /// `execute_rows`.
300    pub fn execute_rows_trusted(&self) -> Result<EntityResponse<E>, QueryError>
301    where
302        E: EntityValue,
303    {
304        self.with_non_paged(DbSession::execute_scalar_query_rows)
305    }
306
307    fn with_admitted_non_paged<T>(
308        &self,
309        map: impl FnOnce(&DbSession<E::Canister>, &Query<E>) -> Result<T, QueryError>,
310    ) -> Result<T, QueryError>
311    where
312        E: EntityValue,
313    {
314        self.ensure_non_paged_mode_ready()?;
315        self.ensure_default_read_admission()?;
316        map(self.session, self.query())
317    }
318
319    // Run one terminal operation through the canonical non-paged fluent policy
320    // gate so execution and explain helpers cannot drift on readiness checks.
321    fn with_non_paged<T>(
322        &self,
323        map: impl FnOnce(&DbSession<E::Canister>, &Query<E>) -> Result<T, QueryError>,
324    ) -> Result<T, QueryError>
325    where
326        E: EntityValue,
327    {
328        self.ensure_non_paged_mode_ready()?;
329        map(self.session, self.query())
330    }
331
332    // Resolve the structural execution descriptor for this fluent load query
333    // through the session-owned visible-index explain path once.
334    fn explain_execution_descriptor(&self) -> Result<ExplainExecutionNodeDescriptor, QueryError>
335    where
336        E: EntityValue,
337    {
338        self.with_non_paged(DbSession::explain_query_execution_with_visible_indexes)
339    }
340
341    // Render one descriptor-derived execution surface so text/json explain
342    // terminals do not each forward the same session explain call ad hoc.
343    fn render_execution_descriptor(
344        &self,
345        render: impl FnOnce(ExplainExecutionNodeDescriptor) -> String,
346    ) -> Result<String, QueryError>
347    where
348        E: EntityValue,
349    {
350        let descriptor = self.explain_execution_descriptor()?;
351
352        Ok(render(descriptor))
353    }
354
355    // Execute one prepared terminal descriptor through the canonical
356    // non-paged fluent policy gate.
357    fn execute_terminal<S>(&self, strategy: S) -> Result<S::Output, QueryError>
358    where
359        E: EntityValue,
360        S: TerminalStrategyDriver<E>,
361    {
362        self.with_admitted_non_paged(|session, query| strategy.execute(session, query))
363    }
364
365    // Explain one prepared terminal strategy through the same non-paged fluent
366    // policy gate used by execution.
367    fn explain_terminal<S>(&self, strategy: &S) -> Result<S::ExplainOutput, QueryError>
368    where
369        E: EntityValue,
370        S: TerminalStrategyDriver<E>,
371    {
372        self.with_non_paged(|session, query| strategy.explain(session, query))
373    }
374
375    fn ensure_exists_intent_owns_limit(&self) -> Result<(), QueryError> {
376        self.ensure_semantic_terminal_owns_limit(IntentError::raw_limit_before_exists_terminal())
377    }
378
379    fn ensure_count_exact_intent_owns_limit(&self) -> Result<(), QueryError> {
380        self.ensure_semantic_terminal_owns_limit(
381            IntentError::raw_limit_before_count_exact_terminal(),
382        )
383    }
384
385    fn ensure_sum_exact_intent_owns_limit(&self) -> Result<(), QueryError> {
386        self.ensure_semantic_terminal_owns_limit(IntentError::raw_limit_before_sum_exact_terminal())
387    }
388
389    fn ensure_collect_complete_intent_owns_limit(&self) -> Result<(), QueryError> {
390        self.ensure_semantic_terminal_owns_limit(
391            IntentError::raw_limit_before_collect_complete_terminal(),
392        )
393    }
394
395    // ------------------------------------------------------------------
396    // Execution terminals — semantic only
397    // ------------------------------------------------------------------
398
399    /// Execute and return whether the result set is empty.
400    pub fn is_empty(&self) -> Result<bool, QueryError>
401    where
402        E: EntityValue,
403    {
404        self.not_exists()
405    }
406
407    /// Execute and return whether no matching row exists.
408    pub fn not_exists(&self) -> Result<bool, QueryError>
409    where
410        E: EntityValue,
411    {
412        Ok(!self.exists()?)
413    }
414
415    /// Execute and return whether at least one matching row exists.
416    pub fn exists(&self) -> Result<bool, QueryError>
417    where
418        E: EntityValue,
419    {
420        self.ensure_exists_intent_owns_limit()?;
421
422        self.execute_terminal(ExistsRowsTerminal::new())
423    }
424
425    /// Execute and return whether at least one matching row exists with
426    /// terminal attribution.
427    #[cfg(feature = "diagnostics")]
428    #[doc(hidden)]
429    pub fn exists_with_attribution(
430        &self,
431    ) -> Result<(bool, FluentTerminalExecutionAttribution), QueryError>
432    where
433        E: EntityValue,
434    {
435        self.ensure_exists_intent_owns_limit()?;
436
437        self.with_admitted_non_paged(|session, query| {
438            session.execute_fluent_exists_rows_terminal_with_attribution(
439                query,
440                ExistsRowsTerminal::new(),
441            )
442        })
443    }
444
445    /// Explain scalar `exists()` routing without executing the terminal.
446    pub fn explain_exists(&self) -> Result<ExplainAggregateTerminalPlan, QueryError>
447    where
448        E: EntityValue,
449    {
450        self.ensure_exists_intent_owns_limit()?;
451
452        self.explain_terminal(&ExistsRowsTerminal::new())
453    }
454
455    /// Execute and return all matching rows if the complete result fits in
456    /// the default public-read small-set cap.
457    ///
458    /// This semantic terminal rejects a prior raw `limit(...)`. It owns an
459    /// internal lookahead limit so it can distinguish a complete small set
460    /// from a silently truncated row window.
461    pub fn collect_complete(&self) -> Result<Vec<E>, QueryError>
462    where
463        E: EntityValue,
464    {
465        self.ensure_collect_complete_intent_owns_limit()?;
466        self.ensure_non_paged_mode_ready()?;
467
468        let query = self.query().with_load_limit(COMPLETE_SMALL_EXECUTION_LIMIT);
469        let policy = QueryAdmissionPolicy::public_read(
470            non_zero_u32(COMPLETE_SMALL_EXECUTION_LIMIT)?,
471            non_zero_u32(PUBLIC_PAGE_MAX_RESPONSE_BYTES)?,
472        );
473
474        self.session
475            .ensure_query_read_admission_policy(&query, &policy)?;
476
477        let response = self.session.execute_scalar_query_rows(&query)?;
478        if response.count() > COMPLETE_SMALL_MAX_ROWS {
479            return Err(QueryError::intent(
480                IntentError::complete_read_too_many_rows(),
481            ));
482        }
483
484        Ok(response.entities())
485    }
486
487    /// Explain scalar `not_exists()` routing without executing the terminal.
488    ///
489    /// This remains an `exists()` execution plan with negated boolean semantics.
490    pub fn explain_not_exists(&self) -> Result<ExplainAggregateTerminalPlan, QueryError>
491    where
492        E: EntityValue,
493    {
494        self.explain_exists()
495    }
496
497    /// Explain scalar load execution shape without executing the query.
498    pub fn explain_execution(&self) -> Result<ExplainExecutionNodeDescriptor, QueryError>
499    where
500        E: EntityValue,
501    {
502        self.explain_execution_descriptor()
503    }
504
505    /// Explain scalar load execution shape as deterministic text.
506    pub fn explain_execution_text(&self) -> Result<String, QueryError>
507    where
508        E: EntityValue,
509    {
510        self.render_execution_descriptor(|descriptor| descriptor.render_text_tree())
511    }
512
513    /// Explain scalar load execution shape as canonical JSON.
514    pub fn explain_execution_json(&self) -> Result<String, QueryError>
515    where
516        E: EntityValue,
517    {
518        self.with_non_paged(DbSession::explain_query_execution_json_with_visible_indexes)
519    }
520
521    /// Explain scalar load execution shape as verbose text with diagnostics.
522    pub fn explain_execution_verbose(&self) -> Result<String, QueryError>
523    where
524        E: EntityValue,
525    {
526        self.with_non_paged(DbSession::explain_query_execution_verbose_with_visible_indexes)
527    }
528
529    /// Execute and return the number of matching rows.
530    pub fn count(&self) -> Result<u32, QueryError>
531    where
532        E: EntityValue,
533    {
534        self.execute_terminal(CountRowsTerminal::new())
535    }
536
537    /// Execute and return the exact number of matching rows.
538    ///
539    /// Unlike `count()`, this semantic aggregate rejects a prior raw
540    /// `limit(...)` so exact counts cannot accidentally mean "count the first
541    /// N rows."
542    pub fn count_exact(&self) -> Result<u32, QueryError>
543    where
544        E: EntityValue,
545    {
546        self.ensure_count_exact_intent_owns_limit()?;
547
548        self.execute_terminal(CountRowsTerminal::new())
549    }
550
551    /// Execute and return the number of matching rows with terminal attribution.
552    #[cfg(feature = "diagnostics")]
553    #[doc(hidden)]
554    pub fn count_with_attribution(
555        &self,
556    ) -> Result<(u32, FluentTerminalExecutionAttribution), QueryError>
557    where
558        E: EntityValue,
559    {
560        self.with_admitted_non_paged(|session, query| {
561            session.execute_fluent_count_rows_terminal_with_attribution(
562                query,
563                CountRowsTerminal::new(),
564            )
565        })
566    }
567
568    /// Execute and return the exact number of matching rows with terminal attribution.
569    #[cfg(feature = "diagnostics")]
570    #[doc(hidden)]
571    pub fn count_exact_with_attribution(
572        &self,
573    ) -> Result<(u32, FluentTerminalExecutionAttribution), QueryError>
574    where
575        E: EntityValue,
576    {
577        self.ensure_count_exact_intent_owns_limit()?;
578
579        self.with_admitted_non_paged(|session, query| {
580            session.execute_fluent_count_rows_terminal_with_attribution(
581                query,
582                CountRowsTerminal::new(),
583            )
584        })
585    }
586
587    /// Execute and return the total persisted payload bytes for the effective
588    /// result window.
589    pub fn bytes(&self) -> Result<u64, QueryError>
590    where
591        E: EntityValue,
592    {
593        self.with_admitted_non_paged(DbSession::execute_fluent_bytes)
594    }
595
596    /// Execute and return the total serialized bytes for `field` over the
597    /// effective result window.
598    pub fn bytes_by(&self, field: impl AsRef<str>) -> Result<u64, QueryError>
599    where
600        E: EntityValue,
601    {
602        let target_slot = self.resolve_non_paged_slot(field)?;
603
604        self.with_admitted_non_paged(|session, query| {
605            session.execute_fluent_bytes_by_slot(query, target_slot)
606        })
607    }
608
609    /// Explain `bytes_by(field)` routing without executing the terminal.
610    pub fn explain_bytes_by(
611        &self,
612        field: impl AsRef<str>,
613    ) -> Result<ExplainExecutionNodeDescriptor, QueryError>
614    where
615        E: EntityValue,
616    {
617        let target_slot = self.resolve_non_paged_slot(field)?;
618
619        self.with_non_paged(|session, query| {
620            session.explain_query_bytes_by_with_visible_indexes(query, target_slot.field())
621        })
622    }
623
624    /// Execute and return the smallest matching identifier, if any.
625    pub fn min(&self) -> Result<Option<Id<E>>, QueryError>
626    where
627        E: EntityValue,
628    {
629        self.execute_terminal(MinIdTerminal::new())
630    }
631
632    /// Explain scalar `min()` routing without executing the terminal.
633    pub fn explain_min(&self) -> Result<ExplainAggregateTerminalPlan, QueryError>
634    where
635        E: EntityValue,
636    {
637        self.explain_terminal(&MinIdTerminal::new())
638    }
639
640    /// Execute and return the id of the row with the smallest value for `field`.
641    ///
642    /// Ties are deterministic: equal field values resolve by primary key ascending.
643    pub fn min_by(&self, field: impl AsRef<str>) -> Result<Option<Id<E>>, QueryError>
644    where
645        E: EntityValue,
646    {
647        let target_slot = self.resolve_non_paged_slot(field)?;
648
649        self.execute_terminal(MinIdBySlotTerminal::new(target_slot))
650    }
651
652    /// Execute and return the largest matching identifier, if any.
653    pub fn max(&self) -> Result<Option<Id<E>>, QueryError>
654    where
655        E: EntityValue,
656    {
657        self.execute_terminal(MaxIdTerminal::new())
658    }
659
660    /// Explain scalar `max()` routing without executing the terminal.
661    pub fn explain_max(&self) -> Result<ExplainAggregateTerminalPlan, QueryError>
662    where
663        E: EntityValue,
664    {
665        self.explain_terminal(&MaxIdTerminal::new())
666    }
667
668    /// Execute and return the id of the row with the largest value for `field`.
669    ///
670    /// Ties are deterministic: equal field values resolve by primary key ascending.
671    pub fn max_by(&self, field: impl AsRef<str>) -> Result<Option<Id<E>>, QueryError>
672    where
673        E: EntityValue,
674    {
675        let target_slot = self.resolve_non_paged_slot(field)?;
676
677        self.execute_terminal(MaxIdBySlotTerminal::new(target_slot))
678    }
679
680    /// Execute and return the id at zero-based ordinal `nth` when rows are
681    /// ordered by `field` ascending, with primary-key ascending tie-breaks.
682    pub fn nth_by(&self, field: impl AsRef<str>, nth: usize) -> Result<Option<Id<E>>, QueryError>
683    where
684        E: EntityValue,
685    {
686        let target_slot = self.resolve_non_paged_slot(field)?;
687
688        self.execute_terminal(NthIdBySlotTerminal::new(target_slot, nth))
689    }
690
691    /// Execute and return the sum of `field` over matching rows.
692    pub fn sum_by(&self, field: impl AsRef<str>) -> Result<Option<Decimal>, QueryError>
693    where
694        E: EntityValue,
695    {
696        let target_slot = self.resolve_non_paged_slot(field)?;
697
698        self.execute_terminal(SumBySlotTerminal::new(target_slot))
699    }
700
701    /// Execute and return the exact sum of `field` over matching rows.
702    ///
703    /// Unlike `sum_by(...)`, this semantic aggregate rejects a prior raw
704    /// `limit(...)` so exact sums cannot accidentally mean "sum the first N
705    /// rows."
706    pub fn sum_exact(&self, field: impl AsRef<str>) -> Result<Option<Decimal>, QueryError>
707    where
708        E: EntityValue,
709    {
710        self.ensure_sum_exact_intent_owns_limit()?;
711
712        let target_slot = self.resolve_non_paged_slot(field)?;
713
714        self.execute_terminal(SumBySlotTerminal::new(target_slot))
715    }
716
717    /// Explain scalar `sum_by(field)` routing without executing the terminal.
718    pub fn explain_sum_by(
719        &self,
720        field: impl AsRef<str>,
721    ) -> Result<ExplainAggregateTerminalPlan, QueryError>
722    where
723        E: EntityValue,
724    {
725        let target_slot = self.resolve_non_paged_slot(field)?;
726
727        self.explain_terminal(&SumBySlotTerminal::new(target_slot))
728    }
729
730    /// Execute and return the sum of distinct `field` values.
731    pub fn sum_distinct_by(&self, field: impl AsRef<str>) -> Result<Option<Decimal>, QueryError>
732    where
733        E: EntityValue,
734    {
735        let target_slot = self.resolve_non_paged_slot(field)?;
736
737        self.execute_terminal(SumDistinctBySlotTerminal::new(target_slot))
738    }
739
740    /// Explain scalar `sum(distinct field)` routing without executing the terminal.
741    pub fn explain_sum_distinct_by(
742        &self,
743        field: impl AsRef<str>,
744    ) -> Result<ExplainAggregateTerminalPlan, QueryError>
745    where
746        E: EntityValue,
747    {
748        let target_slot = self.resolve_non_paged_slot(field)?;
749
750        self.explain_terminal(&SumDistinctBySlotTerminal::new(target_slot))
751    }
752
753    /// Execute and return the average of `field` over matching rows.
754    pub fn avg_by(&self, field: impl AsRef<str>) -> Result<Option<Decimal>, QueryError>
755    where
756        E: EntityValue,
757    {
758        let target_slot = self.resolve_non_paged_slot(field)?;
759
760        self.execute_terminal(AvgBySlotTerminal::new(target_slot))
761    }
762
763    /// Explain scalar `avg_by(field)` routing without executing the terminal.
764    pub fn explain_avg_by(
765        &self,
766        field: impl AsRef<str>,
767    ) -> Result<ExplainAggregateTerminalPlan, QueryError>
768    where
769        E: EntityValue,
770    {
771        let target_slot = self.resolve_non_paged_slot(field)?;
772
773        self.explain_terminal(&AvgBySlotTerminal::new(target_slot))
774    }
775
776    /// Execute and return the average of distinct `field` values.
777    pub fn avg_distinct_by(&self, field: impl AsRef<str>) -> Result<Option<Decimal>, QueryError>
778    where
779        E: EntityValue,
780    {
781        let target_slot = self.resolve_non_paged_slot(field)?;
782
783        self.execute_terminal(AvgDistinctBySlotTerminal::new(target_slot))
784    }
785
786    /// Explain scalar `avg(distinct field)` routing without executing the terminal.
787    pub fn explain_avg_distinct_by(
788        &self,
789        field: impl AsRef<str>,
790    ) -> Result<ExplainAggregateTerminalPlan, QueryError>
791    where
792        E: EntityValue,
793    {
794        let target_slot = self.resolve_non_paged_slot(field)?;
795
796        self.explain_terminal(&AvgDistinctBySlotTerminal::new(target_slot))
797    }
798
799    /// Execute and return the median id by `field` using deterministic ordering
800    /// `(field asc, primary key asc)`.
801    ///
802    /// Even-length windows select the lower median.
803    pub fn median_by(&self, field: impl AsRef<str>) -> Result<Option<Id<E>>, QueryError>
804    where
805        E: EntityValue,
806    {
807        let target_slot = self.resolve_non_paged_slot(field)?;
808
809        self.execute_terminal(MedianIdBySlotTerminal::new(target_slot))
810    }
811
812    /// Execute and return the number of distinct values for `field` over the
813    /// effective result window.
814    pub fn count_distinct_by(&self, field: impl AsRef<str>) -> Result<u32, QueryError>
815    where
816        E: EntityValue,
817    {
818        let target_slot = self.resolve_non_paged_slot(field)?;
819
820        self.execute_terminal(CountDistinctBySlotTerminal::new(target_slot))
821    }
822
823    /// Explain `count_distinct_by(field)` routing without executing the terminal.
824    pub fn explain_count_distinct_by(
825        &self,
826        field: impl AsRef<str>,
827    ) -> Result<ExplainExecutionNodeDescriptor, QueryError>
828    where
829        E: EntityValue,
830    {
831        let target_slot = self.resolve_non_paged_slot(field)?;
832
833        self.explain_terminal(&CountDistinctBySlotTerminal::new(target_slot))
834    }
835
836    /// Execute and return both `(min_by(field), max_by(field))` in one terminal.
837    ///
838    /// Tie handling is deterministic for both extrema: primary key ascending.
839    pub fn min_max_by(&self, field: impl AsRef<str>) -> Result<MinMaxByIds<E>, QueryError>
840    where
841        E: EntityValue,
842    {
843        let target_slot = self.resolve_non_paged_slot(field)?;
844
845        self.execute_terminal(MinMaxIdBySlotTerminal::new(target_slot))
846    }
847
848    /// Execute and return projected field values for the effective result window.
849    pub fn values_by(&self, field: impl AsRef<str>) -> Result<Vec<OutputValue>, QueryError>
850    where
851        E: EntityValue,
852    {
853        let target_slot = self.resolve_non_paged_slot(field)?;
854
855        self.execute_terminal(ValuesBySlotTerminal::new(target_slot))
856            .map(output_values)
857    }
858
859    /// Execute and return projected values for one shared bounded projection
860    /// over the effective response window.
861    pub fn project_values<P>(&self, projection: &P) -> Result<Vec<OutputValue>, QueryError>
862    where
863        E: EntityValue,
864        P: ValueProjectionExpr,
865    {
866        let target_slot = self.resolve_non_paged_slot(projection.field())?;
867
868        self.with_admitted_non_paged(|session, query| {
869            session.execute_fluent_project_values_by_slot(
870                query,
871                target_slot,
872                projection.projection_plan().into_expr(),
873            )
874        })
875        .map(output_values)
876    }
877
878    /// Explain `project_values(projection)` routing without executing it.
879    pub fn explain_project_values<P>(
880        &self,
881        projection: &P,
882    ) -> Result<ExplainExecutionNodeDescriptor, QueryError>
883    where
884        E: EntityValue,
885        P: ValueProjectionExpr,
886    {
887        let target_slot = self.resolve_non_paged_slot(projection.field())?;
888
889        self.explain_terminal(&ValuesBySlotTerminal::new(target_slot))
890    }
891
892    /// Explain `values_by(field)` routing without executing the terminal.
893    pub fn explain_values_by(
894        &self,
895        field: impl AsRef<str>,
896    ) -> Result<ExplainExecutionNodeDescriptor, QueryError>
897    where
898        E: EntityValue,
899    {
900        let target_slot = self.resolve_non_paged_slot(field)?;
901
902        self.explain_terminal(&ValuesBySlotTerminal::new(target_slot))
903    }
904
905    /// Execute and return the first `k` rows from the effective response window.
906    pub fn take(&self, take_count: u32) -> Result<EntityResponse<E>, QueryError>
907    where
908        E: EntityValue,
909    {
910        self.with_admitted_non_paged(|session, query| {
911            session.execute_fluent_take(query, take_count)
912        })
913    }
914
915    /// Execute and return the top `k` rows by `field` under deterministic
916    /// ordering `(field desc, primary_key asc)` over the effective response
917    /// window.
918    ///
919    /// This terminal applies its own ordering and does not preserve query
920    /// `order_term(...)` row order in the returned rows. For `k = 1`, this
921    /// matches `max_by(field)` selection semantics.
922    pub fn top_k_by(
923        &self,
924        field: impl AsRef<str>,
925        take_count: u32,
926    ) -> Result<EntityResponse<E>, QueryError>
927    where
928        E: EntityValue,
929    {
930        let target_slot = self.resolve_non_paged_slot(field)?;
931
932        self.with_admitted_non_paged(|session, query| {
933            session.execute_fluent_top_k_rows_by_slot(query, target_slot, take_count)
934        })
935    }
936
937    /// Execute and return the bottom `k` rows by `field` under deterministic
938    /// ordering `(field asc, primary_key asc)` over the effective response
939    /// window.
940    ///
941    /// This terminal applies its own ordering and does not preserve query
942    /// `order_term(...)` row order in the returned rows. For `k = 1`, this
943    /// matches `min_by(field)` selection semantics.
944    pub fn bottom_k_by(
945        &self,
946        field: impl AsRef<str>,
947        take_count: u32,
948    ) -> Result<EntityResponse<E>, QueryError>
949    where
950        E: EntityValue,
951    {
952        let target_slot = self.resolve_non_paged_slot(field)?;
953
954        self.with_admitted_non_paged(|session, query| {
955            session.execute_fluent_bottom_k_rows_by_slot(query, target_slot, take_count)
956        })
957    }
958
959    /// Execute and return projected values for the top `k` rows by `field`
960    /// under deterministic ordering `(field desc, primary_key asc)` over the
961    /// effective response window.
962    ///
963    /// Ranking is applied before projection and does not preserve query
964    /// `order_term(...)` row order in the returned values. For `k = 1`, this
965    /// matches `max_by(field)` projected to one value.
966    pub fn top_k_by_values(
967        &self,
968        field: impl AsRef<str>,
969        take_count: u32,
970    ) -> Result<Vec<OutputValue>, QueryError>
971    where
972        E: EntityValue,
973    {
974        let target_slot = self.resolve_non_paged_slot(field)?;
975
976        self.with_admitted_non_paged(|session, query| {
977            session
978                .execute_fluent_top_k_values_by_slot(query, target_slot, take_count)
979                .map(output_values)
980        })
981    }
982
983    /// Execute and return projected values for the bottom `k` rows by `field`
984    /// under deterministic ordering `(field asc, primary_key asc)` over the
985    /// effective response window.
986    ///
987    /// Ranking is applied before projection and does not preserve query
988    /// `order_term(...)` row order in the returned values. For `k = 1`, this
989    /// matches `min_by(field)` projected to one value.
990    pub fn bottom_k_by_values(
991        &self,
992        field: impl AsRef<str>,
993        take_count: u32,
994    ) -> Result<Vec<OutputValue>, QueryError>
995    where
996        E: EntityValue,
997    {
998        let target_slot = self.resolve_non_paged_slot(field)?;
999
1000        self.with_admitted_non_paged(|session, query| {
1001            session
1002                .execute_fluent_bottom_k_values_by_slot(query, target_slot, take_count)
1003                .map(output_values)
1004        })
1005    }
1006
1007    /// Execute and return projected id/value pairs for the top `k` rows by
1008    /// `field` under deterministic ordering `(field desc, primary_key asc)`
1009    /// over the effective response window.
1010    ///
1011    /// Ranking is applied before projection and does not preserve query
1012    /// `order_term(...)` row order in the returned values. For `k = 1`, this
1013    /// matches `max_by(field)` projected to one `(id, value)` pair.
1014    pub fn top_k_by_with_ids(
1015        &self,
1016        field: impl AsRef<str>,
1017        take_count: u32,
1018    ) -> Result<Vec<(Id<E>, OutputValue)>, QueryError>
1019    where
1020        E: EntityValue,
1021    {
1022        let target_slot = self.resolve_non_paged_slot(field)?;
1023
1024        self.with_admitted_non_paged(|session, query| {
1025            session
1026                .execute_fluent_top_k_values_with_ids_by_slot(query, target_slot, take_count)
1027                .map(output_values_with_ids)
1028        })
1029    }
1030
1031    /// Execute and return projected id/value pairs for the bottom `k` rows by
1032    /// `field` under deterministic ordering `(field asc, primary_key asc)`
1033    /// over the effective response window.
1034    ///
1035    /// Ranking is applied before projection and does not preserve query
1036    /// `order_term(...)` row order in the returned values. For `k = 1`, this
1037    /// matches `min_by(field)` projected to one `(id, value)` pair.
1038    pub fn bottom_k_by_with_ids(
1039        &self,
1040        field: impl AsRef<str>,
1041        take_count: u32,
1042    ) -> Result<Vec<(Id<E>, OutputValue)>, QueryError>
1043    where
1044        E: EntityValue,
1045    {
1046        let target_slot = self.resolve_non_paged_slot(field)?;
1047
1048        self.with_admitted_non_paged(|session, query| {
1049            session
1050                .execute_fluent_bottom_k_values_with_ids_by_slot(query, target_slot, take_count)
1051                .map(output_values_with_ids)
1052        })
1053    }
1054
1055    /// Execute and return distinct projected field values for the effective
1056    /// result window, preserving first-observed value order.
1057    pub fn distinct_values_by(&self, field: impl AsRef<str>) -> Result<Vec<OutputValue>, QueryError>
1058    where
1059        E: EntityValue,
1060    {
1061        let target_slot = self.resolve_non_paged_slot(field)?;
1062
1063        self.execute_terminal(DistinctValuesBySlotTerminal::new(target_slot))
1064            .map(output_values)
1065    }
1066
1067    /// Explain `distinct_values_by(field)` routing without executing the terminal.
1068    pub fn explain_distinct_values_by(
1069        &self,
1070        field: impl AsRef<str>,
1071    ) -> Result<ExplainExecutionNodeDescriptor, QueryError>
1072    where
1073        E: EntityValue,
1074    {
1075        let target_slot = self.resolve_non_paged_slot(field)?;
1076
1077        self.explain_terminal(&DistinctValuesBySlotTerminal::new(target_slot))
1078    }
1079
1080    /// Execute and return projected field values paired with row ids for the
1081    /// effective result window.
1082    pub fn values_by_with_ids(
1083        &self,
1084        field: impl AsRef<str>,
1085    ) -> Result<Vec<(Id<E>, OutputValue)>, QueryError>
1086    where
1087        E: EntityValue,
1088    {
1089        let target_slot = self.resolve_non_paged_slot(field)?;
1090
1091        self.execute_terminal(ValuesBySlotWithIdsTerminal::new(target_slot))
1092            .map(output_values_with_ids)
1093    }
1094
1095    /// Execute and return projected id/value pairs for one shared bounded
1096    /// projection over the effective response window.
1097    pub fn project_values_with_ids<P>(
1098        &self,
1099        projection: &P,
1100    ) -> Result<Vec<(Id<E>, OutputValue)>, QueryError>
1101    where
1102        E: EntityValue,
1103        P: ValueProjectionExpr,
1104    {
1105        let target_slot = self.resolve_non_paged_slot(projection.field())?;
1106
1107        self.with_admitted_non_paged(|session, query| {
1108            session.execute_fluent_project_values_with_ids_by_slot(
1109                query,
1110                target_slot,
1111                projection.projection_plan().into_expr(),
1112            )
1113        })
1114        .map(output_values_with_ids)
1115    }
1116
1117    /// Explain `values_by_with_ids(field)` routing without executing the terminal.
1118    pub fn explain_values_by_with_ids(
1119        &self,
1120        field: impl AsRef<str>,
1121    ) -> Result<ExplainExecutionNodeDescriptor, QueryError>
1122    where
1123        E: EntityValue,
1124    {
1125        let target_slot = self.resolve_non_paged_slot(field)?;
1126
1127        self.explain_terminal(&ValuesBySlotWithIdsTerminal::new(target_slot))
1128    }
1129
1130    /// Execute and return the first projected field value in effective response
1131    /// order, if any.
1132    pub fn first_value_by(&self, field: impl AsRef<str>) -> Result<Option<OutputValue>, QueryError>
1133    where
1134        E: EntityValue,
1135    {
1136        let target_slot = self.resolve_non_paged_slot(field)?;
1137
1138        self.execute_terminal(FirstValueBySlotTerminal::new(target_slot))
1139            .map(|value| value.map(output))
1140    }
1141
1142    /// Execute and return the first projected value for one shared bounded
1143    /// projection in effective response order, if any.
1144    pub fn project_first_value<P>(&self, projection: &P) -> Result<Option<OutputValue>, QueryError>
1145    where
1146        E: EntityValue,
1147        P: ValueProjectionExpr,
1148    {
1149        let target_slot = self.resolve_non_paged_slot(projection.field())?;
1150
1151        self.with_admitted_non_paged(|session, query| {
1152            session.execute_fluent_project_terminal_value_by_slot(
1153                query,
1154                target_slot,
1155                AggregateKind::First,
1156                projection.projection_plan().into_expr(),
1157            )
1158        })
1159        .map(|value| value.map(output))
1160    }
1161
1162    /// Explain `first_value_by(field)` routing without executing the terminal.
1163    pub fn explain_first_value_by(
1164        &self,
1165        field: impl AsRef<str>,
1166    ) -> Result<ExplainExecutionNodeDescriptor, QueryError>
1167    where
1168        E: EntityValue,
1169    {
1170        let target_slot = self.resolve_non_paged_slot(field)?;
1171
1172        self.explain_terminal(&FirstValueBySlotTerminal::new(target_slot))
1173    }
1174
1175    /// Execute and return the last projected field value in effective response
1176    /// order, if any.
1177    pub fn last_value_by(&self, field: impl AsRef<str>) -> Result<Option<OutputValue>, QueryError>
1178    where
1179        E: EntityValue,
1180    {
1181        let target_slot = self.resolve_non_paged_slot(field)?;
1182
1183        self.execute_terminal(LastValueBySlotTerminal::new(target_slot))
1184            .map(|value| value.map(output))
1185    }
1186
1187    /// Execute and return the last projected value for one shared bounded
1188    /// projection in effective response order, if any.
1189    pub fn project_last_value<P>(&self, projection: &P) -> Result<Option<OutputValue>, QueryError>
1190    where
1191        E: EntityValue,
1192        P: ValueProjectionExpr,
1193    {
1194        let target_slot = self.resolve_non_paged_slot(projection.field())?;
1195
1196        self.with_admitted_non_paged(|session, query| {
1197            session.execute_fluent_project_terminal_value_by_slot(
1198                query,
1199                target_slot,
1200                AggregateKind::Last,
1201                projection.projection_plan().into_expr(),
1202            )
1203        })
1204        .map(|value| value.map(output))
1205    }
1206
1207    /// Explain `last_value_by(field)` routing without executing the terminal.
1208    pub fn explain_last_value_by(
1209        &self,
1210        field: impl AsRef<str>,
1211    ) -> Result<ExplainExecutionNodeDescriptor, QueryError>
1212    where
1213        E: EntityValue,
1214    {
1215        let target_slot = self.resolve_non_paged_slot(field)?;
1216
1217        self.explain_terminal(&LastValueBySlotTerminal::new(target_slot))
1218    }
1219
1220    /// Execute and return the first matching identifier in response order, if any.
1221    pub fn first(&self) -> Result<Option<Id<E>>, QueryError>
1222    where
1223        E: EntityValue,
1224    {
1225        self.execute_terminal(FirstIdTerminal::new())
1226    }
1227
1228    /// Explain scalar `first()` routing without executing the terminal.
1229    pub fn explain_first(&self) -> Result<ExplainAggregateTerminalPlan, QueryError>
1230    where
1231        E: EntityValue,
1232    {
1233        self.explain_terminal(&FirstIdTerminal::new())
1234    }
1235
1236    /// Execute and return the last matching identifier in response order, if any.
1237    pub fn last(&self) -> Result<Option<Id<E>>, QueryError>
1238    where
1239        E: EntityValue,
1240    {
1241        self.execute_terminal(LastIdTerminal::new())
1242    }
1243
1244    /// Explain scalar `last()` routing without executing the terminal.
1245    pub fn explain_last(&self) -> Result<ExplainAggregateTerminalPlan, QueryError>
1246    where
1247        E: EntityValue,
1248    {
1249        self.explain_terminal(&LastIdTerminal::new())
1250    }
1251
1252    /// Execute and require exactly one matching row.
1253    pub fn require_one(&self) -> Result<(), QueryError>
1254    where
1255        E: EntityValue,
1256    {
1257        self.execute_rows()?.require_one()?;
1258        Ok(())
1259    }
1260
1261    /// Execute and require at least one matching row.
1262    pub fn require_some(&self) -> Result<(), QueryError>
1263    where
1264        E: EntityValue,
1265    {
1266        self.execute_rows()?.require_some()?;
1267        Ok(())
1268    }
1269}