address some comments.

Author: lianhuiwang

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/ExpressionInfo.java

@@ -25,6 +25,7 @@ public class ExpressionInfo {
     public enum FunctionType {
         BUILTIN, PERSISTENT, TEMPORARY;
     }
+
     private String className;
     private String usage;
     private String name;
@@ -65,10 +66,6 @@ public ExpressionInfo(String className, String db, String name, String usage, St
         this.functionType = functionType;
     }
 
-    public ExpressionInfo(String className, String db, String name, String usage, String extended) {
-        this(className, db, name, usage, extended, FunctionType.TEMPORARY);
-    }
-
     public ExpressionInfo(String className, String name) {
         this(className, null, name, null, null, FunctionType.TEMPORARY);
     }

sql/catalyst/src/main/scala/org/apache/spark/sql/AnalysisException.scala

@@ -55,3 +55,12 @@ class AnalysisException protected[sql] (
     s"$message;$lineAnnotation$positionAnnotation"
   }
 }
+
+object AnalysisException {
+  /**
+   * Create a no such temporary macro exception.
+   */
+  def noSuchTempMacroException(func: String): AnalysisException = {
+    new AnalysisException(s"Temporary macro '$func' not found")
+  }
+}

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

@@ -144,7 +144,7 @@ class SystemFunctionRegistry(builtin: SimpleFunctionRegistry) extends SimpleFunc
   }
 
   override def listFunction(): Seq[String] = synchronized {
-    (functionBuilders.iterator.map(_._1).toList ++ builtin.listFunction()).sorted
+    (functionBuilders.iterator.map(_._1).toList ++ builtin.listFunction()).distinct.sorted
   }
 
   override def lookupFunction(name: String): Option[ExpressionInfo] = synchronized {
@@ -160,6 +160,7 @@ class SystemFunctionRegistry(builtin: SimpleFunctionRegistry) extends SimpleFunc
   }
 
   override def clear(): Unit = synchronized {
+    builtin.clear()
     functionBuilders.clear()
   }
 

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

@@ -52,6 +52,3 @@ class NoSuchPartitionsException(db: String, table: String, specs: Seq[TableParti
 
 class NoSuchTempFunctionException(func: String)
   extends AnalysisException(s"Temporary function '$func' not found")
-
-class NoSuchTempMacroException(func: String)
-  extends AnalysisException(s"Temporary macro '$func' not found")

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

@@ -1102,7 +1102,7 @@ class SessionCatalog(
   /** Drop a temporary macro. */
   def dropTempMacro(name: String, ignoreIfNotExists: Boolean): Unit = {
     if (!functionRegistry.dropFunction(name) && !ignoreIfNotExists) {
-      throw new NoSuchTempMacroException(name)
+      throw AnalysisException.noSuchTempMacroException(name)
     }
   }
 
@@ -1265,8 +1265,8 @@ class SessionCatalog(
       if (func.database.isDefined) {
         dropFunction(func, ignoreIfNotExists = false)
       } else {
-        val functionType = functionRegistry.lookupFunction(func.funcName).map(_.getFunctionType)
-          .getOrElse(FunctionType.TEMPORARY)
+        val functionType = functionRegistry.lookupFunction(func.funcName)
+          .map(_.getFunctionType).getOrElse(FunctionType.TEMPORARY)
         if (!functionType.equals(FunctionType.BUILTIN)) {
           dropTempFunction(func.funcName, ignoreIfNotExists = false)
         }

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

@@ -724,12 +724,11 @@ class SparkSqlAstBuilder(conf: SQLConf) extends AstBuilder(conf) {
    * }}}
    */
   override def visitCreateMacro(ctx: CreateMacroContext): LogicalPlan = withOrigin(ctx) {
-    val arguments = Option(ctx.colTypeList).map(visitColTypeList(_))
-      .getOrElse(Seq.empty[StructField])
+    val columns = createSchema(ctx.colTypeList)
     val e = expression(ctx.expression)
     CreateMacroCommand(
       ctx.macroName.getText,
-      MacroFunctionWrapper(arguments, e))
+      MacroFunctionWrapper(columns, e))
   }
 
   /**

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

@@ -20,12 +20,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
+import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, Project}
+import org.apache.spark.sql.types.StructType
 
 /**
  * This class provides arguments and body expression of the macro function.
  */
-case class MacroFunctionWrapper(columns: Seq[StructField], macroFunction: Expression)
+case class MacroFunctionWrapper(columns: StructType, macroFunction: Expression)
+
 
 /**
  * The DDL command that creates a macro.
@@ -41,16 +43,33 @@ case class CreateMacroCommand(
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
-    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
+    val columns = funcWrapper.columns
+    val columnAttrs = columns.toAttributes
+    def formatName: (String => String) =
+      if (sparkSession.sessionState.conf.caseSensitiveAnalysis) {
+        (name: String) => name
+      } else {
+        (name: String) => name.toLowerCase
+      }
+    val colToIndex: Map[String, Int] = columnAttrs.map(_.name).map(formatName).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(",")}")
     }
+
+    try {
+      val plan = Project(Seq(Alias(funcWrapper.macroFunction, "m")()), LocalRelation(columnAttrs))
+      val analyzed = sparkSession.sessionState.analyzer.execute(plan)
+      sparkSession.sessionState.analyzer.checkAnalysis(analyzed)
+    } catch {
+      case a: AnalysisException =>
+        throw new AnalysisException(s"CREATE TEMPORARY MACRO $macroName " +
+          s"with exception: ${a.getMessage}")
+    }
+
     val macroFunction = funcWrapper.macroFunction.transform {
       case u: UnresolvedAttribute =>
-        val index = colToIndex.get(u.name).getOrElse(
+        val index = colToIndex.get(formatName(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)
@@ -64,15 +83,15 @@ case class CreateMacroCommand(
           s"for CREATE TEMPORARY MACRO $macroName")
     }
 
-    val macroInfo = columns.mkString(",") + " -> " + funcWrapper.macroFunction.toString
-    val info = new ExpressionInfo(macroInfo, macroName)
+    val columnLength: Int = columns.length
+    val info = new ExpressionInfo(macroName, macroName)
     val builder = (children: Seq[Expression]) => {
-      if (children.size != columns.size) {
+      if (children.size != columnLength) {
         throw new AnalysisException(s"Actual number of columns: ${children.size} != " +
-          s"expected number of columns: ${columns.size} for Macro $macroName")
+          s"expected number of columns: ${columnLength} for Macro $macroName")
       }
       macroFunction.transform {
-        // Skip to validate the input type because check it at runtime.
+        // Skip to validate the input type because check it before.
         case b: BoundReference => children(b.ordinal)
       }
     }

sql/core/src/test/resources/sql-tests/inputs/macro.sql

@@ -0,0 +1,56 @@
+CREATE TEMPORARY MACRO SIGMOID (x DOUBLE) 1.0 / (1.0 + EXP(-x));
+SELECT SIGMOID(2);
+DROP TEMPORARY MACRO SIGMOID;
+
+CREATE TEMPORARY MACRO FIXED_NUMBER() 1;
+SELECT FIXED_NUMBER() + 1;
+DROP TEMPORARY MACRO FIXED_NUMBER;
+
+CREATE TEMPORARY MACRO SIMPLE_ADD (x INT, y INT) x + y;
+SELECT SIMPLE_ADD(1, 9);
+DROP TEMPORARY MACRO SIMPLE_ADD;
+
+CREATE TEMPORARY MACRO flr(d bigint) FLOOR(d/10)*10;
+SELECT flr(12);
+DROP TEMPORARY MACRO flr;
+
+CREATE TEMPORARY MACRO STRING_LEN(x string) length(x);
+CREATE TEMPORARY MACRO STRING_LEN_PLUS_ONE(x string) length(x)+1;
+CREATE TEMPORARY MACRO STRING_LEN_PLUS_TWO(x string) length(x)+2;
+create table macro_test (x string) using parquet;;
+insert into table macro_test values ("bb"), ("a"), ("ccc");
+SELECT CONCAT(STRING_LEN(x), ":", STRING_LEN_PLUS_ONE(x), ":", STRING_LEN_PLUS_TWO(x)) a
+FROM macro_test;
+SELECT CONCAT(STRING_LEN(x), ":", STRING_LEN_PLUS_ONE(x), ":", STRING_LEN_PLUS_TWO(x)) a
+FROM
+macro_test
+sort by a;
+drop table macro_test;
+
+CREATE TABLE macro_testing(a int, b int, c int) using parquet;;
+insert into table macro_testing values (1,2,3);
+insert into table macro_testing values (4,5,6);
+CREATE TEMPORARY MACRO math_square(x int) x*x;
+CREATE TEMPORARY MACRO math_add(x int) x+x;
+select math_square(a), math_square(b),factorial(a), factorial(b), math_add(a), math_add(b),int(c)
+from macro_testing order by int(c);
+drop table macro_testing;
+
+CREATE TEMPORARY MACRO max(x int, y int) x + y;
+SELECT max(1, 2);
+DROP TEMPORARY MACRO max;
+SELECT max(2);
+
+CREATE TEMPORARY MACRO c() 3E9;
+SELECT floor(c()/10);
+DROP TEMPORARY MACRO c;
+
+CREATE TEMPORARY MACRO fixed_number() 42;
+DROP TEMPORARY FUNCTION fixed_number;
+DROP TEMPORARY MACRO IF EXISTS fixed_number;
+
+-- invalid queries
+CREATE TEMPORARY MACRO simple_add_error(x int) x + y;
+CREATE TEMPORARY MACRO simple_add_error(x int, x int) x + y;
+CREATE TEMPORARY MACRO simple_add_error(x int) x NOT IN (select c2);
+DROP TEMPORARY MACRO SOME_MACRO;