Skip to main content

substrait_explain/parser/
relations.rs

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
25/// Parsing context for relations that includes extensions, registry, and optional warning collection
26pub 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    /// Resolve extension detail using registry. Any failure is treated as a hard parse error.
35    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    /// Resolve an advanced-extension detail (enhancement or optimization) using the registry.
58    /// Any failure is treated as a hard parse error.
59    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
82/// A trait for parsing relations with full context for tree building.
83pub trait RelationParsePair: Sized {
84    fn rule() -> Rule;
85    fn message() -> &'static str;
86
87    /// Parse the grammar pair into this relation type and its output field
88    /// count.
89    ///
90    /// Returns `(Self, usize)` where `usize` is the output field count —
91    /// computed during parsing when `input_field_count` is available.
92    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    /// Consume this parsed relation, apply the advanced extension, and produce
100    /// the final `Rel`.
101    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/// This is a utility function for extracting a single child from the list of
181/// children, to be used in the RelationParsePair trait. The RelationParsePair
182/// trait passes a Vec of children, because some relations have multiple
183/// children - but most accept exactly one child.
184#[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
205/// Parse a reference list Pair and return an EmitKind::Emit.
206/// Parse a reference list into field indices for emit mapping.
207fn 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
214/// Build an emit: `Direct` if the mapping is the identity `[0, 1, ..., N-1]`
215/// (where N = `direct_output_count`), otherwise `Emit` with the explicit mapping.
216fn 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
229/// Parse a reference list into an emit and output field count.
230fn 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
237/// Extracts named arguments from pest pairs with duplicate detection and completeness checking.
238///
239/// Usage: `extractor.pop("limit", Rule::fetch_value).0.pop("offset", Rule::fetch_value).0.done()`
240///
241/// The fluent API ensures all arguments are processed exactly once and none are forgotten.
242pub 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    // Returns the pair if it exists and matches the rule, otherwise None.
272    // Asserts that the rule must match the rule of the pair (and therefore
273    // panics in non-release-mode if not)
274    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    // Returns an error if there are any unused arguments.
282    pub fn done(self) -> Result<(), MessageParseError> {
283        if let Some((name, pair)) = self.map.iter().next() {
284            return Err(MessageParseError::invalid(
285                "NamedArgExtractor",
286                // No span available for all unused args; use default.
287                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        // ReadRel is a leaf node - it should have no input children and 0 input fields
312        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
354/// Parsed `Read:Virtual[rows => columns]` relation. Needs a newtype because the
355/// proto type is `ReadRel` (same as `NamedTable`), but the grammar and handling
356/// are different.
357pub(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        // TODO: Validate that each row has the same number of expressions as
392        // columns. Currently no parser-side warning mechanism exists, and while
393        // this is an invalid plan, it is constructible as Substrait. Consider
394        // adding once a warning collection path is available.
395        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
417/// Build a `NamedStruct` from parsed columns.
418fn 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
430/// `Read:Virtual` positional args: either `empty` or a list of row tuples.
431fn 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
450/// Parse a single `virtual_row` (`(expr, expr, ...)` or `()`) into a `nested::Struct`.
451fn 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        // Empty virtual row. An unusual but valid case.
462        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        // references (which become the emit)
494        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                    // Index into the combined schema: [input fields][computed expressions].
556                    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        // Aggregate defines its own output schema (grouping keys + measures),
603        // so the input field count isn't needed for emit construction.
604        _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
646/// Parses the output section of an aggregate (everything after `=>`).
647///
648/// For example, in `Aggregate[($0, $1), _ => sum($2), $0, count($2)]`,
649/// this parses `sum($2), $0, count($2)`.
650fn 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
681/// Parses the grouping section of an aggregate (everything before `=>`).
682///
683/// For example, in `Aggregate[($0, $1), _ => sum($2), $0, count($2)]`,
684/// this parses `($0, $1), _`.
685///
686/// Each inner Vec is one grouping set; an empty vec represents no grouping (global aggregate).
687///
688/// Grammar: `aggregate_group_by = { grouping_set_list | expression_list }`
689fn 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
714/// Parses a single grouping set, e.g. `($0, $1)` or `_`.
715///
716/// Grammar: `grouping_set = { ("(" ~ expression_list ~ ")") | empty }`
717fn 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
733/// Grammar: `expression_list = { expression ~ ("," ~ expression)* }`
734pub(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
743/// Deduplicates expressions across all sets and produces the AggregateRel's
744/// protobuf fields: a flat deduplicated expression list and per-set Grouping
745/// messages with index references into that list.
746fn 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                    // TODO: use a better key here than encoding to bytes.
757                    // Ideally, substrait-rs would support `PartialEq` and `Hash`,
758                    // but as there isn't an easy way to do that now, we'll skip.
759                    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        // Strip the '&' prefix from enum syntax (e.g., "&AscNullsFirst" ->
797        // "AscNullsFirst") The grammar includes '&' to distinguish enums from
798        // identifiers, but the enum variant names don't include it
799        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            // TODO: Add support for SortKind::ComparisonFunctionReference
820            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        // Extract all pairs before any validation: RuleIter's Drop panics on
1010        // incomplete consumption, so we must exhaust the iterator before any
1011        // early return. Validation runs after iter.done() below.
1012        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        // TODO: For semi/anti joins, the direct output width differs from
1130        // left+right — `input_field_count` would misclassify the emit as Direct.
1131        // Revisit when those join types are supported.
1132        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, // not yet represented in the grammar
1145                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        // Removed: test_parse_relation for old Relation struct
1164    }
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    /// Produces a ReadRel with 3 columns: a:i32, b:string?, c:i64
1194    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        // Identity mapping [0, 1, 2] over 3 inputs → Direct
1221        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        // Should have 1 expression (42) and 2 references ($0, $1)
1241        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        // Output mapping should be [0, 1, 3]. References are 0-2; expression is 3.
1249        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, // Assume 5 input fields
1260        )
1261        .unwrap()
1262        .0;
1263
1264        // Should have 2 expressions (42, 100) and 3 references ($0, $2, $1)
1265        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        // Direct mapping: [input_fields..., 42, 100] (input fields first, then expressions)
1273        // Output mapping: [5, 0, 6, 2, 1] (to get: 42, $0, 100, $2, $1)
1274        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        // Should have 2 group-by sets ($0, $1) and an empty group, and emit 2 measures (sum($2), count($2))
1298        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        // Output mapping should be [2, 0, 3] (measures and group-by fields in order)
1315        // sum($2) -> 2, $0 -> 0, count($2) -> 3
1316        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        // Should have 1 group-by field ($0) and 2 measures (sum($1), count($1))
1340        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        // Output mapping should be [1, 0, 2] (grouping fields + measures)
1356        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        // expression_references must be positions [0, 1], not raw field indices [2, 0]
1378        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        // Should have 0 group-by fields and 2 measures
1402        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        // Identity mapping [0, 1] over 2 outputs (0 grouping + 2 measures) → Direct
1408        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        // ($0, $1, $2) -> [0, 1, 2]
1455        assert_eq!(aggregate.groupings[0].expression_references, vec![0, 1, 2]);
1456        // ($2, $0) -> [2, 0] (reuses indices, different order)
1457        assert_eq!(aggregate.groupings[1].expression_references, vec![2, 0]);
1458        // ($1) -> [1]
1459        assert_eq!(aggregate.groupings[2].expression_references, vec![1]);
1460        // _ -> empty
1461        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        // Test valid positive values should work
1470        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        // Verify the limit and offset values are correct
1480        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        // Test that fetch relations with negative limits are properly rejected
1495        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                // Full parse succeeded, now test that validation catches the negative value
1500                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                // If grammar doesn't fully support negative values, that's also acceptable
1511                // since it would prevent negative values at parse time
1512                println!("Grammar prevents negative limit values at parse time");
1513            }
1514        } else {
1515            // Grammar doesn't support negative values in fetch context
1516            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        // Test that fetch relations with negative offsets are properly rejected
1525        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                // Full parse succeeded, now test that validation catches the negative value
1531                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                // If grammar doesn't fully support negative values, that's also acceptable
1542                println!("Grammar prevents negative offset values at parse time");
1543            }
1544        } else {
1545            // Grammar doesn't support negative values in fetch context
1546            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, // left (3) + right (3) = 6 total input fields
1568        )
1569        .unwrap()
1570        .0;
1571
1572        // Should be an Inner join
1573        assert_eq!(join.r#type, join_rel::JoinType::Inner as i32);
1574
1575        // Should have left and right relations
1576        assert!(join.left.is_some());
1577        assert!(join.right.is_some());
1578
1579        // Should have a join condition
1580        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        // Output mapping should be [0, 1, 3, 4] (selected columns)
1588        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        // Should be a Left join
1611        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        // Output mapping should be [0, 1, 2]
1619        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        // Should be a LeftSemi join
1642        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        // Output mapping should be [0, 1] (only left columns for semi join)
1650        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        // Test with 0 children
1658        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        // Test with 1 child
1667        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        // Test with 3 children
1676        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}