1use std::borrow::Cow;
2use std::collections::HashMap;
3use std::convert::TryFrom;
4use std::fmt;
5use std::fmt::Debug;
6
7use prost::{Message, UnknownEnumValue};
8use substrait::proto::fetch_rel::CountMode;
9use substrait::proto::plan_rel::RelType as PlanRelType;
10use substrait::proto::read_rel::ReadType;
11use substrait::proto::rel::RelType;
12use substrait::proto::rel_common::EmitKind;
13use substrait::proto::sort_field::{SortDirection, SortKind};
14use substrait::proto::{
15 AggregateFunction, AggregateRel, CrossRel, Expression, ExtensionLeafRel, ExtensionMultiRel,
16 ExtensionSingleRel, FetchRel, FilterRel, JoinRel, NamedStruct, PlanRel, ProjectRel, ReadRel,
17 Rel, RelCommon, RelRoot, SetRel, SortField, SortRel, Type, join_rel, set_rel,
18};
19
20use super::addenda::AddendumLines;
21use super::expressions::Reference;
22use super::types::Name;
23use super::{PlanError, Scope, Textify};
24use crate::FormatError;
25use crate::extensions::any::AnyRef;
26use crate::extensions::{ExtensionArgs, ExtensionColumn, ExtensionError, ExtensionValue};
27
28pub trait NamedRelation {
29 fn name(&self) -> &'static str;
30}
31
32impl NamedRelation for Rel {
33 fn name(&self) -> &'static str {
34 match self.rel_type.as_ref() {
35 None => "UnknownRel",
36 Some(RelType::Read(_)) => "Read",
37 Some(RelType::Filter(_)) => "Filter",
38 Some(RelType::Project(_)) => "Project",
39 Some(RelType::Fetch(_)) => "Fetch",
40 Some(RelType::Aggregate(_)) => "Aggregate",
41 Some(RelType::Sort(_)) => "Sort",
42 Some(RelType::HashJoin(_)) => "HashJoin",
43 Some(RelType::Exchange(_)) => "Exchange",
44 Some(RelType::Join(_)) => "Join",
45 Some(RelType::Set(_)) => "Set",
46 Some(RelType::ExtensionLeaf(_)) => "ExtensionLeaf",
47 Some(RelType::Cross(_)) => "Cross",
48 Some(RelType::Reference(_)) => "Reference",
49 Some(RelType::ExtensionSingle(_)) => "ExtensionSingle",
50 Some(RelType::ExtensionMulti(_)) => "ExtensionMulti",
51 Some(RelType::Write(_)) => "Write",
52 Some(RelType::Ddl(_)) => "Ddl",
53 Some(RelType::Update(_)) => "Update",
54 Some(RelType::MergeJoin(_)) => "MergeJoin",
55 Some(RelType::NestedLoopJoin(_)) => "NestedLoopJoin",
56 Some(RelType::Window(_)) => "Window",
57 Some(RelType::Expand(_)) => "Expand",
58 }
59 }
60}
61
62impl Textify for Rel {
63 fn name() -> &'static str {
64 "Rel"
65 }
66
67 fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
68 Relation::from_rel(self, ctx).textify(ctx, w)
71 }
72}
73
74pub trait ValueEnum {
80 fn as_enum_str(&self) -> Result<Cow<'static, str>, PlanError>;
81}
82
83#[derive(Debug, Clone)]
84pub struct NamedArg<'a> {
85 pub name: Cow<'a, str>,
86 pub value: Value<'a>,
87}
88
89#[derive(Debug, Clone)]
90pub enum Value<'a> {
91 TableName(Vec<Name<'a>>),
92 Field(Option<Name<'a>>, Option<&'a Type>),
93 Tuple(Vec<Value<'a>>),
94 Reference(i32),
95 Expression(&'a Expression),
96 AggregateFunction(&'a AggregateFunction),
97 Missing(PlanError),
99 Enum(Cow<'a, str>),
101 EmptyGroup,
102 Integer(i64),
103 ExtensionArgument(ExtensionValue),
105 ExtColumn(ExtensionColumn),
107}
108
109impl<'a> Value<'a> {
110 pub fn expect(maybe_value: Option<Self>, f: impl FnOnce() -> PlanError) -> Self {
111 match maybe_value {
112 Some(s) => s,
113 None => Value::Missing(f()),
114 }
115 }
116}
117
118impl<'a> From<Result<Vec<Name<'a>>, PlanError>> for Value<'a> {
119 fn from(token: Result<Vec<Name<'a>>, PlanError>) -> Self {
120 match token {
121 Ok(value) => Value::TableName(value),
122 Err(err) => Value::Missing(err),
123 }
124 }
125}
126
127impl<'a> Textify for Value<'a> {
128 fn name() -> &'static str {
129 "Value"
130 }
131
132 fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
133 match self {
134 Value::TableName(names) => write!(w, "{}", ctx.separated(names, ".")),
135 Value::Field(name, typ) => {
136 write!(w, "{}:{}", ctx.expect(name.as_ref()), ctx.expect(*typ))
137 }
138 Value::Tuple(values) => write!(w, "({})", ctx.separated(values, ", ")),
139 Value::Reference(i) => write!(w, "{}", Reference(*i)),
140 Value::Expression(e) => write!(w, "{}", ctx.display(*e)),
141 Value::AggregateFunction(agg_fn) => agg_fn.textify(ctx, w),
142 Value::Missing(err) => write!(w, "{}", ctx.failure(err.clone())),
143 Value::Enum(res) => write!(w, "&{res}"),
144 Value::Integer(i) => write!(w, "{i}"),
145 Value::EmptyGroup => write!(w, "_"),
146 Value::ExtensionArgument(ev) => ev.textify(ctx, w),
147 Value::ExtColumn(ec) => ec.textify(ctx, w),
148 }
149 }
150}
151
152fn schema_to_values<'a>(schema: &'a NamedStruct) -> Vec<Value<'a>> {
153 let mut fields = schema
154 .r#struct
155 .as_ref()
156 .map(|s| s.types.iter())
157 .into_iter()
158 .flatten();
159 let mut names = schema.names.iter();
160
161 let mut values = Vec::new();
165 loop {
166 let field = fields.next();
167 let name = names.next().map(|n| Name(n));
168 if field.is_none() && name.is_none() {
169 break;
170 }
171
172 values.push(Value::Field(name, field));
173 }
174
175 values
176}
177
178struct Emitted<'a> {
179 pub values: &'a [Value<'a>],
180 pub emit: Option<&'a EmitKind>,
181}
182
183impl<'a> Emitted<'a> {
184 pub fn new(values: &'a [Value<'a>], emit: Option<&'a EmitKind>) -> Self {
185 Self { values, emit }
186 }
187
188 pub fn write_direct<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
189 write!(w, "{}", ctx.separated(self.values.iter(), ", "))
190 }
191}
192
193impl<'a> Textify for Emitted<'a> {
194 fn name() -> &'static str {
195 "Emitted"
196 }
197
198 fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
199 if ctx.options().show_emit {
200 return self.write_direct(ctx, w);
201 }
202
203 let indices = match &self.emit {
204 Some(EmitKind::Emit(e)) => &e.output_mapping,
205 Some(EmitKind::Direct(_)) => return self.write_direct(ctx, w),
206 None => return self.write_direct(ctx, w),
207 };
208
209 for (i, &index) in indices.iter().enumerate() {
210 if i > 0 {
211 write!(w, ", ")?;
212 }
213
214 match self.values.get(index as usize) {
215 Some(value) => write!(w, "{}", ctx.display(value))?,
216 None => write!(w, "{}", ctx.failure(PlanError::invalid(
217 "Emitted",
218 Some("output_mapping"),
219 format!(
220 "Output mapping index {} is out of bounds for values collection of size {}",
221 index, self.values.len()
222 )
223 )))?,
224 }
225 }
226
227 Ok(())
228 }
229}
230
231#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
233pub enum ArgsLayout {
234 #[default]
236 Inline,
237 Rows,
240}
241
242#[derive(Debug, Clone)]
243pub struct Arguments<'a> {
244 pub positional: Vec<Value<'a>>,
246 pub named: Vec<NamedArg<'a>>,
248 layout: ArgsLayout,
251}
252
253impl<'a> Arguments<'a> {
254 pub fn inline(positional: Vec<Value<'a>>, named: Vec<NamedArg<'a>>) -> Self {
257 Arguments {
258 positional,
259 named,
260 layout: ArgsLayout::Inline,
261 }
262 }
263
264 pub fn rows(positional: Vec<Value<'a>>) -> Self {
268 Arguments {
269 positional,
270 named: vec![],
271 layout: ArgsLayout::Rows,
272 }
273 }
274}
275
276impl<'a> Textify for Arguments<'a> {
277 fn name() -> &'static str {
278 "Arguments"
279 }
280 fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
281 if self.positional.is_empty() && self.named.is_empty() {
282 return write!(w, "_");
283 }
284
285 write!(w, "{}", ctx.separated(self.positional.iter(), ", "))?;
286 if !self.positional.is_empty() && !self.named.is_empty() {
287 write!(w, ", ")?;
288 }
289 write!(w, "{}", ctx.separated(self.named.iter(), ", "))
290 }
291}
292
293pub struct Relation<'a> {
294 pub name: Cow<'a, str>,
295 pub arguments: Option<Arguments<'a>>,
304 pub columns: Vec<Value<'a>>,
307 pub emit: Option<&'a EmitKind>,
309 addenda: AddendumLines,
314 pub children: Vec<Option<Relation<'a>>>,
316}
317
318impl Textify for Relation<'_> {
319 fn name() -> &'static str {
320 "Relation"
321 }
322
323 fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
324 self.write_header(ctx, w)?;
325 let child_scope = ctx.push_indent();
326 self.addenda.textify(&child_scope, w)?;
327 self.write_children(ctx, w)?;
328 Ok(())
329 }
330}
331
332impl Relation<'_> {
333 pub fn write_header<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
348 let cols = Emitted::new(&self.columns, self.emit);
349 let indent = ctx.indent();
350 let name = &self.name;
351 let cols = ctx.display(&cols);
352 match &self.arguments {
353 None => {
354 write!(w, "{indent}{name}[{cols}]")
355 }
356 Some(args) if args.layout == ArgsLayout::Rows => {
357 let child = ctx.push_indent();
360 let child_indent = child.indent();
361 writeln!(w, "{indent}{name}[")?;
362 let last = args.positional.len().saturating_sub(1);
363 for (i, row) in args.positional.iter().enumerate() {
364 let row = ctx.display(row);
365 let comma = if i == last { "" } else { "," };
366 writeln!(w, "{child_indent}- {row}{comma}")?;
367 }
368 write!(w, "{child_indent}- => {cols}]")
369 }
370 Some(args) => {
371 let args = ctx.display(args);
372 write!(w, "{indent}{name}[{args} => {cols}]")
373 }
374 }
375 }
376
377 pub fn write_children<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
380 let child_scope = ctx.push_indent();
381 for child in self.children.iter().flatten() {
382 writeln!(w)?;
383 child.textify(&child_scope, w)?;
384 }
385 Ok(())
386 }
387}
388
389impl<'a> Relation<'a> {
390 pub fn emitted(&self) -> usize {
391 match self.emit {
392 Some(EmitKind::Emit(e)) => e.output_mapping.len(),
393 Some(EmitKind::Direct(_)) => self.columns.len(),
394 None => self.columns.len(),
395 }
396 }
397}
398
399impl<'a> Relation<'a> {
400 fn from_read<S: Scope>(rel: &'a ReadRel, ctx: &S) -> Self {
401 let columns = read_columns(rel);
402 let emit = rel.common.as_ref().and_then(|c| c.emit_kind.as_ref());
403
404 match &rel.read_type {
405 Some(ReadType::NamedTable(table)) => {
406 let table_name = Value::TableName(table.names.iter().map(|n| Name(n)).collect());
407 Relation {
408 name: Cow::Borrowed("Read"),
409 arguments: Some(Arguments::inline(vec![table_name], vec![])),
410 columns,
411 emit,
412 addenda: AddendumLines::from_advanced_extension(
413 ctx,
414 rel.advanced_extension.as_ref(),
415 ),
416 children: vec![],
417 }
418 }
419 Some(ReadType::VirtualTable(vt)) => {
420 let positional: Vec<Value> = vt
421 .expressions
422 .iter()
423 .map(|row| Value::Tuple(row.fields.iter().map(Value::Expression).collect()))
424 .collect();
425
426 let multiline = !positional.is_empty()
431 && positional.len() >= ctx.options().virtual_table_multiline_threshold;
432 let arguments = if multiline {
433 Arguments::rows(positional)
434 } else {
435 Arguments::inline(positional, vec![])
436 };
437
438 Relation {
439 name: Cow::Borrowed("Read:Virtual"),
440 arguments: Some(arguments),
441 columns,
442 emit,
443 addenda: AddendumLines::from_advanced_extension(
444 ctx,
445 rel.advanced_extension.as_ref(),
446 ),
447 children: vec![],
448 }
449 }
450 Some(ReadType::ExtensionTable(table)) => {
451 let decoded = match table.detail.as_ref().map(AnyRef::from) {
452 Some(detail) => ctx.extension_registry().decode_extension_table(detail),
453 None => Err(ExtensionError::MissingDetail),
454 };
455
456 Relation {
457 name: Cow::Borrowed("Read:Extension"),
458 arguments: None,
459 columns,
460 emit,
461 addenda: AddendumLines::extension_table(
462 ctx,
463 decoded,
464 rel.advanced_extension.as_ref(),
465 ),
466 children: vec![],
467 }
468 }
469 other => {
470 let err = PlanError::unimplemented(
471 "ReadRel",
472 Some("read_type"),
473 format!("Unsupported read type {other:?}"),
474 );
475 Relation {
476 name: Cow::Borrowed("Read"),
477 arguments: Some(Arguments::inline(vec![Value::Missing(err)], vec![])),
478 columns,
479 emit,
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 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(Arguments::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 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 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
573 children,
574 }
575 }
576
577 pub fn from_rel<S: Scope>(rel: &'a Rel, ctx: &S) -> Self {
578 match rel.rel_type.as_ref() {
579 Some(RelType::Read(r)) => Relation::from_read(r, ctx),
580 Some(RelType::Filter(r)) => Relation::from_filter(r, ctx),
581 Some(RelType::Project(r)) => Relation::from_project(r, ctx),
582 Some(RelType::Aggregate(r)) => Relation::from_aggregate(r, ctx),
583 Some(RelType::Sort(r)) => Relation::from_sort(r, ctx),
584 Some(RelType::Fetch(r)) => Relation::from_fetch(r, ctx),
585 Some(RelType::Join(r)) => Relation::from_join(r, ctx),
586 Some(RelType::Set(r)) => Relation::from_set(r, ctx),
587 Some(RelType::Cross(r)) => Relation::from_cross(r, ctx),
588 Some(RelType::ExtensionLeaf(r)) => Relation::from_extension_leaf(r, ctx),
589 Some(RelType::ExtensionSingle(r)) => Relation::from_extension_single(r, ctx),
590 Some(RelType::ExtensionMulti(r)) => Relation::from_extension_multi(r, ctx),
591 _ => {
592 let name = rel.name();
593 let token = ctx.failure(FormatError::Format(PlanError::unimplemented(
594 "Rel",
595 Some(name),
596 format!("{name} is not yet supported in the text format"),
597 )));
598 Relation {
599 name: Cow::Owned(format!("{token}")),
600 arguments: None,
601 columns: vec![],
602 emit: None,
603 addenda: AddendumLines::none(),
604 children: vec![],
605 }
606 }
607 }
608 }
609
610 fn from_extension_leaf<S: Scope>(rel: &'a ExtensionLeafRel, ctx: &S) -> Self {
611 let detail_ref = rel.detail.as_ref().map(AnyRef::from);
612 let decoded = match detail_ref {
613 Some(d) => ctx.extension_registry().decode(d),
614 None => Err(ExtensionError::MissingDetail),
615 };
616 Relation::from_extension("ExtensionLeaf", decoded, vec![], ctx)
617 }
618
619 fn from_extension_single<S: Scope>(rel: &'a ExtensionSingleRel, ctx: &S) -> Self {
620 let detail_ref = rel.detail.as_ref().map(AnyRef::from);
621 let decoded = match detail_ref {
622 Some(d) => ctx.extension_registry().decode(d),
623 None => Err(ExtensionError::MissingDetail),
624 };
625 Relation::from_extension("ExtensionSingle", decoded, vec![rel.input.as_deref()], ctx)
626 }
627
628 fn from_extension_multi<S: Scope>(rel: &'a ExtensionMultiRel, ctx: &S) -> Self {
629 let detail_ref = rel.detail.as_ref().map(AnyRef::from);
630 let decoded = match detail_ref {
631 Some(d) => ctx.extension_registry().decode(d),
632 None => Err(ExtensionError::MissingDetail),
633 };
634 let mut child_refs: Vec<Option<&'a Rel>> = vec![];
635 for input in &rel.inputs {
636 child_refs.push(Some(input));
637 }
638 Relation::from_extension("ExtensionMulti", decoded, child_refs, ctx)
639 }
640
641 fn from_extension<S: Scope>(
642 ext_type: &'static str,
643 decoded: Result<(String, ExtensionArgs), ExtensionError>,
644 child_refs: Vec<Option<&'a Rel>>,
645 ctx: &S,
646 ) -> Self {
647 match decoded {
648 Ok((name, args)) => {
649 let (children, _) = Relation::convert_children(child_refs, ctx);
650 let mut positional = vec![];
651 for value in args.positional {
652 positional.push(Value::ExtensionArgument(value));
653 }
654 let mut named = vec![];
655 for (key, value) in args.named {
656 named.push(NamedArg {
657 name: Cow::Owned(key),
658 value: Value::ExtensionArgument(value),
659 });
660 }
661 let mut columns = vec![];
662 for col in args.output_columns {
663 columns.push(Value::ExtColumn(col));
664 }
665 Relation {
666 name: Cow::Owned(format!("{}:{}", ext_type, name)),
667 arguments: Some(Arguments::inline(positional, named)),
668 columns,
669 emit: None,
670 addenda: AddendumLines::none(),
674 children,
675 }
676 }
677 Err(error) => {
678 let (children, _) = Relation::convert_children(child_refs, ctx);
679 Relation {
680 name: Cow::Borrowed(ext_type),
681 arguments: None,
682 columns: vec![Value::Missing(PlanError::invalid(
683 "extension",
684 None::<&str>,
685 error.to_string(),
686 ))],
687 emit: None,
688 addenda: AddendumLines::none(),
689 children,
690 }
691 }
692 }
693 }
694
695 fn from_aggregate<S: Scope>(rel: &'a AggregateRel, ctx: &S) -> Self {
705 let mut grouping_sets: Vec<Vec<Value>> = vec![]; let expression_list: Vec<Value>; #[allow(deprecated)]
712 if rel.grouping_expressions.is_empty()
713 && !rel.groupings.is_empty()
714 && !rel.groupings[0].grouping_expressions.is_empty()
715 {
716 (expression_list, grouping_sets) = Relation::get_grouping_sets(rel);
717 } else {
718 expression_list = rel
719 .grouping_expressions
720 .iter()
721 .map(Value::Expression)
722 .collect::<Vec<_>>(); for group in &rel.groupings {
724 let mut grouping_set: Vec<Value> = vec![];
725 for i in &group.expression_references {
726 grouping_set.push(Value::Reference(*i as i32));
727 }
728 grouping_sets.push(grouping_set);
729 }
730 if rel.groupings.is_empty() {
732 grouping_sets.push(vec![]);
733 }
734 }
735
736 let is_single = grouping_sets.len() == 1;
737 let mut positional: Vec<Value> = vec![];
738 for g in grouping_sets {
739 if g.is_empty() {
740 positional.push(Value::EmptyGroup);
741 } else if is_single {
742 positional.extend(g);
744 } else {
745 positional.push(Value::Tuple(g));
746 }
747 }
748
749 let arguments = Some(Arguments::inline(positional, vec![]));
751
752 let mut all_outputs: Vec<Value> = expression_list;
754
755 for m in &rel.measures {
758 if let Some(agg_fn) = m.measure.as_ref() {
759 all_outputs.push(Value::AggregateFunction(agg_fn));
760 }
761 }
762 let emit = get_emit(rel.common.as_ref());
763 let (children, _) = Relation::convert_children(vec![rel.input.as_deref()], ctx);
764
765 Relation {
766 name: Cow::Borrowed("Aggregate"),
767 arguments,
768 columns: all_outputs,
769 emit,
770 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
771 children,
772 }
773 }
774
775 fn get_grouping_sets(rel: &'a AggregateRel) -> (Vec<Value<'a>>, Vec<Vec<Value<'a>>>) {
776 let mut grouping_sets: Vec<Vec<Value>> = vec![];
777 let mut expression_list: Vec<Value> = Vec::new();
778
779 let mut expression_index_map = HashMap::new();
781 let mut i: i32 = 0; for group in &rel.groupings {
784 let mut grouping_set: Vec<Value> = vec![];
785 #[allow(deprecated)]
786 for exp in &group.grouping_expressions {
787 let key = exp.encode_to_vec();
791 expression_index_map.entry(key.clone()).or_insert_with(|| {
792 let value = Value::Expression(exp);
793 expression_list.push(value); let index = i;
796 i += 1;
797 index });
799 grouping_set.push(Value::Reference(expression_index_map[&key]));
800 }
801 grouping_sets.push(grouping_set);
802 }
803 (expression_list, grouping_sets)
804 }
805}
806
807impl Textify for RelRoot {
808 fn name() -> &'static str {
809 "RelRoot"
810 }
811
812 fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
813 let names = self.names.iter().map(|n| Name(n)).collect::<Vec<_>>();
814
815 write!(
816 w,
817 "{}Root[{}]",
818 ctx.indent(),
819 ctx.separated(names.iter(), ", ")
820 )?;
821 let child_scope = ctx.push_indent();
822 for child in self.input.iter() {
823 writeln!(w)?;
824 child.textify(&child_scope, w)?;
825 }
826
827 Ok(())
828 }
829}
830
831impl Textify for PlanRelType {
832 fn name() -> &'static str {
833 "PlanRelType"
834 }
835
836 fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
837 match self {
838 PlanRelType::Rel(rel) => rel.textify(ctx, w),
839 PlanRelType::Root(root) => root.textify(ctx, w),
840 }
841 }
842}
843
844impl Textify for PlanRel {
845 fn name() -> &'static str {
846 "PlanRel"
847 }
848
849 fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
852 write!(w, "{}", ctx.expect(self.rel_type.as_ref()))
853 }
854}
855
856impl<'a> Relation<'a> {
857 fn from_sort<S: Scope>(rel: &'a SortRel, ctx: &S) -> Self {
858 let (children, input_columns) = Relation::convert_children(vec![rel.input.as_deref()], ctx);
859 let mut positional = vec![];
860 for sort_field in &rel.sorts {
861 positional.push(Value::from(sort_field));
862 }
863 let arguments = Some(Arguments::inline(positional, vec![]));
864 let mut col_values = vec![];
866 for i in 0..input_columns {
867 col_values.push(Value::Reference(i as i32));
868 }
869 let emit = get_emit(rel.common.as_ref());
870 Relation {
871 name: Cow::Borrowed("Sort"),
872 arguments,
873 columns: col_values,
874 emit,
875 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
876 children,
877 }
878 }
879
880 fn from_fetch<S: Scope>(rel: &'a FetchRel, ctx: &S) -> Self {
881 let (children, input_columns) = Relation::convert_children(vec![rel.input.as_deref()], ctx);
882 let mut named_args: Vec<NamedArg> = vec![];
883 match &rel.count_mode {
884 Some(CountMode::CountExpr(expr)) => {
885 named_args.push(NamedArg {
886 name: Cow::Borrowed("limit"),
887 value: Value::Expression(expr),
888 });
889 }
890 #[allow(deprecated)]
891 Some(CountMode::Count(val)) => {
892 named_args.push(NamedArg {
893 name: Cow::Borrowed("limit"),
894 value: Value::Integer(*val),
895 });
896 }
897 None => {}
898 }
899 if let Some(offset) = &rel.offset_mode {
900 match offset {
901 substrait::proto::fetch_rel::OffsetMode::OffsetExpr(expr) => {
902 named_args.push(NamedArg {
903 name: Cow::Borrowed("offset"),
904 value: Value::Expression(expr),
905 });
906 }
907 #[allow(deprecated)]
908 substrait::proto::fetch_rel::OffsetMode::Offset(val) => {
909 named_args.push(NamedArg {
910 name: Cow::Borrowed("offset"),
911 value: Value::Integer(*val),
912 });
913 }
914 }
915 }
916
917 let emit = get_emit(rel.common.as_ref());
918 let columns: Vec<Value> = (0..input_columns)
920 .map(|i| Value::Reference(i as i32))
921 .collect();
922 Relation {
923 name: Cow::Borrowed("Fetch"),
924 arguments: Some(Arguments::inline(vec![], named_args)),
925 columns,
926 emit,
927 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
928 children,
929 }
930 }
931}
932
933fn join_output_columns(
934 join_type: join_rel::JoinType,
935 left_columns: usize,
936 right_columns: usize,
937) -> Vec<Value<'static>> {
938 let total_columns = match join_type {
939 join_rel::JoinType::Inner
941 | join_rel::JoinType::Left
942 | join_rel::JoinType::Right
943 | join_rel::JoinType::Outer => left_columns + right_columns,
944
945 join_rel::JoinType::LeftSemi | join_rel::JoinType::LeftAnti => left_columns,
947
948 join_rel::JoinType::RightSemi | join_rel::JoinType::RightAnti => right_columns,
950
951 join_rel::JoinType::LeftSingle => left_columns,
953 join_rel::JoinType::RightSingle => right_columns,
954
955 join_rel::JoinType::LeftMark => left_columns + 1,
957 join_rel::JoinType::RightMark => right_columns + 1,
958
959 join_rel::JoinType::Unspecified => left_columns + right_columns,
961 };
962
963 (0..total_columns)
965 .map(|i| Value::Reference(i as i32))
966 .collect()
967}
968
969impl<'a> Relation<'a> {
970 fn from_join<S: Scope>(rel: &'a JoinRel, ctx: &S) -> Self {
971 let (children, _total_columns) =
972 Relation::convert_children(vec![rel.left.as_deref(), rel.right.as_deref()], ctx);
973
974 assert_eq!(
976 children.len(),
977 2,
978 "convert_children should return same number of elements as input"
979 );
980
981 let left_columns = match &children[0] {
983 Some(child) => child.emitted(),
984 None => 0,
985 };
986 let right_columns = match &children[1] {
987 Some(child) => child.emitted(),
988 None => 0,
989 };
990
991 let (join_type, join_type_value) = match join_rel::JoinType::try_from(rel.r#type) {
994 Ok(join_type) => {
995 let join_type_value = match join_type.as_enum_str() {
996 Ok(s) => Value::Enum(s),
997 Err(e) => Value::Missing(e),
998 };
999 (join_type, join_type_value)
1000 }
1001 Err(_) => {
1002 let join_type_error = Value::Missing(PlanError::invalid(
1004 "JoinRel",
1005 Some("type"),
1006 format!("Unknown join type: {}", rel.r#type),
1007 ));
1008 (join_rel::JoinType::Unspecified, join_type_error)
1009 }
1010 };
1011
1012 let condition = rel
1014 .expression
1015 .as_ref()
1016 .map(|c| Value::Expression(c.as_ref()));
1017 let condition = Value::expect(condition, || {
1018 PlanError::unimplemented("JoinRel", Some("expression"), "Join condition is None")
1019 });
1020
1021 let positional = vec![join_type_value, condition];
1025 let arguments = Some(Arguments::inline(positional, vec![]));
1026
1027 let emit = get_emit(rel.common.as_ref());
1028 let columns = join_output_columns(join_type, left_columns, right_columns);
1029
1030 Relation {
1031 name: Cow::Borrowed("Join"),
1032 arguments,
1033 columns,
1034 emit,
1035 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
1036 children,
1037 }
1038 }
1039
1040 fn from_set<S: Scope>(rel: &'a SetRel, ctx: &S) -> Self {
1041 let child_refs: Vec<Option<&'a Rel>> = rel.inputs.iter().map(Some).collect();
1042 let (children, total_columns) = Relation::convert_children(child_refs, ctx);
1043
1044 let width = if children.is_empty() {
1049 0
1050 } else {
1051 total_columns / children.len()
1052 };
1053
1054 let op_value = match set_rel::SetOp::try_from(rel.op) {
1055 Ok(op) => match op.as_enum_str() {
1056 Ok(s) => Value::Enum(s),
1057 Err(e) => Value::Missing(e),
1058 },
1059 Err(_) => Value::Missing(PlanError::invalid(
1060 "SetRel",
1061 Some("op"),
1062 format!("Unknown set op: {}", rel.op),
1063 )),
1064 };
1065
1066 let arguments = Some(Arguments::inline(vec![op_value], vec![]));
1067 let emit = get_emit(rel.common.as_ref());
1068 let columns = (0..width).map(|i| Value::Reference(i as i32)).collect();
1069
1070 Relation {
1071 name: Cow::Borrowed("Set"),
1072 arguments,
1073 columns,
1074 emit,
1075 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
1076 children,
1077 }
1078 }
1079
1080 fn from_cross<S: Scope>(rel: &'a CrossRel, ctx: &S) -> Self {
1081 let (children, total_columns) =
1082 Relation::convert_children(vec![rel.left.as_deref(), rel.right.as_deref()], ctx);
1083
1084 let columns = (0..total_columns)
1087 .map(|i| Value::Reference(i as i32))
1088 .collect();
1089
1090 Relation {
1091 name: Cow::Borrowed("Cross"),
1092 arguments: None,
1093 columns,
1094 emit: get_emit(rel.common.as_ref()),
1095 addenda: AddendumLines::from_advanced_extension(ctx, rel.advanced_extension.as_ref()),
1096 children,
1097 }
1098 }
1099}
1100
1101impl<'a> From<&'a SortField> for Value<'a> {
1102 fn from(sf: &'a SortField) -> Self {
1103 let field = match &sf.expr {
1104 Some(expr) => match &expr.rex_type {
1105 Some(substrait::proto::expression::RexType::Selection(fref)) => {
1106 if let Some(substrait::proto::expression::field_reference::ReferenceType::DirectReference(seg)) = &fref.reference_type {
1107 if let Some(substrait::proto::expression::reference_segment::ReferenceType::StructField(sf)) = &seg.reference_type {
1108 Value::Reference(sf.field)
1109 } else { Value::Missing(PlanError::unimplemented("SortField", Some("expr"), "Not a struct field")) }
1110 } else { Value::Missing(PlanError::unimplemented("SortField", Some("expr"), "Not a direct reference")) }
1111 }
1112 _ => Value::Missing(PlanError::unimplemented(
1113 "SortField",
1114 Some("expr"),
1115 "Not a selection",
1116 )),
1117 },
1118 None => Value::Missing(PlanError::unimplemented(
1119 "SortField",
1120 Some("expr"),
1121 "Missing expr",
1122 )),
1123 };
1124 let direction = match &sf.sort_kind {
1125 Some(kind) => Value::from(kind),
1126 None => Value::Missing(PlanError::invalid(
1127 "SortKind",
1128 Some(Cow::Borrowed("sort_kind")),
1129 "Missing sort_kind",
1130 )),
1131 };
1132 Value::Tuple(vec![field, direction])
1133 }
1134}
1135
1136impl<'a, T: ValueEnum + ?Sized> From<&'a T> for Value<'a> {
1137 fn from(enum_val: &'a T) -> Self {
1138 match enum_val.as_enum_str() {
1139 Ok(s) => Value::Enum(s),
1140 Err(e) => Value::Missing(e),
1141 }
1142 }
1143}
1144
1145impl ValueEnum for SortKind {
1146 fn as_enum_str(&self) -> Result<Cow<'static, str>, PlanError> {
1147 let d = match self {
1148 &SortKind::Direction(d) => SortDirection::try_from(d),
1149 SortKind::ComparisonFunctionReference(f) => {
1150 return Err(PlanError::invalid(
1151 "SortKind",
1152 Some(Cow::Owned(format!("function reference{f}"))),
1153 "SortKind::ComparisonFunctionReference unimplemented",
1154 ));
1155 }
1156 };
1157 let s = match d {
1158 Err(UnknownEnumValue(d)) => {
1159 return Err(PlanError::invalid(
1160 "SortKind",
1161 Some(Cow::Owned(format!("unknown variant: {d:?}"))),
1162 "Unknown SortDirection",
1163 ));
1164 }
1165 Ok(SortDirection::AscNullsFirst) => "AscNullsFirst",
1166 Ok(SortDirection::AscNullsLast) => "AscNullsLast",
1167 Ok(SortDirection::DescNullsFirst) => "DescNullsFirst",
1168 Ok(SortDirection::DescNullsLast) => "DescNullsLast",
1169 Ok(SortDirection::Clustered) => "Clustered",
1170 Ok(SortDirection::Unspecified) => {
1171 return Err(PlanError::invalid(
1172 "SortKind",
1173 Option::<Cow<str>>::None,
1174 "Unspecified SortDirection",
1175 ));
1176 }
1177 };
1178 Ok(Cow::Borrowed(s))
1179 }
1180}
1181
1182impl ValueEnum for join_rel::JoinType {
1183 fn as_enum_str(&self) -> Result<Cow<'static, str>, PlanError> {
1184 let s = match self {
1185 join_rel::JoinType::Unspecified => {
1186 return Err(PlanError::invalid(
1187 "JoinType",
1188 Option::<Cow<str>>::None,
1189 "Unspecified JoinType",
1190 ));
1191 }
1192 join_rel::JoinType::Inner => "Inner",
1193 join_rel::JoinType::Outer => "Outer",
1194 join_rel::JoinType::Left => "Left",
1195 join_rel::JoinType::Right => "Right",
1196 join_rel::JoinType::LeftSemi => "LeftSemi",
1197 join_rel::JoinType::RightSemi => "RightSemi",
1198 join_rel::JoinType::LeftAnti => "LeftAnti",
1199 join_rel::JoinType::RightAnti => "RightAnti",
1200 join_rel::JoinType::LeftSingle => "LeftSingle",
1201 join_rel::JoinType::RightSingle => "RightSingle",
1202 join_rel::JoinType::LeftMark => "LeftMark",
1203 join_rel::JoinType::RightMark => "RightMark",
1204 };
1205 Ok(Cow::Borrowed(s))
1206 }
1207}
1208
1209impl ValueEnum for set_rel::SetOp {
1210 fn as_enum_str(&self) -> Result<Cow<'static, str>, PlanError> {
1211 let s = match self {
1212 set_rel::SetOp::Unspecified => {
1213 return Err(PlanError::invalid(
1214 "SetOp",
1215 Option::<Cow<str>>::None,
1216 "Unspecified SetOp",
1217 ));
1218 }
1219 set_rel::SetOp::MinusPrimary => "MinusPrimary",
1220 set_rel::SetOp::MinusPrimaryAll => "MinusPrimaryAll",
1221 set_rel::SetOp::MinusMultiset => "MinusMultiset",
1222 set_rel::SetOp::IntersectionPrimary => "IntersectionPrimary",
1223 set_rel::SetOp::IntersectionMultiset => "IntersectionMultiset",
1224 set_rel::SetOp::IntersectionMultisetAll => "IntersectionMultisetAll",
1225 set_rel::SetOp::UnionDistinct => "UnionDistinct",
1226 set_rel::SetOp::UnionAll => "UnionAll",
1227 };
1228 Ok(Cow::Borrowed(s))
1229 }
1230}
1231
1232impl<'a> Textify for NamedArg<'a> {
1233 fn name() -> &'static str {
1234 "NamedArg"
1235 }
1236 fn textify<S: Scope, W: fmt::Write>(&self, ctx: &S, w: &mut W) -> fmt::Result {
1237 write!(w, "{}=", self.name)?;
1238 self.value.textify(ctx, w)
1239 }
1240}
1241
1242#[cfg(test)]
1243mod tests {
1244 use substrait::proto::aggregate_rel::Grouping;
1245 use substrait::proto::expression::literal::LiteralType;
1246 use substrait::proto::expression::{Literal, RexType, ScalarFunction};
1247 use substrait::proto::function_argument::ArgType;
1248 use substrait::proto::read_rel::{NamedTable, ReadType};
1249 use substrait::proto::rel_common::{Direct, Emit};
1250 use substrait::proto::r#type::{self as ptype, Boolean, I64, Kind, Nullability, Struct};
1251 use substrait::proto::{
1252 Expression, FunctionArgument, NamedStruct, ReadRel, Type, aggregate_rel,
1253 };
1254
1255 use super::*;
1256 use crate::fixtures::TestContext;
1257 use crate::parser::expressions::FieldIndex;
1258
1259 #[test]
1260 fn test_read_rel() {
1261 let ctx = TestContext::new();
1262
1263 let read_rel = ReadRel {
1265 common: None,
1266 base_schema: Some(NamedStruct {
1267 names: vec!["col1".into(), "column 2".into()],
1268 r#struct: Some(Struct {
1269 type_variation_reference: 0,
1270 types: vec![
1271 Type {
1272 kind: Some(Kind::I32(ptype::I32 {
1273 type_variation_reference: 0,
1274 nullability: Nullability::Nullable as i32,
1275 })),
1276 },
1277 Type {
1278 kind: Some(Kind::String(ptype::String {
1279 type_variation_reference: 0,
1280 nullability: Nullability::Nullable as i32,
1281 })),
1282 },
1283 ],
1284 nullability: Nullability::Nullable as i32,
1285 }),
1286 }),
1287 filter: None,
1288 best_effort_filter: None,
1289 projection: None,
1290 advanced_extension: None,
1291 read_type: Some(ReadType::NamedTable(NamedTable {
1292 names: vec!["some_db".into(), "test_table".into()],
1293 advanced_extension: None,
1294 })),
1295 };
1296
1297 let rel = Rel {
1298 rel_type: Some(RelType::Read(Box::new(read_rel))),
1299 };
1300 let (result, errors) = ctx.textify(&rel);
1301 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1302 assert_eq!(
1303 result,
1304 "Read[some_db.test_table => col1:i32?, \"column 2\":string?]"
1305 );
1306 }
1307
1308 #[test]
1309 fn test_filter_rel() {
1310 let ctx = TestContext::new()
1311 .with_urn(1, "test_urn")
1312 .with_function(1, 10, "gt");
1313
1314 let read_rel = ReadRel {
1316 common: None,
1317 base_schema: Some(NamedStruct {
1318 names: vec!["col1".into(), "col2".into()],
1319 r#struct: Some(Struct {
1320 type_variation_reference: 0,
1321 types: vec![
1322 Type {
1323 kind: Some(Kind::I32(ptype::I32 {
1324 type_variation_reference: 0,
1325 nullability: Nullability::Nullable as i32,
1326 })),
1327 },
1328 Type {
1329 kind: Some(Kind::I32(ptype::I32 {
1330 type_variation_reference: 0,
1331 nullability: Nullability::Nullable as i32,
1332 })),
1333 },
1334 ],
1335 nullability: Nullability::Nullable as i32,
1336 }),
1337 }),
1338 filter: None,
1339 best_effort_filter: None,
1340 projection: None,
1341 advanced_extension: None,
1342 read_type: Some(ReadType::NamedTable(NamedTable {
1343 names: vec!["test_table".into()],
1344 advanced_extension: None,
1345 })),
1346 };
1347
1348 let filter_expr = Expression {
1350 rex_type: Some(RexType::ScalarFunction(ScalarFunction {
1351 function_reference: 10, arguments: vec![
1353 FunctionArgument {
1354 arg_type: Some(ArgType::Value(Reference(0).into())),
1355 },
1356 FunctionArgument {
1357 arg_type: Some(ArgType::Value(Expression {
1358 rex_type: Some(RexType::Literal(Literal {
1359 literal_type: Some(LiteralType::I32(10)),
1360 nullable: false,
1361 type_variation_reference: 0,
1362 })),
1363 })),
1364 },
1365 ],
1366 options: vec![],
1367 output_type: Some(Type {
1368 kind: Some(Kind::Bool(Boolean {
1369 nullability: Nullability::Required as i32,
1370 type_variation_reference: 0,
1371 })),
1372 }),
1373 #[allow(deprecated)]
1374 args: vec![],
1375 })),
1376 };
1377
1378 let filter_rel = FilterRel {
1379 common: None,
1380 input: Some(Box::new(Rel {
1381 rel_type: Some(RelType::Read(Box::new(read_rel))),
1382 })),
1383 condition: Some(Box::new(filter_expr)),
1384 advanced_extension: None,
1385 };
1386
1387 let rel = Rel {
1388 rel_type: Some(RelType::Filter(Box::new(filter_rel))),
1389 };
1390
1391 let (result, errors) = ctx.textify(&rel);
1392 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1393 let expected = r#"
1394Filter[gt($0, 10:i32):boolean => $0, $1]
1395 Read[test_table => col1:i32?, col2:i32?]"#
1396 .trim_start();
1397 assert_eq!(result, expected);
1398 }
1399
1400 #[test]
1401 fn test_aggregate_function_textify() {
1402 let ctx = TestContext::new()
1403 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1404 .with_function(1, 10, "sum")
1405 .with_function(1, 11, "count");
1406
1407 let agg_fn = get_aggregate_func(10, 1);
1409
1410 let value = Value::AggregateFunction(&agg_fn);
1411 let (result, errors) = ctx.textify(&value);
1412
1413 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1414 assert_eq!(result, "sum($1):i64");
1415 }
1416
1417 #[test]
1418 fn test_aggregate_relation_textify() {
1419 let ctx = TestContext::new()
1420 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1421 .with_function(1, 10, "sum")
1422 .with_function(1, 11, "count");
1423
1424 let agg_fn1 = get_aggregate_func(10, 1);
1426 let agg_fn2 = get_aggregate_func(11, 1);
1427
1428 let grouping_expressions = vec![Expression {
1429 rex_type: Some(RexType::Selection(Box::new(
1430 FieldIndex(0).to_field_reference(),
1431 ))),
1432 }];
1433
1434 let measures = vec![
1435 aggregate_rel::Measure {
1436 measure: Some(agg_fn1),
1437 filter: None,
1438 },
1439 aggregate_rel::Measure {
1440 measure: Some(agg_fn2),
1441 filter: None,
1442 },
1443 ];
1444
1445 let common = Some(RelCommon {
1446 emit_kind: Some(EmitKind::Emit(Emit {
1447 output_mapping: vec![1, 2], })),
1449 ..Default::default()
1450 });
1451
1452 let aggregate_rel = create_aggregate_rel(grouping_expressions, vec![], measures, common);
1453
1454 let rel = Rel {
1455 rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1456 };
1457 let (result, errors) = ctx.textify(&rel);
1458
1459 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1460 assert!(result.contains("Aggregate[_ => sum($1):i64, count($1):i64]"));
1462 }
1463
1464 #[test]
1465 fn test_multiple_groupings_on_aggregate_deprecated() {
1466 let ctx = TestContext::new()
1469 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1470 .with_function(1, 11, "count");
1471
1472 let grouping_expr_0 = create_exp(0);
1473 let grouping_expr_1 = create_exp(1);
1474
1475 let grouping_sets = vec![
1476 aggregate_rel::Grouping {
1477 #[allow(deprecated)]
1478 grouping_expressions: vec![grouping_expr_0.clone()],
1479 expression_references: vec![],
1480 },
1481 aggregate_rel::Grouping {
1482 #[allow(deprecated)]
1483 grouping_expressions: vec![grouping_expr_0.clone(), grouping_expr_1.clone()],
1484 expression_references: vec![],
1485 },
1486 ];
1487
1488 let aggregate_rel = create_aggregate_rel(vec![], grouping_sets, vec![], None);
1489
1490 let rel = Rel {
1491 rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1492 };
1493 let (result, errors) = ctx.textify(&rel);
1494
1495 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1496 assert!(result.contains("Aggregate[($0), ($0, $1) => $0, $1]"));
1497 }
1498
1499 #[test]
1500 fn test_multiple_groupings_with_measure_deprecated() {
1501 let ctx = TestContext::new()
1504 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1505 .with_function(1, 11, "count");
1506
1507 let agg_fn1 = get_aggregate_func(11, 2);
1508
1509 let grouping_expr_0 = create_exp(0);
1510 let grouping_expr_1 = create_exp(1);
1511
1512 let grouping_sets = vec![
1513 aggregate_rel::Grouping {
1514 #[allow(deprecated)]
1515 grouping_expressions: vec![grouping_expr_0.clone()],
1516 expression_references: vec![],
1517 },
1518 aggregate_rel::Grouping {
1519 #[allow(deprecated)]
1520 grouping_expressions: vec![grouping_expr_0.clone(), grouping_expr_1.clone()],
1521 expression_references: vec![],
1522 },
1523 ];
1524
1525 let measures = vec![aggregate_rel::Measure {
1526 measure: Some(agg_fn1),
1527 filter: None,
1528 }];
1529
1530 let aggregate_rel = create_aggregate_rel(vec![], grouping_sets, measures, None);
1531
1532 let rel = Rel {
1533 rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1534 };
1535 let (result, errors) = ctx.textify(&rel);
1536
1537 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1538 assert!(result.contains("($0), ($0, $1) => $0, $1, count($2):i64"));
1539 }
1540
1541 #[test]
1542 fn test_multiple_groupings_on_aggregate() {
1543 let ctx = TestContext::new()
1544 .with_urn(1, "https://github.com/substrait-io/substrait/blob/main/extensions/functions_aggregate.yaml")
1545 .with_function(1, 11, "count");
1546
1547 let agg_fn2 = get_aggregate_func(11, 2);
1548
1549 let grouping_expressions = vec![
1550 Expression {
1551 rex_type: Some(RexType::Selection(Box::new(
1552 FieldIndex(0).to_field_reference(),
1553 ))),
1554 },
1555 Expression {
1556 rex_type: Some(RexType::Selection(Box::new(
1557 FieldIndex(1).to_field_reference(),
1558 ))),
1559 },
1560 ];
1561
1562 let grouping_sets = vec![
1563 Grouping {
1564 #[allow(deprecated)]
1565 grouping_expressions: vec![],
1566 expression_references: vec![0, 1],
1567 },
1568 Grouping {
1569 #[allow(deprecated)]
1570 grouping_expressions: vec![],
1571 expression_references: vec![0, 1],
1572 },
1573 Grouping {
1574 #[allow(deprecated)]
1575 grouping_expressions: vec![],
1576 expression_references: vec![1],
1577 },
1578 Grouping {
1579 #[allow(deprecated)]
1580 grouping_expressions: vec![],
1581 expression_references: vec![1, 1],
1582 },
1583 Grouping {
1584 #[allow(deprecated)]
1585 grouping_expressions: vec![],
1586 expression_references: vec![],
1587 },
1588 ];
1589
1590 let measures = vec![aggregate_rel::Measure {
1591 measure: Some(agg_fn2),
1592 filter: None,
1593 }];
1594
1595 let aggregate_rel =
1596 create_aggregate_rel(grouping_expressions, grouping_sets, measures, None);
1597
1598 let rel = Rel {
1599 rel_type: Some(RelType::Aggregate(Box::new(aggregate_rel))),
1600 };
1601 let (result, errors) = ctx.textify(&rel);
1602
1603 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1604 assert!(
1605 result.contains(
1606 "Aggregate[($0, $1), ($0, $1), ($1), ($1, $1), _ => $0, $1, count($2):i64]"
1607 )
1608 );
1609 }
1610
1611 #[test]
1612 fn test_arguments_textify_positional_only() {
1613 let ctx = TestContext::new();
1614 let args = Arguments::inline(vec![Value::Integer(42), Value::Integer(7)], vec![]);
1615 let (result, errors) = ctx.textify(&args);
1616 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1617 assert_eq!(result, "42, 7");
1618 }
1619
1620 #[test]
1621 fn test_arguments_textify_named_only() {
1622 let ctx = TestContext::new();
1623 let args = Arguments::inline(
1624 vec![],
1625 vec![
1626 NamedArg {
1627 name: Cow::Borrowed("limit"),
1628 value: Value::Integer(10),
1629 },
1630 NamedArg {
1631 name: Cow::Borrowed("offset"),
1632 value: Value::Integer(5),
1633 },
1634 ],
1635 );
1636 let (result, errors) = ctx.textify(&args);
1637 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1638 assert_eq!(result, "limit=10, offset=5");
1639 }
1640
1641 #[test]
1642 fn test_join_relation_unknown_type() {
1643 let ctx = TestContext::new();
1644
1645 let join_rel = JoinRel {
1647 left: Some(Box::new(Rel {
1648 rel_type: Some(RelType::Read(Box::default())),
1649 })),
1650 right: Some(Box::new(Rel {
1651 rel_type: Some(RelType::Read(Box::default())),
1652 })),
1653 expression: Some(Box::new(Expression::default())),
1654 r#type: 999, common: None,
1656 post_join_filter: None,
1657 advanced_extension: None,
1658 };
1659
1660 let rel = Rel {
1661 rel_type: Some(RelType::Join(Box::new(join_rel))),
1662 };
1663 let (result, errors) = ctx.textify(&rel);
1664
1665 assert!(!errors.is_empty(), "Expected errors for unknown join type");
1667 assert!(
1668 result.contains("!{JoinRel}"),
1669 "Expected error token for unknown join type"
1670 );
1671 assert!(
1672 result.contains("Join["),
1673 "Expected Join relation to be formatted"
1674 );
1675 }
1676
1677 #[test]
1678 fn test_set_relation_unknown_op() {
1679 let ctx = TestContext::new();
1680
1681 let set_rel = SetRel {
1682 common: None,
1683 inputs: vec![
1684 Rel {
1685 rel_type: Some(RelType::Read(Box::default())),
1686 },
1687 Rel {
1688 rel_type: Some(RelType::Read(Box::default())),
1689 },
1690 ],
1691 op: 999, advanced_extension: None,
1693 };
1694 let rel = Rel {
1695 rel_type: Some(RelType::Set(set_rel)),
1696 };
1697
1698 let (result, errors) = ctx.textify(&rel);
1699 assert!(!errors.is_empty(), "Expected errors for unknown set op");
1700 assert!(
1701 result.contains("!{SetRel}"),
1702 "Expected error token for unknown set op, got: {result}"
1703 );
1704 assert!(
1705 result.contains("Set["),
1706 "Expected Set relation to be formatted"
1707 );
1708 }
1709
1710 fn basic_read(table: &str) -> Rel {
1711 Rel {
1712 rel_type: Some(RelType::Read(Box::new(ReadRel {
1713 common: None,
1714 base_schema: Some(get_basic_schema()),
1715 filter: None,
1716 best_effort_filter: None,
1717 projection: None,
1718 advanced_extension: None,
1719 read_type: Some(ReadType::NamedTable(NamedTable {
1720 names: vec![table.into()],
1721 advanced_extension: None,
1722 })),
1723 }))),
1724 }
1725 }
1726
1727 #[test]
1728 fn test_cross_relation() {
1729 let ctx = TestContext::new();
1730
1731 let cross = CrossRel {
1734 common: Some(RelCommon {
1735 emit_kind: Some(EmitKind::Direct(Direct {})),
1736 ..Default::default()
1737 }),
1738 left: Some(Box::new(basic_read("left_tbl"))),
1739 right: Some(Box::new(basic_read("right_tbl"))),
1740 advanced_extension: None,
1741 };
1742 let rel = Rel {
1743 rel_type: Some(RelType::Cross(Box::new(cross))),
1744 };
1745
1746 let (result, errors) = ctx.textify(&rel);
1747 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1748 let expected = r#"
1749Cross[$0, $1, $2, $3, $4, $5]
1750 Read[left_tbl => category:string?, amount:fp64?, value:i32?]
1751 Read[right_tbl => category:string?, amount:fp64?, value:i32?]"#
1752 .trim_start();
1753 assert_eq!(result, expected);
1754 }
1755
1756 #[test]
1757 fn test_cross_relation_prunes_columns() {
1758 let ctx = TestContext::new();
1759
1760 let cross = CrossRel {
1762 common: Some(RelCommon {
1763 emit_kind: Some(EmitKind::Emit(Emit {
1764 output_mapping: vec![0, 3],
1765 })),
1766 ..Default::default()
1767 }),
1768 left: Some(Box::new(basic_read("left_tbl"))),
1769 right: Some(Box::new(basic_read("right_tbl"))),
1770 advanced_extension: None,
1771 };
1772 let rel = Rel {
1773 rel_type: Some(RelType::Cross(Box::new(cross))),
1774 };
1775
1776 let (result, errors) = ctx.textify(&rel);
1777 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1778 let expected = r#"
1779Cross[$0, $3]
1780 Read[left_tbl => category:string?, amount:fp64?, value:i32?]
1781 Read[right_tbl => category:string?, amount:fp64?, value:i32?]"#
1782 .trim_start();
1783 assert_eq!(result, expected);
1784 }
1785
1786 #[test]
1787 fn test_arguments_textify_both() {
1788 let ctx = TestContext::new();
1789 let args = Arguments::inline(
1790 vec![Value::Integer(1)],
1791 vec![NamedArg {
1792 name: "foo".into(),
1793 value: Value::Integer(2),
1794 }],
1795 );
1796 let (result, errors) = ctx.textify(&args);
1797 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1798 assert_eq!(result, "1, foo=2");
1799 }
1800
1801 #[test]
1802 fn test_arguments_textify_empty() {
1803 let ctx = TestContext::new();
1804 let args = Arguments::inline(vec![], vec![]);
1805 let (result, errors) = ctx.textify(&args);
1806 assert!(errors.is_empty(), "Expected no errors, got: {errors:?}");
1807 assert_eq!(result, "_");
1808 }
1809
1810 #[test]
1811 fn test_named_arg_textify_error_token() {
1812 let ctx = TestContext::new();
1813 let named_arg = NamedArg {
1814 name: "foo".into(),
1815 value: Value::Missing(PlanError::invalid(
1816 "my_enum",
1817 Some(Cow::Borrowed("my_enum")),
1818 Cow::Borrowed("my_enum"),
1819 )),
1820 };
1821 let (result, errors) = ctx.textify(&named_arg);
1822 assert!(result.contains("foo=!{my_enum}"), "Output: {result}");
1824 assert!(!errors.is_empty(), "Expected error for error token");
1826 }
1827
1828 #[test]
1829 fn test_join_type_enum_textify() {
1830 assert_eq!(join_rel::JoinType::Inner.as_enum_str().unwrap(), "Inner");
1832 assert_eq!(join_rel::JoinType::Left.as_enum_str().unwrap(), "Left");
1833 assert_eq!(
1834 join_rel::JoinType::LeftSemi.as_enum_str().unwrap(),
1835 "LeftSemi"
1836 );
1837 assert_eq!(
1838 join_rel::JoinType::LeftAnti.as_enum_str().unwrap(),
1839 "LeftAnti"
1840 );
1841 }
1842
1843 #[test]
1844 fn test_join_output_columns() {
1845 let inner_cols = super::join_output_columns(join_rel::JoinType::Inner, 2, 3);
1847 assert_eq!(inner_cols.len(), 5); assert!(matches!(inner_cols[0], Value::Reference(0)));
1849 assert!(matches!(inner_cols[4], Value::Reference(4)));
1850
1851 let left_semi_cols = super::join_output_columns(join_rel::JoinType::LeftSemi, 2, 3);
1853 assert_eq!(left_semi_cols.len(), 2); assert!(matches!(left_semi_cols[0], Value::Reference(0)));
1855 assert!(matches!(left_semi_cols[1], Value::Reference(1)));
1856
1857 let right_semi_cols = super::join_output_columns(join_rel::JoinType::RightSemi, 2, 3);
1859 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)));
1862 assert!(matches!(right_semi_cols[2], Value::Reference(2))); let left_mark_cols = super::join_output_columns(join_rel::JoinType::LeftMark, 2, 3);
1866 assert_eq!(left_mark_cols.len(), 3); assert!(matches!(left_mark_cols[0], Value::Reference(0)));
1868 assert!(matches!(left_mark_cols[1], Value::Reference(1)));
1869 assert!(matches!(left_mark_cols[2], Value::Reference(2))); let right_mark_cols = super::join_output_columns(join_rel::JoinType::RightMark, 2, 3);
1873 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)));
1876 assert!(matches!(right_mark_cols[2], Value::Reference(2))); assert!(matches!(right_mark_cols[3], Value::Reference(3))); }
1879
1880 fn get_aggregate_func(func_ref: u32, column_ind: i32) -> AggregateFunction {
1881 AggregateFunction {
1882 function_reference: func_ref,
1883 arguments: vec![FunctionArgument {
1884 arg_type: Some(ArgType::Value(Expression {
1885 rex_type: Some(RexType::Selection(Box::new(
1886 FieldIndex(column_ind).to_field_reference(),
1887 ))),
1888 })),
1889 }],
1890 options: vec![],
1891 output_type: Some(Type {
1892 kind: Some(Kind::I64(I64 {
1893 nullability: Nullability::Required as i32,
1894 type_variation_reference: 0,
1895 })),
1896 }),
1897 invocation: 0,
1898 phase: 0,
1899 sorts: vec![],
1900 #[allow(deprecated)]
1901 args: vec![],
1902 }
1903 }
1904
1905 fn create_aggregate_rel(
1906 grouping_expressions: Vec<Expression>,
1907 grouping_sets: Vec<Grouping>,
1908 measures: Vec<aggregate_rel::Measure>,
1909 common: Option<RelCommon>,
1910 ) -> AggregateRel {
1911 let common = common.or_else(|| {
1912 Some(RelCommon {
1913 emit_kind: Some(EmitKind::Direct(Direct {})),
1914 ..Default::default()
1915 })
1916 });
1917 AggregateRel {
1918 input: Some(Box::new(Rel {
1919 rel_type: Some(RelType::Read(Box::new(ReadRel {
1920 common: None,
1921 base_schema: Some(get_basic_schema()),
1922 filter: None,
1923 best_effort_filter: None,
1924 projection: None,
1925 advanced_extension: None,
1926 read_type: Some(ReadType::NamedTable(NamedTable {
1927 names: vec!["orders".into()],
1928 advanced_extension: None,
1929 })),
1930 }))),
1931 })),
1932 grouping_expressions,
1933 groupings: grouping_sets,
1934 measures,
1935 common,
1936 advanced_extension: None,
1937 }
1938 }
1939
1940 fn get_basic_schema() -> NamedStruct {
1941 NamedStruct {
1942 names: vec!["category".into(), "amount".into(), "value".into()],
1943 r#struct: Some(Struct {
1944 type_variation_reference: 0,
1945 types: vec![
1946 Type {
1947 kind: Some(Kind::String(ptype::String {
1948 type_variation_reference: 0,
1949 nullability: Nullability::Nullable as i32,
1950 })),
1951 },
1952 Type {
1953 kind: Some(Kind::Fp64(ptype::Fp64 {
1954 type_variation_reference: 0,
1955 nullability: Nullability::Nullable as i32,
1956 })),
1957 },
1958 Type {
1959 kind: Some(Kind::I32(ptype::I32 {
1960 type_variation_reference: 0,
1961 nullability: Nullability::Nullable as i32,
1962 })),
1963 },
1964 ],
1965 nullability: Nullability::Nullable as i32,
1966 }),
1967 }
1968 }
1969
1970 fn create_exp(column_ind: i32) -> Expression {
1971 Expression {
1972 rex_type: Some(RexType::Selection(Box::new(
1973 FieldIndex(column_ind).to_field_reference(),
1974 ))),
1975 }
1976 }
1977
1978 #[test]
1979 fn test_unsupported_rel_type_produces_failure_token() {
1980 use substrait::proto::ReferenceRel;
1981
1982 let ctx = TestContext::new();
1983
1984 let rel = Rel {
1988 rel_type: Some(RelType::Reference(ReferenceRel { subtree_ordinal: 0 })),
1989 };
1990
1991 let (result, errors) = ctx.textify(&rel);
1992
1993 assert!(
1995 result.contains("!{Rel}"),
1996 "Expected '!{{Rel}}' in output, got: {result}"
1997 );
1998
1999 assert_eq!(errors.0.len(), 1, "Expected exactly one error: {errors:?}");
2001
2002 match &errors.0[0] {
2004 FormatError::Format(plan_err) => {
2005 assert_eq!(plan_err.message, "Rel");
2006 assert_eq!(
2007 plan_err.error_type,
2008 crate::textify::foundation::FormatErrorType::Unimplemented
2009 );
2010 assert!(
2011 plan_err
2012 .lookup
2013 .as_deref()
2014 .unwrap_or("")
2015 .contains("Reference"),
2016 "Expected lookup to mention 'Reference', got: {:?}",
2017 plan_err.lookup
2018 );
2019 }
2020 other => panic!("Expected FormatError::Format, got: {other:?}"),
2021 }
2022 }
2023}