[CALCITE-7669] Uncollect should support the Trino semantics of UNNEST

core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableUncollect.java

@@ -51,7 +51,16 @@
    * <p>Use {@link #create} unless you know what you're doing. */
   public EnumerableUncollect(RelOptCluster cluster, RelTraitSet traitSet,
       RelNode child, boolean withOrdinality) {
-    super(cluster, traitSet, child, withOrdinality, Collections.emptyList());
+    this(cluster, traitSet, child, withOrdinality, true);
+  }
+
+  /** Creates an EnumerableUncollect.
+   *
+   * <p>Use {@link #create} unless you know what you're doing. */
+  public EnumerableUncollect(RelOptCluster cluster, RelTraitSet traitSet,
+      RelNode child, boolean withOrdinality, boolean expandStructFields) {
+    super(cluster, traitSet, child, withOrdinality, Collections.emptyList(),
+        expandStructFields);
     assert getConvention() instanceof EnumerableConvention;
     assert getConvention() == child.getConvention();
   }
@@ -72,13 +81,30 @@
     return new EnumerableUncollect(cluster, traitSet, input, withOrdinality);
   }
 
+  /**
+   * Creates an EnumerableUncollect.
+   *
+   * @param traitSet           Trait set
+   * @param input              Input relational expression
+   * @param withOrdinality     Whether output should contain an ORDINALITY column
+   * @param expandStructFields If true, a collection whose element type is a struct
+   *                           produces one output column per struct field; if false,
+   *                           a single column typed as the whole element
+   */
+  public static EnumerableUncollect create(RelTraitSet traitSet, RelNode input,
+      boolean withOrdinality, boolean expandStructFields) {
+    final RelOptCluster cluster = input.getCluster();
+    return new EnumerableUncollect(cluster, traitSet, input, withOrdinality,
+        expandStructFields);
+  }
+
   @Override public EnumerableUncollect copy(RelTraitSet traitSet,
       RelNode newInput) {
     return new EnumerableUncollect(getCluster(), traitSet, newInput,
-        withOrdinality);
+        withOrdinality, expandStructFields);
   }
 
   @Override public Result implement(EnumerableRelImplementor implementor, Prefer pref) {
     final BlockBuilder builder = new BlockBuilder();
     final EnumerableRel child = (EnumerableRel) getInput();
     final Result result = implementor.visitChild(this, 0, child, pref);
@@ -105,7 +131,7 @@
         inputTypes.add(FlatProductInputType.MAP);
       } else {
         final RelDataType elementType = getComponentTypeOrThrow(type);
-        if (elementType.isStruct()) {
+        if (elementType.isStruct() && expandStructFields) {
           if (elementType.getFieldCount() == 1 && child.getRowType().getFieldList().size() == 1
               && !withOrdinality) {
             // Solves CALCITE-4063: if we are processing a single field, which is a struct with a
@@ -116,6 +142,12 @@
             fieldCounts.add(elementType.getFieldCount());
             inputTypes.add(FlatProductInputType.LIST);
           }
+        } else if (elementType.isStruct()) {
+          // A struct element kept whole occupies a single output column,
+          // like a scalar element, but its row value must be converted from
+          // the collection's internal list representation to Object[].
+          fieldCounts.add(-1);
+          inputTypes.add(FlatProductInputType.STRUCT);
         } else {
           fieldCounts.add(-1);
           inputTypes.add(FlatProductInputType.SCALAR);

core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableUncollectRule.java

@@ -49,6 +49,6 @@ protected EnumerableUncollectRule(Config config) {
         convert(input,
             input.getTraitSet().replace(EnumerableConvention.INSTANCE));
     return EnumerableUncollect.create(traitSet, newInput,
-        uncollect.withOrdinality);
+        uncollect.withOrdinality, uncollect.expandStructFields);
   }
 }

core/src/main/java/org/apache/calcite/rel/core/Uncollect.java

@@ -50,10 +50,21 @@
  * <p>Like its inverse operation {@link Collect}, Uncollect is generally
  * invoked in a nested loop, driven by
  * {@link org.apache.calcite.rel.logical.LogicalCorrelate} or similar.
+ *
+ * <p>{@code expandStructFields} controls the shape of the element columns:
+ * if {@code true} a collection whose element type is a struct produces one
+ * output column per struct field; if {@code false} it produces a single
+ * column typed as the whole element (Trino semantics). Maps always expand
+ * into a key and a value column, regardless of this flag.
  */
 public class Uncollect extends SingleRel {
   public final boolean withOrdinality;
 
+  /** If true, a collection whose element type is a struct expands into one
+   * output column per struct field; if false, it produces a single column
+   * typed as the whole element. */
+  public final boolean expandStructFields;
+
   // To alias the items in Uncollect list,
   // i.e., "UNNEST(a, b, c) as T(d, e, f)"
   // outputs as row type Record(d, e, f) where the field "d" has element type of "a",
@@ -74,12 +85,30 @@
   /** Creates an Uncollect.
    *
    * <p>Use {@link #create} unless you know what you're doing. */
-  @SuppressWarnings("method.invocation.invalid")
   public Uncollect(RelOptCluster cluster, RelTraitSet traitSet, RelNode input,
       boolean withOrdinality, List<String> itemAliases) {
+    // Non-empty item aliases historically implied that struct elements are not
+    // expanded (Presto dialect), so this constructor derives
+    // {@code expandStructFields} from their absence.
+    this(cluster, traitSet, input, withOrdinality, itemAliases, itemAliases.isEmpty());
+  }
+
+  /** Creates an Uncollect.
+   *
+   * @param input              Input relational expression
+   * @param withOrdinality     Whether output should contain an ORDINALITY column
+   * @param itemAliases        Aliases for the operand items
+   * @param expandStructFields If true, a collection whose element type is a struct
+   *                           produces one output column per struct field; if false,
+   *                           a single column typed as the whole element
+   */
+  @SuppressWarnings("method.invocation.invalid")
+  public Uncollect(RelOptCluster cluster, RelTraitSet traitSet, RelNode input,
+      boolean withOrdinality, List<String> itemAliases, boolean expandStructFields) {
     super(cluster, traitSet, input);
     this.withOrdinality = withOrdinality;
     this.itemAliases = ImmutableList.copyOf(itemAliases);
+    this.expandStructFields = expandStructFields;
     requireNonNull(deriveRowType(), "invalid child rowType");
   }
 
@@ -88,7 +117,8 @@
    */
   public Uncollect(RelInput input) {
     this(input.getCluster(), input.getTraitSet(), input.getInput(),
-        input.getBoolean("withOrdinality", false), Collections.emptyList());
+        input.getBoolean("withOrdinality", false), Collections.emptyList(),
+        input.getBoolean("expandStructFields", true));
   }
 
   /**
@@ -111,6 +141,28 @@
     return new Uncollect(cluster, traitSet, input, withOrdinality, itemAliases);
   }
 
+  /**
+   * Creates an Uncollect.
+   *
+   * @param traitSet           Trait set
+   * @param input              Input relational expression
+   * @param withOrdinality     Whether output should contain an ORDINALITY column
+   * @param itemAliases        Aliases for the operand items
+   * @param expandStructFields If true, a collection whose element type is a struct
+   *                           produces one output column per struct field; if false,
+   *                           a single column typed as the whole element
+   */
+  public static Uncollect create(
+      RelTraitSet traitSet,
+      RelNode input,
+      boolean withOrdinality,
+      List<String> itemAliases,
+      boolean expandStructFields) {
+    final RelOptCluster cluster = input.getCluster();
+    return new Uncollect(cluster, traitSet, input, withOrdinality, itemAliases,
+        expandStructFields);
+  }
+
   //~ Methods ----------------------------------------------------------------
 
   @Override public RelNode accept(RelShuttle shuttle) {
@@ -119,7 +171,8 @@
 
   @Override public RelWriter explainTerms(RelWriter pw) {
     return super.explainTerms(pw)
-        .itemIf("withOrdinality", withOrdinality, withOrdinality);
+        .itemIf("withOrdinality", withOrdinality, withOrdinality)
+        .itemIf("expandStructFields", expandStructFields, !expandStructFields);
   }
 
   @Override public final RelNode copy(RelTraitSet traitSet,
@@ -129,34 +182,47 @@
 
   public RelNode copy(RelTraitSet traitSet, RelNode input) {
     assert traitSet.containsIfApplicable(Convention.NONE);
-    return new Uncollect(getCluster(), traitSet, input, withOrdinality, itemAliases);
-  }
-
-  @Override protected RelDataType deriveRowType() {
-    return deriveUncollectRowType(input, withOrdinality, itemAliases);
+    return new Uncollect(getCluster(), traitSet, input, withOrdinality, itemAliases,
+        expandStructFields);
   }
 
   /**
    * Returns the row type returned by applying the 'UNNEST' operation to a
    * relational expression.
    *
-   * <p>Each column in the relational expression must be a multiset of
-   * structs or an array. The return type is the combination of expanding
-   * element types from each column, plus an ORDINALITY column if {@code
-   * withOrdinality}. If {@code itemAliases} is not empty, the element types
-   * would not expand, each column element outputs as a whole (the return
-   * type has same column types as input type).
+   * @deprecated Construct an {@link Uncollect} and call
+   * {@link #getRowType()} instead.
    */
+  @Deprecated // to be removed before 2.0
   public static RelDataType deriveUncollectRowType(RelNode rel,
       boolean withOrdinality, List<String> itemAliases) {
-    RelDataType inputType = rel.getRowType();
+    return new Uncollect(rel.getCluster(), rel.getTraitSet(), rel,
+        withOrdinality, itemAliases).getRowType();
+  }
+
+  /**
+   * Returns the row type of the 'UNNEST' operation.
+   *
+   * <p>Each column in the input relational expression must be a multiset of
+   * structs or an array. The return type is the combination of expanding
+   * element types from each column, plus an ORDINALITY column if {@code
+   * withOrdinality}.
+   *
+   * <p>{@code expandStructFields} controls the expansion of struct element
+   * types: if {@code true}, one output column per struct field; if {@code
+   * false}, a single column typed as the whole element. Maps always expand
+   * into a key and a value column. {@code itemAliases}, when not empty,
+   * names the non-expanded element columns.
+   */
+  @Override protected RelDataType deriveRowType() {
+    RelDataType inputType = input.getRowType();
     assert inputType.isStruct() : inputType + " is not a struct";
 
     boolean requireAlias = !itemAliases.isEmpty();
     assert !requireAlias || itemAliases.size() == inputType.getFieldCount();
 
     final List<RelDataTypeField> fields = inputType.getFieldList();
-    final RelDataTypeFactory typeFactory = rel.getCluster().getTypeFactory();
+    final RelDataTypeFactory typeFactory = getCluster().getTypeFactory();
     final RelDataTypeFactory.Builder builder = typeFactory.builder();
 
     if (fields.size() == 1
@@ -192,12 +258,7 @@
           throw RESOURCE.unnestArgument().ex();
         }
         boolean isNullable = componentType.isNullable() || padNullable;
-        if (requireAlias) {
-          RelDataType colType = padNullable
-              ? typeFactory.enforceTypeWithNullability(componentType, true)
-              : componentType;
-          builder.add(itemAliases.get(i), colType);
-        } else if (componentType.isStruct()) {
+        if (expandStructFields && componentType.isStruct()) {
           for (RelDataTypeField fieldInfo : componentType.getFieldList()) {
             RelDataType fieldType = fieldInfo.getType();
             if (isNullable) {
@@ -206,11 +267,18 @@
             builder.add(fieldInfo.getName(), fieldType);
           }
         } else {
-          // Element type is not a record, use the field name of the element directly
-          RelDataType colType = padNullable
-              ? typeFactory.enforceTypeWithNullability(componentType, true)
+          // A single column typed as the whole element, named by the item
+          // alias when present, otherwise by the collection field's name.
+          RelDataType elementType = componentType.isStruct()
+              ? typeFactory.builder().kind(componentType.getStructKind())
+                  .addAll(componentType.getFieldList()).build()
               : componentType;
-          builder.add(field.getName(), colType);
+          // A NULL collection element becomes a NULL value in this column, so
+          // the column is nullable whenever the element type is.
+          RelDataType colType = isNullable
+              ? typeFactory.enforceTypeWithNullability(elementType, true)
+              : elementType;
+          builder.add(requireAlias ? itemAliases.get(i) : field.getName(), colType);
         }
       }
     }

core/src/main/java/org/apache/calcite/rel/logical/ToLogicalConverter.java

@@ -41,8 +41,6 @@
 import org.apache.calcite.rel.core.Window;
 import org.apache.calcite.tools.RelBuilder;
 
-import java.util.Collections;
-
 /**
  * Shuttle to convert any rel plan to a plan with all logical nodes.
  */
@@ -191,7 +189,8 @@ public ToLogicalConverter(RelBuilder relBuilder) {
       final Uncollect uncollect = (Uncollect) relNode;
       final RelNode input = visit(uncollect.getInput());
       return Uncollect.create(input.getTraitSet(), input,
-          uncollect.withOrdinality, Collections.emptyList());
+          uncollect.withOrdinality, uncollect.getItemAliases(),
+          uncollect.expandStructFields);
     }
 
     throw new AssertionError("Need to implement logical converter for "

core/src/main/java/org/apache/calcite/rel/mutable/MutableRels.java

@@ -257,7 +257,7 @@ public static RelNode fromMutable(MutableRel node, RelBuilder relBuilder) {
       final MutableUncollect uncollect = (MutableUncollect) node;
       final RelNode child = fromMutable(uncollect.getInput(), relBuilder);
       return Uncollect.create(child.getTraitSet(), child, uncollect.withOrdinality,
-          Collections.emptyList());
+          Collections.emptyList(), uncollect.expandStructFields);
     }
     case WINDOW: {
       final MutableWindow window = (MutableWindow) node;
@@ -378,7 +378,8 @@ public static MutableRel toMutable(RelNode rel) {
     if (rel instanceof Uncollect) {
       final Uncollect uncollect = (Uncollect) rel;
       final MutableRel input = toMutable(uncollect.getInput());
-      return MutableUncollect.of(uncollect.getRowType(), input, uncollect.withOrdinality);
+      return MutableUncollect.of(uncollect.getRowType(), input,
+          uncollect.withOrdinality, uncollect.expandStructFields);
     }
     if (rel instanceof Window) {
       final Window window = (Window) rel;

core/src/main/java/org/apache/calcite/rel/mutable/MutableUncollect.java

@@ -25,15 +25,17 @@
 /** Mutable equivalent of {@link org.apache.calcite.rel.core.Uncollect}. */
 public class MutableUncollect extends MutableSingleRel {
   public final boolean withOrdinality;
+  public final boolean expandStructFields;
 
   private MutableUncollect(RelDataType rowType,
-      MutableRel input, boolean withOrdinality) {
+      MutableRel input, boolean withOrdinality, boolean expandStructFields) {
     super(MutableRelType.UNCOLLECT, rowType, input);
     this.withOrdinality = withOrdinality;
+    this.expandStructFields = expandStructFields;
   }
 
   /**
-   * Creates a MutableUncollect.
+   * Creates a MutableUncollect that expands struct elements.
    *
    * @param rowType         Row type
    * @param input           Input relational expression
@@ -42,26 +44,47 @@ private MutableUncollect(RelDataType rowType,
    */
   public static MutableUncollect of(RelDataType rowType,
       MutableRel input, boolean withOrdinality) {
-    return new MutableUncollect(rowType, input, withOrdinality);
+    return of(rowType, input, withOrdinality, true);
+  }
+
+  /**
+   * Creates a MutableUncollect.
+   *
+   * @param rowType            Row type
+   * @param input              Input relational expression
+   * @param withOrdinality     Whether the output contains an extra
+   *                           {@code ORDINALITY} column
+   * @param expandStructFields If true, a collection whose element type
+   *                           is a struct produces one output column per
+   *                           struct field; if false, a single column
+   *                           typed as the whole element
+   */
+  public static MutableUncollect of(RelDataType rowType,
+      MutableRel input, boolean withOrdinality, boolean expandStructFields) {
+    return new MutableUncollect(rowType, input, withOrdinality,
+        expandStructFields);
   }
 
   @Override public boolean equals(@Nullable Object obj) {
     return obj == this
         || obj instanceof MutableUncollect
         && withOrdinality == ((MutableUncollect) obj).withOrdinality
+        && expandStructFields == ((MutableUncollect) obj).expandStructFields
         && input.equals(((MutableUncollect) obj).input);
   }
 
   @Override public int hashCode() {
-    return Objects.hash(input, withOrdinality);
+    return Objects.hash(input, withOrdinality, expandStructFields);
   }
 
   @Override public StringBuilder digest(StringBuilder buf) {
-    return buf.append("Uncollect(withOrdinality: ")
-        .append(withOrdinality).append(")");
+    return buf.append("Uncollect(withOrdinality: ").append(withOrdinality)
+        .append(", expandStructFields: ").append(expandStructFields)
+        .append(")");
   }
 
   @Override public MutableRel clone() {
-    return MutableUncollect.of(rowType, input.clone(), withOrdinality);
+    return MutableUncollect.of(rowType, input.clone(), withOrdinality,
+        expandStructFields);
   }
 }

core/src/main/java/org/apache/calcite/runtime/CalciteResource.java

@@ -337,6 +337,9 @@ ExInst<CalciteException> invalidCompare(String a0, String a1, String a2,
   @BaseMessage("Cannot specify condition (NATURAL keyword, or ON or USING clause) following CROSS JOIN")
   ExInst<SqlValidatorException> crossJoinDisallowsCondition();
 
+  @BaseMessage("UNNEST is only supported with INNER, LEFT, CROSS, or COMMA join, not ''{0}''")
+  ExInst<SqlValidatorException> unnestInvalidJoinType(String a0);
+
   @BaseMessage("Cannot specify NATURAL keyword with ON or USING clause")
   ExInst<SqlValidatorException> naturalDisallowsOnOrUsing();