[SPARK-13721][SQL] Support outer generators in DataFrame API #16608

bogdanrdc · 2017-01-16T17:11:39Z

What changes were proposed in this pull request?

Added outer_explode, outer_posexplode, outer_inline functions and expressions.
Some bug fixing in GenerateExec.scala for CollectionGenerator. Previously it was not correctly handling the case of outer with empty collections, only with nulls.

How was this patch tested?

New tests added to GeneratorFunctionSuite

hvanhovell · 2017-01-16T17:24:14Z

ok to test

hvanhovell

@bogdanrdc this looks promising! I left some small comments.

hvanhovell · 2017-01-16T17:26:18Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/generators.scala

 }

+case class GeneratorOuter(child: Generator) extends UnaryExpression
+    with Generator {


Add the Unevaluable trait. Then you don't have to implement eval() and doGenCode()

Couldn't make it work with Unevaluable because Generator.eval is not the same as Unevaluable.eval.

Ok, that is fair.

hvanhovell · 2017-01-16T17:27:48Z

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

    expression[NullIf]("nullif"),
    expression[Nvl]("nvl"),
    expression[Nvl2]("nvl2"),
+    expression[OuterExplode]("outer_explode"),


Name these ..._outer, e.g.: explode_outer. That makes them easier to find when someone is looking for them (same goes for naming in functions.scala).

hvanhovell · 2017-01-16T17:29:53Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/generators.scala

        new StructType()
          .add("pos", IntegerType, nullable = false)
-          .add("key", kt, nullable = false)
+          .add("key", kt, nullable = true)


What happens when this is false?

I think we should fix the output of Generate instead of doing this.

I will see if I can fix it from Generate instead.

If it's false then outer_explode with empty map fails with this exception:
java.lang.RuntimeException: Error while decoding: java.lang.NullPointerException
createexternalrow(input[0, string, false].toString, input[1, string, true].toString, StructField(key,StringType,false), StructField(value,StringType,true))
:- input[0, string, false].toString
: +- input[0, string, false]
+- input[1, string, true].toString
+- input[1, string, true]

fixed it in Generate instead. makes more sense from there too.

hvanhovell · 2017-01-16T17:30:53Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/generators.scala

  override val position: Boolean = false
 }
-
+class OuterExplode(child: Expression) extends GeneratorOuter(Explode(child))


Remove this. Favor composition over inheritance. Just create a GeneratorOuter(Explode(child)) whenever you need this.

I made it like this so I can use it in FunctionRegistry.scala:
expression[OuterExplode]("outer_explode")
Is there a way I can still make it work with GeneratorOuter(Explode(child))?

Reverted to GeneratorOuter(Explode()) and created a new helper method in FunctionRegistry instead.
Actually the way I made these classes wasn't working. It was returning strange results, but not failing otherwise. I guess somewhere the OuterExplode class was not caught by a match. The generated code was different.

hvanhovell · 2017-01-16T17:30:59Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/generators.scala

  override val position = true
 }
-
+class OuterPosExplode(child: Expression) extends GeneratorOuter(PosExplode(child))


Remove this.

hvanhovell · 2017-01-16T17:31:05Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/generators.scala

  }
 }
+
+class OuterInline(child: Expression) extends GeneratorOuter(Inline(child))


Remove this.

hvanhovell · 2017-01-16T17:34:38Z

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

+    // array/map contains no input.
+    // generateOuter is an int. it is set to 1 iff outer is true and the input is empty or null.
+    val generateOuter = ctx.freshName("generateOuter")
+    val isOuter = if (outer) {


How about we generate the expressions for the generateOuter variable here instead? It is always 0 when outer == false and $numElements == 0 ? 1 : 0 when it is true. This increases the probability that the generated code gets eliminated at compile time.

I simplified it by using index == -1 like you said and reverting to the old code. It was also easy to make it produce (null, null) for posexplode_outer instead of (-1, null). @rxin preferred to return null and not -1 (or even 0) for the position. So now, everything is null always.

hvanhovell · 2017-01-16T17:35:20Z

sql/core/src/main/scala/org/apache/spark/sql/functions.scala

+   *
+   * @group collection_funcs
+   * @since 2.2.0
+   */


Naming: explode_outer

hvanhovell · 2017-01-16T17:35:30Z

sql/core/src/main/scala/org/apache/spark/sql/functions.scala

+   *
+   * @group collection_funcs
+   * @since 2.2.0
+   */


Naming: posexplode_outer

hvanhovell · 2017-01-16T17:36:06Z

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

    expression[Nvl]("nvl"),
    expression[Nvl2]("nvl2"),
+    expression[OuterExplode]("outer_explode"),
+    expression[OuterInline]("outer_inline"),


Why add this one to the function registry? It is not exposed in functions.scala.

This is to support df.selectExpr("outer_explode(array())").
Should I remove it?

SparkQA · 2017-01-16T19:25:48Z

Test build #71458 has finished for PR 16608 at commit 62f46f6.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class GeneratorOuter(child: Generator) extends UnaryExpression
class OuterExplode(child: Expression) extends GeneratorOuter(Explode(child))
class OuterPosExplode(child: Expression) extends GeneratorOuter(PosExplode(child))
class OuterInline(child: Expression) extends GeneratorOuter(Inline(child))

rxin · 2017-01-16T19:40:16Z

sql/core/src/test/scala/org/apache/spark/sql/GeneratorFunctionSuite.scala

      df.select(explode('intList)),
      Row(1) :: Row(2) :: Row(3) :: Nil)
  }
+  test("single outer_explode") {


add a blank line before this

just add it to all the test cases - have a blank line (only one) separating them.

rxin · 2017-01-16T19:41:26Z

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

-      def unapply(e: Expression): Option[(Generator, Seq[String])] = e match {
-        case Alias(g: Generator, name) if g.resolved => Some((g, name :: Nil))
-        case MultiAlias(g: Generator, names) if g.resolved => Some(g, names)
+      def unapply(e: Expression): Option[(Generator, Seq[String], Boolean)] = e match {


document what the return value means (especially that boolean value, but also the Seq[String] that's preexisting)

hvanhovell · 2017-01-17T00:07:01Z

sql/hive/src/test/scala/org/apache/spark/sql/catalyst/ExpressionToSQLSuite.scala

    checkSqlGeneration("SELECT map(1, 'a', 2, 'b')")
    checkSqlGeneration("SELECT named_struct('c1',1,'c2',2,'c3',3)")
    checkSqlGeneration("SELECT nanvl(a, 5), nanvl(b, 10), nanvl(d, c) from t2")
+    checkSqlGeneration("SELECT outer_explode(array())")


This breaks right?

hvanhovell

I left a few small comments. LGTM pending these & jenkins.

hvanhovell · 2017-01-17T00:10:09Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/generators.scala

 }

+case class GeneratorOuter(child: Generator) extends UnaryExpression
+    with Generator {


Ok, that is fair.

hvanhovell · 2017-01-17T00:11:31Z

...alyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

    // prepend the new qualifier to the existed one
    generatorOutput.map(a => a.withQualifier(Some(q)))
-  }.getOrElse(generatorOutput)
+  }.getOrElse(generatorOutput).map {


Nit style. Can you split this up for readability.

hvanhovell · 2017-01-17T00:11:48Z

sql/core/src/main/scala/org/apache/spark/sql/functions.scala

  def posexplode(e: Column): Column = withExpr { PosExplode(e.expr) }

+  /**
+   * Creates a new row for each element with position in the given array or map column.


update this the result.

hvanhovell · 2017-01-17T00:17:21Z

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

      new ExpressionInfo(clazz.getCanonicalName, name)
    }
  }
+  private def expressionGeneratorOuter[T <: Generator : ClassTag]


Add a new line

hvanhovell · 2017-01-17T00:18:06Z

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

  }
+  private def expressionGeneratorOuter[T <: Generator : ClassTag]
+    (name: String): (String, (ExpressionInfo, FunctionBuilder)) = {
+    val regularGen = expression[T](name)


Deconstruct the tuple here: val (_, (info, generator)) = expression[T](name)

hvanhovell · 2017-01-17T00:18:27Z

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

+    val outerBuilder = (args: Seq[Expression]) => {
+      GeneratorOuter(regularGen._2._2(args).asInstanceOf[Generator])
+    }
+    (name, (expressionInfo[GeneratorOuter](name), outerBuilder))


Let's use the generators info instead of the outer generator info.

SparkQA · 2017-01-17T01:24:06Z

Test build #71466 has finished for PR 16608 at commit 26a1a67.

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

SparkQA · 2017-01-17T01:54:51Z

Test build #71467 has finished for PR 16608 at commit 36ac054.

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

SparkQA · 2017-01-17T18:51:25Z

Test build #71513 has finished for PR 16608 at commit 8a8af55.

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

gatorsmile · 2017-01-17T18:56:52Z

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

  }
+
+  private def expressionGeneratorOuter[T <: Generator : ClassTag]
+    (name: String): (String, (ExpressionInfo, FunctionBuilder)) = {


Nit: Could you follow the indentation style? https://github.com/databricks/scala-style-guide#spacing-and-indentation

private def expressionGeneratorOuter[T <: Generator : ClassTag]( name: String): (String, (ExpressionInfo, FunctionBuilder)) = { val (_, (info, generatorBuilder)) = expression[T](name) ...

SparkQA · 2017-01-17T23:35:27Z

Test build #71529 has finished for PR 16608 at commit 99742be.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class GeneratorOuter(child: Generator) extends UnaryExpression with Generator

hvanhovell · 2017-01-17T23:38:49Z

LGTM - merging to master.

## What changes were proposed in this pull request? Added outer_explode, outer_posexplode, outer_inline functions and expressions. Some bug fixing in GenerateExec.scala for CollectionGenerator. Previously it was not correctly handling the case of outer with empty collections, only with nulls. ## How was this patch tested? New tests added to GeneratorFunctionSuite Author: Bogdan Raducanu <[email protected]> Closes apache#16608 from bogdanrdc/SPARK-13721.

gatorsmile · 2017-02-21T19:47:13Z

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

    expression[Abs]("abs"),
    expression[Coalesce]("coalesce"),
    expression[Explode]("explode"),
+    expressionGeneratorOuter[Explode]("explode_outer"),


We might need an update on ExpressionDescription for these three new expressions.

@gatorsmile it uses the expression description of the underlying expression: https://github.com/apache/spark/pull/16608/files#diff-2c0350957ac4932d3f63796eceaeae08R517

spark-sql> desc function extended explode_outer; Function: explode_outer Class: org.apache.spark.sql.catalyst.expressions.Explode Usage: explode_outer(expr) - Separates the elements of array `expr` into multiple rows, or the elements of map `expr` into multiple rows and columns. Extended Usage: Examples: > SELECT explode_outer(array(10, 20)); 10 20

@hvanhovell explode_outer is sharing the same expression description. Do we need an update on this?

Why would we need an update? What is the extra information you want to convey? Do you want to add a generic line saying that an outer generator might produce nulls instead of filtering out the row?

Yes. Update the description and provide an example to users? Maybe hardcode the function name instead of using _FUNC_?

@ExpressionDescription( usage = "_FUNC_(expr) - Separates the elements of array `expr` into multiple rows, or the elements of map `expr` into multiple rows and columns.", extended = """ Examples: > SELECT _FUNC_(array(10, 20)); 10 20 """)

I am not super enthusiastic about this. We have three options here:

Leave as it is.

Remove the outer_... generators, and make a user use the lateral view outer ... instead.

Create separate OuterGenerator classes for each one, and provide proper documentation.

I am fine with any.

## What changes were proposed in this pull request? Added outer_explode, outer_posexplode, outer_inline functions and expressions. Some bug fixing in GenerateExec.scala for CollectionGenerator. Previously it was not correctly handling the case of outer with empty collections, only with nulls. ## How was this patch tested? New tests added to GeneratorFunctionSuite Author: Bogdan Raducanu <[email protected]> Closes apache#16608 from bogdanrdc/SPARK-13721.

Added outer_explode, outer_posexplode, outer_inline

62f46f6

hvanhovell requested changes Jan 16, 2017

View reviewed changes

rxin reviewed Jan 16, 2017

View reviewed changes

Bogdan Raducanu added 2 commits January 17, 2017 00:51

fixed some review issues

26a1a67

renamed outer_* to *_outer in ExpressionToSQLSuite

36ac054

hvanhovell reviewed Jan 17, 2017

View reviewed changes

hvanhovell requested changes Jan 17, 2017

View reviewed changes

review issues

8a8af55

gatorsmile reviewed Jan 17, 2017

View reviewed changes

Bogdan Raducanu added 2 commits January 17, 2017 21:52

indentation

76f15d1

indentation

99742be

asfgit closed this in 2992a0e Jan 17, 2017

gatorsmile reviewed Feb 21, 2017

View reviewed changes

[SPARK-13721][SQL] Support outer generators in DataFrame API #16608

[SPARK-13721][SQL] Support outer generators in DataFrame API #16608

Uh oh!

Conversation

bogdanrdc commented Jan 16, 2017

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

hvanhovell commented Jan 16, 2017

Uh oh!

hvanhovell left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bogdanrdc Jan 16, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bogdanrdc Jan 16, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jan 16, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hvanhovell left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bogdanrdc Jan 16, 2017 •

edited

Loading

bogdanrdc Jan 16, 2017 •

edited

Loading

gatorsmile Jan 17, 2017 •

edited

Loading

gatorsmile Feb 22, 2017 •

edited

Loading