[CALCITE-7814] Support the SQL-standard CYCLE clause in recursive CTEs

core/src/main/codegen/templates/Parser.jj

@@ -106,6 +106,7 @@ import org.apache.calcite.sql.SqlUtil;
 import org.apache.calcite.sql.SqlWindow;
 import org.apache.calcite.sql.SqlWith;
 import org.apache.calcite.sql.SqlWithItem;
+import org.apache.calcite.sql.SqlCycleClause;
 import org.apache.calcite.sql.fun.SqlCase;
 import org.apache.calcite.sql.fun.SqlInternalOperators;
 import org.apache.calcite.sql.fun.SqlLibraryOperators;
@@ -3929,13 +3930,44 @@ void AddWithItem(List<SqlWithItem> list, SqlLiteral recursive) :
     final SqlIdentifier id;
     final SqlNodeList columnList;
     final SqlNode definition;
+    SqlCycleClause cycle = null;
 }
 {
     id = SimpleIdentifier()
     ( columnList = ParenthesizedSimpleIdentifierList() | { columnList = null; } )
     <AS>
     definition = ParenthesizedExpression(ExprContext.ACCEPT_QUERY)
-    { list.add(new SqlWithItem(id.getParserPosition(), id, columnList, definition, recursive)); }
+    [ cycle = CycleClause() ]
+    { list.add(new SqlWithItem(id.getParserPosition(), id, columnList, definition, recursive, cycle)); }
+}
+
+SqlCycleClause CycleClause() :
+{
+    final Span s;
+    final List<SqlNode> columns = new ArrayList<SqlNode>();
+    SqlIdentifier column;
+    final SqlIdentifier markColumn;
+    final SqlNode markValue;
+    final SqlNode defaultValue;
+    final SqlIdentifier pathColumn;
+}
+{
+    <CYCLE> { s = span(); }
+    column = SimpleIdentifier() { columns.add(column); }
+    ( <COMMA> column = SimpleIdentifier() { columns.add(column); } )*
+    <SET> markColumn = SimpleIdentifier()
+    (
+        <TO> markValue = Expression(ExprContext.ACCEPT_NON_QUERY)
+        <DEFAULT_> defaultValue = Expression(ExprContext.ACCEPT_NON_QUERY)
+    |
+        {
+            markValue = SqlLiteral.createBoolean(true, markColumn.getParserPosition());
+            defaultValue = SqlLiteral.createBoolean(false, markColumn.getParserPosition());
+        }
+    )
+    <USING> pathColumn = SimpleIdentifier()
+    { return new SqlCycleClause(s.end(this), new SqlNodeList(columns, s.pos()),
+        markColumn, markValue, defaultValue, pathColumn); }
 }
 
 /**

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

@@ -587,6 +587,11 @@ private RelDataType toStruct(RelDataType type) {
   @Override public @Nullable Expression comparer() {
     final Expression comparer = format.comparer();
     if (comparer != null) {
+      // ARRAY compares nested arrays by content, but a List or Map compares
+      // the Object[] values it contains by reference.
+      if (format == JavaRowFormat.ARRAY && containsStructInCollection(rowType)) {
+        return Expressions.call(BuiltInMethod.DEEP_COMPARER.method);
+      }
       return comparer;
     }
     if (anyFieldContainsStruct(rowType)) {
@@ -599,6 +604,17 @@ private RelDataType toStruct(RelDataType type) {
     return null;
   }
 
+  /** Returns whether a struct occurs inside a collection or map, including
+   * collections and maps nested in struct fields. Plain nested structs do not
+   * require overriding the ARRAY format's comparer. */
+  private static boolean containsStructInCollection(RelDataType type) {
+    if (type.isStruct()) {
+      return type.getFieldList().stream()
+          .anyMatch(f -> containsStructInCollection(f.getType()));
+    }
+    return containsStruct(type);
+  }
+
   /** Returns whether any field of {@code rowType} contains a struct value:
    * the field is itself a struct, or a collection or map whose elements
    * contain one. */

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

@@ -130,6 +130,8 @@
 import static org.apache.calcite.linq4j.tree.ExpressionType.Subtract;
 import static org.apache.calcite.linq4j.tree.ExpressionType.SubtractChecked;
 import static org.apache.calcite.linq4j.tree.ExpressionType.UnaryPlus;
+import static org.apache.calcite.sql.fun.SqlInternalOperators.CYCLE_PATH_APPEND;
+import static org.apache.calcite.sql.fun.SqlInternalOperators.CYCLE_PATH_CONTAINS;
 import static org.apache.calcite.sql.fun.SqlInternalOperators.LITERAL_AGG;
 import static org.apache.calcite.sql.fun.SqlInternalOperators.THROW_UNLESS;
 import static org.apache.calcite.sql.fun.SqlLibraryOperators.ACOSD;
@@ -1108,6 +1110,8 @@ void populate2() {
       defineMethod(STRUCT_ACCESS, BuiltInMethod.STRUCT_ACCESS.method, NullPolicy.SEMI_STRICT);
       defineMethod(MEMBER_OF, BuiltInMethod.MEMBER_OF.method, NullPolicy.NONE);
       defineMethod(ARRAY_APPEND, BuiltInMethod.ARRAY_APPEND.method, NullPolicy.ARG0);
+      defineMethod(CYCLE_PATH_APPEND, BuiltInMethod.ARRAY_APPEND.method, NullPolicy.ARG0);
+      defineMethod(CYCLE_PATH_CONTAINS, BuiltInMethod.CYCLE_PATH_CONTAINS.method, NullPolicy.NONE);
       defineMethod(ARRAY_COMPACT, BuiltInMethod.ARRAY_COMPACT.method, NullPolicy.STRICT);
       defineMethod(ARRAY_CONTAINS, BuiltInMethod.LIST_CONTAINS.method, NullPolicy.SEMI_STRICT);
       defineMethod(ARRAY_DISTINCT, BuiltInMethod.ARRAY_DISTINCT.method, NullPolicy.STRICT);

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

@@ -935,6 +935,24 @@ ExInst<CalciteException> illegalArgumentForTableFunctionCall(String a0,
   @BaseMessage("A recursive query only supports binary UNION [ALL] operator")
   ExInst<SqlValidatorException> recursiveWithMustHaveTwoChildUnionSetOp();
 
+  @BaseMessage("CYCLE requires WITH RECURSIVE")
+  ExInst<SqlValidatorException> cycleRequiresRecursive();
+
+  @BaseMessage("CYCLE requires a UNION [ALL] with a recursive SELECT containing exactly one direct reference to its WITH item")
+  ExInst<SqlValidatorException> cycleRequiresRecursiveSelect();
+
+  @BaseMessage("CYCLE is not supported with aggregation, DISTINCT, or window functions in the recursive SELECT")
+  ExInst<SqlValidatorException> cycleRecursiveSelectNotSupported();
+
+  @BaseMessage("CYCLE column ''{0}'' is not a column of WITH item ''{1}''")
+  ExInst<SqlValidatorException> cycleColumnNotFound(String column, String withItem);
+
+  @BaseMessage("CYCLE generated column ''{0}'' conflicts with another column")
+  ExInst<SqlValidatorException> cycleColumnConflict(String column);
+
+  @BaseMessage("CYCLE mark and default must be distinct, non-null Boolean or compatible single-character string literals")
+  ExInst<SqlValidatorException> cycleInvalidMarkValues();
+
   @BaseMessage("First column of ORDER BY must be of type TIMESTAMP")
   ExInst<SqlValidatorException> firstColumnOfOrderByMustBeTimestamp();
 

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

@@ -7145,6 +7145,18 @@ public static List compact(List list) {
     return result;
   }
 
+  /** Tests whether a recursive CTE path contains a key tuple.
+   * Implements the condition of {@code CASE WHEN key IN (path) THEN ...}:
+   * only TRUE comparisons close a cycle; UNKNOWN does not. */
+  public static boolean cyclePathContains(List<?> path, Object key) {
+    for (Object previous : path) {
+      if (Boolean.TRUE.equals(rowEq(previous, key))) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   /** Support the ARRAY_APPEND function. */
   public static List arrayAppend(List list, Object element) {
     final List result = new ArrayList(list.size() + 1);

core/src/main/java/org/apache/calcite/sql/SqlCycleClause.java

@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.calcite.sql;
+
+import org.apache.calcite.sql.parser.SqlParserPos;
+
+import com.google.common.collect.ImmutableList;
+
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/** SQL-standard cycle detection clause of a recursive common table expression.
+ *
+ * <p>Syntax: {@code CYCLE columns SET mark [TO value DEFAULT default] USING path}.
+ * Omitted mark values are represented by TRUE and FALSE literals.
+ * The mark and path columns are generated columns, not members of the WITH
+ * item's explicit column list. */
+public class SqlCycleClause extends SqlCall {
+  private static final SqlSpecialOperator OPERATOR =
+      new SqlSpecialOperator("CYCLE", SqlKind.CYCLE) {
+        @Override public SqlCall createCall(@Nullable SqlLiteral qualifier,
+            SqlParserPos pos, @Nullable SqlNode... operands) {
+          return new SqlCycleClause(pos,
+              (SqlNodeList) requireNonNull(operands[0]),
+              (SqlIdentifier) requireNonNull(operands[1]),
+              requireNonNull(operands[2]), requireNonNull(operands[3]),
+              (SqlIdentifier) requireNonNull(operands[4]));
+        }
+      };
+
+  public SqlNodeList columns;
+  public SqlIdentifier markColumn;
+  public SqlNode markValue;
+  public SqlNode defaultValue;
+  public SqlIdentifier pathColumn;
+
+  public SqlCycleClause(SqlParserPos pos, SqlNodeList columns,
+      SqlIdentifier markColumn, SqlNode markValue, SqlNode defaultValue,
+      SqlIdentifier pathColumn) {
+    super(pos);
+    this.columns = columns;
+    this.markColumn = markColumn;
+    this.markValue = markValue;
+    this.defaultValue = defaultValue;
+    this.pathColumn = pathColumn;
+  }
+
+  @Override public SqlOperator getOperator() {
+    return OPERATOR;
+  }
+
+  @Override public List<SqlNode> getOperandList() {
+    return ImmutableList.of(columns, markColumn, markValue, defaultValue, pathColumn);
+  }
+
+  @Override public void setOperand(int i, @Nullable SqlNode operand) {
+    requireNonNull(operand, "operand");
+    switch (i) {
+    case 0:
+      columns = (SqlNodeList) operand;
+      break;
+    case 1:
+      markColumn = (SqlIdentifier) operand;
+      break;
+    case 2:
+      markValue = operand;
+      break;
+    case 3:
+      defaultValue = operand;
+      break;
+    case 4:
+      pathColumn = (SqlIdentifier) operand;
+      break;
+    default:
+      throw new AssertionError(i);
+    }
+  }
+
+  @Override public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
+    writer.keyword("CYCLE");
+    final SqlWriter.Frame frame = writer.startList("", "");
+    for (SqlNode column : columns) {
+      writer.sep(",");
+      column.unparse(writer, 0, 0);
+    }
+    writer.endList(frame);
+    writer.keyword("SET");
+    markColumn.unparse(writer, 0, 0);
+    writer.keyword("TO");
+    markValue.unparse(writer, 0, 0);
+    writer.keyword("DEFAULT");
+    defaultValue.unparse(writer, 0, 0);
+    writer.keyword("USING");
+    pathColumn.unparse(writer, 0, 0);
+  }
+}

core/src/main/java/org/apache/calcite/sql/SqlKind.java

@@ -204,6 +204,9 @@ public enum SqlKind {
   /** Item in WITH clause. */
   WITH_ITEM,
 
+  /** CYCLE clause of a recursive WITH item. */
+  CYCLE,
+
   /** Represents a recursive CTE as a table ref. */
   WITH_ITEM_TABLE_REF,
 
@@ -1492,7 +1495,7 @@ public enum SqlKind {
                   LITERAL_CHAIN, JDBC_FN, PRECEDING, FOLLOWING, ORDER_BY,
                   ORDER_BY_ALL,
                   NULLS_FIRST, NULLS_LAST, COLLECTION_TABLE, TABLESAMPLE,
-                  VALUES, WITH, WITH_ITEM, ITEM, SKIP_TO_FIRST, SKIP_TO_LAST,
+                  VALUES, WITH, WITH_ITEM, CYCLE, ITEM, SKIP_TO_FIRST, SKIP_TO_LAST,
                   JSON_VALUE_EXPRESSION, UNNEST),
               SET_QUERY, AGGREGATE, DML, DDL));
 

core/src/main/java/org/apache/calcite/sql/SqlWithItem.java

@@ -25,13 +25,14 @@
 
 /**
  * An item in a WITH clause of a query.
- * It has a name, an optional column list, and a query.
+ * It has a name, an optional column list, a query, and an optional CYCLE clause.
  */
 public class SqlWithItem extends SqlCall {
   public SqlIdentifier name;
   public @Nullable SqlNodeList columnList; // may be null
   public SqlLiteral recursive;
   public SqlNode query;
+  public @Nullable SqlCycleClause cycleClause;
 
   @Deprecated // to be removed before 2.0
   public SqlWithItem(SqlParserPos pos, SqlIdentifier name,
@@ -43,11 +44,18 @@
   public SqlWithItem(SqlParserPos pos, SqlIdentifier name,
       @Nullable SqlNodeList columnList, SqlNode query,
       SqlLiteral recursive) {
+    this(pos, name, columnList, query, recursive, null);
+  }
+
+  public SqlWithItem(SqlParserPos pos, SqlIdentifier name,
+      @Nullable SqlNodeList columnList, SqlNode query,
+      SqlLiteral recursive, @Nullable SqlCycleClause cycleClause) {
     super(pos);
     this.name = name;
     this.columnList = columnList;
     this.recursive = recursive;
     this.query = query;
+    this.cycleClause = cycleClause;
   }
 
   //~ Methods ----------------------------------------------------------------
@@ -58,7 +66,7 @@
 
   @SuppressWarnings("nullness")
   @Override public List<SqlNode> getOperandList() {
-    return ImmutableNullableList.of(name, columnList, query, recursive);
+    return ImmutableNullableList.of(name, columnList, query, recursive, cycleClause);
   }
 
   @SuppressWarnings("assignment.type.incompatible")
@@ -76,6 +84,9 @@
     case 3:
       recursive = (SqlLiteral) operand;
       break;
+    case 4:
+      cycleClause = (SqlCycleClause) operand;
+      break;
     default:
       throw new AssertionError(i);
     }
@@ -111,15 +122,19 @@
       }
       writer.keyword("AS");
       withItem.query.unparse(writer, MDX_PRECEDENCE, MDX_PRECEDENCE);
+      if (withItem.cycleClause != null) {
+        withItem.cycleClause.unparse(writer, 0, 0);
+      }
     }
 
     @SuppressWarnings("argument.type.incompatible")
     @Override public SqlCall createCall(@Nullable SqlLiteral functionQualifier,
         SqlParserPos pos, @Nullable SqlNode... operands) {
       assert functionQualifier == null;
-      assert operands.length == 4;
+      assert operands.length == 4 || operands.length == 5;
       return new SqlWithItem(pos, (SqlIdentifier) operands[0],
-          (SqlNodeList) operands[1], operands[2], (SqlLiteral) operands[3]);
+          (SqlNodeList) operands[1], operands[2], (SqlLiteral) operands[3],
+          operands.length == 5 ? (SqlCycleClause) operands[4] : null);
     }
   }
 }

core/src/main/java/org/apache/calcite/sql/fun/SqlInternalOperators.java

@@ -49,6 +49,16 @@ public abstract class SqlInternalOperators {
   private SqlInternalOperators() {
   }
 
+  /** Appends a key tuple to the path of a recursive CTE. */
+  public static final SqlInternalOperator CYCLE_PATH_APPEND =
+      new SqlInternalOperator("$CYCLE_PATH_APPEND", SqlKind.OTHER_FUNCTION,
+          2, true, ReturnTypes.ARG0, null, OperandTypes.ANY_ANY);
+
+  /** Tests whether a key tuple occurs in a recursive CTE's path. */
+  public static final SqlInternalOperator CYCLE_PATH_CONTAINS =
+      new SqlInternalOperator("$CYCLE_PATH_CONTAINS", SqlKind.OTHER_FUNCTION,
+          2, true, ReturnTypes.BOOLEAN_NOT_NULL, null, OperandTypes.ANY_ANY);
+
   /** Similar to {@link SqlStdOperatorTable#ROW}, but does not print "ROW".
    *
    * <p>For arguments [1, TRUE], ROW would print "{@code ROW (1, TRUE)}",