1#![deny(rust_2018_idioms)]
2#![deny(clippy::all)]
3
4mod util;
17
18pub mod graph;
19
20use ordered_float::OrderedFloat;
21use partiql_common::catalog::ObjectId;
22use partiql_value::BindingsName;
23use rust_decimal::Decimal as RustDecimal;
63use rustc_hash::FxHashMap;
64#[cfg(feature = "serde")]
65use serde::{Deserialize, Serialize};
66use std::fmt::{Debug, Display, Formatter};
67
68#[derive(Debug, Clone, Eq, PartialEq, Default)]
70#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
71pub struct LogicalPlan<T>
72where
73 T: Default + Debug,
74{
75 nodes: Vec<T>,
76 edges: Vec<(OpId, OpId, u8)>,
78}
79
80impl<T> LogicalPlan<T>
81where
82 T: Default + Debug,
83{
84 #[must_use]
86 pub fn new() -> Self {
87 Self::default()
88 }
89
90 pub fn add_operator(&mut self, op: T) -> OpId {
92 self.nodes.push(op);
93 OpId(self.operator_count())
94 }
95
96 #[inline]
98 pub fn add_flow(&mut self, src: OpId, dst: OpId) {
99 assert!(src.index() <= self.operator_count());
100 assert!(dst.index() <= self.operator_count());
101
102 self.edges.push((src, dst, 0));
103 }
104
105 #[inline]
107 pub fn add_flow_with_branch_num(&mut self, src: OpId, dst: OpId, branch_num: u8) {
108 assert!(src.index() <= self.operator_count());
109 assert!(dst.index() <= self.operator_count());
110
111 self.edges.push((src, dst, branch_num));
112 }
113
114 #[inline]
136 pub fn extend_with_flows(&mut self, flows: &[(OpId, OpId)]) {
137 flows.iter().for_each(|&(s, d)| self.add_flow(s, d));
138 }
139
140 #[inline]
141 pub fn merge_plan(&mut self, other: Self) {
142 let LogicalPlan { nodes, edges } = other;
143 let mut mapping = FxHashMap::default();
144 for (old_id, op) in nodes.into_iter().enumerate().map(|(i, n)| (OpId(i + 1), n)) {
145 let new_id = self.add_operator(op);
146 mapping.insert(old_id, new_id);
147 }
148
149 for (old1, old2, branch) in edges {
150 let new1 = mapping[&old1];
151 let new2 = mapping[&old2];
152 self.edges.push((new1, new2, branch));
153 }
154 }
155
156 #[inline]
158 #[must_use]
159 pub fn operator_count(&self) -> usize {
160 self.nodes.len()
161 }
162
163 #[must_use]
165 pub fn operators(&self) -> &Vec<T> {
166 &self.nodes
167 }
168
169 pub fn operators_by_id(&self) -> impl Iterator<Item = (OpId, &T)> {
171 self.nodes.iter().enumerate().map(|(i, n)| (OpId(i + 1), n))
172 }
173
174 #[must_use]
176 pub fn flows(&self) -> &Vec<(OpId, OpId, u8)> {
177 &self.edges
178 }
179
180 #[must_use]
181 pub fn operator(&self, id: OpId) -> Option<&T> {
182 self.nodes.get(id.0 - 1)
183 }
184
185 pub fn operator_as_mut(&mut self, id: OpId) -> Option<&mut T> {
186 self.nodes.get_mut(id.0 - 1)
187 }
188
189 }
191
192#[derive(Debug, Clone, Eq, PartialEq, Copy, Hash)]
194#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
195pub struct OpId(usize);
196
197impl OpId {
198 #[must_use]
200 pub fn index(&self) -> usize {
201 self.0
202 }
203}
204
205impl<T> Display for LogicalPlan<T>
206where
207 T: Default + Debug,
208{
209 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
210 let flows = self.flows();
211 writeln!(f, "LogicalPlan")?;
212 writeln!(f, "---")?;
213 for (s, d, _w) in flows {
214 let src_node = self.operator(*s).expect("Unable to get the src operator");
215 let dst_node = self.operator(*d).expect("Unable to get the dst operator");
216 writeln!(f, ">>> [{src_node:?}] -> [{dst_node:?}]")?;
217 }
218 writeln!(f)
219 }
220}
221
222#[derive(Debug, Clone, Default, Eq, PartialEq)]
225#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
226#[non_exhaustive]
227pub enum BindingsOp {
228 Scan(Scan),
229 Pivot(Pivot),
230 Unpivot(Unpivot),
231 Filter(Filter),
232 OrderBy(OrderBy),
233 LimitOffset(LimitOffset),
234 Join(Join),
235 BagOp(BagOp),
236 Project(Project),
237 ProjectAll(ProjectAllMode),
238 ProjectValue(ProjectValue),
239 ExprQuery(ExprQuery),
240 Distinct,
241 GroupBy(GroupBy),
242 Having(Having),
243 #[default]
244 Sink,
245}
246
247#[derive(Debug, Clone, Default, Eq, PartialEq)]
248#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
249#[non_exhaustive]
250pub enum ProjectAllMode {
251 #[default]
254 Unwrap,
255 PassThrough,
258}
259
260#[derive(Debug, Clone, Eq, PartialEq)]
262#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
263pub struct Scan {
264 pub expr: ValueExpr,
265 pub as_key: String,
266 pub at_key: Option<String>,
267}
268
269#[derive(Debug, Clone, Eq, PartialEq)]
274#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
275pub struct Pivot {
276 pub key: ValueExpr,
277 pub value: ValueExpr,
278}
279
280#[derive(Debug, Clone, Eq, PartialEq)]
282#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
283pub struct Unpivot {
284 pub expr: ValueExpr,
285 pub as_key: String,
286 pub at_key: Option<String>,
287}
288
289#[derive(Debug, Clone, Eq, PartialEq)]
291#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
292pub struct Filter {
293 pub expr: ValueExpr,
294}
295
296#[derive(Debug, Clone, Eq, PartialEq)]
298#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
299pub struct Having {
300 pub expr: ValueExpr,
301}
302
303#[derive(Debug, Clone, Eq, PartialEq)]
305#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
306pub struct OrderBy {
307 pub specs: Vec<SortSpec>,
308}
309
310#[derive(Clone, Debug, PartialEq, Eq)]
311#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
312#[non_exhaustive]
313pub enum SortSpecOrder {
314 Asc,
315 Desc,
316}
317
318#[derive(Clone, Debug, PartialEq, Eq)]
319#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
320#[non_exhaustive]
321pub enum SortSpecNullOrder {
322 First,
323 Last,
324}
325
326#[derive(Debug, Clone, Eq, PartialEq)]
328#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
329pub struct SortSpec {
330 pub expr: ValueExpr,
331 pub order: SortSpecOrder,
332 pub null_order: SortSpecNullOrder,
333}
334
335#[derive(Debug, Clone, Eq, PartialEq)]
337#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
338pub struct LimitOffset {
339 pub limit: Option<ValueExpr>,
340 pub offset: Option<ValueExpr>,
341}
342
343#[derive(Debug, Clone, Eq, PartialEq)]
345#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
346pub struct BagOp {
347 pub bag_op: BagOperator,
348 pub setq: SetQuantifier,
349}
350
351#[derive(Debug, Clone, Eq, PartialEq)]
353#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
354#[non_exhaustive]
355pub enum BagOperator {
356 Union,
357 Except,
358 Intersect,
359 OuterUnion,
360 OuterExcept,
361 OuterIntersect,
362}
363
364#[derive(Debug, Clone, Eq, PartialEq)]
367#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
368pub struct Join {
369 pub kind: JoinKind,
370 pub left: Box<BindingsOp>,
371 pub right: Box<BindingsOp>,
372 pub on: Option<ValueExpr>,
373}
374
375#[derive(Debug, Clone, Eq, PartialEq)]
377#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
378#[non_exhaustive]
379pub enum JoinKind {
380 Inner,
381 Left,
382 Right,
383 Full,
384 Cross,
385}
386
387#[derive(Debug, Clone, Eq, PartialEq)]
389#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
390pub struct AggregateExpression {
391 pub name: String,
392 pub expr: ValueExpr,
393 pub func: AggFunc,
394 pub setq: SetQuantifier,
395}
396
397#[derive(Debug, Clone, Eq, PartialEq)]
399#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
400#[non_exhaustive]
401pub enum AggFunc {
402 AggAvg,
405 AggCount,
407 AggMax,
409 AggMin,
411 AggSum,
413 AggAny,
415 AggEvery,
417}
418
419#[derive(Debug, Clone, Eq, PartialEq)]
421#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
422pub struct GroupBy {
423 pub strategy: GroupingStrategy,
424 pub exprs: FxHashMap<String, ValueExpr>,
425 pub aggregate_exprs: Vec<AggregateExpression>,
426 pub group_as_alias: Option<String>,
427}
428
429#[derive(Debug, Clone, Eq, PartialEq)]
431#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
432#[non_exhaustive]
433pub enum GroupingStrategy {
434 GroupFull,
435 GroupPartial,
436}
437
438#[derive(Debug, Clone, Eq, PartialEq)]
440#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
441pub struct Project {
442 pub exprs: Vec<(String, ValueExpr)>,
443}
444
445#[derive(Debug, Clone, Eq, PartialEq)]
448#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
449pub struct ProjectValue {
450 pub expr: ValueExpr,
451}
452
453#[derive(Debug, Clone, Eq, PartialEq)]
455#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
456pub struct ExprQuery {
457 pub expr: ValueExpr,
458}
459
460#[derive(Debug, Clone, Eq, PartialEq)]
463#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
464#[non_exhaustive]
465pub enum ValueExpr {
466 UnExpr(UnaryOp, Box<ValueExpr>),
467 BinaryExpr(BinaryOp, Box<ValueExpr>, Box<ValueExpr>),
468 Lit(Box<Lit>),
469 DynamicLookup(Box<Vec<ValueExpr>>),
470 Path(Box<ValueExpr>, Vec<PathComponent>),
471 VarRef(BindingsName<'static>, VarRefType),
472 TupleExpr(TupleExpr),
473 ListExpr(ListExpr),
474 BagExpr(BagExpr),
475 BetweenExpr(BetweenExpr),
476 PatternMatchExpr(PatternMatchExpr),
477 SubQueryExpr(SubQueryExpr),
478 SimpleCase(SimpleCase),
479 SearchedCase(SearchedCase),
480 IsTypeExpr(IsTypeExpr),
481 NullIfExpr(NullIfExpr),
482 CoalesceExpr(CoalesceExpr),
483 Call(CallExpr),
484 GraphMatch(Box<GraphMatchExpr>),
485}
486
487#[derive(Debug, Clone, Eq, PartialEq)]
489#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
490#[non_exhaustive]
491pub enum Lit {
492 Null,
493 Missing,
494 Int8(i8),
495 Int16(i16),
496 Int32(i32),
497 Int64(i64),
498 Decimal(RustDecimal),
499 Double(OrderedFloat<f64>),
500 Bool(bool),
501 String(String),
502 Variant(Vec<u8>, String), Struct(Vec<(String, Lit)>),
504 Bag(Vec<Lit>),
505 List(Vec<Lit>),
506}
507#[derive(Debug, Clone, Eq, PartialEq)]
510#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
511#[non_exhaustive]
512pub enum UnaryOp {
513 Pos,
514 Neg,
515 Not,
516}
517
518#[derive(Debug, Clone, Eq, PartialEq)]
521#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
522#[non_exhaustive]
523pub enum BinaryOp {
524 And,
525 Or,
526 Concat,
527 Eq,
528 Neq,
529 Gt,
530 Gteq,
531 Lt,
532 Lteq,
533
534 Add,
536 Sub,
537 Mul,
538 Div,
539 Mod,
540 Exp,
541
542 In,
543}
544
545#[derive(Debug, Clone, Eq, PartialEq)]
546#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
547#[non_exhaustive]
549pub enum PathComponent {
550 Key(BindingsName<'static>),
552 Index(i64),
554 KeyExpr(Box<ValueExpr>),
555 IndexExpr(Box<ValueExpr>),
556}
557
558#[derive(Clone, Debug, Default, Eq, PartialEq)]
560#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
561pub struct TupleExpr {
562 pub attrs: Vec<ValueExpr>,
563 pub values: Vec<ValueExpr>,
564}
565
566impl TupleExpr {
567 #[must_use]
569 pub fn new() -> Self {
570 Self::default()
571 }
572}
573
574#[derive(Clone, Debug, Default, Eq, PartialEq)]
576#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
577pub struct ListExpr {
578 pub elements: Vec<ValueExpr>,
579}
580
581impl ListExpr {
582 #[must_use]
584 pub fn new() -> Self {
585 Self::default()
586 }
587}
588
589#[derive(Clone, Debug, Default, Eq, PartialEq)]
591#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
592pub struct BagExpr {
593 pub elements: Vec<ValueExpr>,
594}
595
596impl BagExpr {
597 #[must_use]
599 pub fn new() -> Self {
600 Self::default()
601 }
602}
603
604#[derive(Debug, Clone, Eq, PartialEq)]
606#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
607pub struct BetweenExpr {
608 pub value: Box<ValueExpr>,
609 pub from: Box<ValueExpr>,
610 pub to: Box<ValueExpr>,
611}
612
613#[derive(Debug, Clone, Eq, PartialEq)]
615#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
616pub struct PatternMatchExpr {
617 pub value: Box<ValueExpr>,
618 pub pattern: Pattern,
619}
620
621#[derive(Debug, Clone, Eq, PartialEq)]
622#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
623#[non_exhaustive]
624pub enum Pattern {
625 Like(LikeMatch), LikeNonStringNonLiteral(LikeNonStringNonLiteralMatch),
627}
628
629#[derive(Debug, Clone, Eq, PartialEq)]
632#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
633pub struct LikeMatch {
634 pub pattern: String,
635 pub escape: String,
636}
637
638#[derive(Debug, Clone, Eq, PartialEq)]
641#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
642pub struct LikeNonStringNonLiteralMatch {
643 pub pattern: Box<ValueExpr>,
644 pub escape: Box<ValueExpr>,
645}
646
647#[derive(Debug, Clone, Eq, PartialEq)]
649#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
650pub struct GraphMatchExpr {
651 pub value: Box<ValueExpr>,
652 pub pattern: graph::PathPatternMatch,
653}
654#[derive(Debug, Clone, Eq, PartialEq)]
655#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
656pub struct GraphPathPattern {}
657
658#[derive(Debug, Clone, Eq, PartialEq)]
661#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
662pub struct SubQueryExpr {
663 pub plan: LogicalPlan<BindingsOp>,
664}
665
666#[derive(Debug, Clone, Eq, PartialEq)]
669#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
670pub struct SimpleCase {
671 pub expr: Box<ValueExpr>,
672 pub cases: Vec<(Box<ValueExpr>, Box<ValueExpr>)>,
673 pub default: Option<Box<ValueExpr>>,
674}
675
676#[derive(Debug, Clone, Eq, PartialEq)]
679#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
680pub struct SearchedCase {
681 pub cases: Vec<(Box<ValueExpr>, Box<ValueExpr>)>,
682 pub default: Option<Box<ValueExpr>>,
683}
684
685#[derive(Debug, Clone, Eq, PartialEq)]
687#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
688pub struct IsTypeExpr {
689 pub not: bool,
690 pub expr: Box<ValueExpr>,
691 pub is_type: Type,
692}
693
694#[derive(Clone, Debug, PartialEq, Eq)]
696#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
697#[non_exhaustive]
698pub enum Type {
699 NullType,
700 BooleanType,
701 Integer2Type,
702 Integer4Type,
703 Integer8Type,
704 DecimalType,
705 NumericType,
706 RealType,
707 DoublePrecisionType,
708 TimestampType,
709 CharacterType,
710 CharacterVaryingType,
711 MissingType,
712 StringType,
713 SymbolType,
714 BlobType,
715 ClobType,
716 DateType,
717 TimeType,
718 ZonedTimestampType,
719 StructType,
720 TupleType,
721 ListType,
722 SexpType,
723 BagType,
724 AnyType,
725 }
727
728#[derive(Debug, Clone, Eq, PartialEq)]
730#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
731pub struct NullIfExpr {
732 pub lhs: Box<ValueExpr>,
733 pub rhs: Box<ValueExpr>,
734}
735
736#[derive(Debug, Clone, Eq, PartialEq)]
739#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
740pub struct CoalesceExpr {
741 pub elements: Vec<ValueExpr>,
742}
743
744#[derive(Debug, Clone, Eq, PartialEq)]
746#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
747pub struct CallExpr {
748 pub name: CallName,
749 pub arguments: Vec<ValueExpr>,
750}
751
752#[derive(Debug, Clone, Eq, PartialEq)]
754#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
755#[non_exhaustive]
756pub enum CallName {
757 Lower,
758 Upper,
759 CharLength,
760 OctetLength,
761 BitLength,
762 LTrim,
763 BTrim,
764 RTrim,
765 Substring,
766 Position,
767 Overlay,
768 Exists,
769 Abs,
770 Mod,
771 Cardinality,
772 CollToScalar,
773 ExtractYear,
774 ExtractMonth,
775 ExtractDay,
776 ExtractHour,
777 ExtractMinute,
778 ExtractSecond,
779 ExtractTimezoneHour,
780 ExtractTimezoneMinute,
781 CollAvg(SetQuantifier),
782 CollCount(SetQuantifier),
783 CollMax(SetQuantifier),
784 CollMin(SetQuantifier),
785 CollSum(SetQuantifier),
786 CollAny(SetQuantifier),
787 CollEvery(SetQuantifier),
788 ByName(String),
789 ById(String, ObjectId, usize),
790}
791
792#[derive(Debug, Clone, Eq, PartialEq)]
824#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
825#[non_exhaustive]
826pub enum SetQuantifier {
827 All,
828 Distinct,
829}
830
831#[derive(Clone, Debug, PartialEq, Eq)]
833#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
834#[non_exhaustive]
835pub enum VarRefType {
836 Global,
838 Local,
840}
841
842#[cfg(test)]
843mod tests {
844 use super::*;
845
846 #[test]
847 fn test_plan() {
848 let mut p: LogicalPlan<BindingsOp> = LogicalPlan::new();
849 let a = p.add_operator(BindingsOp::OrderBy(OrderBy { specs: vec![] }));
850 let b = p.add_operator(BindingsOp::Sink);
851 let c = p.add_operator(BindingsOp::LimitOffset(LimitOffset {
852 limit: None,
853 offset: None,
854 }));
855 let d = p.add_operator(BindingsOp::GroupBy(GroupBy {
856 strategy: GroupingStrategy::GroupFull,
857 exprs: Default::default(),
858 aggregate_exprs: vec![],
859 group_as_alias: None,
860 }));
861 p.add_flow(a, b);
862 p.add_flow(a, c);
863 p.extend_with_flows(&[(c, d), (b, c)]);
864 assert_eq!(4, p.operators().len());
865 assert_eq!(4, p.flows().len());
866 }
867}