[SPARK-15988] [SQL] Implement DDL commands: Create/Drop temporary macro #13706

lianhuiwang · 2016-06-16T15:17:05Z

What changes were proposed in this pull request?

In https://issues.apache.org/jira/browse/HIVE-2655, Hive have implemented Create/Drop temporary macro. So This PR adds the support of native DDL commands to create and drop temporary macro for SparkSQL.
temporary macro can be considered as a special temporary function that has a FunctionBuilder that transform macro function into expression that can be directly calculated.

How was this patch tested?

add unit tests

SparkQA · 2016-06-16T16:21:31Z

Test build #60642 has finished for PR 13706 at commit f2433c2.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class MacroFunctionWrapper(arguments: Seq[AttributeReference], body: Expression)
- case class CreateMacroCommand(macroName: String, macroFunction: MacroFunctionWrapper)
- case class DropMacroCommand(macroName: String, ifExists: Boolean)

SparkQA · 2016-06-16T19:59:54Z

Test build #60654 has finished for PR 13706 at commit 0b93636.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

hvanhovell · 2016-06-17T04:42:40Z

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -97,6 +97,9 @@ statement
    | CREATE TEMPORARY? FUNCTION qualifiedName AS className=STRING
        (USING resource (',' resource)*)?                              #createFunction
    | DROP TEMPORARY? FUNCTION (IF EXISTS)? qualifiedName              #dropFunction
+    | CREATE TEMPORARY MACRO macroName=identifier
+        '('(columns=colTypeList)?')' expression                        #createMacro


NIT: get rid of the parenthesis, you can also just use the colTypeList?

lianhuiwang · 2016-06-18T10:44:59Z

@hvanhovell I have addressed your comments. Thanks.

SparkQA · 2016-06-18T12:34:31Z

Test build #60780 has finished for PR 13706 at commit 301e950.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class NoSuchTempMacroException(func: String)
- case class CreateMacroCommand(

SparkQA · 2016-06-18T12:43:38Z

Test build #60781 has finished for PR 13706 at commit 808a5fa.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

hvanhovell · 2016-06-20T03:46:35Z

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

+    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


Move this into the CreateMacroCommand command. This would also be relevant if we were to offer a different API for creating macro's.

Why i do not to move this into the CreateMacroCommand? Because analyzer.checkAnalysis() will check if macroFunction of CreateMacroCommand is invalid. macroFunction has UnresolvedAttributes, So analyzer.checkAnalysis() will throw a unresolved exception. If it resolved UnresolvedAttributes before, analyzer.checkAnalysis() does not throw a exception.

@hvanhovell So i think i will create a new Wrapper class to avoid unresolved exception in order to DataFrame can reuse this feature later.

Ah, I see. You could also move this code into the companion object of the CreateMacroCommand. That woud also work. It is just that this code isn't parser specific.

hvanhovell · 2016-06-20T03:54:18Z

@lianhuiwang thanks for updating the PR. Could you implement the Macro removal by pattern matching on a (to be created) MacroFunctionBuilder class. I feel this is simpler, and doesn't touch as much of the API's.

hvanhovell · 2016-06-20T04:15:00Z

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

+          s"expected number of columns: ${columns.size} for Macro $macroName")
+      }
+      macroFunction.transformUp {
+        case b: BoundReference => children(b.ordinal)


We do not validate the input type here. This would be entirely fine if macro arguments were defined without a DataType. I am not sure what we need to do here though. We have two options:

Ignore the DataType and rely on the expressions inputTypes to get casting done. This must be documented though.

Introduce casts to make sure the input conforms to the required input.

What do you think?

@hvanhovell good points. Because Analyzer will check expression's checkInputDataTypes after ResolveFunctions, I think we do not validate input type here. Now i do not think it has benefits if we did casts, but it maybe cause unnecessary casts. I will add some comments for it. Thanks.

Ok that is perfect.

lianhuiwang · 2016-06-20T08:13:32Z

@hvanhovell I have addressed your comments. Thanks.

SparkQA · 2016-06-20T09:48:48Z

Test build #60840 has finished for PR 13706 at commit f4ed3bc.

This patch fails Spark unit tests.
This patch does not merge cleanly.
This patch adds the following public classes (experimental):
- case class MacroFunctionWrapper(columns: Seq[StructField], macroFunction: Expression)

SparkQA · 2016-06-20T10:17:04Z

Test build #60841 has finished for PR 13706 at commit af0136d.

This patch passes all tests.
This patch does not merge cleanly.
This patch adds no public classes.

SparkQA · 2017-05-27T16:57:34Z

Test build #77466 has finished for PR 13706 at commit ad85109.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

lianhuiwang · 2017-05-27T17:15:08Z

@gatorsmile sorry for reply lately. Now i have merged with master. Thanks.

SparkQA · 2017-05-27T19:01:20Z

Test build #77468 has finished for PR 13706 at commit fce1121.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-05-27T19:11:02Z

Test build #77469 has finished for PR 13706 at commit eaff4e9.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-05-27T19:26:29Z

Test build #77467 has finished for PR 13706 at commit b52698f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-05-27T20:44:34Z

Test build #77470 has finished for PR 13706 at commit 97632a9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-05-27T20:48:46Z

Test build #77472 has finished for PR 13706 at commit 4ee32e9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

hvanhovell

This looks pretty good. I left a few comments. I think we need to refine the macro creation path, because the current approach has relatively bad UX when things go wrong.

hvanhovell · 2017-05-28T16:02:15Z

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

+/**
+ * This class provides arguments and body expression of the macro function.
+ */
+case class MacroFunctionWrapper(columns: Seq[StructField], macroFunction: Expression)


Why do we need this?

because Analyzer will check macroFunction that is invalid if I donot use MacroFunctionWrapper.

hvanhovell · 2017-05-28T16:06:17Z

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

+      case s: SubqueryExpression =>
+        throw new AnalysisException(s"Cannot support Subquery: ${s} " +
+          s"for CREATE TEMPORARY MACRO $macroName")
+      case u: UnresolvedGenerator =>


Is this what Hive does? I really don't see why we should not support this.

Please note that we cannot use generators if we decide that an expression has to be a fully resolved expression at creation time.

hvanhovell · 2017-05-28T16:09:58Z

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala

@@ -1516,6 +1516,35 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
    )
  }

+  test("create/drop temporary macro") {


Can you use SQLQueryTestSuite instead?

Can you also add a case for a macro without parameters? E.g.: create temporary macro c() as 3E9

Should we also test a combination of temporary macros/functions...?

hvanhovell · 2017-05-28T16:11:22Z

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

+  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)() }


Nit: put } on a new line

hvanhovell · 2017-05-28T16:12:23Z

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

+    }
+    val macroFunction = funcWrapper.macroFunction.transform {
+      case u: UnresolvedAttribute =>
+        val index = colToIndex.get(u.name).getOrElse(


We should respect the case-sensitivity settings here. So a lookup might not be the best idea.

yes, i will do it, thanks.

hvanhovell · 2017-05-28T16:17:56Z

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

+    val macroInfo = columns.mkString(",") + " -> " + funcWrapper.macroFunction.toString
+    val info = new ExpressionInfo(macroInfo, macroName, true)
+    val builder = (children: Seq[Expression]) => {
+      if (children.size != columns.size) {


It is slightly better to columns.size in a separate variable, so we do not include columns in the closure.

hvanhovell · 2017-05-28T16:19:33Z

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

+   * }}}
+   */
+  override def visitCreateMacro(ctx: CreateMacroContext): LogicalPlan = withOrigin(ctx) {
+    val arguments = Option(ctx.colTypeList).map(visitColTypeList(_))


Nit: you can avoid (_)...

hvanhovell · 2017-05-28T16:32:05Z

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

@@ -107,6 +110,14 @@ class SimpleFunctionRegistry extends FunctionRegistry {
    functionBuilders.remove(name).isDefined
  }

+  override def dropMacro(name: String): Boolean = synchronized {


A drop function can currently also drop a macro. Can you make sure that this cannot happen?

Maybe we should consolidate this into a single drop function with a macro flag. cc @gatorsmile WDYT?

Hive can drop temporary function using command 'drop Macro'. And it also can drop temporary macro using command 'drop temporary function'.

hvanhovell · 2017-05-28T17:27:01Z

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

+      name: String,
+      info: ExpressionInfo,
+      functionBuilder: FunctionBuilder): Unit = {
+    if (functionRegistry.functionExists(name)) {


I am not entirely sure if we should throw an exception here. It unfortunately depends on the semantics you follow, SQL will throw an exception, whereas the Dataframe API will just overwrite the function. Let's follow Hive for now.

hive> create temporary macro max(x int) > x*x; OK Time taken: 0.014 seconds hive> select max(3) from t1; OK 9 Time taken: 0.468 seconds, Fetched: 1 row(s) hive> select max(3,4) from t1; FAILED: SemanticException [Error 10015]: Line 1:7 Arguments length mismatch '4': The macro max accepts exactly 1 arguments.

Hive overwrites the temporary function without issuing an exception.

hvanhovell · 2017-05-28T18:19:44Z

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

+          s"expected number of columns: ${columns.size} for Macro $macroName")
+      }
+      macroFunction.transform {
+        // Skip to validate the input type because check it at runtime.


How do we check at runtime? The current code does not seem to respect the types passed, and rely on the macro's expression to do some type validation, this means you can pass anything to the macro and the user can end up with an unexpected result:

create macro plus(a int, b int) as a + b; select plus(1.0, 1.0) as result -- This returns a decimal, and not an int as expected

So I think we should at least validate the input expressions. The hacky way would be to add casts, and have the analyzer fail if the cast cannot be made (this is terrible UX). A better way to would be to create some sentinel expression that makes sure the analyzer will insert the correct cast, and throws a relevant exception (mentioning the macro) when this blows up...

On a related note, we are currently not sure if the macro produces a valid expression. Maybe we should run analysis on the macro expression to make sure it does not fail every query later on, e.g.:

val resolvedMacroFunction = try { val plan = Project(Alias(macroFunction, "m")() :: Nil, OneRowRelation) val analyzed @ Project(Seq(named), OneRowRelation) = sparkSession.sessionState.analyzer.execute(plan) sparkSession.sessionState.analyzer.checkAnalysis(analyzed) named.children.head } catch { case a: AnalysisException => ... }

Note that we cannot use generators if we use this approach...

Yes, Now i update it with you ideas. Thanks.

hvanhovell · 2017-05-28T18:32:09Z

cc @gatorsmile could you take a look at the way this interacts with the session catalog and the function registry?

gatorsmile · 2017-05-29T18:58:46Z

Sure. Will do it. Thanks!

gatorsmile · 2017-05-29T21:32:52Z

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

@@ -52,3 +52,6 @@ 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)


Please remove it. For reasons, please see the PR #17716.

Yes, Thanks.

gatorsmile · 2017-05-29T22:00:07Z

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

+  /** Drop a temporary macro. */
+  def dropTempMacro(name: String, ignoreIfNotExists: Boolean): Unit = {
+    if (!functionRegistry.dropMacro(name) && !ignoreIfNotExists) {
+      throw new NoSuchTempMacroException(name)


hive> DROP TEMPORARY MACRO max; OK Time taken: 0.01 seconds hive> select max(3) from t1; OK 3

After we drop the macro, the existing function works well. That means, we did not delete the original built-in functions. The built-in function will not be dropped by DROP TEMPORARY MACRO. After we drop the macro with the same name, the original function max is using the original built-in function.

Yes, I have update it with this case.

SparkQA · 2017-05-30T11:12:23Z

Test build #77527 has finished for PR 13706 at commit b539e94.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class SystemFunctionRegistry(builtin: SimpleFunctionRegistry) extends SimpleFunctionRegistry

SparkQA · 2017-05-30T17:09:39Z

Test build #77533 has finished for PR 13706 at commit 1563f12.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class MacroFunctionWrapper(columns: StructType, macroFunction: Expression)

gatorsmile · 2017-05-30T17:47:07Z

Let me first clean up the existing function registry #18142. Will ping you when it is ready. Thanks!

SparkQA · 2017-05-30T19:34:10Z

Test build #77544 has finished for PR 13706 at commit 4d8e843.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

lianhuiwang · 2017-05-31T00:53:23Z

@gatorsmile OK. Thanks.

HyukjinKwon · 2017-07-24T03:48:23Z

@gatorsmile, I just wonder if it is ready to proceed further this PR. It looks the PR you linked is merged properly.

gatorsmile · 2017-10-27T23:38:09Z

@lianhuiwang Could you restart the work? Thanks!

gatorsmile · 2018-12-22T19:34:24Z

@lianhuiwang Please feel free to reopen the PR if you have the bandwidth to restart the work! Thanks!

init commit

f2433c2

lianhuiwang changed the title ~~[SPARK-15988] [SQL] Implement DDL commands: CREATE/DROP TEMPORARY MACRO~~ [SPARK-15988] [SQL] Implement DDL commands: Create/Drop temporary macro Jun 16, 2016

fix unit test

0b93636

hvanhovell reviewed Jun 17, 2016
View reviewed changes

update

301e950

update createTempMacro

808a5fa

hvanhovell reviewed Jun 20, 2016
View reviewed changes

lianhuiwang added 2 commits June 20, 2016 16:03

address comments

f4ed3bc

update

af0136d

add import

b52698f

lianhuiwang added 5 commits May 28, 2017 01:33

treat macro as temp function like hive

3eacebc

add Modifier for FunctionRegistry.

fce1121

update comments.

eaff4e9

add dropMacro().

97632a9

reformat code style

4ee32e9

hvanhovell requested changes May 28, 2017

View reviewed changes

gatorsmile reviewed May 29, 2017

View reviewed changes

address some comments.

b539e94

address some comments.

1563f12

update

4d8e843

gatorsmile closed this Dec 22, 2018

[SPARK-15988] [SQL] Implement DDL commands: Create/Drop temporary macro #13706

[SPARK-15988] [SQL] Implement DDL commands: Create/Drop temporary macro #13706

Conversation

lianhuiwang commented Jun 16, 2016 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Jun 16, 2016

SparkQA commented Jun 16, 2016

Choose a reason for hiding this comment

lianhuiwang commented Jun 18, 2016

SparkQA commented Jun 18, 2016

SparkQA commented Jun 18, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hvanhovell commented Jun 20, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lianhuiwang commented Jun 20, 2016

SparkQA commented Jun 20, 2016

SparkQA commented Jun 20, 2016

SparkQA commented May 27, 2017

lianhuiwang commented May 27, 2017

SparkQA commented May 27, 2017

SparkQA commented May 27, 2017

SparkQA commented May 27, 2017

SparkQA commented May 27, 2017

SparkQA commented May 27, 2017

hvanhovell left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lianhuiwang May 30, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gatorsmile May 29, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hvanhovell commented May 28, 2017

gatorsmile commented May 29, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented May 30, 2017

SparkQA commented May 30, 2017

gatorsmile commented May 30, 2017

SparkQA commented May 30, 2017

lianhuiwang commented May 31, 2017

HyukjinKwon commented Jul 24, 2017

gatorsmile commented Oct 27, 2017

gatorsmile commented Dec 22, 2018

lianhuiwang commented Jun 16, 2016 •

edited

Loading

lianhuiwang May 30, 2017 •

edited

Loading

gatorsmile May 29, 2017 •

edited

Loading