address comments

Author: lianhuiwang

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -733,6 +733,24 @@ class Analyzer(
     }
   }
 
+  protected[sql] def resolveFunction(func: UnresolvedFunction) = {
+    catalog.lookupFunction(func.name, func.children) match {
+      // DISTINCT is not meaningful for a Max or a Min.
+      case max: Max if func.isDistinct =>
+        AggregateExpression(max, Complete, isDistinct = false)
+      case min: Min if func.isDistinct =>
+        AggregateExpression(min, Complete, isDistinct = false)
+      // AggregateWindowFunctions are AggregateFunctions that can only be evaluated within
+      // the context of a Window clause. They do not need to be wrapped in an
+      // AggregateExpression.
+      case wf: AggregateWindowFunction => wf
+      // We get an aggregate function, we need to wrap it in an AggregateExpression.
+      case agg: AggregateFunction => AggregateExpression(agg, Complete, func.isDistinct)
+      // This function is not an aggregate function, just return the resolved one.
+      case other => other
+    }
+  }
+
  /**
   * In many dialects of SQL it is valid to use ordinal positions in order/sort by and group by
   * clauses. This rule is to convert ordinal positions to the corresponding expressions in the
@@ -916,21 +934,7 @@ class Analyzer(
             }
           case u @ UnresolvedFunction(funcId, children, isDistinct) =>
             withPosition(u) {
-              catalog.lookupFunction(funcId, children) match {
-                // DISTINCT is not meaningful for a Max or a Min.
-                case max: Max if isDistinct =>
-                  AggregateExpression(max, Complete, isDistinct = false)
-                case min: Min if isDistinct =>
-                  AggregateExpression(min, Complete, isDistinct = false)
-                // AggregateWindowFunctions are AggregateFunctions that can only be evaluated within
-                // the context of a Window clause. They do not need to be wrapped in an
-                // AggregateExpression.
-                case wf: AggregateWindowFunction => wf
-                // We get an aggregate function, we need to wrap it in an AggregateExpression.
-                case agg: AggregateFunction => AggregateExpression(agg, Complete, isDistinct)
-                // This function is not an aggregate function, just return the resolved one.
-                case other => other
-              }
+              resolveFunction(u)
             }
         }
     }

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala

@@ -25,7 +25,6 @@ import org.antlr.v4.runtime.tree.TerminalNode
 
 import org.apache.spark.sql.SaveMode
 import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier}
-import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.parser._
@@ -601,27 +600,11 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder {
    */
   override def visitCreateMacro(ctx: CreateMacroContext): LogicalPlan = withOrigin(ctx) {
     val arguments = Option(ctx.colTypeList).map(visitColTypeList(_))
-      .getOrElse(Seq.empty[StructField]).map { col =>
-      AttributeReference(col.name, col.dataType, col.nullable, col.metadata)() }
-    val colToIndex: Map[String, Int] = arguments.map(_.name).zipWithIndex.toMap
-    if (colToIndex.size != arguments.size) {
-      throw operationNotAllowed(
-        s"Cannot support duplicate colNames for CREATE TEMPORARY MACRO ", ctx)
-    }
-    val macroFunction = expression(ctx.expression).transformUp {
-      case u: UnresolvedAttribute =>
-        val index = colToIndex.get(u.name).getOrElse(
-          throw new ParseException(
-            s"Cannot find colName: [${u}] for CREATE TEMPORARY MACRO", ctx))
-        BoundReference(index, arguments(index).dataType, arguments(index).nullable)
-      case _: SubqueryExpression =>
-        throw operationNotAllowed(s"Cannot support Subquery for CREATE TEMPORARY MACRO", ctx)
-    }
-
+      .getOrElse(Seq.empty[StructField])
+    val e = expression(ctx.expression)
     CreateMacroCommand(
       ctx.macroName.getText,
-      arguments,
-      macroFunction)
+        MacroFunctionWrapper(arguments, e))
   }
 
   /**

sql/core/src/main/scala/org/apache/spark/sql/execution/command/macros.scala

@@ -18,7 +18,14 @@
 package org.apache.spark.sql.execution.command
 
 import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.types.StructField
+
+/**
+ * This class provides arguments and body expression of the macro function.
+ */
+case class MacroFunctionWrapper(columns: Seq[StructField], macroFunction: Expression)
 
 /**
  * The DDL command that creates a macro.
@@ -29,20 +36,54 @@ import org.apache.spark.sql.catalyst.expressions._
  */
 case class CreateMacroCommand(
     macroName: String,
-    columns: Seq[AttributeReference],
-    macroFunction: Expression)
+    funcWrapper: MacroFunctionWrapper)
   extends RunnableCommand {
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    val macroInfo = columns.mkString(",") + " -> " + macroFunction.toString
+    val columns = funcWrapper.columns.map { col =>
+      AttributeReference(col.name, col.dataType, col.nullable, col.metadata)() }
+    val colToIndex: Map[String, Int] = columns.map(_.name).zipWithIndex.toMap
+    if (colToIndex.size != columns.size) {
+      throw new AnalysisException(s"Cannot support duplicate colNames " +
+        s"for CREATE TEMPORARY MACRO $macroName, actual columns: ${columns.mkString(",")}")
+    }
+    val macroFunction = funcWrapper.macroFunction.transformDown {
+      case u: UnresolvedAttribute =>
+        val index = colToIndex.get(u.name).getOrElse(
+          throw new AnalysisException(s"Cannot find colName: ${u} " +
+            s"for CREATE TEMPORARY MACRO $macroName, actual columns: ${columns.mkString(",")}"))
+        BoundReference(index, columns(index).dataType, columns(index).nullable)
+      case u: UnresolvedFunction =>
+        sparkSession.sessionState.analyzer.resolveFunction(u)
+      case s: SubqueryExpression =>
+        throw new AnalysisException(s"Cannot support Subquery: ${s} " +
+          s"for CREATE TEMPORARY MACRO $macroName")
+      case u: UnresolvedGenerator =>
+        throw new AnalysisException(s"Cannot support Generator: ${u} " +
+          s"for CREATE TEMPORARY MACRO $macroName")
+    }
+    if (!macroFunction.resolved) {
+      if (macroFunction.checkInputDataTypes().isFailure) {
+        macroFunction.checkInputDataTypes() match {
+          case TypeCheckResult.TypeCheckFailure(message) =>
+            throw new AnalysisException(s"Cannot resolve '${macroFunction.sql}' " +
+              s"for CREATE TEMPORARY MACRO $macroName, due to data type mismatch: $message")
+        }
+      } else {
+        throw new AnalysisException(s"Cannot resolve '${macroFunction.sql}' " +
+          s"for CREATE TEMPORARY MACRO $macroName")
+      }
+    }
+    val macroInfo = columns.mkString(",") + " -> " + funcWrapper.macroFunction.toString
     val info = new ExpressionInfo(macroInfo, macroName)
     val builder = (children: Seq[Expression]) => {
       if (children.size != columns.size) {
         throw new AnalysisException(s"Actual number of columns: ${children.size} != " +
           s"expected number of columns: ${columns.size} for Macro $macroName")
       }
       macroFunction.transformUp {
+        // Skip to validate the input type because Analyzer will check it after ResolveFunctions.
         case b: BoundReference => children(b.ordinal)
       }
     }