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

teaql_core/
query.rs
