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            kind: crate::TraceKind::Comment,
698            entity_type: self.entity.clone(),
699            entity_id: None,
700            comment: comment_str,
701        });
702        self
703    }
704
705    pub fn raw_sql(mut self, raw_sql: impl Into<String>) -> Self {
706        self.raw_sql = Some(raw_sql.into());
707        self
708    }
709
710    pub fn raw_sql_search_criteria(mut self, raw_sql: impl Into<String>) -> Self {
711        self.raw_sql_search_criteria.push(raw_sql.into());
712        self
713    }
714
715    pub fn object_group_by(
716        mut self,
717        property_name: impl Into<String>,
718        storage_field: impl Into<String>,
719        query: SelectQuery,
720    ) -> Self {
721        self.object_group_bys
722            .push(ObjectGroupBy::new(property_name, storage_field, query));
723        self
724    }
725
726    pub fn child_enhancement(mut self, query: SelectQuery) -> Self {
727        self.child_enhancements.push(query);
728        self
729    }
730
731    pub fn relation(mut self, name: impl Into<String>) -> Self {
732        self.relations.push(RelationLoad::new(name));
733        self
734    }
735
736    pub fn relation_query(mut self, name: impl Into<String>, query: SelectQuery) -> Self {
737        self.relations.push(RelationLoad::with_query(name, query));
738        self
739    }
740
741    pub fn limit(mut self, limit: u64) -> Self {
742        let slice = self.slice.get_or_insert(Slice {
743            limit: None,
744            offset: 0,
745        });
746        slice.limit = Some(limit);
747        self
748    }
749
750    /// Override the outer materialized-list ceiling. Most callers should keep 10,000.
751    pub fn hard_limit(mut self, hard_limit: u64) -> Self {
752        assert!(hard_limit > 0, "hard_limit must be positive");
753        self.hard_limit = hard_limit;
754        self
755    }
756
757    /// Apply and validate list-materialization limits. This is intentionally not
758    /// used by streaming execution.
759    pub fn prepare_for_list(mut self) -> Result<Self, String> {
760        self.apply_list_limit(self.hard_limit, true)?;
761        Ok(self)
762    }
763
764    fn apply_list_limit(&mut self, ceiling: u64, outer: bool) -> Result<(), String> {
765        let slice = self.slice.get_or_insert(Slice {
766            limit: None,
767            offset: 0,
768        });
769        match slice.limit {
770            Some(limit) if limit > ceiling => {
771                return Err(format!(
772                    "QUERY_HARD_LIMIT_EXCEEDED: requested limit {limit} exceeds hard limit {ceiling}"
773                ));
774            }
775            None => slice.limit = Some(ceiling),
776            _ => {}
777        }
778        for relation in &mut self.relations {
779            if let Some(query) = relation.query.as_mut() {
780                query.apply_list_limit(10_000, false)?;
781            }
782        }
783        for query in &mut self.child_enhancements {
784            query.apply_list_limit(10_000, false)?;
785        }
786        let _ = outer;
787        Ok(())
788    }
789
790    pub fn offset(mut self, offset: u64) -> Self {
791        let slice = self.slice.get_or_insert(Slice {
792            limit: None,
793            offset: 0,
794        });
795        slice.offset = offset;
796        self
797    }
798
799    pub fn page(self, offset: u64, limit: u64) -> Self {
800        self.offset(offset).limit(limit)
801    }
802
803    pub fn optimize_for_continuous_page_fetch(mut self) -> Self {
804        self.continuous_page_fetch = Some(ContinuousPageFetchOptions::new(
805            "default",
806            ContinuousPageFetchOptions::DEFAULT_TTL_SECONDS,
807        ));
808        self
809    }
810
811    pub fn optimize_for_continuous_page_fetch_with(
812        mut self,
813        namespace: impl Into<String>,
814        ttl_seconds: u64,
815    ) -> Self {
816        self.continuous_page_fetch = Some(ContinuousPageFetchOptions::new(namespace, ttl_seconds));
817        self
818    }
819
820    pub fn optimize_pagination_with_id_set(mut self) -> Self {
821        self.id_set_pagination = Some(IdSetPaginationOptions::new(
822            "default",
823            IdSetPaginationOptions::DEFAULT_TTL_SECONDS,
824            IdSetPaginationOptions::DEFAULT_MAX_IDS,
825        ));
826        self
827    }
828
829    pub fn optimize_pagination_with_id_set_config(
830        mut self,
831        namespace: impl Into<String>,
832        ttl_seconds: u64,
833        max_ids: u64,
834    ) -> Self {
835        self.id_set_pagination = Some(IdSetPaginationOptions::new(namespace, ttl_seconds, max_ids));
836        self
837    }
838
839    /// Scope pagination to each distinct value of `field`.
840    ///
841    /// Relation loading sets this automatically. Most application queries
842    /// should use a generated relation selector instead of calling this
843    /// method directly.
844    pub fn partition_by(mut self, field: impl Into<String>) -> Self {
845        self.partition_by = Some(field.into());
846        self
847    }
848
849    /// Override the provider's per-parent Top-N execution policy.
850    ///
851    /// `0` selects the window plan. A positive threshold selects bounded
852    /// probes only when the parent rows already in memory are at or below it.
853    pub fn top_n_probe_parent_threshold(mut self, threshold: usize) -> Self {
854        self.top_n_probe_parent_threshold = Some(threshold);
855        self
856    }
857
858    /// Enable streaming mode with the given chunk size.
859    /// When streaming, rows are fetched and enhanced in batches rather than all at once.
860    pub fn stream(mut self, chunk_size: usize) -> Self {
861        self.stream_config = Some(StreamConfig { chunk_size });
862        self
863    }
864
865    /// Enable streaming mode with default chunk size (1000).
866    pub fn stream_default(mut self) -> Self {
867        self.stream_config = Some(StreamConfig::default());
868        self
869    }
870}
871
872pub type Record = BTreeMap<String, Value>;
873
874/// A database result row whose column names are shared by the whole result set.
875///
876/// Providers use this representation for typed decoding so a 100-row result does
877/// not allocate 100 copies of every projected column name (or 100 B-trees).
878#[derive(Debug, Clone, PartialEq)]
879pub struct CompactRow {
880    columns: Arc<[String]>,
881    values: Vec<Value>,
882}
883
884impl CompactRow {
885    pub fn new(columns: Arc<[String]>, values: Vec<Value>) -> Self {
886        debug_assert_eq!(columns.len(), values.len());
887        Self { columns, values }
888    }
889
890    pub fn get(&self, name: &str) -> Option<&Value> {
891        self.columns
892            .iter()
893            .position(|column| column == name)
894            .and_then(|index| self.values.get(index))
895    }
896
897    pub fn shared_columns(&self) -> Arc<[String]> {
898        self.columns.clone()
899    }
900
901    pub fn get_mut(&mut self, name: &str) -> Option<&mut Value> {
902        self.columns
903            .iter()
904            .position(|column| column == name)
905            .and_then(|index| self.values.get_mut(index))
906    }
907
908    /// Adds or replaces a projected value. Column layouts remain shared until
909    /// an enhancement actually changes the shape of this row.
910    pub fn insert(&mut self, name: String, value: Value) -> Option<Value> {
911        if let Some(index) = self.columns.iter().position(|column| column == &name) {
912            return Some(std::mem::replace(&mut self.values[index], value));
913        }
914        let mut columns = self.columns.to_vec();
915        columns.push(name);
916        self.columns = columns.into();
917        self.values.push(value);
918        None
919    }
920
921    pub fn remove(&mut self, name: &str) -> Option<Value> {
922        let index = self.columns.iter().position(|column| column == name)?;
923        let mut columns = self.columns.to_vec();
924        columns.remove(index);
925        self.columns = columns.into();
926        Some(self.values.remove(index))
927    }
928
929    pub fn extend(&mut self, other: CompactRow) {
930        for (name, value) in other.columns.iter().cloned().zip(other.values) {
931            self.insert(name, value);
932        }
933    }
934
935    pub fn contains_key(&self, name: &str) -> bool {
936        self.columns.iter().any(|column| column == name)
937    }
938
939    pub fn len(&self) -> usize {
940        self.values.len()
941    }
942
943    pub fn is_empty(&self) -> bool {
944        self.values.is_empty()
945    }
946
947    pub fn iter(&self) -> impl Iterator<Item = (&String, &Value)> {
948        self.columns.iter().zip(self.values.iter())
949    }
950
951    pub fn keys(&self) -> impl Iterator<Item = &String> {
952        self.columns.iter()
953    }
954
955    pub fn values(&self) -> impl Iterator<Item = &Value> {
956        self.values.iter()
957    }
958
959    pub fn into_map(self) -> BTreeMap<String, Value> {
960        self.columns.iter().cloned().zip(self.values).collect()
961    }
962
963    /// Transitional boundary adapter. Core query providers should construct
964    /// compact rows directly instead of routing through this function.
965    pub fn from_map(values_by_name: BTreeMap<String, Value>) -> Self {
966        let (columns, values): (Vec<_>, Vec<_>) = values_by_name.into_iter().unzip();
967        Self::new(columns.into(), values)
968    }
969}
970
971impl From<BTreeMap<String, Value>> for CompactRow {
972    fn from(values: BTreeMap<String, Value>) -> Self {
973        Self::from_map(values)
974    }
975}
976
977/// Internal projection used to implement per-parent pagination for relation
978/// loads. Runtime relation attachment removes it before exposing child rows.
979pub const PARTITION_RANK_PROPERTY: &str = "__teaql_partition_rank";
980
981pub fn record_to_json_value(record: &Record) -> serde_json::Value {
982    serde_json::Value::Object(
983        record
984            .iter()
985            .map(|(key, value)| (key.clone(), value.to_json_value()))
986            .collect(),
987    )
988}
989
990pub fn compact_row_to_json_value(row: &CompactRow) -> serde_json::Value {
991    serde_json::Value::Object(
992        row.iter()
993            .map(|(key, value)| (key.clone(), value.to_json_value()))
994            .collect(),
995    )
996}
teaql_core/query.rs

teaql_core/
query.rs
