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, OffsetMode};
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::{ExtensionContext, ExtensionError, ExtensionInput};
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 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 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#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
100enum OutputSyntax {
101 #[default]
103 Implicit,
104 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#[derive(Debug, Clone)]
220pub enum RelationArgs<'a> {
221 Inline(Arguments<'a>),
223 Rows {
227 rows: Vec<Value<'a>>,
228 named: Vec<NamedArg<'a>>,
229 },
230}
231
232impl<'a> RelationArgs<'a> {
233 pub fn inline(positional: Vec<Value<'a>>, named: Vec<NamedArg<'a>>) -> Self {
236 RelationArgs::Inline(Arguments::new(positional, named))
237 }
238
239 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 pub arguments: Option<RelationArgs<'a>>,
260 pub columns: Vec<Value<'a>>,
263 pub emit: Option<&'a EmitKind>,
265 output_syntax: OutputSyntax,
268 addenda: AddendumLines,
273 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 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 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 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 let output_syntax = match emit {
388 Some(EmitKind::Emit(_)) => OutputSyntax::Explicit,
389 Some(EmitKind::Direct(_)) | None => OutputSyntax::Implicit,
390 };
391 Relation {
392 name: Cow::Borrowed("Read"),
393 arguments: Some(RelationArgs::inline(vec![table_name], vec![])),
394 columns,
395 emit,
396 output_syntax,
397 addenda: AddendumLines::from_advanced_extension(
398 ctx,
399 rel.advanced_extension.as_ref(),
400 ),
401 children: vec![],
402 }
403 }
404 Some(ReadType::VirtualTable(vt)) => {
405 let row_count = vt.expressions.len();
406 let mut positional: Vec<Value> = vt
407 .expressions
408 .iter()
409 .map(|row| Value::Tuple(row.fields.iter().map(Value::Expression).collect()))
410 .collect();
411 let mut named = vec![];
412 if let Some(filter) = rel.filter.as_ref() {
413 named.push(NamedArg {
414 name: Cow::Borrowed("filter"),
415 value: Value::Expression(filter.as_ref()),
416 });
417 }
418 if positional.is_empty() && !named.is_empty() {
419 positional.push(Value::EmptyGroup);
420 }
421
422 let multiline =
427 row_count > 0 && row_count >= ctx.options().virtual_table_multiline_threshold;
428 let arguments = if multiline {
429 RelationArgs::rows(positional, named)
430 } else {
431 RelationArgs::inline(positional, named)
432 };
433
434 Relation {
435 name: Cow::Borrowed("Read:Virtual"),
436 arguments: Some(arguments),
437 columns,
438 emit,
439 output_syntax: OutputSyntax::Implicit,
440 addenda: AddendumLines::from_advanced_extension(
441 ctx,
442 rel.advanced_extension.as_ref(),
443 ),
444 children: vec![],
445 }
446 }
447 Some(ReadType::ExtensionTable(table)) => {
448 let decoded = match table.detail.as_ref().map(AnyRef::from) {
449 Some(detail) => ctx.extension_registry().decode_extension_table(detail),
450 None => Err(ExtensionError::MissingDetail),
451 };
452
453 Relation {
454 name: Cow::Borrowed("Read:Extension"),
455 arguments: None,
456 columns,
457 emit,
458 output_syntax: OutputSyntax::Implicit,
459 addenda: AddendumLines::extension_table(
460 ctx,
461 decoded,
462 rel.advanced_extension.as_ref(),
463 ),
464 children: vec![],
465 }
466 }
467 other => {
468 let err = PlanError::unimplemented(
469 "ReadRel",
470 Some("read_type"),
471 format!("Unsupported read type {other:?}"),
472 );
473 Relation {
474 name: Cow::Borrowed("Read"),
475 arguments: Some(RelationArgs::inline(vec![Value::Missing(err)], vec![])),
476 columns,
477 emit,
478 output_syntax: OutputSyntax::Implicit,
479 addenda: AddendumLines::from_advanced_extension(
480 ctx,
481 rel.advanced_extension.as_ref(),
482 ),
483 children: vec![],
484 }
485 }
486 }
487 }
488}
489
490fn read_columns<'a>(rel: &'a ReadRel) -> Vec<Value<'a>> {
491 match rel.base_schema {
492 Some(ref schema) => schema_to_values(schema),
493 None => {
494 let err =
495 PlanError::unimplemented("ReadRel", Some("base_schema"), "Base schema is required");
496 vec![Value::Missing(err)]
497 }
498 }
499}
500
501pub fn get_emit(rel: Option<&RelCommon>) -> Option<&EmitKind> {
502 rel.as_ref().and_then(|c| c.emit_kind.as_ref())
503}
504
505impl<'a> Relation<'a> {
506 pub fn convert_children<S: Scope>(
510 refs: Vec<Option<&'a Rel>>,
511 ctx: &S,
512 ) -> (Vec<Option<Relation<'a>>>, usize) {
513 let mut children = vec![];
514 let mut inputs = 0;
515
516 for maybe_rel in refs {
517 match maybe_rel {
518 Some(rel) => {
519 let child = Relation::from_rel(rel, ctx);
520 inputs += child.emitted();
521 children.push(Some(child));
522 }
523 None => children.push(None),
524 }
525 }
526
527 (children, inputs)
528 }
529}
530
531impl<'a> Relation<'a> {
532 fn from_filter<S: Scope>(rel: &'a FilterRel, ctx: &S) -> Self {
533 let condition = rel
534 .condition
535 .as_ref()
536 .map(|c| Value::Expression(c.as_ref()));
537 let condition = Value::expect(condition, || {
538 PlanError::unimplemented("FilterRel", Some("condition"), "Condition is None")
539 });
540 let positional = vec![condition];
541 let arguments = Some(RelationArgs::inline(positional, vec![]));
542 let emit = get_emit(rel.common.as_ref());
543 let (children, columns) = Relation::convert_children(vec![rel.input.as_deref()], ctx);
544 let columns = (0..columns).map(|i| Value::Reference(i as i32)).collect();
545
546 Relation {
547 name: Cow::Borrowed("Filter"),
548 arguments,
549 columns,
550 emit,
551 output_syntax: OutputSyntax::Implicit,
552 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
553 children,
554 }
555 }
556
557 fn from_project<S: Scope>(rel: &'a ProjectRel, ctx: &S) -> Self {
558 let (children, input_columns) = Relation::convert_children(vec![rel.input.as_deref()], ctx);
559 let mut columns: Vec<Value> = vec![];
560 for i in 0..input_columns {
561 columns.push(Value::Reference(i as i32));
562 }
563 for expr in &rel.expressions {
564 columns.push(Value::Expression(expr));
565 }
566
567 Relation {
568 name: Cow::Borrowed("Project"),
569 arguments: None,
570 columns,
571 emit: get_emit(rel.common.as_ref()),
572 output_syntax: OutputSyntax::Implicit,
573 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
574 children,
575 }
576 }
577
578 pub fn from_rel<S: Scope>(rel: &'a Rel, ctx: &S) -> Self {
579 match rel.rel_type.as_ref() {
580 Some(RelType::Read(r)) => Relation::from_read(r, ctx),
581 Some(RelType::Filter(r)) => Relation::from_filter(r, ctx),
582 Some(RelType::Project(r)) => Relation::from_project(r, ctx),
583 Some(RelType::Aggregate(r)) => Relation::from_aggregate(r, ctx),
584 Some(RelType::Sort(r)) => Relation::from_sort(r, ctx),
585 Some(RelType::Fetch(r)) => Relation::from_fetch(r, ctx),
586 Some(RelType::Join(r)) => Relation::from_join(r, ctx),
587 Some(RelType::Set(r)) => Relation::from_set(r, ctx),
588 Some(RelType::Cross(r)) => Relation::from_cross(r, ctx),
589 Some(RelType::ExtensionLeaf(r)) => Relation::from_extension_leaf(r, ctx),
590 Some(RelType::ExtensionSingle(r)) => Relation::from_extension_single(r, ctx),
591 Some(RelType::ExtensionMulti(r)) => Relation::from_extension_multi(r, ctx),
592 _ => {
593 let name = rel.name();
594 let token = ctx.failure(FormatError::Format(PlanError::unimplemented(
595 "Rel",
596 Some(name),
597 format!("{name} is not yet supported in the text format"),
598 )));
599 Relation {
600 name: Cow::Owned(format!("{token}")),
601 arguments: None,
602 columns: vec![],
603 emit: None,
604 output_syntax: OutputSyntax::Implicit,
605 addenda: AddendumLines::none(),
606 children: vec![],
607 }
608 }
609 }
610 }
611
612 fn from_extension_leaf<S: Scope>(rel: &'a ExtensionLeafRel, ctx: &S) -> Self {
613 Relation::from_extension(
614 "ExtensionLeaf",
615 rel.detail.as_ref().map(AnyRef::from),
616 vec![],
617 ctx,
618 )
619 }
620
621 fn from_extension_single<S: Scope>(rel: &'a ExtensionSingleRel, ctx: &S) -> Self {
622 Relation::from_extension(
623 "ExtensionSingle",
624 rel.detail.as_ref().map(AnyRef::from),
625 vec![rel.input.as_deref()],
626 ctx,
627 )
628 }
629
630 fn from_extension_multi<S: Scope>(rel: &'a ExtensionMultiRel, ctx: &S) -> Self {
631 let mut child_refs: Vec<Option<&'a Rel>> = vec![];
632 for input in &rel.inputs {
633 child_refs.push(Some(input));
634 }
635 Relation::from_extension(
636 "ExtensionMulti",
637 rel.detail.as_ref().map(AnyRef::from),
638 child_refs,
639 ctx,
640 )
641 }
642
643 fn from_extension<S: Scope>(
644 ext_type: &'static str,
645 detail: Option<AnyRef<'a>>,
646 child_refs: Vec<Option<&'a Rel>>,
647 ctx: &S,
648 ) -> Self {
649 let (children, _) = Relation::convert_children(child_refs, ctx);
650 let inputs = children
651 .iter()
652 .filter_map(|child| {
653 child
654 .as_ref()
655 .map(|child| ExtensionInput::new(child.emitted()))
656 })
657 .collect::<Vec<_>>();
658 let context = ExtensionContext::new(&inputs);
659 let decoded = match detail {
660 Some(detail) => ctx
661 .extension_registry()
662 .decode_with_context(detail, &context),
663 None => Err(ExtensionError::MissingDetail),
664 };
665
666 match decoded {
667 Ok((name, args)) => {
668 let mut positional = vec![];
669 for value in args.positional {
670 positional.push(Value::ExtensionArgument(value));
671 }
672 let mut named = vec![];
673 for (key, value) in args.named {
674 named.push(NamedArg {
675 name: Cow::Owned(key),
676 value: Value::ExtensionArgument(value),
677 });
678 }
679 let columns = args
680 .output_columns
681 .into_iter()
682 .map(Value::ExtColumn)
683 .collect();
684 Relation {
685 name: Cow::Owned(format!("{}:{}", ext_type, name)),
686 arguments: Some(RelationArgs::inline(positional, named)),
687 columns,
688 emit: None,
689 output_syntax: OutputSyntax::Implicit,
690 addenda: AddendumLines::none(),
694 children,
695 }
696 }
697 Err(error) => Relation {
698 name: Cow::Borrowed(ext_type),
699 arguments: None,
700 columns: vec![Value::Missing(PlanError::invalid(
701 "extension",
702 None::<&str>,
703 error.to_string(),
704 ))],
705 emit: None,
706 output_syntax: OutputSyntax::Implicit,
707 addenda: AddendumLines::none(),
708 children,
709 },
710 }
711 }
712
713 fn from_aggregate<S: Scope>(rel: &'a AggregateRel, ctx: &S) -> Self {
723 let mut grouping_sets: Vec<Vec<Value>> = vec![]; let expression_list: Vec<Value>; #[allow(deprecated)]
730 if rel.grouping_expressions.is_empty()
731 && !rel.groupings.is_empty()
732 && !rel.groupings[0].grouping_expressions.is_empty()
733 {
734 (expression_list, grouping_sets) = Relation::get_grouping_sets(rel);
735 } else {
736 expression_list = rel
737 .grouping_expressions
738 .iter()
739 .map(Value::Expression)
740 .collect::<Vec<_>>(); for group in &rel.groupings {
742 let mut grouping_set: Vec<Value> = vec![];
743 for i in &group.expression_references {
744 let value = match rel.grouping_expressions.get(*i as usize) {
745 Some(expr) => Value::Expression(expr),
746 None => Value::Missing(PlanError::invalid(
747 "AggregateRel",
748 Some("groupings.expression_references"),
749 format!(
750 "expression_reference {i} is out of bounds for grouping_expressions of length {}",
751 rel.grouping_expressions.len()
752 ),
753 )),
754 };
755 grouping_set.push(value);
756 }
757 grouping_sets.push(grouping_set);
758 }
759 if rel.groupings.is_empty() {
761 grouping_sets.push(vec![]);
762 }
763 }
764
765 let is_single = grouping_sets.len() == 1;
766 let mut positional: Vec<Value> = vec![];
767 for g in grouping_sets {
768 if g.is_empty() {
769 positional.push(Value::EmptyGroup);
770 } else if is_single {
771 positional.extend(g);
773 } else {
774 positional.push(Value::Tuple(g));
775 }
776 }
777
778 let arguments = Some(RelationArgs::inline(positional, vec![]));
780
781 let mut all_outputs: Vec<Value> = expression_list;
783
784 for m in &rel.measures {
787 if let Some(agg_fn) = m.measure.as_ref() {
788 all_outputs.push(Value::AggregateFunction(agg_fn));
789 }
790 }
791 let emit = get_emit(rel.common.as_ref());
792 let (children, _) = Relation::convert_children(vec![rel.input.as_deref()], ctx);
793
794 Relation {
795 name: Cow::Borrowed("Aggregate"),
796 arguments,
797 columns: all_outputs,
798 emit,
799 output_syntax: OutputSyntax::Implicit,
800 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
801 children,
802 }
803 }
804
805 fn get_grouping_sets(rel: &'a AggregateRel) -> (Vec<Value<'a>>, Vec<Vec<Value<'a>>>) {
806 let mut grouping_sets: Vec<Vec<Value>> = vec![];
807 let mut expression_list: Vec<Value> = Vec::new();
808
809 let mut seen_expressions = HashSet::new();
813
814 for group in &rel.groupings {
815 let mut grouping_set: Vec<Value> = vec![];
816 #[allow(deprecated)]
817 for exp in &group.grouping_expressions {
818 if seen_expressions.insert(exp.encode_to_vec()) {
822 expression_list.push(Value::Expression(exp)); }
824 grouping_set.push(Value::Expression(exp));
825 }
826 grouping_sets.push(grouping_set);
827 }
828 (expression_list, grouping_sets)
829 }
830}
831
832impl Textify for RelRoot {
833 fn name() -> &'static str {
834 "RelRoot"
835 }
836
837 fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
838 let names = self.names.iter().map(|n| Name(n)).collect::<Vec<_>>();
839
840 write!(
841 w,
842 "{}Root[{}]",
843 ctx.indent(),
844 ctx.separated(names.iter(), ", ")
845 )?;
846 let child_scope = ctx.push_indent();
847 for child in self.input.iter() {
848 writeln!(w)?;
849 child.textify(&child_scope, w)?;
850 }
851
852 Ok(())
853 }
854}
855
856impl Textify for PlanRelType {
857 fn name() -> &'static str {
858 "PlanRelType"
859 }
860
861 fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
862 match self {
863 PlanRelType::Rel(rel) => rel.textify(ctx, w),
864 PlanRelType::Root(root) => root.textify(ctx, w),
865 }
866 }
867}
868
869impl Textify for PlanRel {
870 fn name() -> &'static str {
871 "PlanRel"
872 }
873
874 fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
877 write!(w, "{}", ctx.expect(self.rel_type.as_ref()))
878 }
879}
880
881impl<'a> Relation<'a> {
882 fn from_sort<S: Scope>(rel: &'a SortRel, ctx: &S) -> Self {
883 let (children, input_columns) = Relation::convert_children(vec![rel.input.as_deref()], ctx);
884 let mut positional = vec![];
885 for sort_field in &rel.sorts {
886 positional.push(Value::from(sort_field));
887 }
888 let arguments = Some(RelationArgs::inline(positional, vec![]));
889 let mut col_values = vec![];
891 for i in 0..input_columns {
892 col_values.push(Value::Reference(i as i32));
893 }
894 let emit = get_emit(rel.common.as_ref());
895 Relation {
896 name: Cow::Borrowed("Sort"),
897 arguments,
898 columns: col_values,
899 emit,
900 output_syntax: OutputSyntax::Implicit,
901 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
902 children,
903 }
904 }
905
906 fn from_fetch<S: Scope>(rel: &'a FetchRel, ctx: &S) -> Self {
907 let (children, input_columns) = Relation::convert_children(vec![rel.input.as_deref()], ctx);
908 let mut named_args: Vec<NamedArg> = vec![];
909 match &rel.count_mode {
910 Some(CountMode::CountExpr(expr)) => {
911 named_args.push(NamedArg {
912 name: Cow::Borrowed("limit"),
913 value: Value::Expression(expr),
914 });
915 }
916 #[allow(deprecated)]
917 Some(CountMode::Count(val)) => {
918 named_args.push(NamedArg {
919 name: Cow::Borrowed("limit"),
920 value: Value::Integer(*val),
921 });
922 }
923 None => {}
924 }
925 if let Some(offset) = &rel.offset_mode {
926 match offset {
927 OffsetMode::OffsetExpr(expr) => {
928 named_args.push(NamedArg {
929 name: Cow::Borrowed("offset"),
930 value: Value::Expression(expr),
931 });
932 }
933 #[allow(deprecated)]
934 OffsetMode::Offset(val) => {
935 named_args.push(NamedArg {
936 name: Cow::Borrowed("offset"),
937 value: Value::Integer(*val),
938 });
939 }
940 }
941 }
942
943 let emit = get_emit(rel.common.as_ref());
944 let columns: Vec<Value> = (0..input_columns)
946 .map(|i| Value::Reference(i as i32))
947 .collect();
948 Relation {
949 name: Cow::Borrowed("Fetch"),
950 arguments: Some(RelationArgs::inline(vec![], named_args)),
951 columns,
952 emit,
953 output_syntax: OutputSyntax::Implicit,
954 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
955 children,
956 }
957 }
958}
959
960fn join_output_columns(
961 join_type: join_rel::JoinType,
962 left_columns: usize,
963 right_columns: usize,
964) -> Vec<Value<'static>> {
965 let total_columns = match join_type {
966 join_rel::JoinType::Inner
968 | join_rel::JoinType::Left
969 | join_rel::JoinType::Right
970 | join_rel::JoinType::Outer => left_columns + right_columns,
971
972 join_rel::JoinType::LeftSemi | join_rel::JoinType::LeftAnti => left_columns,
974
975 join_rel::JoinType::RightSemi | join_rel::JoinType::RightAnti => right_columns,
977
978 join_rel::JoinType::LeftSingle => left_columns,
980 join_rel::JoinType::RightSingle => right_columns,
981
982 join_rel::JoinType::LeftMark => left_columns + 1,
984 join_rel::JoinType::RightMark => right_columns + 1,
985
986 join_rel::JoinType::Unspecified => left_columns + right_columns,
988 };
989
990 (0..total_columns)
992 .map(|i| Value::Reference(i as i32))
993 .collect()
994}
995
996impl<'a> Relation<'a> {
997 fn from_join<S: Scope>(rel: &'a JoinRel, ctx: &S) -> Self {
998 let (children, _total_columns) =
999 Relation::convert_children(vec![rel.left.as_deref(), rel.right.as_deref()], ctx);
1000
1001 assert_eq!(
1003 children.len(),
1004 2,
1005 "convert_children should return same number of elements as input"
1006 );
1007
1008 let left_columns = match &children[0] {
1010 Some(child) => child.emitted(),
1011 None => 0,
1012 };
1013 let right_columns = match &children[1] {
1014 Some(child) => child.emitted(),
1015 None => 0,
1016 };
1017
1018 let (join_type, join_type_value) = match join_rel::JoinType::try_from(rel.r#type) {
1021 Ok(join_type) => {
1022 let join_type_value = match join_type.as_enum_str() {
1023 Ok(s) => Value::Enum(s),
1024 Err(e) => Value::Missing(e),
1025 };
1026 (join_type, join_type_value)
1027 }
1028 Err(_) => {
1029 let join_type_error = Value::Missing(PlanError::invalid(
1031 "JoinRel",
1032 Some("type"),
1033 format!("Unknown join type: {}", rel.r#type),
1034 ));
1035 (join_rel::JoinType::Unspecified, join_type_error)
1036 }
1037 };
1038
1039 let condition = rel
1041 .expression
1042 .as_ref()
1043 .map(|c| Value::Expression(c.as_ref()));
1044 let condition = Value::expect(condition, || {
1045 PlanError::unimplemented("JoinRel", Some("expression"), "Join condition is None")
1046 });
1047
1048 let positional = vec![join_type_value, condition];
1049 let mut named = vec![];
1050 if let Some(post_join_filter) = rel.post_join_filter.as_ref() {
1051 named.push(NamedArg {
1052 name: Cow::Borrowed("post_filter"),
1053 value: Value::Expression(post_join_filter.as_ref()),
1054 });
1055 }
1056 let arguments = Some(RelationArgs::inline(positional, named));
1057
1058 let emit = get_emit(rel.common.as_ref());
1059 let columns = join_output_columns(join_type, left_columns, right_columns);
1060
1061 Relation {
1062 name: Cow::Borrowed("Join"),
1063 arguments,
1064 columns,
1065 emit,
1066 output_syntax: OutputSyntax::Implicit,
1067 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
1068 children,
1069 }
1070 }
1071
1072 fn from_set<S: Scope>(rel: &'a SetRel, ctx: &S) -> Self {
1073 let child_refs: Vec<Option<&'a Rel>> = rel.inputs.iter().map(Some).collect();
1074 let (children, total_columns) = Relation::convert_children(child_refs, ctx);
1075
1076 let width = if children.is_empty() {
1081 0
1082 } else {
1083 total_columns / children.len()
1084 };
1085
1086 let op_value = decode_enum_field::<set_rel::SetOp>(rel.op, "SetRel", "op");
1087
1088 let arguments = Some(RelationArgs::inline(vec![op_value], vec![]));
1089 let emit = get_emit(rel.common.as_ref());
1090 let columns = (0..width).map(|i| Value::Reference(i as i32)).collect();
1091
1092 Relation {
1093 name: Cow::Borrowed("Set"),
1094 arguments,
1095 columns,
1096 emit,
1097 output_syntax: OutputSyntax::Implicit,
1098 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
1099 children,
1100 }
1101 }
1102
1103 fn from_cross<S: Scope>(rel: &'a CrossRel, ctx: &S) -> Self {
1104 let (children, total_columns) =
1105 Relation::convert_children(vec![rel.left.as_deref(), rel.right.as_deref()], ctx);
1106
1107 let columns = (0..total_columns)
1110 .map(|i| Value::Reference(i as i32))
1111 .collect();
1112
1113 Relation {
1114 name: Cow::Borrowed("Cross"),
1115 arguments: None,
1116 columns,
1117 emit: get_emit(rel.common.as_ref()),
1118 output_syntax: OutputSyntax::Implicit,
1119 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
1120 children,
1121 }
1122 }
1123}
1124
1125#[cfg(test)]
1126mod tests {
1127 use substrait::proto::aggregate_rel::Grouping;
1128 use substrait::proto::expression::literal::LiteralType;
1129 use substrait::proto::expression::{Literal, RexType, ScalarFunction};
1130 use substrait::proto::function_argument::ArgType;
1131 use substrait::proto::read_rel::{NamedTable, ReadType};
1132 use substrait::proto::rel_common::{Direct, Emit};
1133 use substrait::proto::r#type::{self as ptype, Boolean, I64, Kind, Nullability, Struct};
1134 use substrait::proto::{
1135 AggregateFunction, Expression, FunctionArgument, NamedStruct, ReadRel, ReferenceRel, Type,
1136 aggregate_rel,
1137 };
1138
1139 use super::*;
1140 use crate::fixtures::TestContext;
1141 use crate::parser::expressions::FieldIndex;
1142 use crate::textify::expressions::Reference;
1143 use crate::textify::foundation::FormatErrorType;
1144
1145 #[test]
1146 fn test_read_rel() {
1147 let ctx = TestContext::new();
1148
1149 let read_rel = ReadRel {
1151 common: None,
1152 base_schema: Some(NamedStruct {
1153 names: vec!["col1".into(), "column 2".into()],
1154 r#struct: Some(Struct {
1155 type_variation_reference: 0,
1156 types: vec![
1157 Type {
1158 kind: Some(Kind::I32(ptype::I32 {
1159 type_variation_reference: 0,
1160 nullability: Nullability::Nullable as i32,
1161 })),
1162 },
1163 Type {
1164 kind: Some(Kind::String(ptype::String {
1165 type_variation_reference: 0,
1166 nullability: Nullability::Nullable as i32,
1167 })),
1168 },
1169 ],
1170 nullability: Nullability::Nullable as i32,
1171 }),
1172 }),
1173 filter: None,
1174 best_effort_filter: None,
1175 projection: None,
1176 advanced_extension: None,
1177 read_type: Some(ReadType::NamedTable(NamedTable {
1178 names: vec!["some_db".into(), "test_table".into()],
1179 advanced_extension: None,
1180 })),
1181 };
1182
1183 let rel = Rel {
1184 rel_type: Some(RelType::Read(Box::new(read_rel))),
1185 };
1186 let (result, errors) = ctx.textify(&rel);
1187 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1188 assert_eq!(
1189 result,
1190 "Read[some_db.test_table => col1:i32?, \"column 2\":string?]"
1191 );
1192 }
1193
1194 #[test]
1195 fn test_filter_rel() {
1196 let ctx = TestContext::new()
1197 .with_urn(1, "test_urn")
1198 .with_function(1, 10, "gt");
1199
1200 let read_rel = ReadRel {
1202 common: None,
1203 base_schema: Some(NamedStruct {
1204 names: vec!["col1".into(), "col2".into()],
1205 r#struct: Some(Struct {
1206 type_variation_reference: 0,
1207 types: vec![
1208 Type {
1209 kind: Some(Kind::I32(ptype::I32 {
1210 type_variation_reference: 0,
1211 nullability: Nullability::Nullable as i32,
1212 })),
1213 },
1214 Type {
1215 kind: Some(Kind::I32(ptype::I32 {
1216 type_variation_reference: 0,
1217 nullability: Nullability::Nullable as i32,
1218 })),
1219 },
1220 ],
1221 nullability: Nullability::Nullable as i32,
1222 }),
1223 }),
1224 filter: None,
1225 best_effort_filter: None,
1226 projection: None,
1227 advanced_extension: None,
1228 read_type: Some(ReadType::NamedTable(NamedTable {
1229 names: vec!["test_table".into()],
1230 advanced_extension: None,
1231 })),
1232 };
1233
1234 let filter_expr = Expression {
1236 rex_type: Some(RexType::ScalarFunction(ScalarFunction {
1237 function_reference: 10, arguments: vec![
1239 FunctionArgument {
1240 arg_type: Some(ArgType::Value(Reference(0).into())),
1241 },
1242 FunctionArgument {
1243 arg_type: Some(ArgType::Value(Expression {
1244 rex_type: Some(RexType::Literal(Literal {
1245 literal_type: Some(LiteralType::I32(10)),
1246 nullable: false,
1247 type_variation_reference: 0,
1248 })),
1249 })),
1250 },
1251 ],
1252 options: vec![],
1253 output_type: Some(Type {
1254 kind: Some(Kind::Bool(Boolean {
1255 nullability: Nullability::Required as i32,
1256 type_variation_reference: 0,
1257 })),
1258 }),
1259 #[allow(deprecated)]
1260 args: vec![],
1261 })),
1262 };
1263
1264 let filter_rel = FilterRel {
1265 common: None,
1266 input: Some(Box::new(Rel {
1267 rel_type: Some(RelType::Read(Box::new(read_rel))),
1268 })),
1269 condition: Some(Box::new(filter_expr)),
1270 advanced_extension: None,
1271 };
1272
1273 let rel = Rel {
1274 rel_type: Some(RelType::Filter(Box::new(filter_rel))),
1275 };
1276
1277 let (result, errors) = ctx.textify(&rel);
1278 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1279 let expected = r#"
1280Filter[gt($0, 10:i32):boolean => $0, $1]
1281 Read[test_table => col1:i32?, col2:i32?]"#
1282 .trim_start();
1283 assert_eq!(result, expected);
1284 }
1285
1286 #[test]
1287 fn test_aggregate_function_textify() {
1288 let ctx = TestContext::new()
1289 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1290 .with_function(1, 10, "sum")
1291 .with_function(1, 11, "count");
1292
1293 let agg_fn = get_aggregate_func(10, 1);
1295
1296 let value = Value::AggregateFunction(&agg_fn);
1297 let (result, errors) = ctx.textify(&value);
1298
1299 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1300 assert_eq!(result, "sum($1):i64");
1301 }
1302
1303 #[test]
1304 fn test_aggregate_relation_textify() {
1305 let ctx = TestContext::new()
1306 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1307 .with_function(1, 10, "sum")
1308 .with_function(1, 11, "count");
1309
1310 let agg_fn1 = get_aggregate_func(10, 1);
1312 let agg_fn2 = get_aggregate_func(11, 1);
1313
1314 let grouping_expressions = vec![Expression {
1315 rex_type: Some(RexType::Selection(Box::new(
1316 FieldIndex(0).to_field_reference(),
1317 ))),
1318 }];
1319
1320 let measures = vec![
1321 aggregate_rel::Measure {
1322 measure: Some(agg_fn1),
1323 filter: None,
1324 },
1325 aggregate_rel::Measure {
1326 measure: Some(agg_fn2),
1327 filter: None,
1328 },
1329 ];
1330
1331 let common = Some(RelCommon {
1332 emit_kind: Some(EmitKind::Emit(Emit {
1333 output_mapping: vec![1, 2], })),
1335 ..Default::default()
1336 });
1337
1338 let aggregate_rel = create_aggregate_rel(grouping_expressions, vec![], measures, common);
1339
1340 let rel = Rel {
1341 rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1342 };
1343 let (result, errors) = ctx.textify(&rel);
1344
1345 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1346 assert!(result.contains("Aggregate[_ => sum($1):i64, count($1):i64]"));
1348 }
1349
1350 #[test]
1351 fn test_multiple_groupings_on_aggregate_deprecated() {
1352 let ctx = TestContext::new()
1355 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1356 .with_function(1, 11, "count");
1357
1358 let grouping_expr_0 = create_exp(0);
1359 let grouping_expr_1 = create_exp(1);
1360
1361 let grouping_sets = vec![
1362 aggregate_rel::Grouping {
1363 #[allow(deprecated)]
1364 grouping_expressions: vec![grouping_expr_0.clone()],
1365 expression_references: vec![],
1366 },
1367 aggregate_rel::Grouping {
1368 #[allow(deprecated)]
1369 grouping_expressions: vec![grouping_expr_0.clone(), grouping_expr_1.clone()],
1370 expression_references: vec![],
1371 },
1372 ];
1373
1374 let aggregate_rel = create_aggregate_rel(vec![], grouping_sets, vec![], None);
1375
1376 let rel = Rel {
1377 rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1378 };
1379 let (result, errors) = ctx.textify(&rel);
1380
1381 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1382 assert!(result.contains("Aggregate[($0), ($0, $1) => $0, $1]"));
1383 }
1384
1385 #[test]
1386 fn test_multiple_groupings_with_measure_deprecated() {
1387 let ctx = TestContext::new()
1390 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1391 .with_function(1, 11, "count");
1392
1393 let agg_fn1 = get_aggregate_func(11, 2);
1394
1395 let grouping_expr_0 = create_exp(0);
1396 let grouping_expr_1 = create_exp(1);
1397
1398 let grouping_sets = vec![
1399 aggregate_rel::Grouping {
1400 #[allow(deprecated)]
1401 grouping_expressions: vec![grouping_expr_0.clone()],
1402 expression_references: vec![],
1403 },
1404 aggregate_rel::Grouping {
1405 #[allow(deprecated)]
1406 grouping_expressions: vec![grouping_expr_0.clone(), grouping_expr_1.clone()],
1407 expression_references: vec![],
1408 },
1409 ];
1410
1411 let measures = vec![aggregate_rel::Measure {
1412 measure: Some(agg_fn1),
1413 filter: None,
1414 }];
1415
1416 let aggregate_rel = create_aggregate_rel(vec![], grouping_sets, measures, None);
1417
1418 let rel = Rel {
1419 rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1420 };
1421 let (result, errors) = ctx.textify(&rel);
1422
1423 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1424 assert!(result.contains("($0), ($0, $1) => $0, $1, count($2):i64"));
1425 }
1426
1427 #[test]
1428 fn test_multiple_groupings_on_aggregate() {
1429 let ctx = TestContext::new()
1430 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1431 .with_function(1, 11, "count");
1432
1433 let agg_fn2 = get_aggregate_func(11, 2);
1434
1435 let grouping_expressions = vec![
1436 Expression {
1437 rex_type: Some(RexType::Selection(Box::new(
1438 FieldIndex(0).to_field_reference(),
1439 ))),
1440 },
1441 Expression {
1442 rex_type: Some(RexType::Selection(Box::new(
1443 FieldIndex(1).to_field_reference(),
1444 ))),
1445 },
1446 ];
1447
1448 let grouping_sets = vec![
1449 Grouping {
1450 #[allow(deprecated)]
1451 grouping_expressions: vec![],
1452 expression_references: vec![0, 1],
1453 },
1454 Grouping {
1455 #[allow(deprecated)]
1456 grouping_expressions: vec![],
1457 expression_references: vec![0, 1],
1458 },
1459 Grouping {
1460 #[allow(deprecated)]
1461 grouping_expressions: vec![],
1462 expression_references: vec![1],
1463 },
1464 Grouping {
1465 #[allow(deprecated)]
1466 grouping_expressions: vec![],
1467 expression_references: vec![1, 1],
1468 },
1469 Grouping {
1470 #[allow(deprecated)]
1471 grouping_expressions: vec![],
1472 expression_references: vec![],
1473 },
1474 ];
1475
1476 let measures = vec![aggregate_rel::Measure {
1477 measure: Some(agg_fn2),
1478 filter: None,
1479 }];
1480
1481 let aggregate_rel =
1482 create_aggregate_rel(grouping_expressions, grouping_sets, measures, None);
1483
1484 let rel = Rel {
1485 rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1486 };
1487 let (result, errors) = ctx.textify(&rel);
1488
1489 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1490 assert!(
1491 result.contains(
1492 "Aggregate[($0, $1), ($0, $1), ($1), ($1, $1), _ => $0, $1, count($2):i64]"
1493 )
1494 );
1495 }
1496
1497 #[test]
1498 fn test_deprecated_reordered_grouping() {
1499 let ctx = TestContext::new();
1507 let grouping_expr_5 = create_exp(5);
1508
1509 let grouping_sets = vec![aggregate_rel::Grouping {
1510 #[allow(deprecated)]
1511 grouping_expressions: vec![grouping_expr_5],
1512 expression_references: vec![],
1513 }];
1514
1515 let aggregate_rel = create_aggregate_rel(vec![], grouping_sets, vec![], None);
1516 let rel = Rel {
1517 rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1518 };
1519 let (result, errors) = ctx.textify(&rel);
1520
1521 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1522 assert!(result.contains("Aggregate[$5 => $5]"));
1523 }
1524
1525 #[test]
1526 fn test_reordered_grouping_textifies_expression_not_raw_index() {
1527 let ctx = TestContext::new()
1528 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1529 .with_function(1, 11, "count");
1530
1531 let agg_fn2 = get_aggregate_func(11, 1);
1532
1533 let grouping_expressions = vec![
1539 Expression {
1540 rex_type: Some(RexType::Selection(Box::new(
1541 FieldIndex(2).to_field_reference(),
1542 ))),
1543 },
1544 Expression {
1545 rex_type: Some(RexType::Selection(Box::new(
1546 FieldIndex(0).to_field_reference(),
1547 ))),
1548 },
1549 ];
1550
1551 let grouping_sets = vec![Grouping {
1552 #[allow(deprecated)]
1553 grouping_expressions: vec![],
1554 expression_references: vec![0, 1],
1555 }];
1556
1557 let measures = vec![aggregate_rel::Measure {
1558 measure: Some(agg_fn2),
1559 filter: None,
1560 }];
1561
1562 let aggregate_rel =
1563 create_aggregate_rel(grouping_expressions, grouping_sets, measures, None);
1564
1565 let rel = Rel {
1566 rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1567 };
1568 let (result, errors) = ctx.textify(&rel);
1569
1570 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1571 assert!(result.contains("Aggregate[$2, $0 => $2, $0, count($1):i64]"));
1572 }
1573
1574 #[test]
1575 fn test_join_relation_unknown_type() {
1576 let ctx = TestContext::new();
1577
1578 let join_rel = JoinRel {
1580 left: Some(Box::new(Rel {
1581 rel_type: Some(RelType::Read(Box::default())),
1582 })),
1583 right: Some(Box::new(Rel {
1584 rel_type: Some(RelType::Read(Box::default())),
1585 })),
1586 expression: Some(Box::new(Expression::default())),
1587 r#type: 999, common: None,
1589 post_join_filter: None,
1590 advanced_extension: None,
1591 };
1592
1593 let rel = Rel {
1594 rel_type: Some(RelType::Join(Box::new(join_rel))),
1595 };
1596 let (result, errors) = ctx.textify(&rel);
1597
1598 assert!(!errors.is_empty(), "Expected errors for unknown join type");
1600 assert!(
1601 result.contains("!{JoinRel}"),
1602 "Expected error token for unknown join type"
1603 );
1604 assert!(
1605 result.contains("Join["),
1606 "Expected Join relation to be formatted"
1607 );
1608 }
1609
1610 #[test]
1611 fn test_set_relation_unknown_op() {
1612 let ctx = TestContext::new();
1613
1614 let set_rel = SetRel {
1615 common: None,
1616 inputs: vec![
1617 Rel {
1618 rel_type: Some(RelType::Read(Box::default())),
1619 },
1620 Rel {
1621 rel_type: Some(RelType::Read(Box::default())),
1622 },
1623 ],
1624 op: 999, advanced_extension: None,
1626 };
1627 let rel = Rel {
1628 rel_type: Some(RelType::Set(set_rel)),
1629 };
1630
1631 let (result, errors) = ctx.textify(&rel);
1632 assert!(!errors.is_empty(), "Expected errors for unknown set op");
1633 assert!(
1634 result.contains("!{SetRel}"),
1635 "Expected error token for unknown set op, got: {result}"
1636 );
1637 assert!(
1638 result.contains("Set["),
1639 "Expected Set relation to be formatted"
1640 );
1641 }
1642
1643 fn basic_read(table: &str) -> Rel {
1644 Rel {
1645 rel_type: Some(RelType::Read(Box::new(ReadRel {
1646 common: None,
1647 base_schema: Some(get_basic_schema()),
1648 filter: None,
1649 best_effort_filter: None,
1650 projection: None,
1651 advanced_extension: None,
1652 read_type: Some(ReadType::NamedTable(NamedTable {
1653 names: vec![table.into()],
1654 advanced_extension: None,
1655 })),
1656 }))),
1657 }
1658 }
1659
1660 #[test]
1661 fn test_cross_relation() {
1662 let ctx = TestContext::new();
1663
1664 let cross = CrossRel {
1667 common: Some(RelCommon {
1668 emit_kind: Some(EmitKind::Direct(Direct {})),
1669 ..Default::default()
1670 }),
1671 left: Some(Box::new(basic_read("left_tbl"))),
1672 right: Some(Box::new(basic_read("right_tbl"))),
1673 advanced_extension: None,
1674 };
1675 let rel = Rel {
1676 rel_type: Some(RelType::Cross(Box::new(cross))),
1677 };
1678
1679 let (result, errors) = ctx.textify(&rel);
1680 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1681 let expected = r#"
1682Cross[$0, $1, $2, $3, $4, $5]
1683 Read[left_tbl => category:string?, amount:fp64?, value:i32?]
1684 Read[right_tbl => category:string?, amount:fp64?, value:i32?]"#
1685 .trim_start();
1686 assert_eq!(result, expected);
1687 }
1688
1689 #[test]
1690 fn test_cross_relation_prunes_columns() {
1691 let ctx = TestContext::new();
1692
1693 let cross = CrossRel {
1695 common: Some(RelCommon {
1696 emit_kind: Some(EmitKind::Emit(Emit {
1697 output_mapping: vec![0, 3],
1698 })),
1699 ..Default::default()
1700 }),
1701 left: Some(Box::new(basic_read("left_tbl"))),
1702 right: Some(Box::new(basic_read("right_tbl"))),
1703 advanced_extension: None,
1704 };
1705 let rel = Rel {
1706 rel_type: Some(RelType::Cross(Box::new(cross))),
1707 };
1708
1709 let (result, errors) = ctx.textify(&rel);
1710 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1711 let expected = r#"
1712Cross[$0, $3]
1713 Read[left_tbl => category:string?, amount:fp64?, value:i32?]
1714 Read[right_tbl => category:string?, amount:fp64?, value:i32?]"#
1715 .trim_start();
1716 assert_eq!(result, expected);
1717 }
1718
1719 #[test]
1720 fn test_join_type_enum_textify() {
1721 assert_eq!(join_rel::JoinType::Inner.as_enum_str().unwrap(), "Inner");
1723 assert_eq!(join_rel::JoinType::Left.as_enum_str().unwrap(), "Left");
1724 assert_eq!(
1725 join_rel::JoinType::LeftSemi.as_enum_str().unwrap(),
1726 "LeftSemi"
1727 );
1728 assert_eq!(
1729 join_rel::JoinType::LeftAnti.as_enum_str().unwrap(),
1730 "LeftAnti"
1731 );
1732 }
1733
1734 #[test]
1735 fn test_join_output_columns() {
1736 let inner_cols = super::join_output_columns(join_rel::JoinType::Inner, 2, 3);
1738 assert_eq!(inner_cols.len(), 5); assert!(matches!(inner_cols[0], Value::Reference(0)));
1740 assert!(matches!(inner_cols[4], Value::Reference(4)));
1741
1742 let left_semi_cols = super::join_output_columns(join_rel::JoinType::LeftSemi, 2, 3);
1744 assert_eq!(left_semi_cols.len(), 2); assert!(matches!(left_semi_cols[0], Value::Reference(0)));
1746 assert!(matches!(left_semi_cols[1], Value::Reference(1)));
1747
1748 let right_semi_cols = super::join_output_columns(join_rel::JoinType::RightSemi, 2, 3);
1750 assert_eq!(right_semi_cols.len(), 3); assert!(matches!(right_semi_cols[0], Value::Reference(0))); assert!(matches!(right_semi_cols[1], Value::Reference(1)));
1753 assert!(matches!(right_semi_cols[2], Value::Reference(2))); let left_mark_cols = super::join_output_columns(join_rel::JoinType::LeftMark, 2, 3);
1757 assert_eq!(left_mark_cols.len(), 3); assert!(matches!(left_mark_cols[0], Value::Reference(0)));
1759 assert!(matches!(left_mark_cols[1], Value::Reference(1)));
1760 assert!(matches!(left_mark_cols[2], Value::Reference(2))); let right_mark_cols = super::join_output_columns(join_rel::JoinType::RightMark, 2, 3);
1764 assert_eq!(right_mark_cols.len(), 4); assert!(matches!(right_mark_cols[0], Value::Reference(0))); assert!(matches!(right_mark_cols[1], Value::Reference(1)));
1767 assert!(matches!(right_mark_cols[2], Value::Reference(2))); assert!(matches!(right_mark_cols[3], Value::Reference(3))); }
1770
1771 fn get_aggregate_func(func_ref: u32, column_ind: i32) -> AggregateFunction {
1772 AggregateFunction {
1773 function_reference: func_ref,
1774 arguments: vec![FunctionArgument {
1775 arg_type: Some(ArgType::Value(Expression {
1776 rex_type: Some(RexType::Selection(Box::new(
1777 FieldIndex(column_ind).to_field_reference(),
1778 ))),
1779 })),
1780 }],
1781 options: vec![],
1782 output_type: Some(Type {
1783 kind: Some(Kind::I64(I64 {
1784 nullability: Nullability::Required as i32,
1785 type_variation_reference: 0,
1786 })),
1787 }),
1788 invocation: 0,
1789 phase: 0,
1790 sorts: vec![],
1791 #[allow(deprecated)]
1792 args: vec![],
1793 }
1794 }
1795
1796 fn create_aggregate_rel(
1797 grouping_expressions: Vec<Expression>,
1798 grouping_sets: Vec<Grouping>,
1799 measures: Vec<aggregate_rel::Measure>,
1800 common: Option<RelCommon>,
1801 ) -> AggregateRel {
1802 let common = common.or_else(|| {
1803 Some(RelCommon {
1804 emit_kind: Some(EmitKind::Direct(Direct {})),
1805 ..Default::default()
1806 })
1807 });
1808 AggregateRel {
1809 input: Some(Box::new(Rel {
1810 rel_type: Some(RelType::Read(Box::new(ReadRel {
1811 common: None,
1812 base_schema: Some(get_basic_schema()),
1813 filter: None,
1814 best_effort_filter: None,
1815 projection: None,
1816 advanced_extension: None,
1817 read_type: Some(ReadType::NamedTable(NamedTable {
1818 names: vec!["orders".into()],
1819 advanced_extension: None,
1820 })),
1821 }))),
1822 })),
1823 grouping_expressions,
1824 groupings: grouping_sets,
1825 measures,
1826 common,
1827 advanced_extension: None,
1828 }
1829 }
1830
1831 fn get_basic_schema() -> NamedStruct {
1832 NamedStruct {
1833 names: vec!["category".into(), "amount".into(), "value".into()],
1834 r#struct: Some(Struct {
1835 type_variation_reference: 0,
1836 types: vec![
1837 Type {
1838 kind: Some(Kind::String(ptype::String {
1839 type_variation_reference: 0,
1840 nullability: Nullability::Nullable as i32,
1841 })),
1842 },
1843 Type {
1844 kind: Some(Kind::Fp64(ptype::Fp64 {
1845 type_variation_reference: 0,
1846 nullability: Nullability::Nullable as i32,
1847 })),
1848 },
1849 Type {
1850 kind: Some(Kind::I32(ptype::I32 {
1851 type_variation_reference: 0,
1852 nullability: Nullability::Nullable as i32,
1853 })),
1854 },
1855 ],
1856 nullability: Nullability::Nullable as i32,
1857 }),
1858 }
1859 }
1860
1861 fn create_exp(column_ind: i32) -> Expression {
1862 Expression {
1863 rex_type: Some(RexType::Selection(Box::new(
1864 FieldIndex(column_ind).to_field_reference(),
1865 ))),
1866 }
1867 }
1868
1869 #[test]
1870 fn test_unsupported_rel_type_produces_failure_token() {
1871 let ctx = TestContext::new();
1872
1873 let rel = Rel {
1877 rel_type: Some(RelType::Reference(ReferenceRel { subtree_ordinal: 0 })),
1878 };
1879
1880 let (result, errors) = ctx.textify(&rel);
1881
1882 assert!(
1884 result.contains("!{Rel}"),
1885 "Expected '!{{Rel}}' in output, got: {result}"
1886 );
1887
1888 assert_eq!(errors.0.len(), 1, "Expected exactly one error: {errors:?}");
1890
1891 match &errors.0[0] {
1893 FormatError::Format(plan_err) => {
1894 assert_eq!(plan_err.message, "Rel");
1895 assert_eq!(plan_err.error_type, FormatErrorType::Unimplemented);
1896 assert!(
1897 plan_err
1898 .lookup
1899 .as_deref()
1900 .unwrap_or("")
1901 .contains("Reference"),
1902 "Expected lookup to mention 'Reference', got: {:?}",
1903 plan_err.lookup
1904 );
1905 }
1906 other => panic!("Expected FormatError::Format, got: {other:?}"),
1907 }
1908 }
1909}