substrait_explain/textify/
rels.rs

1use std::borrow::Cow;
2use std::collections::HashSet;
3use std::convert::TryFrom;
4use std::fmt;
5
6use prost::Message;
7use substrait::proto::fetch_rel::CountMode;
8use substrait::proto::plan_rel::RelType as PlanRelType;
9use substrait::proto::read_rel::ReadType;
10use substrait::proto::rel::RelType;
11use substrait::proto::rel_common::EmitKind;
12use substrait::proto::{
13    AggregateRel, CrossRel, ExtensionLeafRel, ExtensionMultiRel, ExtensionSingleRel, FetchRel,
14    FilterRel, JoinRel, NamedStruct, PlanRel, ProjectRel, ReadRel, Rel, RelCommon, RelRoot, SetRel,
15    SortRel, join_rel, set_rel,
16};
17
18use super::addenda::AddendumLines;
19use super::types::Name;
20use super::values::{Arguments, NamedArg, Value, ValueEnum, decode_enum_field};
21use super::{PlanError, Scope, Textify};
22use crate::FormatError;
23use crate::extensions::any::AnyRef;
24use crate::extensions::{ExtensionArgs, ExtensionError};
25
26pub trait NamedRelation {
27    fn name(&self) -> &'static str;
28}
29
30impl NamedRelation for Rel {
31    fn name(&self) -> &'static str {
32        match self.rel_type.as_ref() {
33            None => "UnknownRel",
34            Some(RelType::Read(_)) => "Read",
35            Some(RelType::Filter(_)) => "Filter",
36            Some(RelType::Project(_)) => "Project",
37            Some(RelType::Fetch(_)) => "Fetch",
38            Some(RelType::Aggregate(_)) => "Aggregate",
39            Some(RelType::Sort(_)) => "Sort",
40            Some(RelType::HashJoin(_)) => "HashJoin",
41            Some(RelType::Exchange(_)) => "Exchange",
42            Some(RelType::Join(_)) => "Join",
43            Some(RelType::Set(_)) => "Set",
44            Some(RelType::ExtensionLeaf(_)) => "ExtensionLeaf",
45            Some(RelType::Cross(_)) => "Cross",
46            Some(RelType::Reference(_)) => "Reference",
47            Some(RelType::ExtensionSingle(_)) => "ExtensionSingle",
48            Some(RelType::ExtensionMulti(_)) => "ExtensionMulti",
49            Some(RelType::Write(_)) => "Write",
50            Some(RelType::Ddl(_)) => "Ddl",
51            Some(RelType::Update(_)) => "Update",
52            Some(RelType::MergeJoin(_)) => "MergeJoin",
53            Some(RelType::NestedLoopJoin(_)) => "NestedLoopJoin",
54            Some(RelType::Window(_)) => "Window",
55            Some(RelType::Expand(_)) => "Expand",
56        }
57    }
58}
59
60impl Textify for Rel {
61    fn name() -> &'static str {
62        "Rel"
63    }
64
65    fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
66        // delegates to `Relation` which carries `advanced_extension`, so the full
67        // header → enhancement → children sequence is handled uniformly there.
68        Relation::from_rel(self, ctx).textify(ctx, w)
69    }
70}
71
72fn schema_to_values<'a>(schema: &'a NamedStruct) -> Vec<Value<'a>> {
73    let mut fields = schema
74        .r#struct
75        .as_ref()
76        .map(|s| s.types.iter())
77        .into_iter()
78        .flatten();
79    let mut names = schema.names.iter();
80
81    // let field_count = schema.r#struct.as_ref().map(|s| s.types.len()).unwrap_or(0);
82    // let name_count = schema.names.len();
83
84    let mut values = Vec::new();
85    loop {
86        let field = fields.next();
87        let name = names.next().map(|n| Name(n));
88        if field.is_none() && name.is_none() {
89            break;
90        }
91
92        values.push(Value::Field(name, field));
93    }
94
95    values
96}
97
98/// How a relation header renders its output.
99#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
100enum OutputSyntax {
101    /// Output columns are rendered as final visible output: `=> output_columns`.
102    #[default]
103    Implicit,
104    /// Output columns are rendered as the direct output domain: `+> columns`,
105    /// with `|> order` appended when an explicit emit mapping is present.
106    Explicit,
107}
108
109struct Emitted<'a> {
110    values: &'a [Value<'a>],
111    emit: Option<&'a EmitKind>,
112    output_syntax: Option<OutputSyntax>,
113}
114
115impl<'a> Emitted<'a> {
116    pub fn columns(values: &'a [Value<'a>], emit: Option<&'a EmitKind>) -> Self {
117        Self {
118            values,
119            emit,
120            output_syntax: None,
121        }
122    }
123
124    pub fn output_clause(
125        values: &'a [Value<'a>],
126        emit: Option<&'a EmitKind>,
127        output_syntax: OutputSyntax,
128    ) -> Self {
129        Self {
130            values,
131            emit,
132            output_syntax: Some(output_syntax),
133        }
134    }
135
136    fn write_output_clause<S: Scope, W: fmt::Write>(
137        &self,
138        ctx: &S,
139        w: &mut W,
140        output_syntax: OutputSyntax,
141    ) -> fmt::Result {
142        match output_syntax {
143            OutputSyntax::Implicit => {
144                write!(w, "=> ")?;
145                self.write_implicit_columns(ctx, w)
146            }
147            OutputSyntax::Explicit => {
148                write!(w, "+> ")?;
149                self.write_direct_columns(ctx, w)?;
150                self.write_emit_suffix(ctx, w)
151            }
152        }
153    }
154
155    fn write_direct_columns<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
156        write!(w, "{}", ctx.separated(self.values.iter(), ", "))
157    }
158
159    fn write_implicit_columns<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
160        if ctx.options().show_emit {
161            return self.write_direct_columns(ctx, w);
162        }
163
164        let indices = match &self.emit {
165            Some(EmitKind::Emit(e)) => &e.output_mapping,
166            Some(EmitKind::Direct(_)) => return self.write_direct_columns(ctx, w),
167            None => return self.write_direct_columns(ctx, w),
168        };
169
170        for (i, &index) in indices.iter().enumerate() {
171            if i > 0 {
172                write!(w, ", ")?;
173            }
174
175            match self.values.get(index as usize) {
176                Some(value) => write!(w, "{}", ctx.display(value))?,
177                None => write!(w, "{}", ctx.failure(PlanError::invalid(
178                    "Emitted",
179                    Some("output_mapping"),
180                    format!(
181                        "Output mapping index {} is out of bounds for values collection of size {}",
182                        index, self.values.len()
183                    )
184                )))?,
185            }
186        }
187
188        Ok(())
189    }
190
191    fn write_emit_suffix<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
192        let Some(EmitKind::Emit(emit)) = self.emit else {
193            return Ok(());
194        };
195        let mapping = emit
196            .output_mapping
197            .iter()
198            .copied()
199            .map(Value::Reference)
200            .collect::<Vec<_>>();
201        write!(w, " |> {}", ctx.separated(mapping.iter(), ", "))
202    }
203}
204
205impl<'a> Textify for Emitted<'a> {
206    fn name() -> &'static str {
207        "Emitted"
208    }
209
210    fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
211        match self.output_syntax {
212            Some(output_syntax) => self.write_output_clause(ctx, w, output_syntax),
213            None => self.write_implicit_columns(ctx, w),
214        }
215    }
216}
217
218/// The argument section of a relation header.
219#[derive(Debug, Clone)]
220pub enum RelationArgs<'a> {
221    /// `arg, arg, name=arg` inline inside the header's `[...]`.
222    Inline(Arguments<'a>),
223    /// One `- row` per line, used by `Read:Virtual` once it has enough rows to
224    /// be worth spreading out. Any named arguments follow the rows, one per
225    /// line, in the same `- name=value` form.
226    Rows {
227        rows: Vec<Value<'a>>,
228        named: Vec<NamedArg<'a>>,
229    },
230}
231
232impl<'a> RelationArgs<'a> {
233    /// An inline argument list, the layout used by every relation but a
234    /// multi-row `Read:Virtual`.
235    pub fn inline(positional: Vec<Value<'a>>, named: Vec<NamedArg<'a>>) -> Self {
236        RelationArgs::Inline(Arguments::new(positional, named))
237    }
238
239    /// A row-per-line argument list (`- arg` per line) used for `Read:Virtual`
240    /// with many rows. Named arguments, if any, follow the rows.
241    pub fn rows(rows: Vec<Value<'a>>, named: Vec<NamedArg<'a>>) -> Self {
242        RelationArgs::Rows { rows, named }
243    }
244}
245
246pub struct Relation<'a> {
247    pub name: Cow<'a, str>,
248    /// Arguments to the relation, if any.
249    ///
250    /// - `None` means this relation does not take arguments, and the argument
251    ///   section is omitted entirely.
252    /// - `Some(RelationArgs::Inline(args))` with both vectors empty means the
253    ///   relation takes arguments, but none are provided; this will print as
254    ///   `_ => ...`.
255    /// - `Some(RelationArgs::Inline(args))` with non-empty vectors will print
256    ///   with positional arguments first, then named arguments, separated by commas.
257    /// - `Some(RelationArgs::Rows { .. })` prints one row per line, followed by
258    ///   any named arguments, one per line.
259    pub arguments: Option<RelationArgs<'a>>,
260    /// The columns emitted by this relation, pre-emit - the 'direct' column
261    /// output.
262    pub columns: Vec<Value<'a>>,
263    /// The emit kind, if any. If none, use the columns directly.
264    pub emit: Option<&'a EmitKind>,
265    /// Whether output columns are rendered as visible output or as a direct
266    /// output domain plus optional explicit emit mapping.
267    output_syntax: OutputSyntax,
268    /// `+`-prefixed addendum lines to emit between this relation's header and
269    /// children.  This owns the canonical ordering for `+ Ext`, `+ Enh`, and
270    /// `+ Opt` lines rather than making the generic relation shape grow one
271    /// field per addendum kind.
272    addenda: AddendumLines,
273    /// The input relations.
274    pub children: Vec<Option<Relation<'a>>>,
275}
276
277impl Textify for Relation<'_> {
278    fn name() -> &'static str {
279        "Relation"
280    }
281
282    fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
283        self.write_header(ctx, w)?;
284        let child_scope = ctx.push_indent();
285        self.addenda.textify(&child_scope, w)?;
286        self.write_children(ctx, w)?;
287        Ok(())
288    }
289}
290
291impl Relation<'_> {
292    /// Write the header for this relation, e.g. `Filter[$0 => $0]`.
293    ///
294    /// Usually a single line, but an argument list of [`RelationArgs::Rows`]
295    /// (used by `Read:Virtual` with many rows) spans several lines:
296    ///
297    /// ```text
298    /// Read:Virtual[
299    ///   - (1, 'alice'),
300    ///   - (2, 'bob')
301    ///   - => id:i64, name:string]
302    /// ```
303    ///
304    /// Does not write a trailing newline; callers are responsible for any
305    /// newline that follows (either from an addendum or from the next child).
306    pub fn write_header<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
307        let indent = ctx.indent();
308        let name = &self.name;
309        match &self.arguments {
310            None => {
311                let cols = Emitted::columns(&self.columns, self.emit);
312                let cols = ctx.display(&cols);
313                write!(w, "{indent}{name}[{cols}]")
314            }
315            Some(RelationArgs::Rows { rows, named }) => {
316                // One `- row` per line, one indent level deeper, with a
317                // trailing comma when another row or named argument follows,
318                // then `- <output> cols]`.
319                let child = ctx.push_indent();
320                let child_indent = child.indent();
321                writeln!(w, "{indent}{name}[")?;
322                let last = rows.len().saturating_sub(1);
323                for (i, row) in rows.iter().enumerate() {
324                    let row = ctx.display(row);
325                    let comma = if i == last && named.is_empty() {
326                        ""
327                    } else {
328                        ","
329                    };
330                    writeln!(w, "{child_indent}- {row}{comma}")?;
331                }
332                let last = named.len().saturating_sub(1);
333                for (i, named_arg) in named.iter().enumerate() {
334                    let named_arg = ctx.display(named_arg);
335                    let comma = if i == last { "" } else { "," };
336                    writeln!(w, "{child_indent}- {named_arg}{comma}")?;
337                }
338                let output = Emitted::output_clause(&self.columns, self.emit, self.output_syntax);
339                let output = ctx.display(&output);
340                write!(w, "{child_indent}- {output}]")
341            }
342            Some(RelationArgs::Inline(args)) => {
343                let args = ctx.display(args);
344                let output = Emitted::output_clause(&self.columns, self.emit, self.output_syntax);
345                let output = ctx.display(&output);
346                write!(w, "{indent}{name}[{args} {output}]")
347            }
348        }
349    }
350
351    /// Write each child relation at one indent level deeper than `ctx`.
352    /// Each child is preceded by a newline.
353    pub fn write_children<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
354        let child_scope = ctx.push_indent();
355        for child in self.children.iter().flatten() {
356            writeln!(w)?;
357            child.textify(&child_scope, w)?;
358        }
359        Ok(())
360    }
361}
362
363impl<'a> Relation<'a> {
364    pub fn emitted(&self) -> usize {
365        match self.emit {
366            Some(EmitKind::Emit(e)) => e.output_mapping.len(),
367            Some(EmitKind::Direct(_)) => self.columns.len(),
368            None => self.columns.len(),
369        }
370    }
371}
372
373impl<'a> Relation<'a> {
374    fn from_read<S: Scope>(rel: &'a ReadRel, ctx: &S) -> Self {
375        let columns = read_columns(rel);
376        let emit = rel.common.as_ref().and_then(|c| c.emit_kind.as_ref());
377
378        match &rel.read_type {
379            Some(ReadType::NamedTable(table)) => {
380                let table_name = Value::TableName(table.names.iter().map(|n| Name(n)).collect());
381                // XXX: For `ReadRel`s, we use `=>` if emit is None, `+>` if
382                // `emit` is `Some(Direct)`, and `+> … |>` if emit is
383                // `Some(Remap(…))`. However, we ignore the
384                // `OutputOptions::show_emit` option, and we haven't yet
385                // supported other operators; at some point, we should figure
386                // out our policy here and clean this up.
387                let output_syntax = if emit.is_some() {
388                    OutputSyntax::Explicit
389                } else {
390                    OutputSyntax::Implicit
391                };
392                Relation {
393                    name: Cow::Borrowed("Read"),
394                    arguments: Some(RelationArgs::inline(vec![table_name], vec![])),
395                    columns,
396                    emit,
397                    output_syntax,
398                    addenda: AddendumLines::from_advanced_extension(
399                        ctx,
400                        rel.advanced_extension.as_ref(),
401                    ),
402                    children: vec![],
403                }
404            }
405            Some(ReadType::VirtualTable(vt)) => {
406                let row_count = vt.expressions.len();
407                let mut positional: Vec<Value> = vt
408                    .expressions
409                    .iter()
410                    .map(|row| Value::Tuple(row.fields.iter().map(Value::Expression).collect()))
411                    .collect();
412                let mut named = vec![];
413                if let Some(filter) = rel.filter.as_ref() {
414                    named.push(NamedArg {
415                        name: Cow::Borrowed("filter"),
416                        value: Value::Expression(filter.as_ref()),
417                    });
418                }
419                if positional.is_empty() && !named.is_empty() {
420                    positional.push(Value::EmptyGroup);
421                }
422
423                // Emit many rows across multiple lines for readability, based on
424                // a configurable threshold (default = 3). An empty table has no
425                // rows to spread out and is written `_`, so it stays inline
426                // regardless of the threshold — the row layout has no `_` form.
427                let multiline =
428                    row_count > 0 && row_count >= ctx.options().virtual_table_multiline_threshold;
429                let arguments = if multiline {
430                    RelationArgs::rows(positional, named)
431                } else {
432                    RelationArgs::inline(positional, named)
433                };
434
435                Relation {
436                    name: Cow::Borrowed("Read:Virtual"),
437                    arguments: Some(arguments),
438                    columns,
439                    emit,
440                    output_syntax: OutputSyntax::Implicit,
441                    addenda: AddendumLines::from_advanced_extension(
442                        ctx,
443                        rel.advanced_extension.as_ref(),
444                    ),
445                    children: vec![],
446                }
447            }
448            Some(ReadType::ExtensionTable(table)) => {
449                let decoded = match table.detail.as_ref().map(AnyRef::from) {
450                    Some(detail) => ctx.extension_registry().decode_extension_table(detail),
451                    None => Err(ExtensionError::MissingDetail),
452                };
453
454                Relation {
455                    name: Cow::Borrowed("Read:Extension"),
456                    arguments: None,
457                    columns,
458                    emit,
459                    output_syntax: OutputSyntax::Implicit,
460                    addenda: AddendumLines::extension_table(
461                        ctx,
462                        decoded,
463                        rel.advanced_extension.as_ref(),
464                    ),
465                    children: vec![],
466                }
467            }
468            other => {
469                let err = PlanError::unimplemented(
470                    "ReadRel",
471                    Some("read_type"),
472                    format!("Unsupported read type {other:?}"),
473                );
474                Relation {
475                    name: Cow::Borrowed("Read"),
476                    arguments: Some(RelationArgs::inline(vec![Value::Missing(err)], vec![])),
477                    columns,
478                    emit,
479                    output_syntax: OutputSyntax::Implicit,
480                    addenda: AddendumLines::from_advanced_extension(
481                        ctx,
482                        rel.advanced_extension.as_ref(),
483                    ),
484                    children: vec![],
485                }
486            }
487        }
488    }
489}
490
491fn read_columns<'a>(rel: &'a ReadRel) -> Vec<Value<'a>> {
492    match rel.base_schema {
493        Some(ref schema) => schema_to_values(schema),
494        None => {
495            let err =
496                PlanError::unimplemented("ReadRel", Some("base_schema"), "Base schema is required");
497            vec![Value::Missing(err)]
498        }
499    }
500}
501
502pub fn get_emit(rel: Option<&RelCommon>) -> Option<&EmitKind> {
503    rel.as_ref().and_then(|c| c.emit_kind.as_ref())
504}
505
506impl<'a> Relation<'a> {
507    /// Convert a vector of relation references into their structured form.
508    ///
509    /// Returns a list of children (with None for ones missing), and a count of input columns.
510    pub fn convert_children<S: Scope>(
511        refs: Vec<Option<&'a Rel>>,
512        ctx: &S,
513    ) -> (Vec<Option<Relation<'a>>>, usize) {
514        let mut children = vec![];
515        let mut inputs = 0;
516
517        for maybe_rel in refs {
518            match maybe_rel {
519                Some(rel) => {
520                    let child = Relation::from_rel(rel, ctx);
521                    inputs += child.emitted();
522                    children.push(Some(child));
523                }
524                None => children.push(None),
525            }
526        }
527
528        (children, inputs)
529    }
530}
531
532impl<'a> Relation<'a> {
533    fn from_filter<S: Scope>(rel: &'a FilterRel, ctx: &S) -> Self {
534        let condition = rel
535            .condition
536            .as_ref()
537            .map(|c| Value::Expression(c.as_ref()));
538        let condition = Value::expect(condition, || {
539            PlanError::unimplemented("FilterRel", Some("condition"), "Condition is None")
540        });
541        let positional = vec![condition];
542        let arguments = Some(RelationArgs::inline(positional, vec![]));
543        let emit = get_emit(rel.common.as_ref());
544        let (children, columns) = Relation::convert_children(vec![rel.input.as_deref()], ctx);
545        let columns = (0..columns).map(|i| Value::Reference(i as i32)).collect();
546
547        Relation {
548            name: Cow::Borrowed("Filter"),
549            arguments,
550            columns,
551            emit,
552            output_syntax: OutputSyntax::Implicit,
553            addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
554            children,
555        }
556    }
557
558    fn from_project<S: Scope>(rel: &'a ProjectRel, ctx: &S) -> Self {
559        let (children, input_columns) = Relation::convert_children(vec![rel.input.as_deref()], ctx);
560        let mut columns: Vec<Value> = vec![];
561        for i in 0..input_columns {
562            columns.push(Value::Reference(i as i32));
563        }
564        for expr in &rel.expressions {
565            columns.push(Value::Expression(expr));
566        }
567
568        Relation {
569            name: Cow::Borrowed("Project"),
570            arguments: None,
571            columns,
572            emit: get_emit(rel.common.as_ref()),
573            output_syntax: OutputSyntax::Implicit,
574            addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
575            children,
576        }
577    }
578
579    pub fn from_rel<S: Scope>(rel: &'a Rel, ctx: &S) -> Self {
580        match rel.rel_type.as_ref() {
581            Some(RelType::Read(r)) => Relation::from_read(r, ctx),
582            Some(RelType::Filter(r)) => Relation::from_filter(r, ctx),
583            Some(RelType::Project(r)) => Relation::from_project(r, ctx),
584            Some(RelType::Aggregate(r)) => Relation::from_aggregate(r, ctx),
585            Some(RelType::Sort(r)) => Relation::from_sort(r, ctx),
586            Some(RelType::Fetch(r)) => Relation::from_fetch(r, ctx),
587            Some(RelType::Join(r)) => Relation::from_join(r, ctx),
588            Some(RelType::Set(r)) => Relation::from_set(r, ctx),
589            Some(RelType::Cross(r)) => Relation::from_cross(r, ctx),
590            Some(RelType::ExtensionLeaf(r)) => Relation::from_extension_leaf(r, ctx),
591            Some(RelType::ExtensionSingle(r)) => Relation::from_extension_single(r, ctx),
592            Some(RelType::ExtensionMulti(r)) => Relation::from_extension_multi(r, ctx),
593            _ => {
594                let name = rel.name();
595                let token = ctx.failure(FormatError::Format(PlanError::unimplemented(
596                    "Rel",
597                    Some(name),
598                    format!("{name} is not yet supported in the text format"),
599                )));
600                Relation {
601                    name: Cow::Owned(format!("{token}")),
602                    arguments: None,
603                    columns: vec![],
604                    emit: None,
605                    output_syntax: OutputSyntax::Implicit,
606                    addenda: AddendumLines::none(),
607                    children: vec![],
608                }
609            }
610        }
611    }
612
613    fn from_extension_leaf<S: Scope>(rel: &'a ExtensionLeafRel, ctx: &S) -> Self {
614        let detail_ref = rel.detail.as_ref().map(AnyRef::from);
615        let decoded = match detail_ref {
616            Some(d) => ctx.extension_registry().decode(d),
617            None => Err(ExtensionError::MissingDetail),
618        };
619        Relation::from_extension("ExtensionLeaf", decoded, vec![], ctx)
620    }
621
622    fn from_extension_single<S: Scope>(rel: &'a ExtensionSingleRel, ctx: &S) -> Self {
623        let detail_ref = rel.detail.as_ref().map(AnyRef::from);
624        let decoded = match detail_ref {
625            Some(d) => ctx.extension_registry().decode(d),
626            None => Err(ExtensionError::MissingDetail),
627        };
628        Relation::from_extension("ExtensionSingle", decoded, vec![rel.input.as_deref()], ctx)
629    }
630
631    fn from_extension_multi<S: Scope>(rel: &'a ExtensionMultiRel, ctx: &S) -> Self {
632        let detail_ref = rel.detail.as_ref().map(AnyRef::from);
633        let decoded = match detail_ref {
634            Some(d) => ctx.extension_registry().decode(d),
635            None => Err(ExtensionError::MissingDetail),
636        };
637        let mut child_refs: Vec<Option<&'a Rel>> = vec![];
638        for input in &rel.inputs {
639            child_refs.push(Some(input));
640        }
641        Relation::from_extension("ExtensionMulti", decoded, child_refs, ctx)
642    }
643
644    fn from_extension<S: Scope>(
645        ext_type: &'static str,
646        decoded: Result<(String, ExtensionArgs), ExtensionError>,
647        child_refs: Vec<Option<&'a Rel>>,
648        ctx: &S,
649    ) -> Self {
650        match decoded {
651            Ok((name, args)) => {
652                let (children, _) = Relation::convert_children(child_refs, ctx);
653                let mut positional = vec![];
654                for value in args.positional {
655                    positional.push(Value::ExtensionArgument(value));
656                }
657                let mut named = vec![];
658                for (key, value) in args.named {
659                    named.push(NamedArg {
660                        name: Cow::Owned(key),
661                        value: Value::ExtensionArgument(value),
662                    });
663                }
664                let mut columns = vec![];
665                for col in args.output_columns {
666                    columns.push(Value::ExtColumn(col));
667                }
668                Relation {
669                    name: Cow::Owned(format!("{}:{}", ext_type, name)),
670                    arguments: Some(RelationArgs::inline(positional, named)),
671                    columns,
672                    emit: None,
673                    output_syntax: OutputSyntax::Implicit,
674                    // Extension relations use `detail` rather than
675                    // `advanced_extension`; the field does not exist on these
676                    // proto types.
677                    addenda: AddendumLines::none(),
678                    children,
679                }
680            }
681            Err(error) => {
682                let (children, _) = Relation::convert_children(child_refs, ctx);
683                Relation {
684                    name: Cow::Borrowed(ext_type),
685                    arguments: None,
686                    columns: vec![Value::Missing(PlanError::invalid(
687                        "extension",
688                        None::<&str>,
689                        error.to_string(),
690                    ))],
691                    emit: None,
692                    output_syntax: OutputSyntax::Implicit,
693                    addenda: AddendumLines::none(),
694                    children,
695                }
696            }
697        }
698    }
699
700    /// Convert an AggregateRel to a Relation for textification.
701    ///
702    /// The conversion follows this logic:
703    /// 1. Arguments: Group-by expressions (as Value::Expression)
704    /// 2. Columns: All possible outputs in order:
705    ///    - First: Group-by field references (Value::Reference)
706    ///    - Then: Aggregate function measures (Value::AggregateFunction)
707    /// 3. Emit: Uses the relation's emit mapping to select which outputs to display
708    /// 4. Children: The input relation
709    fn from_aggregate<S: Scope>(rel: &'a AggregateRel, ctx: &S) -> Self {
710        let mut grouping_sets: Vec<Vec<Value>> = vec![]; // the Groupings in the Aggregate
711        let expression_list: Vec<Value>; // grouping_expressions defined on Aggregate
712
713        // if rel.grouping_expressions is empty, the deprecated rel.groupings.grouping_expressions might be set
714        // If *both* the deprecated `rel.groupings.grouping_expressions` and `rel.grouping_expressions` are
715        // set, then we silently ignore the deprecated one.
716        #[allow(deprecated)]
717        if rel.grouping_expressions.is_empty()
718            && !rel.groupings.is_empty()
719            && !rel.groupings[0].grouping_expressions.is_empty()
720        {
721            (expression_list, grouping_sets) = Relation::get_grouping_sets(rel);
722        } else {
723            expression_list = rel
724                .grouping_expressions
725                .iter()
726                .map(Value::Expression)
727                .collect::<Vec<_>>(); // already a list of the unique expressions
728            for group in &rel.groupings {
729                let mut grouping_set: Vec<Value> = vec![];
730                for i in &group.expression_references {
731                    let value = match rel.grouping_expressions.get(*i as usize) {
732                        Some(expr) => Value::Expression(expr),
733                        None => Value::Missing(PlanError::invalid(
734                            "AggregateRel",
735                            Some("groupings.expression_references"),
736                            format!(
737                                "expression_reference {i} is out of bounds for grouping_expressions of length {}",
738                                rel.grouping_expressions.len()
739                            ),
740                        )),
741                    };
742                    grouping_set.push(value);
743                }
744                grouping_sets.push(grouping_set);
745            }
746            // no defined groupings means there is global group by
747            if rel.groupings.is_empty() {
748                grouping_sets.push(vec![]);
749            }
750        }
751
752        let is_single = grouping_sets.len() == 1;
753        let mut positional: Vec<Value> = vec![];
754        for g in grouping_sets {
755            if g.is_empty() {
756                positional.push(Value::EmptyGroup);
757            } else if is_single {
758                // Single non-empty grouping set: spread expressions directly without parens
759                positional.extend(g);
760            } else {
761                positional.push(Value::Tuple(g));
762            }
763        }
764
765        // adding the grouping_sets as a list of Arguments to Aggregate Rel
766        let arguments = Some(RelationArgs::inline(positional, vec![]));
767
768        // The columns are the direct outputs of this relation (before emit)
769        let mut all_outputs: Vec<Value> = expression_list;
770
771        // Then, add all measures (aggregate functions)
772        // These are indexed after the group-by fields
773        for m in &rel.measures {
774            if let Some(agg_fn) = m.measure.as_ref() {
775                all_outputs.push(Value::AggregateFunction(agg_fn));
776            }
777        }
778        let emit = get_emit(rel.common.as_ref());
779        let (children, _) = Relation::convert_children(vec![rel.input.as_deref()], ctx);
780
781        Relation {
782            name: Cow::Borrowed("Aggregate"),
783            arguments,
784            columns: all_outputs,
785            emit,
786            output_syntax: OutputSyntax::Implicit,
787            addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
788            children,
789        }
790    }
791
792    fn get_grouping_sets(rel: &'a AggregateRel) -> (Vec<Value<'a>>, Vec<Vec<Value<'a>>>) {
793        let mut grouping_sets: Vec<Vec<Value>> = vec![];
794        let mut expression_list: Vec<Value> = Vec::new();
795
796        // groupings might have the same expressions in their set, so we track
797        // which byte-encoded expressions have already been added to
798        // `expression_list` to keep it deduplicated.
799        let mut seen_expressions = HashSet::new();
800
801        for group in &rel.groupings {
802            let mut grouping_set: Vec<Value> = vec![];
803            #[allow(deprecated)]
804            for exp in &group.grouping_expressions {
805                // TODO: use a better key here than encoding to bytes.
806                // Ideally, substrait-rs would support `PartialEq` and `Hash`,
807                // but as there isn't an easy way to do that now, we'll skip.
808                if seen_expressions.insert(exp.encode_to_vec()) {
809                    expression_list.push(Value::Expression(exp)); // new unique expression found
810                }
811                grouping_set.push(Value::Expression(exp));
812            }
813            grouping_sets.push(grouping_set);
814        }
815        (expression_list, grouping_sets)
816    }
817}
818
819impl Textify for RelRoot {
820    fn name() -> &'static str {
821        "RelRoot"
822    }
823
824    fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
825        let names = self.names.iter().map(|n| Name(n)).collect::<Vec<_>>();
826
827        write!(
828            w,
829            "{}Root[{}]",
830            ctx.indent(),
831            ctx.separated(names.iter(), ", ")
832        )?;
833        let child_scope = ctx.push_indent();
834        for child in self.input.iter() {
835            writeln!(w)?;
836            child.textify(&child_scope, w)?;
837        }
838
839        Ok(())
840    }
841}
842
843impl Textify for PlanRelType {
844    fn name() -> &'static str {
845        "PlanRelType"
846    }
847
848    fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
849        match self {
850            PlanRelType::Rel(rel) => rel.textify(ctx, w),
851            PlanRelType::Root(root) => root.textify(ctx, w),
852        }
853    }
854}
855
856impl Textify for PlanRel {
857    fn name() -> &'static str {
858        "PlanRel"
859    }
860
861    /// Write the relation as a string. Inputs are ignored - those are handled
862    /// separately.
863    fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
864        write!(w, "{}", ctx.expect(self.rel_type.as_ref()))
865    }
866}
867
868impl<'a> Relation<'a> {
869    fn from_sort<S: Scope>(rel: &'a SortRel, ctx: &S) -> Self {
870        let (children, input_columns) = Relation::convert_children(vec![rel.input.as_deref()], ctx);
871        let mut positional = vec![];
872        for sort_field in &rel.sorts {
873            positional.push(Value::from(sort_field));
874        }
875        let arguments = Some(RelationArgs::inline(positional, vec![]));
876        // The columns are the direct outputs of this relation (before emit)
877        let mut col_values = vec![];
878        for i in 0..input_columns {
879            col_values.push(Value::Reference(i as i32));
880        }
881        let emit = get_emit(rel.common.as_ref());
882        Relation {
883            name: Cow::Borrowed("Sort"),
884            arguments,
885            columns: col_values,
886            emit,
887            output_syntax: OutputSyntax::Implicit,
888            addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
889            children,
890        }
891    }
892
893    fn from_fetch<S: Scope>(rel: &'a FetchRel, ctx: &S) -> Self {
894        let (children, input_columns) = Relation::convert_children(vec![rel.input.as_deref()], ctx);
895        let mut named_args: Vec<NamedArg> = vec![];
896        match &rel.count_mode {
897            Some(CountMode::CountExpr(expr)) => {
898                named_args.push(NamedArg {
899                    name: Cow::Borrowed("limit"),
900                    value: Value::Expression(expr),
901                });
902            }
903            #[allow(deprecated)]
904            Some(CountMode::Count(val)) => {
905                named_args.push(NamedArg {
906                    name: Cow::Borrowed("limit"),
907                    value: Value::Integer(*val),
908                });
909            }
910            None => {}
911        }
912        if let Some(offset) = &rel.offset_mode {
913            match offset {
914                substrait::proto::fetch_rel::OffsetMode::OffsetExpr(expr) => {
915                    named_args.push(NamedArg {
916                        name: Cow::Borrowed("offset"),
917                        value: Value::Expression(expr),
918                    });
919                }
920                #[allow(deprecated)]
921                substrait::proto::fetch_rel::OffsetMode::Offset(val) => {
922                    named_args.push(NamedArg {
923                        name: Cow::Borrowed("offset"),
924                        value: Value::Integer(*val),
925                    });
926                }
927            }
928        }
929
930        let emit = get_emit(rel.common.as_ref());
931        // Fetch is passthrough — direct output is all input columns.
932        let columns: Vec<Value> = (0..input_columns)
933            .map(|i| Value::Reference(i as i32))
934            .collect();
935        Relation {
936            name: Cow::Borrowed("Fetch"),
937            arguments: Some(RelationArgs::inline(vec![], named_args)),
938            columns,
939            emit,
940            output_syntax: OutputSyntax::Implicit,
941            addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
942            children,
943        }
944    }
945}
946
947fn join_output_columns(
948    join_type: join_rel::JoinType,
949    left_columns: usize,
950    right_columns: usize,
951) -> Vec<Value<'static>> {
952    let total_columns = match join_type {
953        // Inner, Left, Right, Outer joins output columns from both sides
954        join_rel::JoinType::Inner
955        | join_rel::JoinType::Left
956        | join_rel::JoinType::Right
957        | join_rel::JoinType::Outer => left_columns + right_columns,
958
959        // Left semi/anti joins only output columns from the left side
960        join_rel::JoinType::LeftSemi | join_rel::JoinType::LeftAnti => left_columns,
961
962        // Right semi/anti joins output columns from the right side
963        join_rel::JoinType::RightSemi | join_rel::JoinType::RightAnti => right_columns,
964
965        // Single joins behave like semi joins
966        join_rel::JoinType::LeftSingle => left_columns,
967        join_rel::JoinType::RightSingle => right_columns,
968
969        // Mark joins output base columns plus one mark column
970        join_rel::JoinType::LeftMark => left_columns + 1,
971        join_rel::JoinType::RightMark => right_columns + 1,
972
973        // Unspecified - fallback to all columns
974        join_rel::JoinType::Unspecified => left_columns + right_columns,
975    };
976
977    // Output is always a contiguous range starting from $0
978    (0..total_columns)
979        .map(|i| Value::Reference(i as i32))
980        .collect()
981}
982
983impl<'a> Relation<'a> {
984    fn from_join<S: Scope>(rel: &'a JoinRel, ctx: &S) -> Self {
985        let (children, _total_columns) =
986            Relation::convert_children(vec![rel.left.as_deref(), rel.right.as_deref()], ctx);
987
988        // convert_children should preserve input vector length
989        assert_eq!(
990            children.len(),
991            2,
992            "convert_children should return same number of elements as input"
993        );
994
995        // Calculate left and right column counts separately
996        let left_columns = match &children[0] {
997            Some(child) => child.emitted(),
998            None => 0,
999        };
1000        let right_columns = match &children[1] {
1001            Some(child) => child.emitted(),
1002            None => 0,
1003        };
1004
1005        // Convert join type from protobuf i32 to enum value
1006        // JoinType is stored as i32 in protobuf, convert to typed enum for processing
1007        let (join_type, join_type_value) = match join_rel::JoinType::try_from(rel.r#type) {
1008            Ok(join_type) => {
1009                let join_type_value = match join_type.as_enum_str() {
1010                    Ok(s) => Value::Enum(s),
1011                    Err(e) => Value::Missing(e),
1012                };
1013                (join_type, join_type_value)
1014            }
1015            Err(_) => {
1016                // Use Unspecified for the join_type but create an error for the join_type_value
1017                let join_type_error = Value::Missing(PlanError::invalid(
1018                    "JoinRel",
1019                    Some("type"),
1020                    format!("Unknown join type: {}", rel.r#type),
1021                ));
1022                (join_rel::JoinType::Unspecified, join_type_error)
1023            }
1024        };
1025
1026        // Join condition
1027        let condition = rel
1028            .expression
1029            .as_ref()
1030            .map(|c| Value::Expression(c.as_ref()));
1031        let condition = Value::expect(condition, || {
1032            PlanError::unimplemented("JoinRel", Some("expression"), "Join condition is None")
1033        });
1034
1035        let positional = vec![join_type_value, condition];
1036        let mut named = vec![];
1037        if let Some(post_join_filter) = rel.post_join_filter.as_ref() {
1038            named.push(NamedArg {
1039                name: Cow::Borrowed("post_filter"),
1040                value: Value::Expression(post_join_filter.as_ref()),
1041            });
1042        }
1043        let arguments = Some(RelationArgs::inline(positional, named));
1044
1045        let emit = get_emit(rel.common.as_ref());
1046        let columns = join_output_columns(join_type, left_columns, right_columns);
1047
1048        Relation {
1049            name: Cow::Borrowed("Join"),
1050            arguments,
1051            columns,
1052            emit,
1053            output_syntax: OutputSyntax::Implicit,
1054            addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
1055            children,
1056        }
1057    }
1058
1059    fn from_set<S: Scope>(rel: &'a SetRel, ctx: &S) -> Self {
1060        let child_refs: Vec<Option<&'a Rel>> = rel.inputs.iter().map(Some).collect();
1061        let (children, total_columns) = Relation::convert_children(child_refs, ctx);
1062
1063        // Set relation output has the same width as any one of its inputs
1064        // (it's a pass-through, not a concatenation like Join).
1065        // TODO: we may want to validate that all inputs have the same width
1066        // (and schema, if possible...), and provide a warning if they do not.
1067        let width = if children.is_empty() {
1068            0
1069        } else {
1070            total_columns / children.len()
1071        };
1072
1073        let op_value = decode_enum_field::<set_rel::SetOp>(rel.op, "SetRel", "op");
1074
1075        let arguments = Some(RelationArgs::inline(vec![op_value], vec![]));
1076        let emit = get_emit(rel.common.as_ref());
1077        let columns = (0..width).map(|i| Value::Reference(i as i32)).collect();
1078
1079        Relation {
1080            name: Cow::Borrowed("Set"),
1081            arguments,
1082            columns,
1083            emit,
1084            output_syntax: OutputSyntax::Implicit,
1085            addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
1086            children,
1087        }
1088    }
1089
1090    fn from_cross<S: Scope>(rel: &'a CrossRel, ctx: &S) -> Self {
1091        let (children, total_columns) =
1092            Relation::convert_children(vec![rel.left.as_deref(), rel.right.as_deref()], ctx);
1093
1094        // Output columns concatenate the left and right inputs; there is no
1095        // join-type column dropping, since CrossRel has none.
1096        let columns = (0..total_columns)
1097            .map(|i| Value::Reference(i as i32))
1098            .collect();
1099
1100        Relation {
1101            name: Cow::Borrowed("Cross"),
1102            arguments: None,
1103            columns,
1104            emit: get_emit(rel.common.as_ref()),
1105            output_syntax: OutputSyntax::Implicit,
1106            addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
1107            children,
1108        }
1109    }
1110}
1111
1112#[cfg(test)]
1113mod tests {
1114    use substrait::proto::aggregate_rel::Grouping;
1115    use substrait::proto::expression::literal::LiteralType;
1116    use substrait::proto::expression::{Literal, RexType, ScalarFunction};
1117    use substrait::proto::function_argument::ArgType;
1118    use substrait::proto::read_rel::{NamedTable, ReadType};
1119    use substrait::proto::rel_common::{Direct, Emit};
1120    use substrait::proto::r#type::{self as ptype, Boolean, I64, Kind, Nullability, Struct};
1121    use substrait::proto::{
1122        AggregateFunction, Expression, FunctionArgument, NamedStruct, ReadRel, Type, aggregate_rel,
1123    };
1124
1125    use super::*;
1126    use crate::fixtures::TestContext;
1127    use crate::parser::expressions::FieldIndex;
1128    use crate::textify::expressions::Reference;
1129
1130    #[test]
1131    fn test_read_rel() {
1132        let ctx = TestContext::new();
1133
1134        // Create a simple ReadRel with a NamedStruct schema
1135        let read_rel = ReadRel {
1136            common: None,
1137            base_schema: Some(NamedStruct {
1138                names: vec!["col1".into(), "column 2".into()],
1139                r#struct: Some(Struct {
1140                    type_variation_reference: 0,
1141                    types: vec![
1142                        Type {
1143                            kind: Some(Kind::I32(ptype::I32 {
1144                                type_variation_reference: 0,
1145                                nullability: Nullability::Nullable as i32,
1146                            })),
1147                        },
1148                        Type {
1149                            kind: Some(Kind::String(ptype::String {
1150                                type_variation_reference: 0,
1151                                nullability: Nullability::Nullable as i32,
1152                            })),
1153                        },
1154                    ],
1155                    nullability: Nullability::Nullable as i32,
1156                }),
1157            }),
1158            filter: None,
1159            best_effort_filter: None,
1160            projection: None,
1161            advanced_extension: None,
1162            read_type: Some(ReadType::NamedTable(NamedTable {
1163                names: vec!["some_db".into(), "test_table".into()],
1164                advanced_extension: None,
1165            })),
1166        };
1167
1168        let rel = Rel {
1169            rel_type: Some(RelType::Read(Box::new(read_rel))),
1170        };
1171        let (result, errors) = ctx.textify(&rel);
1172        assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1173        assert_eq!(
1174            result,
1175            "Read[some_db.test_table => col1:i32?, \"column 2\":string?]"
1176        );
1177    }
1178
1179    #[test]
1180    fn test_filter_rel() {
1181        let ctx = TestContext::new()
1182            .with_urn(1, "test_urn")
1183            .with_function(1, 10, "gt");
1184
1185        // Create a simple FilterRel with a ReadRel input and a filter expression
1186        let read_rel = ReadRel {
1187            common: None,
1188            base_schema: Some(NamedStruct {
1189                names: vec!["col1".into(), "col2".into()],
1190                r#struct: Some(Struct {
1191                    type_variation_reference: 0,
1192                    types: vec![
1193                        Type {
1194                            kind: Some(Kind::I32(ptype::I32 {
1195                                type_variation_reference: 0,
1196                                nullability: Nullability::Nullable as i32,
1197                            })),
1198                        },
1199                        Type {
1200                            kind: Some(Kind::I32(ptype::I32 {
1201                                type_variation_reference: 0,
1202                                nullability: Nullability::Nullable as i32,
1203                            })),
1204                        },
1205                    ],
1206                    nullability: Nullability::Nullable as i32,
1207                }),
1208            }),
1209            filter: None,
1210            best_effort_filter: None,
1211            projection: None,
1212            advanced_extension: None,
1213            read_type: Some(ReadType::NamedTable(NamedTable {
1214                names: vec!["test_table".into()],
1215                advanced_extension: None,
1216            })),
1217        };
1218
1219        // Create a filter expression: col1 > 10
1220        let filter_expr = Expression {
1221            rex_type: Some(RexType::ScalarFunction(ScalarFunction {
1222                function_reference: 10, // gt function
1223                arguments: vec![
1224                    FunctionArgument {
1225                        arg_type: Some(ArgType::Value(Reference(0).into())),
1226                    },
1227                    FunctionArgument {
1228                        arg_type: Some(ArgType::Value(Expression {
1229                            rex_type: Some(RexType::Literal(Literal {
1230                                literal_type: Some(LiteralType::I32(10)),
1231                                nullable: false,
1232                                type_variation_reference: 0,
1233                            })),
1234                        })),
1235                    },
1236                ],
1237                options: vec![],
1238                output_type: Some(Type {
1239                    kind: Some(Kind::Bool(Boolean {
1240                        nullability: Nullability::Required as i32,
1241                        type_variation_reference: 0,
1242                    })),
1243                }),
1244                #[allow(deprecated)]
1245                args: vec![],
1246            })),
1247        };
1248
1249        let filter_rel = FilterRel {
1250            common: None,
1251            input: Some(Box::new(Rel {
1252                rel_type: Some(RelType::Read(Box::new(read_rel))),
1253            })),
1254            condition: Some(Box::new(filter_expr)),
1255            advanced_extension: None,
1256        };
1257
1258        let rel = Rel {
1259            rel_type: Some(RelType::Filter(Box::new(filter_rel))),
1260        };
1261
1262        let (result, errors) = ctx.textify(&rel);
1263        assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1264        let expected = r#"
1265Filter[gt($0, 10:i32):boolean => $0, $1]
1266  Read[test_table => col1:i32?, col2:i32?]"#
1267            .trim_start();
1268        assert_eq!(result, expected);
1269    }
1270
1271    #[test]
1272    fn test_aggregate_function_textify() {
1273        let ctx = TestContext::new()
1274        .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1275        .with_function(1, 10, "sum")
1276        .with_function(1, 11, "count");
1277
1278        // Create a simple AggregateFunction
1279        let agg_fn = get_aggregate_func(10, 1);
1280
1281        let value = Value::AggregateFunction(&agg_fn);
1282        let (result, errors) = ctx.textify(&value);
1283
1284        assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1285        assert_eq!(result, "sum($1):i64");
1286    }
1287
1288    #[test]
1289    fn test_aggregate_relation_textify() {
1290        let ctx = TestContext::new()
1291        .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1292        .with_function(1, 10, "sum")
1293        .with_function(1, 11, "count");
1294
1295        // Create a simple AggregateRel
1296        let agg_fn1 = get_aggregate_func(10, 1);
1297        let agg_fn2 = get_aggregate_func(11, 1);
1298
1299        let grouping_expressions = vec![Expression {
1300            rex_type: Some(RexType::Selection(Box::new(
1301                FieldIndex(0).to_field_reference(),
1302            ))),
1303        }];
1304
1305        let measures = vec![
1306            aggregate_rel::Measure {
1307                measure: Some(agg_fn1),
1308                filter: None,
1309            },
1310            aggregate_rel::Measure {
1311                measure: Some(agg_fn2),
1312                filter: None,
1313            },
1314        ];
1315
1316        let common = Some(RelCommon {
1317            emit_kind: Some(EmitKind::Emit(Emit {
1318                output_mapping: vec![1, 2], // measures only
1319            })),
1320            ..Default::default()
1321        });
1322
1323        let aggregate_rel = create_aggregate_rel(grouping_expressions, vec![], measures, common);
1324
1325        let rel = Rel {
1326            rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1327        };
1328        let (result, errors) = ctx.textify(&rel);
1329
1330        assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1331        // Expected: Aggregate[_ => sum($1):i64, count($1):i64] we chose to emit only measures
1332        assert!(result.contains("Aggregate[_ => sum($1):i64, count($1):i64]"));
1333    }
1334
1335    #[test]
1336    fn test_multiple_groupings_on_aggregate_deprecated() {
1337        // Protobuf plan that uses AggregateRel.groupings with deprecated
1338        // grouping_expressions, leaving AggregateRel.grouping_expressions empty.
1339        let ctx = TestContext::new()
1340        .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1341        .with_function(1, 11, "count");
1342
1343        let grouping_expr_0 = create_exp(0);
1344        let grouping_expr_1 = create_exp(1);
1345
1346        let grouping_sets = vec![
1347            aggregate_rel::Grouping {
1348                #[allow(deprecated)]
1349                grouping_expressions: vec![grouping_expr_0.clone()],
1350                expression_references: vec![],
1351            },
1352            aggregate_rel::Grouping {
1353                #[allow(deprecated)]
1354                grouping_expressions: vec![grouping_expr_0.clone(), grouping_expr_1.clone()],
1355                expression_references: vec![],
1356            },
1357        ];
1358
1359        let aggregate_rel = create_aggregate_rel(vec![], grouping_sets, vec![], None);
1360
1361        let rel = Rel {
1362            rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1363        };
1364        let (result, errors) = ctx.textify(&rel);
1365
1366        assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1367        assert!(result.contains("Aggregate[($0), ($0, $1) => $0, $1]"));
1368    }
1369
1370    #[test]
1371    fn test_multiple_groupings_with_measure_deprecated() {
1372        // Protobuf plan that uses AggregateRel.groupings with deprecated
1373        // grouping_expressions, leaving AggregateRel.grouping_expressions empty.
1374        let ctx = TestContext::new()
1375        .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1376        .with_function(1, 11, "count");
1377
1378        let agg_fn1 = get_aggregate_func(11, 2);
1379
1380        let grouping_expr_0 = create_exp(0);
1381        let grouping_expr_1 = create_exp(1);
1382
1383        let grouping_sets = vec![
1384            aggregate_rel::Grouping {
1385                #[allow(deprecated)]
1386                grouping_expressions: vec![grouping_expr_0.clone()],
1387                expression_references: vec![],
1388            },
1389            aggregate_rel::Grouping {
1390                #[allow(deprecated)]
1391                grouping_expressions: vec![grouping_expr_0.clone(), grouping_expr_1.clone()],
1392                expression_references: vec![],
1393            },
1394        ];
1395
1396        let measures = vec![aggregate_rel::Measure {
1397            measure: Some(agg_fn1),
1398            filter: None,
1399        }];
1400
1401        let aggregate_rel = create_aggregate_rel(vec![], grouping_sets, measures, None);
1402
1403        let rel = Rel {
1404            rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1405        };
1406        let (result, errors) = ctx.textify(&rel);
1407
1408        assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1409        assert!(result.contains("($0), ($0, $1) => $0, $1, count($2):i64"));
1410    }
1411
1412    #[test]
1413    fn test_multiple_groupings_on_aggregate() {
1414        let ctx = TestContext::new()
1415        .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1416        .with_function(1, 11, "count");
1417
1418        let agg_fn2 = get_aggregate_func(11, 2);
1419
1420        let grouping_expressions = vec![
1421            Expression {
1422                rex_type: Some(RexType::Selection(Box::new(
1423                    FieldIndex(0).to_field_reference(),
1424                ))),
1425            },
1426            Expression {
1427                rex_type: Some(RexType::Selection(Box::new(
1428                    FieldIndex(1).to_field_reference(),
1429                ))),
1430            },
1431        ];
1432
1433        let grouping_sets = vec![
1434            Grouping {
1435                #[allow(deprecated)]
1436                grouping_expressions: vec![],
1437                expression_references: vec![0, 1],
1438            },
1439            Grouping {
1440                #[allow(deprecated)]
1441                grouping_expressions: vec![],
1442                expression_references: vec![0, 1],
1443            },
1444            Grouping {
1445                #[allow(deprecated)]
1446                grouping_expressions: vec![],
1447                expression_references: vec![1],
1448            },
1449            Grouping {
1450                #[allow(deprecated)]
1451                grouping_expressions: vec![],
1452                expression_references: vec![1, 1],
1453            },
1454            Grouping {
1455                #[allow(deprecated)]
1456                grouping_expressions: vec![],
1457                expression_references: vec![],
1458            },
1459        ];
1460
1461        let measures = vec![aggregate_rel::Measure {
1462            measure: Some(agg_fn2),
1463            filter: None,
1464        }];
1465
1466        let aggregate_rel =
1467            create_aggregate_rel(grouping_expressions, grouping_sets, measures, None);
1468
1469        let rel = Rel {
1470            rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1471        };
1472        let (result, errors) = ctx.textify(&rel);
1473
1474        assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1475        assert!(
1476            result.contains(
1477                "Aggregate[($0, $1), ($0, $1), ($1), ($1, $1), _ => $0, $1, count($2):i64]"
1478            )
1479        );
1480    }
1481
1482    #[test]
1483    fn test_deprecated_reordered_grouping() {
1484        // Protobuf plan that uses the deprecated per-Grouping
1485        // `grouping_expressions`, leaving `AggregateRel.grouping_expressions`
1486        // empty. The lone unique expression here is $5, but it is the first
1487        // (index 0) expression discovered during deduplication - so if the
1488        // grouping set were rendered from that dedup index rather than from
1489        // the expression itself, it would wrongly print as `$0` instead of
1490        // `$5`.
1491        let ctx = TestContext::new();
1492        let grouping_expr_5 = create_exp(5);
1493
1494        let grouping_sets = vec![aggregate_rel::Grouping {
1495            #[allow(deprecated)]
1496            grouping_expressions: vec![grouping_expr_5],
1497            expression_references: vec![],
1498        }];
1499
1500        let aggregate_rel = create_aggregate_rel(vec![], grouping_sets, vec![], None);
1501        let rel = Rel {
1502            rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1503        };
1504        let (result, errors) = ctx.textify(&rel);
1505
1506        assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1507        assert!(result.contains("Aggregate[$5 => $5]"));
1508    }
1509
1510    #[test]
1511    fn test_reordered_grouping_textifies_expression_not_raw_index() {
1512        let ctx = TestContext::new()
1513            .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1514            .with_function(1, 11, "count");
1515
1516        let agg_fn2 = get_aggregate_func(11, 1);
1517
1518        // grouping_expressions is [$2, $0] (textual order); the single
1519        // grouping set references both by index into grouping_expressions:
1520        // [0, 1]. Those indexes must resolve back through
1521        // grouping_expressions ([$2, $0]), not be printed directly as `$0,
1522        // $1`.
1523        let grouping_expressions = vec![
1524            Expression {
1525                rex_type: Some(RexType::Selection(Box::new(
1526                    FieldIndex(2).to_field_reference(),
1527                ))),
1528            },
1529            Expression {
1530                rex_type: Some(RexType::Selection(Box::new(
1531                    FieldIndex(0).to_field_reference(),
1532                ))),
1533            },
1534        ];
1535
1536        let grouping_sets = vec![Grouping {
1537            #[allow(deprecated)]
1538            grouping_expressions: vec![],
1539            expression_references: vec![0, 1],
1540        }];
1541
1542        let measures = vec![aggregate_rel::Measure {
1543            measure: Some(agg_fn2),
1544            filter: None,
1545        }];
1546
1547        let aggregate_rel =
1548            create_aggregate_rel(grouping_expressions, grouping_sets, measures, None);
1549
1550        let rel = Rel {
1551            rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1552        };
1553        let (result, errors) = ctx.textify(&rel);
1554
1555        assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1556        assert!(result.contains("Aggregate[$2, $0 => $2, $0, count($1):i64]"));
1557    }
1558
1559    #[test]
1560    fn test_join_relation_unknown_type() {
1561        let ctx = TestContext::new();
1562
1563        // Create a join with an unknown/invalid type
1564        let join_rel = JoinRel {
1565            left: Some(Box::new(Rel {
1566                rel_type: Some(RelType::Read(Box::default())),
1567            })),
1568            right: Some(Box::new(Rel {
1569                rel_type: Some(RelType::Read(Box::default())),
1570            })),
1571            expression: Some(Box::new(Expression::default())),
1572            r#type: 999, // Invalid join type
1573            common: None,
1574            post_join_filter: None,
1575            advanced_extension: None,
1576        };
1577
1578        let rel = Rel {
1579            rel_type: Some(RelType::Join(Box::new(join_rel))),
1580        };
1581        let (result, errors) = ctx.textify(&rel);
1582
1583        // Should contain error for unknown join type but still show condition and columns
1584        assert!(!errors.is_empty(), "Expected errors for unknown join type");
1585        assert!(
1586            result.contains("!{JoinRel}"),
1587            "Expected error token for unknown join type"
1588        );
1589        assert!(
1590            result.contains("Join["),
1591            "Expected Join relation to be formatted"
1592        );
1593    }
1594
1595    #[test]
1596    fn test_set_relation_unknown_op() {
1597        let ctx = TestContext::new();
1598
1599        let set_rel = SetRel {
1600            common: None,
1601            inputs: vec![
1602                Rel {
1603                    rel_type: Some(RelType::Read(Box::default())),
1604                },
1605                Rel {
1606                    rel_type: Some(RelType::Read(Box::default())),
1607                },
1608            ],
1609            op: 999, // Invalid set op
1610            advanced_extension: None,
1611        };
1612        let rel = Rel {
1613            rel_type: Some(RelType::Set(set_rel)),
1614        };
1615
1616        let (result, errors) = ctx.textify(&rel);
1617        assert!(!errors.is_empty(), "Expected errors for unknown set op");
1618        assert!(
1619            result.contains("!{SetRel}"),
1620            "Expected error token for unknown set op, got: {result}"
1621        );
1622        assert!(
1623            result.contains("Set["),
1624            "Expected Set relation to be formatted"
1625        );
1626    }
1627
1628    fn basic_read(table: &str) -> Rel {
1629        Rel {
1630            rel_type: Some(RelType::Read(Box::new(ReadRel {
1631                common: None,
1632                base_schema: Some(get_basic_schema()),
1633                filter: None,
1634                best_effort_filter: None,
1635                projection: None,
1636                advanced_extension: None,
1637                read_type: Some(ReadType::NamedTable(NamedTable {
1638                    names: vec![table.into()],
1639                    advanced_extension: None,
1640                })),
1641            }))),
1642        }
1643    }
1644
1645    #[test]
1646    fn test_cross_relation() {
1647        let ctx = TestContext::new();
1648
1649        // Two 3-column reads: the cross output concatenates both, giving 6
1650        // columns ($0..$5), with no arguments.
1651        let cross = CrossRel {
1652            common: Some(RelCommon {
1653                emit_kind: Some(EmitKind::Direct(Direct {})),
1654                ..Default::default()
1655            }),
1656            left: Some(Box::new(basic_read("left_tbl"))),
1657            right: Some(Box::new(basic_read("right_tbl"))),
1658            advanced_extension: None,
1659        };
1660        let rel = Rel {
1661            rel_type: Some(RelType::Cross(Box::new(cross))),
1662        };
1663
1664        let (result, errors) = ctx.textify(&rel);
1665        assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1666        let expected = r#"
1667Cross[$0, $1, $2, $3, $4, $5]
1668  Read[left_tbl => category:string?, amount:fp64?, value:i32?]
1669  Read[right_tbl => category:string?, amount:fp64?, value:i32?]"#
1670            .trim_start();
1671        assert_eq!(result, expected);
1672    }
1673
1674    #[test]
1675    fn test_cross_relation_prunes_columns() {
1676        let ctx = TestContext::new();
1677
1678        // A non-identity emit selects only two of the six columns.
1679        let cross = CrossRel {
1680            common: Some(RelCommon {
1681                emit_kind: Some(EmitKind::Emit(Emit {
1682                    output_mapping: vec![0, 3],
1683                })),
1684                ..Default::default()
1685            }),
1686            left: Some(Box::new(basic_read("left_tbl"))),
1687            right: Some(Box::new(basic_read("right_tbl"))),
1688            advanced_extension: None,
1689        };
1690        let rel = Rel {
1691            rel_type: Some(RelType::Cross(Box::new(cross))),
1692        };
1693
1694        let (result, errors) = ctx.textify(&rel);
1695        assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1696        let expected = r#"
1697Cross[$0, $3]
1698  Read[left_tbl => category:string?, amount:fp64?, value:i32?]
1699  Read[right_tbl => category:string?, amount:fp64?, value:i32?]"#
1700            .trim_start();
1701        assert_eq!(result, expected);
1702    }
1703
1704    #[test]
1705    fn test_join_type_enum_textify() {
1706        // Test that JoinType enum values convert correctly to their string representation
1707        assert_eq!(join_rel::JoinType::Inner.as_enum_str().unwrap(), "Inner");
1708        assert_eq!(join_rel::JoinType::Left.as_enum_str().unwrap(), "Left");
1709        assert_eq!(
1710            join_rel::JoinType::LeftSemi.as_enum_str().unwrap(),
1711            "LeftSemi"
1712        );
1713        assert_eq!(
1714            join_rel::JoinType::LeftAnti.as_enum_str().unwrap(),
1715            "LeftAnti"
1716        );
1717    }
1718
1719    #[test]
1720    fn test_join_output_columns() {
1721        // Test Inner join - outputs all columns from both sides
1722        let inner_cols = super::join_output_columns(join_rel::JoinType::Inner, 2, 3);
1723        assert_eq!(inner_cols.len(), 5); // 2 + 3 = 5 columns
1724        assert!(matches!(inner_cols[0], Value::Reference(0)));
1725        assert!(matches!(inner_cols[4], Value::Reference(4)));
1726
1727        // Test LeftSemi join - outputs only left columns
1728        let left_semi_cols = super::join_output_columns(join_rel::JoinType::LeftSemi, 2, 3);
1729        assert_eq!(left_semi_cols.len(), 2); // Only left columns
1730        assert!(matches!(left_semi_cols[0], Value::Reference(0)));
1731        assert!(matches!(left_semi_cols[1], Value::Reference(1)));
1732
1733        // Test RightSemi join - outputs right columns as contiguous range starting from $0
1734        let right_semi_cols = super::join_output_columns(join_rel::JoinType::RightSemi, 2, 3);
1735        assert_eq!(right_semi_cols.len(), 3); // Only right columns
1736        assert!(matches!(right_semi_cols[0], Value::Reference(0))); // Contiguous range starts at $0
1737        assert!(matches!(right_semi_cols[1], Value::Reference(1)));
1738        assert!(matches!(right_semi_cols[2], Value::Reference(2))); // Last right column
1739
1740        // Test LeftMark join - outputs left columns plus a mark column as contiguous range
1741        let left_mark_cols = super::join_output_columns(join_rel::JoinType::LeftMark, 2, 3);
1742        assert_eq!(left_mark_cols.len(), 3); // 2 left + 1 mark
1743        assert!(matches!(left_mark_cols[0], Value::Reference(0)));
1744        assert!(matches!(left_mark_cols[1], Value::Reference(1)));
1745        assert!(matches!(left_mark_cols[2], Value::Reference(2))); // Mark column at contiguous position
1746
1747        // Test RightMark join - outputs right columns plus a mark column as contiguous range
1748        let right_mark_cols = super::join_output_columns(join_rel::JoinType::RightMark, 2, 3);
1749        assert_eq!(right_mark_cols.len(), 4); // 3 right + 1 mark
1750        assert!(matches!(right_mark_cols[0], Value::Reference(0))); // Contiguous range starts at $0
1751        assert!(matches!(right_mark_cols[1], Value::Reference(1)));
1752        assert!(matches!(right_mark_cols[2], Value::Reference(2))); // Last right column
1753        assert!(matches!(right_mark_cols[3], Value::Reference(3))); // Mark column at contiguous position
1754    }
1755
1756    fn get_aggregate_func(func_ref: u32, column_ind: i32) -> AggregateFunction {
1757        AggregateFunction {
1758            function_reference: func_ref,
1759            arguments: vec![FunctionArgument {
1760                arg_type: Some(ArgType::Value(Expression {
1761                    rex_type: Some(RexType::Selection(Box::new(
1762                        FieldIndex(column_ind).to_field_reference(),
1763                    ))),
1764                })),
1765            }],
1766            options: vec![],
1767            output_type: Some(Type {
1768                kind: Some(Kind::I64(I64 {
1769                    nullability: Nullability::Required as i32,
1770                    type_variation_reference: 0,
1771                })),
1772            }),
1773            invocation: 0,
1774            phase: 0,
1775            sorts: vec![],
1776            #[allow(deprecated)]
1777            args: vec![],
1778        }
1779    }
1780
1781    fn create_aggregate_rel(
1782        grouping_expressions: Vec<Expression>,
1783        grouping_sets: Vec<Grouping>,
1784        measures: Vec<aggregate_rel::Measure>,
1785        common: Option<RelCommon>,
1786    ) -> AggregateRel {
1787        let common = common.or_else(|| {
1788            Some(RelCommon {
1789                emit_kind: Some(EmitKind::Direct(Direct {})),
1790                ..Default::default()
1791            })
1792        });
1793        AggregateRel {
1794            input: Some(Box::new(Rel {
1795                rel_type: Some(RelType::Read(Box::new(ReadRel {
1796                    common: None,
1797                    base_schema: Some(get_basic_schema()),
1798                    filter: None,
1799                    best_effort_filter: None,
1800                    projection: None,
1801                    advanced_extension: None,
1802                    read_type: Some(ReadType::NamedTable(NamedTable {
1803                        names: vec!["orders".into()],
1804                        advanced_extension: None,
1805                    })),
1806                }))),
1807            })),
1808            grouping_expressions,
1809            groupings: grouping_sets,
1810            measures,
1811            common,
1812            advanced_extension: None,
1813        }
1814    }
1815
1816    fn get_basic_schema() -> NamedStruct {
1817        NamedStruct {
1818            names: vec!["category".into(), "amount".into(), "value".into()],
1819            r#struct: Some(Struct {
1820                type_variation_reference: 0,
1821                types: vec![
1822                    Type {
1823                        kind: Some(Kind::String(ptype::String {
1824                            type_variation_reference: 0,
1825                            nullability: Nullability::Nullable as i32,
1826                        })),
1827                    },
1828                    Type {
1829                        kind: Some(Kind::Fp64(ptype::Fp64 {
1830                            type_variation_reference: 0,
1831                            nullability: Nullability::Nullable as i32,
1832                        })),
1833                    },
1834                    Type {
1835                        kind: Some(Kind::I32(ptype::I32 {
1836                            type_variation_reference: 0,
1837                            nullability: Nullability::Nullable as i32,
1838                        })),
1839                    },
1840                ],
1841                nullability: Nullability::Nullable as i32,
1842            }),
1843        }
1844    }
1845
1846    fn create_exp(column_ind: i32) -> Expression {
1847        Expression {
1848            rex_type: Some(RexType::Selection(Box::new(
1849                FieldIndex(column_ind).to_field_reference(),
1850            ))),
1851        }
1852    }
1853
1854    #[test]
1855    fn test_unsupported_rel_type_produces_failure_token() {
1856        use substrait::proto::ReferenceRel;
1857
1858        let ctx = TestContext::new();
1859
1860        // ReferenceRel is a valid Substrait relation type that the textifier
1861        // does not yet support.  Wrapping it in a Rel and textifying should
1862        // produce a `!{Rel}` failure token rather than panicking.
1863        let rel = Rel {
1864            rel_type: Some(RelType::Reference(ReferenceRel { subtree_ordinal: 0 })),
1865        };
1866
1867        let (result, errors) = ctx.textify(&rel);
1868
1869        // The output should contain the failure token, not an empty string.
1870        assert!(
1871            result.contains("!{Rel}"),
1872            "Expected '!{{Rel}}' in output, got: {result}"
1873        );
1874
1875        // Exactly one error should have been collected.
1876        assert_eq!(errors.0.len(), 1, "Expected exactly one error: {errors:?}");
1877
1878        // The error should be a Format / Unimplemented error mentioning ReferenceRel.
1879        match &errors.0[0] {
1880            FormatError::Format(plan_err) => {
1881                assert_eq!(plan_err.message, "Rel");
1882                assert_eq!(
1883                    plan_err.error_type,
1884                    crate::textify::foundation::FormatErrorType::Unimplemented
1885                );
1886                assert!(
1887                    plan_err
1888                        .lookup
1889                        .as_deref()
1890                        .unwrap_or("")
1891                        .contains("Reference"),
1892                    "Expected lookup to mention 'Reference', got: {:?}",
1893                    plan_err.lookup
1894                );
1895            }
1896            other => panic!("Expected FormatError::Format, got: {other:?}"),
1897        }
1898    }
1899}
substrait_explain/textify/rels.rs

substrait_explain/textify/
rels.rs
