teaql_core/
query.rs

1use std::collections::BTreeMap;
2use std::sync::Arc;
3
4use crate::{Expr, Value};
5
6#[derive(Debug, Clone, Copy, PartialEq, Eq)]
7pub enum SortDirection {
8    Asc,
9    Desc,
10}
11
12#[cfg(test)]
13mod hard_limit_tests {
14    use super::*;
15
16    #[test]
17    fn list_limit_defaults_rejects_and_allows_explicit_override() {
18        assert_eq!(
19            SelectQuery::new("Order")
20                .prepare_for_list()
21                .unwrap()
22                .slice
23                .unwrap()
24                .limit,
25            Some(10_000)
26        );
27        assert!(
28            SelectQuery::new("Order")
29                .limit(10_001)
30                .prepare_for_list()
31                .is_err()
32        );
33        assert!(
34            SelectQuery::new("Order")
35                .limit(10_001)
36                .hard_limit(20_000)
37                .prepare_for_list()
38                .is_ok()
39        );
40    }
41
42    #[test]
43    fn continuous_page_fetch_is_explicit_and_validated() {
44        assert!(SelectQuery::new("Order").continuous_page_fetch.is_none());
45        let query =
46            SelectQuery::new("Order").optimize_for_continuous_page_fetch_with("recent-orders", 30);
47        let options = query.continuous_page_fetch.unwrap();
48        assert_eq!(options.namespace, "recent-orders");
49        assert_eq!(options.ttl_seconds, 30);
50    }
51
52    #[test]
53    #[should_panic(expected = "continuous page namespace must not be empty")]
54    fn continuous_page_fetch_rejects_empty_namespace() {
55        let _ = SelectQuery::new("Order").optimize_for_continuous_page_fetch_with(" ", 30);
56    }
57
58    #[test]
59    fn id_set_pagination_is_explicit_and_validated() {
60        assert!(SelectQuery::new("Order").id_set_pagination.is_none());
61        let query = SelectQuery::new("Order").optimize_pagination_with_id_set_config(
62            "recent-orders",
63            30,
64            5_000,
65        );
66        let options = query.id_set_pagination.expect("ID set options");
67        assert_eq!(options.namespace, "recent-orders");
68        assert_eq!(options.ttl_seconds, 30);
69        assert_eq!(options.max_ids, 5_000);
70    }
71
72    #[test]
73    #[should_panic(expected = "ID set pagination max_ids must be positive")]
74    fn id_set_pagination_rejects_zero_limit() {
75        let _ = SelectQuery::new("Order").optimize_pagination_with_id_set_config("orders", 30, 0);
76    }
77
78    #[test]
79    fn topn_001_002_003_probe_threshold_is_explicit_and_bounded() {
80        assert!(
81            SelectQuery::new("Trip")
82                .top_n_probe_parent_threshold
83                .is_none()
84        );
85        assert_eq!(
86            SelectQuery::new("Trip")
87                .top_n_probe_parent_threshold(32)
88                .top_n_probe_parent_threshold,
89            Some(32)
90        );
91        assert_eq!(
92            SelectQuery::new("Trip")
93                .top_n_probe_parent_threshold(0)
94                .top_n_probe_parent_threshold,
95            Some(0)
96        );
97    }
98}
99
100#[derive(Debug, Clone, PartialEq)]
101pub struct NamedExpr {
102    pub alias: String,
103    pub expr: Expr,
104}
105
106impl NamedExpr {
107    pub fn new(alias: impl Into<String>, expr: Expr) -> Self {
108        Self {
109            alias: alias.into(),
110            expr,
111        }
112    }
113}
114
115#[derive(Debug, Clone, PartialEq)]
116pub struct OrderBy {
117    pub field: String,
118    pub expr: Option<Expr>,
119    pub direction: SortDirection,
120}
121
122impl OrderBy {
123    pub fn new(field: impl Into<String>, direction: SortDirection) -> Self {
124        Self {
125            field: field.into(),
126            expr: None,
127            direction,
128        }
129    }
130
131    pub fn expr(expr: Expr, direction: SortDirection) -> Self {
132        Self {
133            field: String::new(),
134            expr: Some(expr),
135            direction,
136        }
137    }
138
139    pub fn asc(field: impl Into<String>) -> Self {
140        Self::new(field, SortDirection::Asc)
141    }
142
143    pub fn desc(field: impl Into<String>) -> Self {
144        Self::new(field, SortDirection::Desc)
145    }
146
147    pub fn asc_expr(expr: Expr) -> Self {
148        Self::expr(expr, SortDirection::Asc)
149    }
150
151    pub fn desc_expr(expr: Expr) -> Self {
152        Self::expr(expr, SortDirection::Desc)
153    }
154
155    pub fn asc_gbk(field: impl Into<String>) -> Self {
156        Self::asc_expr(Expr::gbk(Expr::column(field)))
157    }
158
159    pub fn desc_gbk(field: impl Into<String>) -> Self {
160        Self::desc_expr(Expr::gbk(Expr::column(field)))
161    }
162}
163
164#[derive(Debug, Clone, Copy, PartialEq, Eq)]
165pub enum AggregateFunction {
166    Count,
167    Sum,
168    Avg,
169    Min,
170    Max,
171    Stddev,
172    StddevPop,
173    VarSamp,
174    VarPop,
175    BitAnd,
176    BitOr,
177    BitXor,
178}
179
180#[derive(Debug, Clone, PartialEq, Eq)]
181pub struct Aggregate {
182    pub function: AggregateFunction,
183    pub field: String,
184    pub alias: String,
185}
186
187impl Aggregate {
188    pub fn new(
189        function: AggregateFunction,
190        field: impl Into<String>,
191        alias: impl Into<String>,
192    ) -> Self {
193        Self {
194            function,
195            field: field.into(),
196            alias: alias.into(),
197        }
198    }
199
200    pub fn count(alias: impl Into<String>) -> Self {
201        Self::new(AggregateFunction::Count, "*", alias)
202    }
203
204    pub fn count_field(field: impl Into<String>, alias: impl Into<String>) -> Self {
205        Self::new(AggregateFunction::Count, field, alias)
206    }
207
208    pub fn sum(field: impl Into<String>, alias: impl Into<String>) -> Self {
209        Self::new(AggregateFunction::Sum, field, alias)
210    }
211
212    pub fn avg(field: impl Into<String>, alias: impl Into<String>) -> Self {
213        Self::new(AggregateFunction::Avg, field, alias)
214    }
215
216    pub fn min(field: impl Into<String>, alias: impl Into<String>) -> Self {
217        Self::new(AggregateFunction::Min, field, alias)
218    }
219
220    pub fn max(field: impl Into<String>, alias: impl Into<String>) -> Self {
221        Self::new(AggregateFunction::Max, field, alias)
222    }
223
224    pub fn stddev(field: impl Into<String>, alias: impl Into<String>) -> Self {
225        Self::new(AggregateFunction::Stddev, field, alias)
226    }
227
228    pub fn stddev_pop(field: impl Into<String>, alias: impl Into<String>) -> Self {
229        Self::new(AggregateFunction::StddevPop, field, alias)
230    }
231
232    pub fn var_samp(field: impl Into<String>, alias: impl Into<String>) -> Self {
233        Self::new(AggregateFunction::VarSamp, field, alias)
234    }
235
236    pub fn var_pop(field: impl Into<String>, alias: impl Into<String>) -> Self {
237        Self::new(AggregateFunction::VarPop, field, alias)
238    }
239
240    pub fn bit_and(field: impl Into<String>, alias: impl Into<String>) -> Self {
241        Self::new(AggregateFunction::BitAnd, field, alias)
242    }
243
244    pub fn bit_or(field: impl Into<String>, alias: impl Into<String>) -> Self {
245        Self::new(AggregateFunction::BitOr, field, alias)
246    }
247
248    pub fn bit_xor(field: impl Into<String>, alias: impl Into<String>) -> Self {
249        Self::new(AggregateFunction::BitXor, field, alias)
250    }
251}
252
253#[derive(Debug, Clone, Copy, PartialEq, Eq)]
254pub struct Slice {
255    pub limit: Option<u64>,
256    pub offset: u64,
257}
258
259#[derive(Debug, Clone, PartialEq)]
260pub struct RelationLoad {
261    pub name: String,
262    pub query: Option<Box<SelectQuery>>,
263}
264
265impl RelationLoad {
266    pub fn new(name: impl Into<String>) -> Self {
267        Self {
268            name: name.into(),
269            query: None,
270        }
271    }
272
273    pub fn with_query(name: impl Into<String>, query: SelectQuery) -> Self {
274        Self {
275            name: name.into(),
276            query: Some(Box::new(query)),
277        }
278    }
279}
280
281#[derive(Debug, Clone, PartialEq)]
282pub struct RelationAggregate {
283    pub relation_name: String,
284    pub alias: String,
285    pub query: SelectQuery,
286    pub single_result: bool,
287}
288
289impl RelationAggregate {
290    pub fn new(
291        relation_name: impl Into<String>,
292        alias: impl Into<String>,
293        query: SelectQuery,
294        single_result: bool,
295    ) -> Self {
296        Self {
297            relation_name: relation_name.into(),
298            alias: alias.into(),
299            query,
300            single_result,
301        }
302    }
303}
304
305#[derive(Debug, Clone, PartialEq, Eq)]
306pub struct RawSqlProjection {
307    pub property_name: String,
308    pub raw_sql_segment: String,
309}
310
311impl RawSqlProjection {
312    pub fn new(property_name: impl Into<String>, raw_sql_segment: impl Into<String>) -> Self {
313        Self {
314            property_name: property_name.into(),
315            raw_sql_segment: raw_sql_segment.into(),
316        }
317    }
318}
319
320#[derive(Debug, Clone, PartialEq)]
321pub struct ObjectGroupBy {
322    pub property_name: String,
323    pub storage_field: String,
324    pub query: SelectQuery,
325}
326
327impl ObjectGroupBy {
328    pub fn new(
329        property_name: impl Into<String>,
330        storage_field: impl Into<String>,
331        query: SelectQuery,
332    ) -> Self {
333        Self {
334            property_name: property_name.into(),
335            storage_field: storage_field.into(),
336            query,
337        }
338    }
339}
340
341#[derive(Debug, Clone, Copy, PartialEq, Eq)]
342pub struct AggregationCacheOptions {
343    pub enabled: bool,
344    pub cache_expired_millis: u64,
345    pub propagate: bool,
346    pub propagate_cache_expired_millis: u64,
347}
348
349impl AggregationCacheOptions {
350    pub fn enabled(cache_expired_millis: u64) -> Self {
351        Self {
352            enabled: true,
353            cache_expired_millis,
354            propagate: false,
355            propagate_cache_expired_millis: 0,
356        }
357    }
358
359    pub fn propagate(mut self, cache_expired_millis: u64) -> Self {
360        self.propagate = true;
361        self.propagate_cache_expired_millis = cache_expired_millis;
362        self
363    }
364}
365
366#[derive(Debug, Clone, PartialEq)]
367pub struct StreamConfig {
368    pub chunk_size: usize,
369}
370
371#[derive(Debug, Clone, PartialEq, Eq)]
372pub struct ContinuousPageFetchOptions {
373    pub namespace: String,
374    pub ttl_seconds: u64,
375}
376
377#[derive(Debug, Clone, PartialEq, Eq)]
378pub struct IdSetPaginationOptions {
379    pub namespace: String,
380    pub ttl_seconds: u64,
381    pub max_ids: u64,
382}
383
384impl IdSetPaginationOptions {
385    pub const DEFAULT_TTL_SECONDS: u64 = 600;
386    pub const DEFAULT_MAX_IDS: u64 = 3_000_000;
387
388    pub fn new(namespace: impl Into<String>, ttl_seconds: u64, max_ids: u64) -> Self {
389        let namespace = namespace.into();
390        assert!(
391            !namespace.trim().is_empty(),
392            "ID set pagination namespace must not be empty"
393        );
394        assert!(
395            ttl_seconds > 0,
396            "ID set pagination ttl_seconds must be positive"
397        );
398        assert!(max_ids > 0, "ID set pagination max_ids must be positive");
399        Self {
400            namespace,
401            ttl_seconds,
402            max_ids,
403        }
404    }
405}
406
407impl ContinuousPageFetchOptions {
408    pub const DEFAULT_TTL_SECONDS: u64 = 600;
409
410    pub fn new(namespace: impl Into<String>, ttl_seconds: u64) -> Self {
411        let namespace = namespace.into();
412        assert!(
413            !namespace.trim().is_empty(),
414            "continuous page namespace must not be empty"
415        );
416        assert!(
417            ttl_seconds > 0,
418            "continuous page ttl_seconds must be positive"
419        );
420        Self {
421            namespace,
422            ttl_seconds,
423        }
424    }
425}
426
427impl Default for StreamConfig {
428    fn default() -> Self {
429        Self { chunk_size: 1000 }
430    }
431}
432
433#[derive(Debug, Clone, PartialEq)]
434pub struct SelectQuery {
435    /// Safety ceiling for a fully materialized outer query.
436    pub hard_limit: u64,
437    pub entity: String,
438    pub projection: Vec<String>,
439    pub expr_projection: Vec<NamedExpr>,
440    pub search_with_text: Option<String>,
441    pub filter: Option<Expr>,
442    pub having: Option<Expr>,
443    pub order_by: Vec<OrderBy>,
444    pub slice: Option<Slice>,
445    /// Apply `slice` independently inside each value of this property.
446    pub partition_by: Option<String>,
447    /// Optional bounded-probe override for per-parent Top-N relation loading.
448    /// Zero forces the provider's window plan; a positive value permits probes
449    /// only when the already-loaded parent count does not exceed the value.
450    pub top_n_probe_parent_threshold: Option<usize>,
451    pub aggregates: Vec<Aggregate>,
452    pub group_by: Vec<String>,
453    pub relations: Vec<RelationLoad>,
454    pub aggregation_cache: Option<AggregationCacheOptions>,
455    pub comment: Option<String>,
456    pub trace_chain: Vec<crate::TraceNode>,
457    pub raw_sql: Option<String>,
458    pub raw_sql_search_criteria: Vec<String>,
459    pub dynamic_properties: Vec<RawSqlProjection>,
460    pub raw_projections: Vec<RawSqlProjection>,
461    pub object_group_bys: Vec<ObjectGroupBy>,
462    pub child_enhancements: Vec<SelectQuery>,
463    pub stream_config: Option<StreamConfig>,
464    /// Explicit, process-local hint for transparent seek pagination of outer list queries.
465    pub continuous_page_fetch: Option<ContinuousPageFetchOptions>,
466    /// Explicit hint to retain the complete ordered ID sequence for pagination.
467    pub id_set_pagination: Option<IdSetPaginationOptions>,
468}
469
470impl SelectQuery {
471    pub fn new(entity: impl Into<String>) -> Self {
472        Self {
473            hard_limit: 10_000,
474            entity: entity.into(),
475            projection: Vec::new(),
476            expr_projection: Vec::new(),
477            search_with_text: None,
478            filter: None,
479            having: None,
480            order_by: Vec::new(),
481            slice: None,
482            partition_by: None,
483            top_n_probe_parent_threshold: None,
484            aggregates: Vec::new(),
485            group_by: Vec::new(),
486            relations: Vec::new(),
487            aggregation_cache: None,
488            comment: None,
489            trace_chain: Vec::new(),
490            raw_sql: None,
491            raw_sql_search_criteria: Vec::new(),
492            dynamic_properties: Vec::new(),
493            raw_projections: Vec::new(),
494            object_group_bys: Vec::new(),
495            child_enhancements: Vec::new(),
496            stream_config: None,
497            continuous_page_fetch: None,
498            id_set_pagination: None,
499        }
500    }
501
502    pub fn project(mut self, field: impl Into<String>) -> Self {
503        self.projection.push(field.into());
504        self
505    }
506
507    pub fn projects(mut self, fields: impl IntoIterator<Item = impl Into<String>>) -> Self {
508        self.projection.extend(fields.into_iter().map(Into::into));
509        self
510    }
511
512    pub fn project_expr(mut self, alias: impl Into<String>, expr: Expr) -> Self {
513        self.expr_projection.push(NamedExpr::new(alias, expr));
514        self
515    }
516
517    pub fn project_raw(
518        mut self,
519        alias: impl Into<String>,
520        raw_sql_segment: impl Into<String>,
521    ) -> Self {
522        self.raw_projections
523            .push(RawSqlProjection::new(alias, raw_sql_segment));
524        self
525    }
526
527    pub fn dynamic_property_raw(
528        mut self,
529        alias: impl Into<String>,
530        raw_sql_segment: impl Into<String>,
531    ) -> Self {
532        self.dynamic_properties
533            .push(RawSqlProjection::new(alias, raw_sql_segment));
534        self
535    }
536
537    pub fn search_with_text(mut self, text: impl Into<String>) -> Self {
538        self.search_with_text = Some(text.into());
539        self
540    }
541
542    pub fn filter(mut self, filter: Expr) -> Self {
543        self.filter = Some(filter);
544        self
545    }
546
547    pub fn and_filter(mut self, filter: Expr) -> Self {
548        self.filter = Some(match self.filter.take() {
549            Some(existing) => existing.and_expr(filter),
550            None => filter,
551        });
552        self
553    }
554
555    pub fn or_filter(mut self, filter: Expr) -> Self {
556        self.filter = Some(match self.filter.take() {
557            Some(existing) => existing.or_expr(filter),
558            None => filter,
559        });
560        self
561    }
562
563    pub fn having(mut self, having: Expr) -> Self {
564        self.having = Some(having);
565        self
566    }
567
568    pub fn and_having(mut self, having: Expr) -> Self {
569        self.having = Some(match self.having.take() {
570            Some(existing) => existing.and_expr(having),
571            None => having,
572        });
573        self
574    }
575
576    pub fn or_having(mut self, having: Expr) -> Self {
577        self.having = Some(match self.having.take() {
578            Some(existing) => existing.or_expr(having),
579            None => having,
580        });
581        self
582    }
583
584    pub fn order_by(mut self, order: OrderBy) -> Self {
585        self.order_by.push(order);
586        self
587    }
588
589    pub fn order_asc(self, field: impl Into<String>) -> Self {
590        self.order_by(OrderBy::asc(field))
591    }
592
593    pub fn order_desc(self, field: impl Into<String>) -> Self {
594        self.order_by(OrderBy::desc(field))
595    }
596
597    pub fn order_expr_asc(self, expr: Expr) -> Self {
598        self.order_by(OrderBy::asc_expr(expr))
599    }
600
601    pub fn order_expr_desc(self, expr: Expr) -> Self {
602        self.order_by(OrderBy::desc_expr(expr))
603    }
604
605    pub fn order_gbk_asc(self, field: impl Into<String>) -> Self {
606        self.order_by(OrderBy::asc_gbk(field))
607    }
608
609    pub fn order_gbk_desc(self, field: impl Into<String>) -> Self {
610        self.order_by(OrderBy::desc_gbk(field))
611    }
612
613    pub fn group_by(mut self, field: impl Into<String>) -> Self {
614        self.group_by.push(field.into());
615        self
616    }
617
618    pub fn aggregate(mut self, aggregate: Aggregate) -> Self {
619        self.aggregates.push(aggregate);
620        self
621    }
622
623    pub fn count(self, alias: impl Into<String>) -> Self {
624        self.aggregate(Aggregate::count(alias))
625    }
626
627    pub fn count_field(self, field: impl Into<String>, alias: impl Into<String>) -> Self {
628        self.aggregate(Aggregate::count_field(field, alias))
629    }
630
631    pub fn sum(self, field: impl Into<String>, alias: impl Into<String>) -> Self {
632        self.aggregate(Aggregate::sum(field, alias))
633    }
634
635    pub fn avg(self, field: impl Into<String>, alias: impl Into<String>) -> Self {
636        self.aggregate(Aggregate::avg(field, alias))
637    }
638
639    pub fn min(self, field: impl Into<String>, alias: impl Into<String>) -> Self {
640        self.aggregate(Aggregate::min(field, alias))
641    }
642
643    pub fn max(self, field: impl Into<String>, alias: impl Into<String>) -> Self {
644        self.aggregate(Aggregate::max(field, alias))
645    }
646
647    pub fn stddev(self, field: impl Into<String>, alias: impl Into<String>) -> Self {
648        self.aggregate(Aggregate::stddev(field, alias))
649    }
650
651    pub fn stddev_pop(self, field: impl Into<String>, alias: impl Into<String>) -> Self {
652        self.aggregate(Aggregate::stddev_pop(field, alias))
653    }
654
655    pub fn var_samp(self, field: impl Into<String>, alias: impl Into<String>) -> Self {
656        self.aggregate(Aggregate::var_samp(field, alias))
657    }
658
659    pub fn var_pop(self, field: impl Into<String>, alias: impl Into<String>) -> Self {
660        self.aggregate(Aggregate::var_pop(field, alias))
661    }
662
663    pub fn bit_and(self, field: impl Into<String>, alias: impl Into<String>) -> Self {
664        self.aggregate(Aggregate::bit_and(field, alias))
665    }
666
667    pub fn bit_or(self, field: impl Into<String>, alias: impl Into<String>) -> Self {
668        self.aggregate(Aggregate::bit_or(field, alias))
669    }
670
671    pub fn bit_xor(self, field: impl Into<String>, alias: impl Into<String>) -> Self {
672        self.aggregate(Aggregate::bit_xor(field, alias))
673    }
674
675    pub fn enable_aggregation_cache(self) -> Self {
676        self.enable_aggregation_cache_for(0)
677    }
678
679    pub fn enable_aggregation_cache_for(mut self, cache_expired_millis: u64) -> Self {
680        self.aggregation_cache = Some(AggregationCacheOptions::enabled(cache_expired_millis));
681        self
682    }
683
684    pub fn propagate_aggregation_cache(mut self, cache_expired_millis: u64) -> Self {
685        self.aggregation_cache = Some(
686            self.aggregation_cache
687                .unwrap_or_else(|| AggregationCacheOptions::enabled(0))
688                .propagate(cache_expired_millis),
689        );
690        self
691    }
692
693    pub fn comment(mut self, comment: impl Into<String>) -> Self {
694        let comment_str = comment.into();
695        self.comment = Some(comment_str.clone());
696        self.trace_chain.push(crate::TraceNode {
697            entity_type: self.entity.clone(),
698            entity_id: None,
699            comment: comment_str,
700        });
701        self
702    }
703
704    pub fn raw_sql(mut self, raw_sql: impl Into<String>) -> Self {
705        self.raw_sql = Some(raw_sql.into());
706        self
707    }
708
709    pub fn raw_sql_search_criteria(mut self, raw_sql: impl Into<String>) -> Self {
710        self.raw_sql_search_criteria.push(raw_sql.into());
711        self
712    }
713
714    pub fn object_group_by(
715        mut self,
716        property_name: impl Into<String>,
717        storage_field: impl Into<String>,
718        query: SelectQuery,
719    ) -> Self {
720        self.object_group_bys
721            .push(ObjectGroupBy::new(property_name, storage_field, query));
722        self
723    }
724
725    pub fn child_enhancement(mut self, query: SelectQuery) -> Self {
726        self.child_enhancements.push(query);
727        self
728    }
729
730    pub fn relation(mut self, name: impl Into<String>) -> Self {
731        self.relations.push(RelationLoad::new(name));
732        self
733    }
734
735    pub fn relation_query(mut self, name: impl Into<String>, query: SelectQuery) -> Self {
736        self.relations.push(RelationLoad::with_query(name, query));
737        self
738    }
739
740    pub fn limit(mut self, limit: u64) -> Self {
741        let slice = self.slice.get_or_insert(Slice {
742            limit: None,
743            offset: 0,
744        });
745        slice.limit = Some(limit);
746        self
747    }
748
749    /// Override the outer materialized-list ceiling. Most callers should keep 10,000.
750    pub fn hard_limit(mut self, hard_limit: u64) -> Self {
751        assert!(hard_limit > 0, "hard_limit must be positive");
752        self.hard_limit = hard_limit;
753        self
754    }
755
756    /// Apply and validate list-materialization limits. This is intentionally not
757    /// used by streaming execution.
758    pub fn prepare_for_list(mut self) -> Result<Self, String> {
759        self.apply_list_limit(self.hard_limit, true)?;
760        Ok(self)
761    }
762
763    fn apply_list_limit(&mut self, ceiling: u64, outer: bool) -> Result<(), String> {
764        let slice = self.slice.get_or_insert(Slice {
765            limit: None,
766            offset: 0,
767        });
768        match slice.limit {
769            Some(limit) if limit > ceiling => {
770                return Err(format!(
771                    "QUERY_HARD_LIMIT_EXCEEDED: requested limit {limit} exceeds hard limit {ceiling}"
772                ));
773            }
774            None => slice.limit = Some(ceiling),
775            _ => {}
776        }
777        for relation in &mut self.relations {
778            if let Some(query) = relation.query.as_mut() {
779                query.apply_list_limit(10_000, false)?;
780            }
781        }
782        for query in &mut self.child_enhancements {
783            query.apply_list_limit(10_000, false)?;
784        }
785        let _ = outer;
786        Ok(())
787    }
788
789    pub fn offset(mut self, offset: u64) -> Self {
790        let slice = self.slice.get_or_insert(Slice {
791            limit: None,
792            offset: 0,
793        });
794        slice.offset = offset;
795        self
796    }
797
798    pub fn page(self, offset: u64, limit: u64) -> Self {
799        self.offset(offset).limit(limit)
800    }
801
802    pub fn optimize_for_continuous_page_fetch(mut self) -> Self {
803        self.continuous_page_fetch = Some(ContinuousPageFetchOptions::new(
804            "default",
805            ContinuousPageFetchOptions::DEFAULT_TTL_SECONDS,
806        ));
807        self
808    }
809
810    pub fn optimize_for_continuous_page_fetch_with(
811        mut self,
812        namespace: impl Into<String>,
813        ttl_seconds: u64,
814    ) -> Self {
815        self.continuous_page_fetch = Some(ContinuousPageFetchOptions::new(namespace, ttl_seconds));
816        self
817    }
818
819    pub fn optimize_pagination_with_id_set(mut self) -> Self {
820        self.id_set_pagination = Some(IdSetPaginationOptions::new(
821            "default",
822            IdSetPaginationOptions::DEFAULT_TTL_SECONDS,
823            IdSetPaginationOptions::DEFAULT_MAX_IDS,
824        ));
825        self
826    }
827
828    pub fn optimize_pagination_with_id_set_config(
829        mut self,
830        namespace: impl Into<String>,
831        ttl_seconds: u64,
832        max_ids: u64,
833    ) -> Self {
834        self.id_set_pagination = Some(IdSetPaginationOptions::new(namespace, ttl_seconds, max_ids));
835        self
836    }
837
838    /// Scope pagination to each distinct value of `field`.
839    ///
840    /// Relation loading sets this automatically. Most application queries
841    /// should use a generated relation selector instead of calling this
842    /// method directly.
843    pub fn partition_by(mut self, field: impl Into<String>) -> Self {
844        self.partition_by = Some(field.into());
845        self
846    }
847
848    /// Override the provider's per-parent Top-N execution policy.
849    ///
850    /// `0` selects the window plan. A positive threshold selects bounded
851    /// probes only when the parent rows already in memory are at or below it.
852    pub fn top_n_probe_parent_threshold(mut self, threshold: usize) -> Self {
853        self.top_n_probe_parent_threshold = Some(threshold);
854        self
855    }
856
857    /// Enable streaming mode with the given chunk size.
858    /// When streaming, rows are fetched and enhanced in batches rather than all at once.
859    pub fn stream(mut self, chunk_size: usize) -> Self {
860        self.stream_config = Some(StreamConfig { chunk_size });
861        self
862    }
863
864    /// Enable streaming mode with default chunk size (1000).
865    pub fn stream_default(mut self) -> Self {
866        self.stream_config = Some(StreamConfig::default());
867        self
868    }
869}
870
871pub type Record = BTreeMap<String, Value>;
872
873/// A database result row whose column names are shared by the whole result set.
874///
875/// Providers use this representation for typed decoding so a 100-row result does
876/// not allocate 100 copies of every projected column name (or 100 B-trees).
877#[derive(Debug, Clone, PartialEq)]
878pub struct CompactRow {
879    columns: Arc<[String]>,
880    values: Vec<Value>,
881}
882
883impl CompactRow {
884    pub fn new(columns: Arc<[String]>, values: Vec<Value>) -> Self {
885        debug_assert_eq!(columns.len(), values.len());
886        Self { columns, values }
887    }
888
889    pub fn get(&self, name: &str) -> Option<&Value> {
890        self.columns
891            .iter()
892            .position(|column| column == name)
893            .and_then(|index| self.values.get(index))
894    }
895
896    pub fn shared_columns(&self) -> Arc<[String]> {
897        self.columns.clone()
898    }
899
900    pub fn get_mut(&mut self, name: &str) -> Option<&mut Value> {
901        self.columns
902            .iter()
903            .position(|column| column == name)
904            .and_then(|index| self.values.get_mut(index))
905    }
906
907    /// Adds or replaces a projected value. Column layouts remain shared until
908    /// an enhancement actually changes the shape of this row.
909    pub fn insert(&mut self, name: String, value: Value) -> Option<Value> {
910        if let Some(index) = self.columns.iter().position(|column| column == &name) {
911            return Some(std::mem::replace(&mut self.values[index], value));
912        }
913        let mut columns = self.columns.to_vec();
914        columns.push(name);
915        self.columns = columns.into();
916        self.values.push(value);
917        None
918    }
919
920    pub fn remove(&mut self, name: &str) -> Option<Value> {
921        let index = self.columns.iter().position(|column| column == name)?;
922        let mut columns = self.columns.to_vec();
923        columns.remove(index);
924        self.columns = columns.into();
925        Some(self.values.remove(index))
926    }
927
928    pub fn extend(&mut self, other: CompactRow) {
929        for (name, value) in other.columns.iter().cloned().zip(other.values) {
930            self.insert(name, value);
931        }
932    }
933
934    pub fn contains_key(&self, name: &str) -> bool {
935        self.columns.iter().any(|column| column == name)
936    }
937
938    pub fn len(&self) -> usize {
939        self.values.len()
940    }
941
942    pub fn is_empty(&self) -> bool {
943        self.values.is_empty()
944    }
945
946    pub fn iter(&self) -> impl Iterator<Item = (&String, &Value)> {
947        self.columns.iter().zip(self.values.iter())
948    }
949
950    pub fn keys(&self) -> impl Iterator<Item = &String> {
951        self.columns.iter()
952    }
953
954    pub fn values(&self) -> impl Iterator<Item = &Value> {
955        self.values.iter()
956    }
957
958    pub fn into_map(self) -> BTreeMap<String, Value> {
959        self.columns.iter().cloned().zip(self.values).collect()
960    }
961
962    /// Transitional boundary adapter. Core query providers should construct
963    /// compact rows directly instead of routing through this function.
964    pub fn from_map(values_by_name: BTreeMap<String, Value>) -> Self {
965        let (columns, values): (Vec<_>, Vec<_>) = values_by_name.into_iter().unzip();
966        Self::new(columns.into(), values)
967    }
968}
969
970impl From<BTreeMap<String, Value>> for CompactRow {
971    fn from(values: BTreeMap<String, Value>) -> Self {
972        Self::from_map(values)
973    }
974}
975
976/// Internal projection used to implement per-parent pagination for relation
977/// loads. Runtime relation attachment removes it before exposing child rows.
978pub const PARTITION_RANK_PROPERTY: &str = "__teaql_partition_rank";
979
980pub fn record_to_json_value(record: &Record) -> serde_json::Value {
981    serde_json::Value::Object(
982        record
983            .iter()
984            .map(|(key, value)| (key.clone(), value.to_json_value()))
985            .collect(),
986    )
987}
988
989pub fn compact_row_to_json_value(row: &CompactRow) -> serde_json::Value {
990    serde_json::Value::Object(
991        row.iter()
992            .map(|(key, value)| (key.clone(), value.to_json_value()))
993            .collect(),
994    )
995}
teaql_core/query.rs

teaql_core/
query.rs
