[CALCITE-7670] Uncollect should support LEFT JOIN UNNEST

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

@@ -51,16 +51,20 @@ public EnumerableUncollect(RelOptCluster cluster, RelTraitSet traitSet,
    * <p>Use {@link #create} unless you know what you're doing. */
   public EnumerableUncollect(RelOptCluster cluster, RelTraitSet traitSet,
       RelNode child, boolean withOrdinality) {
-    this(cluster, traitSet, child, withOrdinality, true);
+    this(cluster, traitSet, child, withOrdinality, true, false);
   }
 
   /** Creates an EnumerableUncollect.
    *
-   * <p>Use {@link #create} unless you know what you're doing. */
+   * <p>Use {@link #create} unless you know what you're doing.
+   *
+   * @param isOuter If true, an empty or NULL collection yields one row of
+   *                NULLs (LEFT JOIN); if false, it yields no rows (INNER) */
   public EnumerableUncollect(RelOptCluster cluster, RelTraitSet traitSet,
-      RelNode child, boolean withOrdinality, boolean expandStructFields) {
+      RelNode child, boolean withOrdinality, boolean expandStructFields,
+      boolean isOuter) {
     super(cluster, traitSet, child, withOrdinality, Collections.emptyList(),
-        expandStructFields);
+        expandStructFields, isOuter);
     assert getConvention() instanceof EnumerableConvention;
     assert getConvention() == child.getConvention();
   }
@@ -90,18 +94,20 @@ public static EnumerableUncollect create(RelTraitSet traitSet, RelNode input,
    * @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
+   * @param isOuter            If true, an empty or NULL collection yields one row of
+   *                           NULLs (LEFT JOIN); if false, it yields no rows (INNER)
    */
   public static EnumerableUncollect create(RelTraitSet traitSet, RelNode input,
-      boolean withOrdinality, boolean expandStructFields) {
+      boolean withOrdinality, boolean expandStructFields, boolean isOuter) {
     final RelOptCluster cluster = input.getCluster();
     return new EnumerableUncollect(cluster, traitSet, input, withOrdinality,
-        expandStructFields);
+        expandStructFields, isOuter);
   }
 
   @Override public EnumerableUncollect copy(RelTraitSet traitSet,
       RelNode newInput) {
     return new EnumerableUncollect(getCluster(), traitSet, newInput,
-        withOrdinality, expandStructFields);
+        withOrdinality, expandStructFields, isOuter);
   }
 
   @Override public Result implement(EnumerableRelImplementor implementor, Prefer pref) {
@@ -136,8 +142,11 @@ public static EnumerableUncollect create(RelTraitSet traitSet, RelNode input,
               && !withOrdinality) {
             // Solves CALCITE-4063: if we are processing a single field, which is a struct with a
             // single item inside, and no ordinality; the result must be a scalar, hence use a
-            // special lambda that does not return lists, but the (single) items within those lists
-            lambdaForStructWithSingleItem = Expressions.call(BuiltInMethod.FLAT_LIST.method);
+            // special lambda that does not return lists, but the (single) items within those
+            // lists. The outer variant returns one NULL scalar for an empty or NULL collection.
+            lambdaForStructWithSingleItem =
+                Expressions.call(isOuter ? BuiltInMethod.FLAT_LIST_OUTER.method
+                    : BuiltInMethod.FLAT_LIST.method);
           } else {
             fieldCounts.add(elementType.getFieldCount());
             inputTypes.add(FlatProductInputType.LIST);
@@ -161,7 +170,8 @@ public static EnumerableUncollect create(RelTraitSet traitSet, RelNode input,
             Expressions.constant(Ints.toArray(fieldCounts)),
             Expressions.constant(withOrdinality),
             Expressions.constant(
-                inputTypes.toArray(new FlatProductInputType[0])));
+                inputTypes.toArray(new FlatProductInputType[0])),
+            Expressions.constant(isOuter));
     builder.add(
         Expressions.return_(null,
             Expressions.call(child_,

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

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

core/src/main/java/org/apache/calcite/interpreter/UncollectNode.java

@@ -33,15 +33,26 @@
   }
 
   @Override public void run() throws InterruptedException {
+    // Under isOuter an empty or NULL collection still produces one row, with
+    // every column NULL.
+    final int width = rel.getRowType().getFieldCount();
     Row row = null;
     while ((row = source.receive()) != null) {
       for (Object value : row.getValues()) {
         if (value == null) {
+          if (rel.isOuter) {
+            sink.send(Row.of(new Object[width]));
+            continue;
+          }
           throw new NullPointerException("NULL value for unnest.");
         }
         int i = 1;
         if (value instanceof List) {
           List list = (List) value;
+          if (list.isEmpty() && rel.isOuter) {
+            sink.send(Row.of(new Object[width]));
+            continue;
+          }
           for (Object o : list) {
             if (rel.withOrdinality) {
               sink.send(Row.of(o, i++));
@@ -51,6 +62,10 @@
           }
         } else if (value instanceof Map) {
           Map map = (Map) value;
+          if (map.isEmpty() && rel.isOuter) {
+            sink.send(Row.of(new Object[width]));
+            continue;
+          }
           for (Object key : map.keySet()) {
             if (rel.withOrdinality) {
               sink.send(Row.of(key, map.get(key), i++));

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

@@ -56,10 +56,20 @@
  * 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.
+ *
+ * <p>{@code isOuter} controls what happens to an empty or {@code NULL}
+ * collection: if {@code true} (LEFT JOIN semantics) one row is emitted with
+ * every element column set to {@code NULL}; if {@code false} (INNER
+ * semantics) no row is emitted. Every element column is therefore nullable
+ * when {@code isOuter}.
  */
 public class Uncollect extends SingleRel {
   public final boolean withOrdinality;
 
+  /** If true, an empty or NULL collection yields a single row whose element
+   * columns are all NULL, rather than no rows at all. */
+  public final boolean isOuter;
+
   /** 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. */
@@ -90,7 +100,8 @@ public Uncollect(RelOptCluster cluster, RelTraitSet traitSet, RelNode input,
     // 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());
+    this(cluster, traitSet, input, withOrdinality, itemAliases, itemAliases.isEmpty(),
+        false);
   }
 
   /** Creates an Uncollect.
@@ -101,14 +112,18 @@ public Uncollect(RelOptCluster cluster, RelTraitSet traitSet, RelNode input,
    * @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
+   * @param isOuter            If true, an empty or NULL collection yields one row of
+   *                           NULLs (LEFT JOIN); if false, it yields no rows (INNER)
    */
   @SuppressWarnings("method.invocation.invalid")
   public Uncollect(RelOptCluster cluster, RelTraitSet traitSet, RelNode input,
-      boolean withOrdinality, List<String> itemAliases, boolean expandStructFields) {
+      boolean withOrdinality, List<String> itemAliases, boolean expandStructFields,
+      boolean isOuter) {
     super(cluster, traitSet, input);
     this.withOrdinality = withOrdinality;
     this.itemAliases = ImmutableList.copyOf(itemAliases);
     this.expandStructFields = expandStructFields;
+    this.isOuter = isOuter;
     requireNonNull(deriveRowType(), "invalid child rowType");
   }
 
@@ -118,7 +133,8 @@ public Uncollect(RelOptCluster cluster, RelTraitSet traitSet, RelNode input,
   public Uncollect(RelInput input) {
     this(input.getCluster(), input.getTraitSet(), input.getInput(),
         input.getBoolean("withOrdinality", false), Collections.emptyList(),
-        input.getBoolean("expandStructFields", true));
+        input.getBoolean("expandStructFields", true),
+        input.getBoolean("isOuter", false));
   }
 
   /**
@@ -151,16 +167,19 @@ public static Uncollect create(
    * @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
+   * @param isOuter            If true, an empty or NULL collection yields one row of
+   *                           NULLs (LEFT JOIN); if false, it yields no rows (INNER)
    */
   public static Uncollect create(
       RelTraitSet traitSet,
       RelNode input,
       boolean withOrdinality,
       List<String> itemAliases,
-      boolean expandStructFields) {
+      boolean expandStructFields,
+      boolean isOuter) {
     final RelOptCluster cluster = input.getCluster();
     return new Uncollect(cluster, traitSet, input, withOrdinality, itemAliases,
-        expandStructFields);
+        expandStructFields, isOuter);
   }
 
   //~ Methods ----------------------------------------------------------------
@@ -172,7 +191,8 @@ public static Uncollect create(
   @Override public RelWriter explainTerms(RelWriter pw) {
     return super.explainTerms(pw)
         .itemIf("withOrdinality", withOrdinality, withOrdinality)
-        .itemIf("expandStructFields", expandStructFields, !expandStructFields);
+        .itemIf("expandStructFields", expandStructFields, !expandStructFields)
+        .itemIf("isOuter", isOuter, isOuter);
   }
 
   @Override public final RelNode copy(RelTraitSet traitSet,
@@ -183,7 +203,7 @@ public static Uncollect create(
   public RelNode copy(RelTraitSet traitSet, RelNode input) {
     assert traitSet.containsIfApplicable(Convention.NONE);
     return new Uncollect(getCluster(), traitSet, input, withOrdinality, itemAliases,
-        expandStructFields);
+        expandStructFields, isOuter);
   }
 
   /**
@@ -287,7 +307,18 @@ public static RelDataType deriveUncollectRowType(RelNode rel,
       builder.add(SqlUnnestOperator.ORDINALITY_COLUMN_NAME,
           SqlTypeName.INTEGER);
     }
-    return builder.build();
+    final RelDataType rowType = builder.build();
+    if (!isOuter) {
+      return rowType;
+    }
+    // Under isOuter an empty or NULL collection yields a row of NULLs, so
+    // every output column is nullable, including the ordinality column.
+    final RelDataTypeFactory.Builder outerBuilder = typeFactory.builder();
+    for (RelDataTypeField field : rowType.getFieldList()) {
+      outerBuilder.add(field.getName(),
+          typeFactory.createTypeWithNullability(field.getType(), true));
+    }
+    return outerBuilder.build();
   }
 
   /** Gets the aliases for the unnest items. */

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

@@ -190,7 +190,7 @@ public ToLogicalConverter(RelBuilder relBuilder) {
       final RelNode input = visit(uncollect.getInput());
       return Uncollect.create(input.getTraitSet(), input,
           uncollect.withOrdinality, uncollect.getItemAliases(),
-          uncollect.expandStructFields);
+          uncollect.expandStructFields, uncollect.isOuter);
     }
 
     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(), uncollect.expandStructFields);
+          Collections.emptyList(), uncollect.expandStructFields, uncollect.isOuter);
     }
     case WINDOW: {
       final MutableWindow window = (MutableWindow) node;
@@ -379,7 +379,7 @@ public static MutableRel toMutable(RelNode rel) {
       final Uncollect uncollect = (Uncollect) rel;
       final MutableRel input = toMutable(uncollect.getInput());
       return MutableUncollect.of(uncollect.getRowType(), input,
-          uncollect.withOrdinality, uncollect.expandStructFields);
+          uncollect.withOrdinality, uncollect.expandStructFields, uncollect.isOuter);
     }
     if (rel instanceof Window) {
       final Window window = (Window) rel;

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

@@ -26,12 +26,15 @@
 public class MutableUncollect extends MutableSingleRel {
   public final boolean withOrdinality;
   public final boolean expandStructFields;
+  public final boolean isOuter;
 
   private MutableUncollect(RelDataType rowType,
-      MutableRel input, boolean withOrdinality, boolean expandStructFields) {
+      MutableRel input, boolean withOrdinality, boolean expandStructFields,
+      boolean isOuter) {
     super(MutableRelType.UNCOLLECT, rowType, input);
     this.withOrdinality = withOrdinality;
     this.expandStructFields = expandStructFields;
+    this.isOuter = isOuter;
   }
 
   /**
@@ -44,7 +47,7 @@
    */
   public static MutableUncollect of(RelDataType rowType,
       MutableRel input, boolean withOrdinality) {
-    return of(rowType, input, withOrdinality, true);
+    return of(rowType, input, withOrdinality, true, false);
   }
 
   /**
@@ -59,32 +62,49 @@
    *                           struct field; if false, a single column
    *                           typed as the whole element
    */
+  /**
+   * 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
+   * @param isOuter            If true, an empty or NULL collection yields one
+   *                           row of NULLs; if false, it yields no rows
+   */
   public static MutableUncollect of(RelDataType rowType,
-      MutableRel input, boolean withOrdinality, boolean expandStructFields) {
+      MutableRel input, boolean withOrdinality, boolean expandStructFields,
+      boolean isOuter) {
     return new MutableUncollect(rowType, input, withOrdinality,
-        expandStructFields);
+        expandStructFields, isOuter);
   }
 
   @Override public boolean equals(@Nullable Object obj) {
     return obj == this
         || obj instanceof MutableUncollect
         && withOrdinality == ((MutableUncollect) obj).withOrdinality
         && expandStructFields == ((MutableUncollect) obj).expandStructFields
+        && isOuter == ((MutableUncollect) obj).isOuter
         && input.equals(((MutableUncollect) obj).input);
   }
 
   @Override public int hashCode() {
-    return Objects.hash(input, withOrdinality, expandStructFields);
+    return Objects.hash(input, withOrdinality, expandStructFields, isOuter);
   }
 
   @Override public StringBuilder digest(StringBuilder buf) {
     return buf.append("Uncollect(withOrdinality: ").append(withOrdinality)
         .append(", expandStructFields: ").append(expandStructFields)
+        .append(", isOuter: ").append(isOuter)
         .append(")");
   }
 
   @Override public MutableRel clone() {
     return MutableUncollect.of(rowType, input.clone(), withOrdinality,
-        expandStructFields);
+        expandStructFields, isOuter);
   }
 }

core/src/main/java/org/apache/calcite/rel/rules/CoreRules.java

@@ -995,6 +995,12 @@ private CoreRules() {}
   public static final AggregateRemoveLiteralAggRule AGGREGATE_REMOVE_LITERAL_AGG =
       AggregateRemoveLiteralAggRule.Config.DEFAULT.toRule();
 
+  /** Rule that moves the outer join semantics of a {@link Correlate} over an
+   * {@link Uncollect} onto the {@code Uncollect}, leaving an inner
+   * {@code Correlate} that {@link #UNNEST_DECORRELATE} may then remove. */
+  public static final CorrelateUncollectOuterRule CORRELATE_UNCOLLECT_OUTER =
+      CorrelateUncollectOuterRule.Config.DEFAULT.toRule();
+
   /** Rule that converts a {@link Correlate} after an {@link Uncollect} into a simple
    * Uncollect, if possible. */
   public static final RelOptRule UNNEST_DECORRELATE =