1use std::collections::HashMap;
2
3use pest::iterators::Pair;
4use prost::Message;
5use substrait::proto::aggregate_rel::Grouping;
6use substrait::proto::expression::literal::LiteralType;
7use substrait::proto::expression::{Literal, RexType, nested};
8use substrait::proto::extensions::AdvancedExtension;
9use substrait::proto::fetch_rel::{CountMode, OffsetMode};
10use substrait::proto::rel::RelType;
11use substrait::proto::rel_common::{Direct, Emit, EmitKind};
12use substrait::proto::sort_field::{SortDirection, SortKind};
13use substrait::proto::{
14 AggregateRel, Expression, FetchRel, FilterRel, JoinRel, NamedStruct, ProjectRel, ReadRel, Rel,
15 RelCommon, SortField, SortRel, Type, aggregate_rel, join_rel, read_rel, r#type,
16};
17
18use super::{MessageParseError, ParsePair, Rule, RuleIter, ScopedParsePair, unwrap_single_pair};
19use crate::extensions::any::Any;
20use crate::extensions::registry::ExtensionError;
21use crate::extensions::{AddendumKind, ExtensionArgs, ExtensionRegistry, SimpleExtensions};
22use crate::parser::errors::{ParseContext, ParseError};
23use crate::parser::expressions::{FieldIndex, Name};
24
25pub struct RelationParsingContext<'a> {
27 pub extensions: &'a SimpleExtensions,
28 pub registry: &'a ExtensionRegistry,
29 pub line_no: i64,
30 pub line: &'a str,
31}
32
33impl<'a> RelationParsingContext<'a> {
34 pub fn resolve_extension_detail(
36 &self,
37 extension_name: &str,
38 extension_args: &ExtensionArgs,
39 ) -> Result<Option<Any>, ParseError> {
40 let detail = self
41 .registry
42 .parse_extension(extension_name, extension_args);
43
44 match detail {
45 Ok(any) => Ok(Some(any)),
46 Err(ExtensionError::NotFound { .. }) => Err(ParseError::UnregisteredExtension {
47 name: extension_name.to_string(),
48 context: ParseContext::new(self.line_no, self.line.to_string()),
49 }),
50 Err(err) => Err(ParseError::ExtensionDetail(
51 ParseContext::new(self.line_no, self.line.to_string()),
52 err,
53 )),
54 }
55 }
56
57 pub(crate) fn resolve_adv_ext_detail(
60 &self,
61 kind: AddendumKind,
62 name: &str,
63 args: &ExtensionArgs,
64 ) -> Result<Any, ParseError> {
65 let result = match kind {
66 AddendumKind::Enhancement => self.registry.parse_enhancement(name, args),
67 AddendumKind::Optimization => self.registry.parse_optimization(name, args),
68 };
69 result.map_err(|err| match err {
70 ExtensionError::NotFound { .. } => ParseError::UnregisteredExtension {
71 name: name.to_string(),
72 context: ParseContext::new(self.line_no, self.line.to_string()),
73 },
74 err => ParseError::ExtensionDetail(
75 ParseContext::new(self.line_no, self.line.to_string()),
76 err,
77 ),
78 })
79 }
80}
81
82pub trait RelationParsePair: Sized {
84 fn rule() -> Rule;
85 fn message() -> &'static str;
86
87 fn parse_pair_with_context(
93 extensions: &SimpleExtensions,
94 pair: Pair<Rule>,
95 input_children: Vec<Box<Rel>>,
96 input_field_count: usize,
97 ) -> Result<(Self, usize), MessageParseError>;
98
99 fn into_rel(self, adv_ext: Option<AdvancedExtension>) -> Rel;
102}
103
104pub struct TableName(Vec<String>);
105
106impl ParsePair for TableName {
107 fn rule() -> Rule {
108 Rule::table_name
109 }
110
111 fn message() -> &'static str {
112 "TableName"
113 }
114
115 fn parse_pair(pair: Pair<Rule>) -> Self {
116 assert_eq!(pair.as_rule(), Self::rule());
117 let pairs = pair.into_inner();
118 let mut names = Vec::with_capacity(pairs.len());
119 let mut iter = RuleIter::from(pairs);
120 while let Some(name) = iter.parse_if_next::<Name>() {
121 names.push(name.0);
122 }
123 iter.done();
124 Self(names)
125 }
126}
127
128#[derive(Debug, Clone)]
129pub struct Column {
130 pub name: String,
131 pub typ: Type,
132}
133
134impl ScopedParsePair for Column {
135 fn rule() -> Rule {
136 Rule::named_column
137 }
138
139 fn message() -> &'static str {
140 "Column"
141 }
142
143 fn parse_pair(
144 extensions: &SimpleExtensions,
145 pair: Pair<Rule>,
146 ) -> Result<Self, MessageParseError> {
147 assert_eq!(pair.as_rule(), Self::rule());
148 let mut iter = RuleIter::from(pair.into_inner());
149 let name = iter.parse_next::<Name>().0;
150 let typ = iter.parse_next_scoped(extensions)?;
151 iter.done();
152 Ok(Self { name, typ })
153 }
154}
155
156pub struct NamedColumnList(Vec<Column>);
157
158impl ScopedParsePair for NamedColumnList {
159 fn rule() -> Rule {
160 Rule::named_column_list
161 }
162
163 fn message() -> &'static str {
164 "NamedColumnList"
165 }
166
167 fn parse_pair(
168 extensions: &SimpleExtensions,
169 pair: Pair<Rule>,
170 ) -> Result<Self, MessageParseError> {
171 assert_eq!(pair.as_rule(), Self::rule());
172 let mut columns = Vec::new();
173 for col in pair.into_inner() {
174 columns.push(Column::parse_pair(extensions, col)?);
175 }
176 Ok(Self(columns))
177 }
178}
179
180#[allow(clippy::vec_box)]
185pub(crate) fn expect_one_child(
186 message: &'static str,
187 pair: &Pair<Rule>,
188 mut input_children: Vec<Box<Rel>>,
189) -> Result<Box<Rel>, MessageParseError> {
190 match input_children.len() {
191 0 => Err(MessageParseError::invalid(
192 message,
193 pair.as_span(),
194 format!("{message} missing child"),
195 )),
196 1 => Ok(input_children.pop().unwrap()),
197 n => Err(MessageParseError::invalid(
198 message,
199 pair.as_span(),
200 format!("{message} should have 1 input child, got {n}"),
201 )),
202 }
203}
204
205fn parse_output_mapping(pair: Pair<Rule>) -> Vec<i32> {
208 assert_eq!(pair.as_rule(), Rule::reference_list);
209 pair.into_inner()
210 .map(|p| FieldIndex::parse_pair(p).0)
211 .collect()
212}
213
214fn make_emit(output_mapping: Vec<i32>, direct_output_count: usize) -> EmitKind {
217 let is_identity = output_mapping.len() == direct_output_count
218 && output_mapping
219 .iter()
220 .enumerate()
221 .all(|(i, &v)| v == i as i32);
222 if is_identity {
223 EmitKind::Direct(Direct {})
224 } else {
225 EmitKind::Emit(Emit { output_mapping })
226 }
227}
228
229fn parse_emit(reference_list: Pair<Rule>, direct_output_count: usize) -> (EmitKind, usize) {
231 let output_mapping = parse_output_mapping(reference_list);
232 let output_count = output_mapping.len();
233 let emit = make_emit(output_mapping, direct_output_count);
234 (emit, output_count)
235}
236
237pub struct ParsedNamedArgs<'a> {
243 map: HashMap<&'a str, Pair<'a, Rule>>,
244}
245
246impl<'a> ParsedNamedArgs<'a> {
247 pub fn new(
248 pairs: pest::iterators::Pairs<'a, Rule>,
249 rule: Rule,
250 ) -> Result<Self, MessageParseError> {
251 let mut map = HashMap::new();
252 for pair in pairs {
253 assert_eq!(pair.as_rule(), rule);
254 let mut inner = pair.clone().into_inner();
255 let name_pair = inner.next().unwrap();
256 let value_pair = inner.next().unwrap();
257 assert_eq!(inner.next(), None);
258 let name = name_pair.as_str();
259 if map.contains_key(name) {
260 return Err(MessageParseError::invalid(
261 "NamedArg",
262 name_pair.as_span(),
263 format!("Duplicate argument: {name}"),
264 ));
265 }
266 map.insert(name, value_pair);
267 }
268 Ok(Self { map })
269 }
270
271 pub fn pop(mut self, name: &str, rule: Rule) -> (Self, Option<Pair<'a, Rule>>) {
275 let pair = self.map.remove(name).inspect(|pair| {
276 assert_eq!(pair.as_rule(), rule, "Rule mismatch for argument {name}");
277 });
278 (self, pair)
279 }
280
281 pub fn done(self) -> Result<(), MessageParseError> {
283 if let Some((name, pair)) = self.map.iter().next() {
284 return Err(MessageParseError::invalid(
285 "NamedArgExtractor",
286 pair.as_span(),
288 format!("Unknown argument: {name}"),
289 ));
290 }
291 Ok(())
292 }
293}
294
295impl RelationParsePair for ReadRel {
296 fn rule() -> Rule {
297 Rule::read_relation
298 }
299
300 fn message() -> &'static str {
301 "ReadRel"
302 }
303
304 fn parse_pair_with_context(
305 extensions: &SimpleExtensions,
306 pair: Pair<Rule>,
307 input_children: Vec<Box<Rel>>,
308 input_field_count: usize,
309 ) -> Result<(Self, usize), MessageParseError> {
310 assert_eq!(pair.as_rule(), Self::rule());
311 if !input_children.is_empty() {
313 return Err(MessageParseError::invalid(
314 Self::message(),
315 pair.as_span(),
316 "ReadRel should have no input children",
317 ));
318 }
319 if input_field_count != 0 {
320 return Err(MessageParseError::invalid(
321 "ReadRel",
322 pair.as_span(),
323 "ReadRel should have 0 input fields",
324 ));
325 }
326
327 let mut iter = RuleIter::from(pair.into_inner());
328 let table = iter.parse_next::<TableName>().0;
329 let columns = iter.parse_next_scoped::<NamedColumnList>(extensions)?.0;
330 iter.done();
331
332 let output_count = columns.len();
333 Ok((
334 ReadRel {
335 base_schema: Some(build_named_struct(columns)),
336 read_type: Some(read_rel::ReadType::NamedTable(read_rel::NamedTable {
337 names: table,
338 advanced_extension: None,
339 })),
340 ..Default::default()
341 },
342 output_count,
343 ))
344 }
345
346 fn into_rel(mut self, adv_ext: Option<AdvancedExtension>) -> Rel {
347 self.advanced_extension = adv_ext;
348 Rel {
349 rel_type: Some(RelType::Read(Box::new(self))),
350 }
351 }
352}
353
354pub(crate) struct VirtualReadRel(ReadRel);
358
359impl RelationParsePair for VirtualReadRel {
360 fn rule() -> Rule {
361 Rule::virtual_read_relation
362 }
363
364 fn message() -> &'static str {
365 "VirtualReadRel"
366 }
367
368 fn parse_pair_with_context(
369 extensions: &SimpleExtensions,
370 pair: Pair<Rule>,
371 input_children: Vec<Box<Rel>>,
372 _input_field_count: usize,
373 ) -> Result<(Self, usize), MessageParseError> {
374 assert_eq!(pair.as_rule(), Self::rule());
375 if !input_children.is_empty() {
376 return Err(MessageParseError::invalid(
377 Self::message(),
378 pair.as_span(),
379 "Read:Virtual should have no input children",
380 ));
381 }
382
383 let mut iter = RuleIter::from(pair.into_inner());
384 let args_pair = iter.pop(Rule::virtual_read_args);
385 let columns_pair = iter.pop(Rule::named_column_list);
386 iter.done();
387
388 let rows = parse_virtual_read_args(extensions, args_pair)?;
389 let columns = NamedColumnList::parse_pair(extensions, columns_pair)?.0;
390
391 let output_count = columns.len();
396 Ok((
397 VirtualReadRel(ReadRel {
398 base_schema: Some(build_named_struct(columns)),
399 read_type: Some(read_rel::ReadType::VirtualTable(read_rel::VirtualTable {
400 expressions: rows,
401 ..Default::default()
402 })),
403 ..Default::default()
404 }),
405 output_count,
406 ))
407 }
408
409 fn into_rel(mut self, adv_ext: Option<AdvancedExtension>) -> Rel {
410 self.0.advanced_extension = adv_ext;
411 Rel {
412 rel_type: Some(RelType::Read(Box::new(self.0))),
413 }
414 }
415}
416
417fn build_named_struct(columns: Vec<Column>) -> NamedStruct {
419 let (names, types): (Vec<_>, Vec<_>) = columns.into_iter().map(|c| (c.name, c.typ)).unzip();
420 NamedStruct {
421 names,
422 r#struct: Some(r#type::Struct {
423 types,
424 type_variation_reference: 0,
425 nullability: r#type::Nullability::Required as i32,
426 }),
427 }
428}
429
430fn parse_virtual_read_args(
432 extensions: &SimpleExtensions,
433 pair: Pair<Rule>,
434) -> Result<Vec<nested::Struct>, MessageParseError> {
435 assert_eq!(pair.as_rule(), Rule::virtual_read_args);
436 let inner = unwrap_single_pair(pair);
437 match inner.as_rule() {
438 Rule::empty => Ok(vec![]),
439 Rule::virtual_row_list => inner
440 .into_inner()
441 .map(|row| parse_virtual_row(extensions, row))
442 .collect(),
443 _ => unreachable!(
444 "Unexpected rule in virtual_read_args: {:?}",
445 inner.as_rule()
446 ),
447 }
448}
449
450fn parse_virtual_row(
452 extensions: &SimpleExtensions,
453 pair: Pair<Rule>,
454) -> Result<nested::Struct, MessageParseError> {
455 assert_eq!(pair.as_rule(), Rule::virtual_row);
456 let fields = match pair.into_inner().next() {
457 Some(expression_list) => {
458 assert_eq!(expression_list.as_rule(), Rule::expression_list);
459 parse_expression_list(extensions, expression_list)?
460 }
461 None => vec![],
463 };
464 Ok(nested::Struct { fields })
465}
466
467impl RelationParsePair for FilterRel {
468 fn rule() -> Rule {
469 Rule::filter_relation
470 }
471
472 fn message() -> &'static str {
473 "FilterRel"
474 }
475
476 fn into_rel(mut self, adv_ext: Option<AdvancedExtension>) -> Rel {
477 self.advanced_extension = adv_ext;
478 Rel {
479 rel_type: Some(RelType::Filter(Box::new(self))),
480 }
481 }
482
483 fn parse_pair_with_context(
484 extensions: &SimpleExtensions,
485 pair: Pair<Rule>,
486 input_children: Vec<Box<Rel>>,
487 input_field_count: usize,
488 ) -> Result<(Self, usize), MessageParseError> {
489 assert_eq!(pair.as_rule(), Self::rule());
490 let input = expect_one_child(Self::message(), &pair, input_children)?;
491 let mut iter = RuleIter::from(pair.into_inner());
492 let condition = iter.parse_next_scoped::<Expression>(extensions)?;
493 let references_pair = iter.pop(Rule::reference_list);
495 iter.done();
496
497 let (emit, output_count) = parse_emit(references_pair, input_field_count);
498 let common = RelCommon {
499 emit_kind: Some(emit),
500 ..Default::default()
501 };
502
503 Ok((
504 FilterRel {
505 input: Some(input),
506 condition: Some(Box::new(condition)),
507 common: Some(common),
508 advanced_extension: None,
509 },
510 output_count,
511 ))
512 }
513}
514
515impl RelationParsePair for ProjectRel {
516 fn rule() -> Rule {
517 Rule::project_relation
518 }
519
520 fn message() -> &'static str {
521 "ProjectRel"
522 }
523
524 fn into_rel(mut self, adv_ext: Option<AdvancedExtension>) -> Rel {
525 self.advanced_extension = adv_ext;
526 Rel {
527 rel_type: Some(RelType::Project(Box::new(self))),
528 }
529 }
530
531 fn parse_pair_with_context(
532 extensions: &SimpleExtensions,
533 pair: Pair<Rule>,
534 input_children: Vec<Box<Rel>>,
535 input_field_count: usize,
536 ) -> Result<(Self, usize), MessageParseError> {
537 assert_eq!(pair.as_rule(), Self::rule());
538 let input = expect_one_child(Self::message(), &pair, input_children)?;
539
540 let arguments_pair = unwrap_single_pair(pair);
541
542 let mut expressions = Vec::new();
543 let mut output_mapping = Vec::new();
544
545 for arg in arguments_pair.into_inner() {
546 let inner_arg = unwrap_single_pair(arg);
547 match inner_arg.as_rule() {
548 Rule::reference => {
549 let field_index = FieldIndex::parse_pair(inner_arg);
550 output_mapping.push(field_index.0);
551 }
552 Rule::expression => {
553 let expr = Expression::parse_pair(extensions, inner_arg)?;
554 expressions.push(expr);
555 output_mapping.push(input_field_count as i32 + (expressions.len() as i32 - 1));
557 }
558 _ => panic!("Unexpected inner argument rule: {:?}", inner_arg.as_rule()),
559 }
560 }
561
562 let output_count = output_mapping.len();
563 let direct_count = input_field_count + expressions.len();
564 let emit = make_emit(output_mapping, direct_count);
565 let common = RelCommon {
566 emit_kind: Some(emit),
567 ..Default::default()
568 };
569
570 Ok((
571 ProjectRel {
572 input: Some(input),
573 expressions,
574 common: Some(common),
575 advanced_extension: None,
576 },
577 output_count,
578 ))
579 }
580}
581
582impl RelationParsePair for AggregateRel {
583 fn rule() -> Rule {
584 Rule::aggregate_relation
585 }
586
587 fn message() -> &'static str {
588 "AggregateRel"
589 }
590
591 fn into_rel(mut self, adv_ext: Option<AdvancedExtension>) -> Rel {
592 self.advanced_extension = adv_ext;
593 Rel {
594 rel_type: Some(RelType::Aggregate(Box::new(self))),
595 }
596 }
597
598 fn parse_pair_with_context(
599 extensions: &SimpleExtensions,
600 pair: Pair<Rule>,
601 input_children: Vec<Box<Rel>>,
602 _input_field_count: usize,
605 ) -> Result<(Self, usize), MessageParseError> {
606 assert_eq!(pair.as_rule(), Self::rule());
607 let input = expect_one_child(Self::message(), &pair, input_children)?;
608 let mut iter = RuleIter::from(pair.into_inner());
609 let group_by_pair = iter.pop(Rule::aggregate_group_by);
610 let output_pair = iter.pop(Rule::aggregate_output);
611 iter.done();
612
613 let inner = group_by_pair
614 .into_inner()
615 .next()
616 .expect("aggregate_group_by must have one inner item");
617
618 let grouping_sets = parse_grouping_sets(extensions, inner)?;
619 let (groupings, grouping_expressions) = build_grouping_fields(&grouping_sets);
620
621 let (measures, output_mapping) =
622 parse_aggregate_measures(extensions, output_pair, &grouping_expressions)?;
623
624 let output_count = output_mapping.len();
625 let direct_count = grouping_expressions.len() + measures.len();
626 let emit = make_emit(output_mapping, direct_count);
627 let common = RelCommon {
628 emit_kind: Some(emit),
629 ..Default::default()
630 };
631
632 Ok((
633 AggregateRel {
634 input: Some(input),
635 grouping_expressions,
636 groupings,
637 measures,
638 common: Some(common),
639 advanced_extension: None,
640 },
641 output_count,
642 ))
643 }
644}
645
646fn parse_aggregate_measures(
651 extensions: &SimpleExtensions,
652 output_pair: Pair<'_, Rule>,
653 grouping_expressions: &[Expression],
654) -> Result<(Vec<aggregate_rel::Measure>, Vec<i32>), MessageParseError> {
655 assert_eq!(output_pair.as_rule(), Rule::aggregate_output);
656 let mut measures = Vec::new();
657 let mut output_mapping = Vec::new();
658
659 for aggregate_output_item in output_pair.into_inner() {
660 let inner_item = unwrap_single_pair(aggregate_output_item);
661 match inner_item.as_rule() {
662 Rule::reference => {
663 let field_index = FieldIndex::parse_pair(inner_item);
664 output_mapping.push(field_index.0);
665 }
666 Rule::aggregate_measure => {
667 let measure = aggregate_rel::Measure::parse_pair(extensions, inner_item)?;
668 output_mapping.push(grouping_expressions.len() as i32 + measures.len() as i32);
669 measures.push(measure);
670 }
671 _ => panic!(
672 "Unexpected inner output item rule: {:?}",
673 inner_item.as_rule()
674 ),
675 }
676 }
677
678 Ok((measures, output_mapping))
679}
680
681fn parse_grouping_sets(
690 extensions: &SimpleExtensions,
691 inner: Pair<'_, Rule>,
692) -> Result<Vec<Vec<Expression>>, MessageParseError> {
693 assert!(
694 matches!(
695 inner.as_rule(),
696 Rule::expression_list | Rule::grouping_set_list
697 ),
698 "Expected expression_list or grouping_set_list, got {:?}",
699 inner.as_rule()
700 );
701 match inner.as_rule() {
702 Rule::expression_list => Ok(vec![parse_expression_list(extensions, inner)?]),
703 Rule::grouping_set_list => inner
704 .into_inner()
705 .map(|pair| parse_grouping_set(extensions, pair))
706 .collect(),
707 _ => unreachable!(
708 "Unexpected rule in aggregate_group_by: {:?}",
709 inner.as_rule()
710 ),
711 }
712}
713
714fn parse_grouping_set(
718 extensions: &SimpleExtensions,
719 pair: Pair<'_, Rule>,
720) -> Result<Vec<Expression>, MessageParseError> {
721 assert_eq!(pair.as_rule(), Rule::grouping_set);
722 let inner = pair
723 .into_inner()
724 .next()
725 .expect("grouping_set must have one inner item");
726 match inner.as_rule() {
727 Rule::empty => Ok(vec![]),
728 Rule::expression_list => parse_expression_list(extensions, inner),
729 _ => unreachable!("Unexpected item in grouping_set: {:?}", inner.as_rule()),
730 }
731}
732
733pub(crate) fn parse_expression_list(
735 extensions: &SimpleExtensions,
736 pair: Pair<'_, Rule>,
737) -> Result<Vec<Expression>, MessageParseError> {
738 pair.into_inner()
739 .map(|expr_pair| Expression::parse_pair(extensions, expr_pair))
740 .collect()
741}
742
743fn build_grouping_fields(expression_sets: &[Vec<Expression>]) -> (Vec<Grouping>, Vec<Expression>) {
747 let mut expressions: Vec<Expression> = Vec::new();
748 let mut seen: HashMap<Vec<u8>, u32> = HashMap::new();
749
750 let groupings = expression_sets
751 .iter()
752 .map(|set| {
753 let expression_references = set
754 .iter()
755 .map(|exp| {
756 let key = exp.encode_to_vec();
760 let next_idx = expressions.len() as u32;
761 *seen.entry(key).or_insert_with(|| {
762 expressions.push(exp.clone());
763 next_idx
764 })
765 })
766 .collect();
767 Grouping {
768 expression_references,
769 #[allow(deprecated)]
770 grouping_expressions: vec![],
771 }
772 })
773 .collect();
774
775 (groupings, expressions)
776}
777
778impl ScopedParsePair for SortField {
779 fn rule() -> Rule {
780 Rule::sort_field
781 }
782
783 fn message() -> &'static str {
784 "SortField"
785 }
786
787 fn parse_pair(
788 _extensions: &SimpleExtensions,
789 pair: Pair<Rule>,
790 ) -> Result<Self, MessageParseError> {
791 assert_eq!(pair.as_rule(), Self::rule());
792 let mut iter = RuleIter::from(pair.into_inner());
793 let reference_pair = iter.pop(Rule::reference);
794 let field_index = FieldIndex::parse_pair(reference_pair);
795 let direction_pair = iter.pop(Rule::sort_direction);
796 let direction = match direction_pair.as_str().trim_start_matches('&') {
800 "AscNullsFirst" => SortDirection::AscNullsFirst,
801 "AscNullsLast" => SortDirection::AscNullsLast,
802 "DescNullsFirst" => SortDirection::DescNullsFirst,
803 "DescNullsLast" => SortDirection::DescNullsLast,
804 other => {
805 return Err(MessageParseError::invalid(
806 "SortDirection",
807 direction_pair.as_span(),
808 format!("Unknown sort direction: {other}"),
809 ));
810 }
811 };
812 iter.done();
813 Ok(SortField {
814 expr: Some(Expression {
815 rex_type: Some(substrait::proto::expression::RexType::Selection(Box::new(
816 field_index.to_field_reference(),
817 ))),
818 }),
819 sort_kind: Some(SortKind::Direction(direction as i32)),
821 })
822 }
823}
824
825impl RelationParsePair for SortRel {
826 fn rule() -> Rule {
827 Rule::sort_relation
828 }
829
830 fn message() -> &'static str {
831 "SortRel"
832 }
833
834 fn into_rel(mut self, adv_ext: Option<AdvancedExtension>) -> Rel {
835 self.advanced_extension = adv_ext;
836 Rel {
837 rel_type: Some(RelType::Sort(Box::new(self))),
838 }
839 }
840
841 fn parse_pair_with_context(
842 extensions: &SimpleExtensions,
843 pair: Pair<Rule>,
844 input_children: Vec<Box<Rel>>,
845 input_field_count: usize,
846 ) -> Result<(Self, usize), MessageParseError> {
847 assert_eq!(pair.as_rule(), Self::rule());
848 let input = expect_one_child(Self::message(), &pair, input_children)?;
849 let mut iter = RuleIter::from(pair.into_inner());
850 let sort_field_list_pair = iter.pop(Rule::sort_field_list);
851 let reference_list_pair = iter.pop(Rule::reference_list);
852 let mut sorts = Vec::new();
853 for sort_field_pair in sort_field_list_pair.into_inner() {
854 let sort_field = SortField::parse_pair(extensions, sort_field_pair)?;
855 sorts.push(sort_field);
856 }
857 let (emit, output_count) = parse_emit(reference_list_pair, input_field_count);
858 let common = RelCommon {
859 emit_kind: Some(emit),
860 ..Default::default()
861 };
862 iter.done();
863 Ok((
864 SortRel {
865 input: Some(input),
866 sorts,
867 common: Some(common),
868 advanced_extension: None,
869 },
870 output_count,
871 ))
872 }
873}
874
875impl ScopedParsePair for CountMode {
876 fn rule() -> Rule {
877 Rule::fetch_value
878 }
879 fn message() -> &'static str {
880 "CountMode"
881 }
882 fn parse_pair(
883 extensions: &SimpleExtensions,
884 pair: Pair<Rule>,
885 ) -> Result<Self, MessageParseError> {
886 assert_eq!(pair.as_rule(), Self::rule());
887 let mut arg_inner = RuleIter::from(pair.into_inner());
888 let value_pair = if let Some(int_pair) = arg_inner.try_pop(Rule::integer) {
889 int_pair
890 } else {
891 arg_inner.pop(Rule::expression)
892 };
893 match value_pair.as_rule() {
894 Rule::integer => {
895 let value = value_pair.as_str().parse::<i64>().map_err(|e| {
896 MessageParseError::invalid(
897 Self::message(),
898 value_pair.as_span(),
899 format!("Invalid integer: {e}"),
900 )
901 })?;
902 if value < 0 {
903 return Err(MessageParseError::invalid(
904 Self::message(),
905 value_pair.as_span(),
906 format!("Fetch limit must be non-negative, got: {value}"),
907 ));
908 }
909 Ok(CountMode::CountExpr(i64_literal_expr(value)))
910 }
911 Rule::expression => {
912 let expr = Expression::parse_pair(extensions, value_pair)?;
913 Ok(CountMode::CountExpr(Box::new(expr)))
914 }
915 _ => Err(MessageParseError::invalid(
916 Self::message(),
917 value_pair.as_span(),
918 format!("Unexpected rule for CountMode: {:?}", value_pair.as_rule()),
919 )),
920 }
921 }
922}
923
924fn i64_literal_expr(value: i64) -> Box<Expression> {
925 Box::new(Expression {
926 rex_type: Some(RexType::Literal(Literal {
927 nullable: false,
928 type_variation_reference: 0,
929 literal_type: Some(LiteralType::I64(value)),
930 })),
931 })
932}
933
934impl ScopedParsePair for OffsetMode {
935 fn rule() -> Rule {
936 Rule::fetch_value
937 }
938 fn message() -> &'static str {
939 "OffsetMode"
940 }
941 fn parse_pair(
942 extensions: &SimpleExtensions,
943 pair: Pair<Rule>,
944 ) -> Result<Self, MessageParseError> {
945 assert_eq!(pair.as_rule(), Self::rule());
946 let mut arg_inner = RuleIter::from(pair.into_inner());
947 let value_pair = if let Some(int_pair) = arg_inner.try_pop(Rule::integer) {
948 int_pair
949 } else {
950 arg_inner.pop(Rule::expression)
951 };
952 match value_pair.as_rule() {
953 Rule::integer => {
954 let value = value_pair.as_str().parse::<i64>().map_err(|e| {
955 MessageParseError::invalid(
956 Self::message(),
957 value_pair.as_span(),
958 format!("Invalid integer: {e}"),
959 )
960 })?;
961 if value < 0 {
962 return Err(MessageParseError::invalid(
963 Self::message(),
964 value_pair.as_span(),
965 format!("Fetch offset must be non-negative, got: {value}"),
966 ));
967 }
968 Ok(OffsetMode::OffsetExpr(i64_literal_expr(value)))
969 }
970 Rule::expression => {
971 let expr = Expression::parse_pair(extensions, value_pair)?;
972 Ok(OffsetMode::OffsetExpr(Box::new(expr)))
973 }
974 _ => Err(MessageParseError::invalid(
975 Self::message(),
976 value_pair.as_span(),
977 format!("Unexpected rule for OffsetMode: {:?}", value_pair.as_rule()),
978 )),
979 }
980 }
981}
982
983impl RelationParsePair for FetchRel {
984 fn rule() -> Rule {
985 Rule::fetch_relation
986 }
987
988 fn message() -> &'static str {
989 "FetchRel"
990 }
991
992 fn into_rel(mut self, adv_ext: Option<AdvancedExtension>) -> Rel {
993 self.advanced_extension = adv_ext;
994 Rel {
995 rel_type: Some(RelType::Fetch(Box::new(self))),
996 }
997 }
998
999 fn parse_pair_with_context(
1000 extensions: &SimpleExtensions,
1001 pair: Pair<Rule>,
1002 input_children: Vec<Box<Rel>>,
1003 input_field_count: usize,
1004 ) -> Result<(Self, usize), MessageParseError> {
1005 assert_eq!(pair.as_rule(), Self::rule());
1006 let input = expect_one_child(Self::message(), &pair, input_children)?;
1007 let mut iter = RuleIter::from(pair.into_inner());
1008
1009 let (limit_pair, offset_pair) = match iter.try_pop(Rule::fetch_named_arg_list) {
1013 None => {
1014 iter.pop(Rule::empty);
1015 (None, None)
1016 }
1017 Some(fetch_args_pair) => {
1018 let extractor =
1019 ParsedNamedArgs::new(fetch_args_pair.into_inner(), Rule::fetch_named_arg)?;
1020 let (extractor, limit_pair) = extractor.pop("limit", Rule::fetch_value);
1021 let (extractor, offset_pair) = extractor.pop("offset", Rule::fetch_value);
1022 extractor.done()?;
1023 (limit_pair, offset_pair)
1024 }
1025 };
1026
1027 let reference_list_pair = iter.pop(Rule::reference_list);
1028 let (emit, output_count) = parse_emit(reference_list_pair, input_field_count);
1029 let common = RelCommon {
1030 emit_kind: Some(emit),
1031 ..Default::default()
1032 };
1033 iter.done();
1034
1035 let count_mode = limit_pair
1036 .map(|pair| CountMode::parse_pair(extensions, pair))
1037 .transpose()?;
1038 let offset_mode = offset_pair
1039 .map(|pair| OffsetMode::parse_pair(extensions, pair))
1040 .transpose()?;
1041 Ok((
1042 FetchRel {
1043 input: Some(input),
1044 common: Some(common),
1045 advanced_extension: None,
1046 offset_mode,
1047 count_mode,
1048 },
1049 output_count,
1050 ))
1051 }
1052}
1053
1054impl ParsePair for join_rel::JoinType {
1055 fn rule() -> Rule {
1056 Rule::join_type
1057 }
1058
1059 fn message() -> &'static str {
1060 "JoinType"
1061 }
1062
1063 fn parse_pair(pair: Pair<Rule>) -> Self {
1064 assert_eq!(pair.as_rule(), Self::rule());
1065 let join_type_str = pair.as_str().trim_start_matches('&');
1066 match join_type_str {
1067 "Inner" => join_rel::JoinType::Inner,
1068 "Left" => join_rel::JoinType::Left,
1069 "Right" => join_rel::JoinType::Right,
1070 "Outer" => join_rel::JoinType::Outer,
1071 "LeftSemi" => join_rel::JoinType::LeftSemi,
1072 "RightSemi" => join_rel::JoinType::RightSemi,
1073 "LeftAnti" => join_rel::JoinType::LeftAnti,
1074 "RightAnti" => join_rel::JoinType::RightAnti,
1075 "LeftSingle" => join_rel::JoinType::LeftSingle,
1076 "RightSingle" => join_rel::JoinType::RightSingle,
1077 "LeftMark" => join_rel::JoinType::LeftMark,
1078 "RightMark" => join_rel::JoinType::RightMark,
1079 _ => panic!("Unknown join type: {join_type_str} (this should be caught by grammar)"),
1080 }
1081 }
1082}
1083
1084impl RelationParsePair for JoinRel {
1085 fn rule() -> Rule {
1086 Rule::join_relation
1087 }
1088
1089 fn message() -> &'static str {
1090 "JoinRel"
1091 }
1092
1093 fn into_rel(mut self, adv_ext: Option<AdvancedExtension>) -> Rel {
1094 self.advanced_extension = adv_ext;
1095 Rel {
1096 rel_type: Some(RelType::Join(Box::new(self))),
1097 }
1098 }
1099
1100 fn parse_pair_with_context(
1101 extensions: &SimpleExtensions,
1102 pair: Pair<Rule>,
1103 input_children: Vec<Box<Rel>>,
1104 input_field_count: usize,
1105 ) -> Result<(Self, usize), MessageParseError> {
1106 assert_eq!(pair.as_rule(), Self::rule());
1107
1108 if input_children.len() != 2 {
1109 return Err(MessageParseError::invalid(
1110 Self::message(),
1111 pair.as_span(),
1112 format!(
1113 "JoinRel should have exactly 2 input children, got {}",
1114 input_children.len()
1115 ),
1116 ));
1117 }
1118
1119 let mut children_iter = input_children.into_iter();
1120 let left = children_iter.next().unwrap();
1121 let right = children_iter.next().unwrap();
1122
1123 let mut iter = RuleIter::from(pair.into_inner());
1124 let join_type = iter.parse_next::<join_rel::JoinType>();
1125 let condition = iter.parse_next_scoped::<Expression>(extensions)?;
1126 let reference_list_pair = iter.pop(Rule::reference_list);
1127 iter.done();
1128
1129 let (emit, output_count) = parse_emit(reference_list_pair, input_field_count);
1133 let common = RelCommon {
1134 emit_kind: Some(emit),
1135 ..Default::default()
1136 };
1137
1138 Ok((
1139 JoinRel {
1140 common: Some(common),
1141 left: Some(left),
1142 right: Some(right),
1143 expression: Some(Box::new(condition)),
1144 post_join_filter: None, r#type: join_type as i32,
1146 advanced_extension: None,
1147 },
1148 output_count,
1149 ))
1150 }
1151}
1152
1153#[cfg(test)]
1154mod tests {
1155 use pest::Parser;
1156
1157 use super::*;
1158 use crate::fixtures::TestContext;
1159 use crate::parser::{ExpressionParser, Rule};
1160
1161 #[test]
1162 fn test_parse_relation() {
1163 }
1165
1166 #[test]
1167 fn test_parse_read_relation() {
1168 let extensions = SimpleExtensions::default();
1169 let read = ReadRel::parse_pair_with_context(
1170 &extensions,
1171 parse_exact(Rule::read_relation, "Read[ab.cd.ef => a:i32, b:string?]"),
1172 vec![],
1173 0,
1174 )
1175 .unwrap()
1176 .0;
1177 let names = match &read.read_type {
1178 Some(read_rel::ReadType::NamedTable(table)) => &table.names,
1179 _ => panic!("Expected NamedTable"),
1180 };
1181 assert_eq!(names, &["ab", "cd", "ef"]);
1182 let columns = &read
1183 .base_schema
1184 .as_ref()
1185 .unwrap()
1186 .r#struct
1187 .as_ref()
1188 .unwrap()
1189 .types;
1190 assert_eq!(columns.len(), 2);
1191 }
1192
1193 fn example_read_relation() -> ReadRel {
1195 let extensions = SimpleExtensions::default();
1196 ReadRel::parse_pair_with_context(
1197 &extensions,
1198 parse_exact(
1199 Rule::read_relation,
1200 "Read[ab.cd.ef => a:i32, b:string?, c:i64]",
1201 ),
1202 vec![],
1203 0,
1204 )
1205 .unwrap()
1206 .0
1207 }
1208
1209 #[test]
1210 fn test_parse_filter_relation() {
1211 let extensions = SimpleExtensions::default();
1212 let filter = FilterRel::parse_pair_with_context(
1213 &extensions,
1214 parse_exact(Rule::filter_relation, "Filter[$1 => $0, $1, $2]"),
1215 vec![Box::new(example_read_relation().into_rel(None))],
1216 3,
1217 )
1218 .unwrap()
1219 .0;
1220 let emit_kind = filter.common.as_ref().unwrap().emit_kind.as_ref().unwrap();
1222 assert!(
1223 matches!(emit_kind, EmitKind::Direct(_)),
1224 "Expected Direct for identity emit, got {emit_kind:?}"
1225 );
1226 }
1227
1228 #[test]
1229 fn test_parse_project_relation() {
1230 let extensions = SimpleExtensions::default();
1231 let project = ProjectRel::parse_pair_with_context(
1232 &extensions,
1233 parse_exact(Rule::project_relation, "Project[$0, $1, 42]"),
1234 vec![Box::new(example_read_relation().into_rel(None))],
1235 3,
1236 )
1237 .unwrap()
1238 .0;
1239
1240 assert_eq!(project.expressions.len(), 1);
1242
1243 let emit_kind = &project.common.as_ref().unwrap().emit_kind.as_ref().unwrap();
1244 let emit = match emit_kind {
1245 EmitKind::Emit(emit) => &emit.output_mapping,
1246 _ => panic!("Expected EmitKind::Emit, got {emit_kind:?}"),
1247 };
1248 assert_eq!(emit, &[0, 1, 3]);
1250 }
1251
1252 #[test]
1253 fn test_parse_project_relation_complex() {
1254 let extensions = SimpleExtensions::default();
1255 let project = ProjectRel::parse_pair_with_context(
1256 &extensions,
1257 parse_exact(Rule::project_relation, "Project[42, $0, 100, $2, $1]"),
1258 vec![Box::new(example_read_relation().into_rel(None))],
1259 5, )
1261 .unwrap()
1262 .0;
1263
1264 assert_eq!(project.expressions.len(), 2);
1266
1267 let emit_kind = &project.common.as_ref().unwrap().emit_kind.as_ref().unwrap();
1268 let emit = match emit_kind {
1269 EmitKind::Emit(emit) => &emit.output_mapping,
1270 _ => panic!("Expected EmitKind::Emit, got {emit_kind:?}"),
1271 };
1272 assert_eq!(emit, &[5, 0, 6, 2, 1]);
1275 }
1276
1277 #[test]
1278 fn test_parse_aggregate_relation() {
1279 let extensions = TestContext::new()
1280 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1281 .with_function(1, 10, "sum")
1282 .with_function(1, 11, "count")
1283 .extensions;
1284
1285 let aggregate = AggregateRel::parse_pair_with_context(
1286 &extensions,
1287 parse_exact(
1288 Rule::aggregate_relation,
1289 "Aggregate[($0, $1), _ => sum($2), $0, count($2)]",
1290 ),
1291 vec![Box::new(example_read_relation().into_rel(None))],
1292 3,
1293 )
1294 .unwrap()
1295 .0;
1296
1297 assert_eq!(aggregate.grouping_expressions.len(), 2);
1299 assert_eq!(aggregate.groupings[0].expression_references.len(), 2);
1300 assert_eq!(aggregate.groupings.len(), 2);
1301 assert_eq!(aggregate.measures.len(), 2);
1302
1303 let emit_kind = &aggregate
1304 .common
1305 .as_ref()
1306 .unwrap()
1307 .emit_kind
1308 .as_ref()
1309 .unwrap();
1310 let emit = match emit_kind {
1311 EmitKind::Emit(emit) => &emit.output_mapping,
1312 _ => panic!("Expected EmitKind::Emit, got {emit_kind:?}"),
1313 };
1314 assert_eq!(emit, &[2, 0, 3]);
1317 }
1318
1319 #[test]
1320 fn test_parse_aggregate_relation_maintain_column_order() {
1321 let extensions = TestContext::new()
1322 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1323 .with_function(1, 10, "sum")
1324 .with_function(1, 11, "count")
1325 .extensions;
1326
1327 let aggregate = AggregateRel::parse_pair_with_context(
1328 &extensions,
1329 parse_exact(
1330 Rule::aggregate_relation,
1331 "Aggregate[$0 => sum($1), $0, count($1)]",
1332 ),
1333 vec![Box::new(example_read_relation().into_rel(None))],
1334 3,
1335 )
1336 .unwrap()
1337 .0;
1338
1339 assert_eq!(aggregate.grouping_expressions.len(), 1);
1341 assert_eq!(aggregate.groupings.len(), 1);
1342 assert_eq!(aggregate.measures.len(), 2);
1343
1344 let emit_kind = &aggregate
1345 .common
1346 .as_ref()
1347 .unwrap()
1348 .emit_kind
1349 .as_ref()
1350 .unwrap();
1351 let emit = match emit_kind {
1352 EmitKind::Emit(emit) => &emit.output_mapping,
1353 _ => panic!("Expected EmitKind::Emit, got {emit_kind:?}"),
1354 };
1355 assert_eq!(emit, &[1, 0, 2]);
1357 }
1358
1359 #[test]
1360 fn test_parse_aggregate_relation_simple() {
1361 let extensions = TestContext::new()
1362 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1363 .with_function(1, 10, "sum")
1364 .extensions;
1365
1366 let aggregate = AggregateRel::parse_pair_with_context(
1367 &extensions,
1368 parse_exact(Rule::aggregate_relation, "Aggregate[$2, $0 => sum($1)]"),
1369 vec![Box::new(example_read_relation().into_rel(None))],
1370 3,
1371 )
1372 .unwrap()
1373 .0;
1374
1375 assert_eq!(aggregate.grouping_expressions.len(), 2);
1376 assert_eq!(aggregate.groupings.len(), 1);
1377 assert_eq!(aggregate.groupings[0].expression_references, vec![0, 1]);
1379 }
1380
1381 #[test]
1382 fn test_parse_aggregate_relation_global_aggregate() {
1383 let extensions = TestContext::new()
1384 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1385 .with_function(1, 10, "sum")
1386 .with_function(1, 11, "count")
1387 .extensions;
1388
1389 let aggregate = AggregateRel::parse_pair_with_context(
1390 &extensions,
1391 parse_exact(
1392 Rule::aggregate_relation,
1393 "Aggregate[_ => sum($0), count($1)]",
1394 ),
1395 vec![Box::new(example_read_relation().into_rel(None))],
1396 3,
1397 )
1398 .unwrap()
1399 .0;
1400
1401 assert_eq!(aggregate.grouping_expressions.len(), 0);
1403 assert_eq!(aggregate.groupings.len(), 1);
1404 assert_eq!(aggregate.groupings[0].expression_references.len(), 0);
1405 assert_eq!(aggregate.measures.len(), 2);
1406
1407 let emit_kind = aggregate
1409 .common
1410 .as_ref()
1411 .unwrap()
1412 .emit_kind
1413 .as_ref()
1414 .unwrap();
1415 assert!(
1416 matches!(emit_kind, EmitKind::Direct(_)),
1417 "Expected Direct for identity emit, got {emit_kind:?}"
1418 );
1419 }
1420
1421 #[test]
1422 fn test_parse_aggregate_relation_grouping_sets() {
1423 let extensions = TestContext::new()
1424 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1425 .with_function(1, 11, "count")
1426 .extensions;
1427
1428 let read_rel = ReadRel::parse_pair_with_context(
1429 &extensions,
1430 parse_exact(
1431 Rule::read_relation,
1432 "Read[ab.cd.ef => a:i32, b:string?, c:i64, d:i64]",
1433 ),
1434 vec![],
1435 0,
1436 )
1437 .unwrap()
1438 .0;
1439
1440 let aggregate = AggregateRel::parse_pair_with_context(
1441 &extensions,
1442 parse_exact(
1443 Rule::aggregate_relation,
1444 "Aggregate[($0, $1, $2), ($2, $0), ($1), _ => $0, $1, $2, count($3)]",
1445 ),
1446 vec![Box::new(read_rel.into_rel(None))],
1447 4,
1448 )
1449 .unwrap()
1450 .0;
1451
1452 assert_eq!(aggregate.grouping_expressions.len(), 3);
1453 assert_eq!(aggregate.groupings.len(), 4);
1454 assert_eq!(aggregate.groupings[0].expression_references, vec![0, 1, 2]);
1456 assert_eq!(aggregate.groupings[1].expression_references, vec![2, 0]);
1458 assert_eq!(aggregate.groupings[2].expression_references, vec![1]);
1460 assert!(aggregate.groupings[3].expression_references.is_empty());
1462 assert_eq!(aggregate.measures.len(), 1);
1463 }
1464
1465 #[test]
1466 fn test_fetch_relation_positive_values() {
1467 let extensions = SimpleExtensions::default();
1468
1469 let fetch_rel = FetchRel::parse_pair_with_context(
1471 &extensions,
1472 parse_exact(Rule::fetch_relation, "Fetch[limit=10, offset=5 => $0]"),
1473 vec![Box::new(example_read_relation().into_rel(None))],
1474 3,
1475 )
1476 .unwrap()
1477 .0;
1478
1479 assert_eq!(
1481 fetch_rel.count_mode,
1482 Some(CountMode::CountExpr(i64_literal_expr(10)))
1483 );
1484 assert_eq!(
1485 fetch_rel.offset_mode,
1486 Some(OffsetMode::OffsetExpr(i64_literal_expr(5)))
1487 );
1488 }
1489
1490 #[test]
1491 fn test_fetch_relation_negative_limit_rejected() {
1492 let extensions = SimpleExtensions::default();
1493
1494 let parsed_result = ExpressionParser::parse(Rule::fetch_relation, "Fetch[limit=-5 => $0]");
1496 if let Ok(mut pairs) = parsed_result {
1497 let pair = pairs.next().unwrap();
1498 if pair.as_str() == "Fetch[limit=-5 => $0]" {
1499 let result = FetchRel::parse_pair_with_context(
1501 &extensions,
1502 pair,
1503 vec![Box::new(example_read_relation().into_rel(None))],
1504 3,
1505 );
1506 assert!(result.is_err());
1507 let error_msg = result.unwrap_err().to_string();
1508 assert!(error_msg.contains("Fetch limit must be non-negative"));
1509 } else {
1510 println!("Grammar prevents negative limit values at parse time");
1513 }
1514 } else {
1515 println!("Grammar prevents negative limit values at parse time");
1517 }
1518 }
1519
1520 #[test]
1521 fn test_fetch_relation_negative_offset_rejected() {
1522 let extensions = SimpleExtensions::default();
1523
1524 let parsed_result =
1526 ExpressionParser::parse(Rule::fetch_relation, "Fetch[offset=-10 => $0]");
1527 if let Ok(mut pairs) = parsed_result {
1528 let pair = pairs.next().unwrap();
1529 if pair.as_str() == "Fetch[offset=-10 => $0]" {
1530 let result = FetchRel::parse_pair_with_context(
1532 &extensions,
1533 pair,
1534 vec![Box::new(example_read_relation().into_rel(None))],
1535 3,
1536 );
1537 assert!(result.is_err());
1538 let error_msg = result.unwrap_err().to_string();
1539 assert!(error_msg.contains("Fetch offset must be non-negative"));
1540 } else {
1541 println!("Grammar prevents negative offset values at parse time");
1543 }
1544 } else {
1545 println!("Grammar prevents negative offset values at parse time");
1547 }
1548 }
1549
1550 #[test]
1551 fn test_parse_join_relation() {
1552 let extensions = TestContext::new()
1553 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_comparison.yaml")
1554 .with_function(1, 10, "eq")
1555 .extensions;
1556
1557 let left_rel = example_read_relation().into_rel(None);
1558 let right_rel = example_read_relation().into_rel(None);
1559
1560 let join = JoinRel::parse_pair_with_context(
1561 &extensions,
1562 parse_exact(
1563 Rule::join_relation,
1564 "Join[&Inner, eq($0, $3) => $0, $1, $3, $4]",
1565 ),
1566 vec![Box::new(left_rel), Box::new(right_rel)],
1567 6, )
1569 .unwrap()
1570 .0;
1571
1572 assert_eq!(join.r#type, join_rel::JoinType::Inner as i32);
1574
1575 assert!(join.left.is_some());
1577 assert!(join.right.is_some());
1578
1579 assert!(join.expression.is_some());
1581
1582 let emit_kind = &join.common.as_ref().unwrap().emit_kind.as_ref().unwrap();
1583 let emit = match emit_kind {
1584 EmitKind::Emit(emit) => &emit.output_mapping,
1585 _ => panic!("Expected EmitKind::Emit, got {emit_kind:?}"),
1586 };
1587 assert_eq!(emit, &[0, 1, 3, 4]);
1589 }
1590
1591 #[test]
1592 fn test_parse_join_relation_left_outer() {
1593 let extensions = TestContext::new()
1594 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_comparison.yaml")
1595 .with_function(1, 10, "eq")
1596 .extensions;
1597
1598 let left_rel = example_read_relation().into_rel(None);
1599 let right_rel = example_read_relation().into_rel(None);
1600
1601 let join = JoinRel::parse_pair_with_context(
1602 &extensions,
1603 parse_exact(Rule::join_relation, "Join[&Left, eq($0, $3) => $0, $1, $2]"),
1604 vec![Box::new(left_rel), Box::new(right_rel)],
1605 6,
1606 )
1607 .unwrap()
1608 .0;
1609
1610 assert_eq!(join.r#type, join_rel::JoinType::Left as i32);
1612
1613 let emit_kind = &join.common.as_ref().unwrap().emit_kind.as_ref().unwrap();
1614 let emit = match emit_kind {
1615 EmitKind::Emit(emit) => &emit.output_mapping,
1616 _ => panic!("Expected EmitKind::Emit, got {emit_kind:?}"),
1617 };
1618 assert_eq!(emit, &[0, 1, 2]);
1620 }
1621
1622 #[test]
1623 fn test_parse_join_relation_left_semi() {
1624 let extensions = TestContext::new()
1625 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_comparison.yaml")
1626 .with_function(1, 10, "eq")
1627 .extensions;
1628
1629 let left_rel = example_read_relation().into_rel(None);
1630 let right_rel = example_read_relation().into_rel(None);
1631
1632 let join = JoinRel::parse_pair_with_context(
1633 &extensions,
1634 parse_exact(Rule::join_relation, "Join[&LeftSemi, eq($0, $3) => $0, $1]"),
1635 vec![Box::new(left_rel), Box::new(right_rel)],
1636 6,
1637 )
1638 .unwrap()
1639 .0;
1640
1641 assert_eq!(join.r#type, join_rel::JoinType::LeftSemi as i32);
1643
1644 let emit_kind = &join.common.as_ref().unwrap().emit_kind.as_ref().unwrap();
1645 let emit = match emit_kind {
1646 EmitKind::Emit(emit) => &emit.output_mapping,
1647 _ => panic!("Expected EmitKind::Emit, got {emit_kind:?}"),
1648 };
1649 assert_eq!(emit, &[0, 1]);
1651 }
1652
1653 #[test]
1654 fn test_parse_join_relation_requires_two_children() {
1655 let extensions = SimpleExtensions::default();
1656
1657 let result = JoinRel::parse_pair_with_context(
1659 &extensions,
1660 parse_exact(Rule::join_relation, "Join[&Inner, eq($0, $1) => $0, $1]"),
1661 vec![],
1662 0,
1663 );
1664 assert!(result.is_err());
1665
1666 let result = JoinRel::parse_pair_with_context(
1668 &extensions,
1669 parse_exact(Rule::join_relation, "Join[&Inner, eq($0, $1) => $0, $1]"),
1670 vec![Box::new(example_read_relation().into_rel(None))],
1671 3,
1672 );
1673 assert!(result.is_err());
1674
1675 let result = JoinRel::parse_pair_with_context(
1677 &extensions,
1678 parse_exact(Rule::join_relation, "Join[&Inner, eq($0, $1) => $0, $1]"),
1679 vec![
1680 Box::new(example_read_relation().into_rel(None)),
1681 Box::new(example_read_relation().into_rel(None)),
1682 Box::new(example_read_relation().into_rel(None)),
1683 ],
1684 9,
1685 );
1686 assert!(result.is_err());
1687 }
1688
1689 fn parse_exact(rule: Rule, input: &'_ str) -> pest::iterators::Pair<'_, Rule> {
1690 let mut pairs = ExpressionParser::parse(rule, input).unwrap();
1691 assert_eq!(pairs.as_str(), input);
1692 let pair = pairs.next().unwrap();
1693 assert_eq!(pairs.next(), None);
1694 pair
1695 }
1696}