[CALCITE-1581] UDTF like in hive

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

@@ -1728,15 +1728,38 @@ List<SqlNode> SelectList() :
 SqlNode SelectItem() :
 {
     SqlNode e;
-    final SqlIdentifier id;
+    SqlIdentifier columnAlias;
+    final List<SqlNode> columnAliases = new ArrayList();
+    final Span s = span();
 }
 {
     e = SelectExpression()
     [
-        [ <AS> ]
-        id = SimpleIdentifier() {
-            e = SqlStdOperatorTable.AS.createCall(span().end(e), e, id);
-        }
+      (
+       LOOKAHEAD(2) (
+          [ <AS> ]
+          columnAlias = SimpleIdentifier()
+          {
+            e = SqlStdOperatorTable.AS.createCall(s.end(e), e, columnAlias);
+          }
+        )
+      |
+         (
+          <AS> <LPAREN>
+          columnAlias = SimpleIdentifier() {
+             columnAliases.add(columnAlias);
+          }
+          ( <COMMA> columnAlias = SimpleIdentifier() { columnAliases.add(columnAlias);} )*
+            <RPAREN> {
+             if (!this.conformance.allowSelectTableFunction()) {
+               throw SqlUtil.newContextException(getPos(),
+                 RESOURCE.notAllowTableFunctionInSelect());
+             }
+             e = SqlStdOperatorTable.AS.createCall(s.end(e), e,
+               new SqlNodeList(columnAliases, s.end(e)));
+           }
+        )
+      )
     ]
     {
         return e;

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

@@ -900,6 +900,18 @@ ExInst<CalciteException> invalidTypesForComparison(String clazzName0, String op,
 
   @BaseMessage("Not a valid input for REGEXP_REPLACE: ''{0}''")
   ExInst<CalciteException> invalidInputForRegexpReplace(String value);
+
+  @BaseMessage("Table function is not allowed in select list in current SQL conformance level")
+  ExInst<SqlValidatorException> notAllowTableFunctionInSelect();
+
+  @BaseMessage("''{0}'' should be a table function")
+  ExInst<SqlValidatorException> exceptTableFunction(String name);
+
+  @BaseMessage("Only one table function is allowed in select list")
+  ExInst<SqlValidatorException> onlyOneTableFunctionAllowedInSelect();
+
+  @BaseMessage("Table function is not allowed in aggregate statement")
+  ExInst<SqlValidatorException> notAllowTableFunctionInAggregate();
 }
 
 // End CalciteResource.java

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

@@ -30,6 +30,7 @@
 import org.apache.calcite.sql.validate.SqlValidatorScope;
 import org.apache.calcite.util.Util;
 
+import java.util.ArrayList;
 import java.util.List;
 
 import static org.apache.calcite.util.Static.RESOURCE;
@@ -102,12 +103,26 @@ public void validateCall(
     // we don't want to validate the identifier.
     final List<SqlNode> operands = call.getOperandList();
     assert operands.size() == 2;
-    assert operands.get(1) instanceof SqlIdentifier;
     operands.get(0).validateExpr(validator, scope);
-    SqlIdentifier id = (SqlIdentifier) operands.get(1);
-    if (!id.isSimple()) {
-      throw validator.newValidationError(id,
-          RESOURCE.aliasMustBeSimpleIdentifier());
+
+    SqlNode asIdentifier = operands.get(1);
+    assert asIdentifier instanceof SqlIdentifier
+        || (asIdentifier instanceof SqlNodeList
+            && validator.getConformance().allowSelectTableFunction());
+
+    List<SqlNode> ids = new ArrayList<>();
+    if (asIdentifier instanceof SqlIdentifier) {
+      ids.add(operands.get(1));
+    } else {
+      ids.addAll(((SqlNodeList) operands.get(1)).getList());
+    }
+    for (int i = 0; i < ids.size(); i++) {
+      assert ids.get(i) instanceof SqlIdentifier;
+      SqlIdentifier id = (SqlIdentifier) ids.get(i);
+      if (!id.isSimple()) {
+        throw validator.newValidationError(id,
+              RESOURCE.aliasMustBeSimpleIdentifier());
+      }
     }
   }
 

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

@@ -1143,6 +1143,21 @@ private void visitChild(SqlNode node) {
       return check(type);
     }
   }
+
+  /**
+   * Whether the selectItem is a table function node in the select.
+   * eg. "select table_func(1) as (f0,f1)"
+   *
+   * @param selectItem select item
+   * @return true if this selectItem is a table function call
+   */
+  public static boolean isTableFunctionInSelect(SqlNode selectItem) {
+    if (selectItem.getKind() == SqlKind.AS) {
+      SqlBasicCall call = (SqlBasicCall) selectItem;
+      return call.getOperands()[1] instanceof SqlNodeList;
+    }
+    return false;
+  }
 }
 
 // End SqlUtil.java

core/src/main/java/org/apache/calcite/sql/validate/AliasNamespace.java

@@ -22,6 +22,7 @@
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlUtil;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.calcite.util.Util;
@@ -69,7 +70,12 @@ protected RelDataType validateImpl(RelDataType targetRowType) {
     final SqlValidatorNamespace childNs =
         validator.getNamespace(operands.get(0));
     final RelDataType rowType = childNs.getRowTypeSansSystemColumns();
-    final List<SqlNode> columnNames = Util.skip(operands, 2);
+    List<SqlNode> columnNames;
+    if (SqlUtil.isTableFunctionInSelect(call)) {
+      columnNames = ((SqlNodeList) operands.get(1)).getList();
+    } else {
+      columnNames = Util.skip(operands, 2);
+    }
     for (final SqlNode operand : columnNames) {
       String name = ((SqlIdentifier) operand).getSimple();
       if (nameList.contains(name)) {

core/src/main/java/org/apache/calcite/sql/validate/SqlAbstractConformance.java

@@ -106,6 +106,10 @@ public boolean allowExtendedTrim() {
   public boolean allowPluralTimeUnits() {
     return SqlConformanceEnum.DEFAULT.allowPluralTimeUnits();
   }
+
+  public boolean allowSelectTableFunction() {
+    return SqlConformanceEnum.DEFAULT.allowSelectTableFunction();
+  }
 }
 
 // End SqlAbstractConformance.java

core/src/main/java/org/apache/calcite/sql/validate/SqlConformance.java

@@ -61,6 +61,7 @@ public interface SqlConformance {
   @Deprecated // to be removed before 2.0
   SqlConformanceEnum PRAGMATIC_2003 = SqlConformanceEnum.PRAGMATIC_2003;
 
+  SqlConformanceEnum HIVE = SqlConformanceEnum.HIVE;
   /**
    * Whether this dialect supports features from a wide variety of
    * dialects. This is enabled for the Babel parser, disabled otherwise.
@@ -408,6 +409,13 @@ public interface SqlConformance {
    * false otherwise.
    */
   boolean allowPluralTimeUnits();
+
+  /**
+   * Whether SELECT can contain a table function.
+   * <p>For example, consider the query
+   * <blockquote><pre> SELECT SPLIT(col) AS (F0, F1) FROM A </pre> </blockquote>
+   */
+  boolean allowSelectTableFunction();
 }
 
 // End SqlConformance.java

core/src/main/java/org/apache/calcite/sql/validate/SqlConformanceEnum.java

@@ -69,7 +69,11 @@ public enum SqlConformanceEnum implements SqlConformance {
 
   /** Conformance value that instructs Calcite to use SQL semantics
    * consistent with Microsoft SQL Server version 2008. */
-  SQL_SERVER_2008;
+  SQL_SERVER_2008,
+
+  /** Conformance value that instructs Calcite to use SQL semantics
+   * consistent with Hive version. */
+  HIVE;
 
   public boolean isLiberal() {
     switch (this) {
@@ -314,6 +318,14 @@ public boolean allowExtendedTrim() {
     }
   }
 
+  public boolean allowSelectTableFunction() {
+    switch (this) {
+    case HIVE:
+      return true;
+    default:
+      return false;
+    }
+  }
 }
 
 // End SqlConformanceEnum.java

core/src/main/java/org/apache/calcite/sql/validate/SqlDelegatingConformance.java

@@ -73,6 +73,9 @@ protected SqlDelegatingConformance(SqlConformance delegate) {
     return delegate.allowNiladicParentheses();
   }
 
+  @Override public boolean allowSelectTableFunction() {
+    return delegate.allowSelectTableFunction();
+  }
 }
 
 // End SqlDelegatingConformance.java

core/src/main/java/org/apache/calcite/sql/validate/SqlValidator.java

@@ -23,6 +23,7 @@
 import org.apache.calcite.runtime.CalciteContextException;
 import org.apache.calcite.runtime.CalciteException;
 import org.apache.calcite.runtime.Resources;
+import org.apache.calcite.sql.SqlBasicCall;
 import org.apache.calcite.sql.SqlCall;
 import org.apache.calcite.sql.SqlDataTypeSpec;
 import org.apache.calcite.sql.SqlDelete;
@@ -749,6 +750,14 @@ CalciteException handleUnresolvedFunction(SqlCall call,
    */
   SqlValidatorScope getOverScope(SqlNode node);
 
+  /**
+   * Returns the table function SqlBasicCall in SqlSelect.
+   *
+   * @param select      The select node
+   * @return            The table function node  associate with the select node
+   */
+  SqlBasicCall getTableFunctionInSelect(SqlSelect select);
+
   /**
    * Validates that a query is capable of producing a return of given modality
    * (relational or streaming).

core/src/main/java/org/apache/calcite/sql/validate/SqlValidatorImpl.java

@@ -294,6 +294,9 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
   // Flag saying if we enable the implicit type coercion.
   private boolean enableTypeCoercion;
 
+  // Mapping the table function and the select node.
+  private final Map<SqlSelect, SqlBasicCall> selectTableFunctions =
+      new IdentityHashMap<>();
   //~ Constructors -----------------------------------------------------------
 
   /**
@@ -1067,6 +1070,10 @@ public SqlValidatorScope getOverScope(SqlNode node) {
     return scopes.get(node);
   }
 
+  public SqlBasicCall getTableFunctionInSelect(SqlSelect select) {
+    return selectTableFunctions.get(select);
+  }
+
   private SqlValidatorNamespace getNamespace(SqlNode node,
       SqlValidatorScope scope) {
     if (node instanceof SqlIdentifier && scope instanceof DelegatingScope) {
@@ -4137,6 +4144,14 @@ protected RelDataType validateSelectList(
             expandedSelectItems,
             aliases,
             fieldList);
+      } else if (SqlUtil.isTableFunctionInSelect(selectItem)) {
+        handleTableFunctionInSelect(
+            select,
+            (SqlBasicCall) selectItem,
+            expandedSelectItems,
+            aliases,
+            fieldList
+        );
       } else {
         expandSelectItem(
             selectItem,
@@ -4247,6 +4262,114 @@ private void handleScalarSubQuery(
     fieldList.add(Pair.of(alias, nodeType));
   }
 
+  /**
+   * Process table function found in select list.Checks that is
+   * actually a table function and validate the table function
+   * count in Select list.
+   *
+   * @param parentSelect        Base SqlSelect
+   * @param selectItem          Child select items from select list
+   * @param expandedSelectItems Select items after processing
+   * @param aliases             Built from user or system values
+   * @param fields              Built up entries for each select list entry
+   */
+  private void handleTableFunctionInSelect(
+      SqlSelect parentSelect,
+      SqlBasicCall selectItem,
+      List<SqlNode> expandedSelectItems,
+      Set<String> aliases,
+      List<Map.Entry<String, RelDataType>> fields) {
+    SqlBasicCall functionCall = (SqlBasicCall) selectItem.getOperands()[0];
+    SqlFunction function = (SqlFunction) functionCall.getOperator();
+    // Check whether there are more than one table function in select list.
+    for (SqlNode item : parentSelect.getSelectList()) {
+      if (SqlUtil.isTableFunctionInSelect(item)
+          && item != selectItem) {
+        throw newValidationError(parentSelect.getSelectList(),
+            RESOURCE.onlyOneTableFunctionAllowedInSelect());
+      }
+    }
+
+    // Change the function category to USER_DEFINED_TABLE_FUNCTION.
+    // It is because that in sql-select list, the SqlFunctionCategory is USER_DEFINED_FUNCTION
+    // for a SqlUnresolvedFunction,so we should do this change.
+    if (function instanceof SqlUnresolvedFunction) {
+      if (!function.getFunctionType().isTableFunction()) {
+        SqlFunction newFunction =
+            new SqlUnresolvedFunction(function.getNameAsId(),
+                function.getReturnTypeInference(),
+                function.getOperandTypeInference(),
+                function.getOperandTypeChecker(),
+                function.getParamTypes(),
+                SqlFunctionCategory.USER_DEFINED_TABLE_FUNCTION);
+        functionCall.setOperator(newFunction);
+        function = newFunction;
+      }
+    }
+    // Check functionCall whether is a table function
+    List<SqlOperator> overloads  = new ArrayList<>();
+    opTab.lookupOperatorOverloads(function.getNameAsId(),
+        function.getFunctionType(),
+        function.getSyntax(), overloads, catalogReader.nameMatcher());
+    if (overloads.size() == 0) {
+      throw newValidationError(functionCall,
+          RESOURCE.exceptTableFunction(function.getName()));
+    }
+    // Check the parent select whether is a aggregate statement
+    if (isAggregate(parentSelect)) {
+      throw newValidationError(functionCall,
+          RESOURCE.notAllowTableFunctionInAggregate());
+    }
+    SqlNodeList aliasNodes
+        = (SqlNodeList) selectItem.getOperands()[1];
+    List<String> aliasList = new ArrayList<>();
+    for (SqlNode aliasNode : aliasNodes) {
+      aliasList.add(deriveAlias(aliasNode, aliasList.size()));
+    }
+    aliases.addAll(aliasList);
+
+    String tableAlias = functionCall.getOperator().getName();
+    // Expand the table function alias
+    for (String alias : aliasList) {
+      SqlIdentifier id = new SqlIdentifier(Lists.newArrayList
+          (tableAlias, alias), SqlParserPos.ZERO);
+      expandedSelectItems.add(id);
+    }
+
+    // Register namespace for table function
+    SqlValidatorScope fromScope = getFromScope(parentSelect);
+    ProcedureNamespace tableNs = new ProcedureNamespace(this,
+        fromScope, functionCall, selectItem);
+    tableNs.validateImpl(unknownType);
+    registerNamespace(null, null,
+        tableNs, false);
+    AliasNamespace aliasNs = new AliasNamespace(this,
+        selectItem, parentSelect);
+    aliasNs.validateImpl(unknownType);
+    registerNamespace(getSelectScope(parentSelect),
+        tableAlias, aliasNs, false);
+
+    // Create a table scope for table function
+    TableScope tableScope = new TableScope(fromScope, parentSelect);
+    if (fromScope instanceof ListScope) {
+      for (ScopeChild child : ((ListScope) fromScope).children) {
+        tableScope.addChild(child.namespace, child.name, child.nullable);
+      }
+    }
+    scopes.put(functionCall, tableScope);
+    // Associate the select with the table function
+    selectTableFunctions.put(parentSelect, functionCall);
+
+    RelDataType type = aliasNs.getRowType();
+    setValidatedNodeType(selectItem, type);
+    for (int i = 0; i < aliasList.size(); i++) {
+      fields.add(
+          Pair.of(
+            aliasList.get(i), type.getFieldList()
+                .get(i).getType()));
+    }
+  }
+
   /**
    * Derives a row-type for INSERT and UPDATE operations.
    *