[SPARK-23939][SQL] Add transform_keys function #22013

codeatri · 2018-08-06T19:04:45Z

What changes were proposed in this pull request?

This pr adds transform_keys function which applies the function to each entry of the map and transforms the keys.

> SELECT transform_keys(map(array(1, 2, 3), array(1, 2, 3)), (k,v) -> k + 1);
       map(2->1, 3->2, 4->3)

> SELECT transform_keys(map(array(1, 2, 3), array(1, 2, 3)), (k,v) -> k + v);
       map(2->1, 4->2, 6->3)

How was this patch tested?

Added tests.

hvanhovell · 2018-08-06T20:22:26Z

ok to test

SparkQA · 2018-08-07T00:11:44Z

Test build #94315 has finished for PR 22013 at commit 0a19cc4.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class TransformKeys(

hvanhovell · 2018-08-07T08:12:39Z

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

+      while (i < arr.numElements) {
+        keyVar.value.set(arr.keyArray().get(i, keyVar.dataType))
+        valueVar.value.set(arr.valueArray().get(i, valueVar.dataType))
+        resultKeys.update(i, f.eval(input))


This assumes that the transformation will return a unique key right? If it doesn't you'll break the map semantics. For example: transform_keys(some_map, (k, v) -> 0)

I'm not a fun of duplicated keys either, but other functions transforming maps have the same problem. See the discussions here and here.

Example:

scala> spark.range(1).selectExpr("map(0,1,0,2)").show() +----------------+ | map(0, 1, 0, 2)| +----------------+ |[0 -> 1, 0 -> 2]| +----------------+

mgaido91 · 2018-08-07T09:26:35Z

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

+
+  override def dataType: DataType = {
+    val valueType = input.dataType.asInstanceOf[MapType].valueType
+    MapType(function.dataType, valueType, input.nullable)


I think here input.nullable is wrong. This should indicate whether the value contains null, not whether the returned object can be null or not.

mn-mikke · 2018-08-07T09:29:17Z

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

+  usage = "_FUNC_(expr, func) - Transforms elements in a map using the function.",
+  examples = """
+    Examples:
+      > SELECT _FUNC_(map(array(1, 2, 3), array(1, 2, 3), (k,v) -> k + 1);


nit: missing space -> k, v

mn-mikke · 2018-08-07T09:32:05Z

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

@@ -365,3 +365,69 @@ case class ArrayAggregate(

  override def prettyName: String = "aggregate"
 }
+
+/**
+ * Transform Keys in a map using the transform_keys function.


maybe a better comment?

mn-mikke · 2018-08-07T09:52:41Z

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

+      while (i < arr.numElements) {
+        keyVar.value.set(arr.keyArray().get(i, keyVar.dataType))
+        valueVar.value.set(arr.valueArray().get(i, valueVar.dataType))
+        resultKeys.update(i, f.eval(input))


Maybe I'm missing something, but couldn't f.eval(input) be evaluated to null? Keys are not allowed to benull. Other functions have usually a null check and throw RuntimeException for such cases.

mn-mikke · 2018-08-07T09:57:08Z

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

+    dfExample5.cache()
+    dfExample6.cache()
+    // Test with cached relation, the Project will be evaluated with codegen
+    testMapOfPrimitiveTypesCombination()


Do we have do that if the expression implements CodegenFallback?

mn-mikke · 2018-08-07T09:59:42Z

...yst/src/test/scala/org/apache/spark/sql/catalyst/expressions/HigherOrderFunctionsSuite.scala

+
+  test("TransformKeys") {
+    val ai0 = Literal.create(
+      Map(1 -> 1, 2 -> 2, 3 -> 3),


It's maybe irrelevant but WDYT about adding test cases with null values?

Thanks for catching this!
Included test cases, both here and in DataFrameFunctionsSuite.

SparkQA · 2018-08-08T07:04:45Z

Test build #94404 has finished for PR 22013 at commit 5806ac4.

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

SparkQA · 2018-08-08T07:05:02Z

Test build #94405 has finished for PR 22013 at commit 150a6a5.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-08T20:51:18Z

Test build #94451 has finished for PR 22013 at commit 9f6a8ab.

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

SparkQA · 2018-08-09T00:06:46Z

Test build #94455 has finished for PR 22013 at commit 6526630.

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

SparkQA · 2018-08-09T00:07:23Z

Test build #94457 has finished for PR 22013 at commit f7fd231.

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

codeatri · 2018-08-09T03:44:49Z

@hvanhovell @mn-mikke @mgaido91 Thanks for the review! I have addressed all your comments and added appropriate test cases for the same.

ueshin · 2018-08-09T05:39:31Z

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

+  usage = "_FUNC_(expr, func) - Transforms elements in a map using the function.",
+  examples = """
+    Examples:
+      > SELECT _FUNC_(map(array(1, 2, 3), array(1, 2, 3), (k, v) -> k + 1);


nit: we need one more right parenthesis after the second array(1, 2, 3)?

ueshin · 2018-08-09T05:39:37Z

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

+    Examples:
+      > SELECT _FUNC_(map(array(1, 2, 3), array(1, 2, 3), (k, v) -> k + 1);
+       map(array(2, 3, 4), array(1, 2, 3))
+      > SELECT _FUNC_(map(array(1, 2, 3), array(1, 2, 3), (k, v) -> k + v);


ueshin · 2018-08-09T05:44:13Z

...yst/src/test/scala/org/apache/spark/sql/catalyst/expressions/HigherOrderFunctionsSuite.scala

+  def transformKeys(expr: Expression, f: (Expression, Expression) => Expression): Expression = {
+    val valueType = expr.dataType.asInstanceOf[MapType].valueType
+    val keyType = expr.dataType.asInstanceOf[MapType].keyType
+    TransformKeys(expr, createLambda(keyType, false, valueType, true, f))


We should use valueContainsNull instead of true?

ueshin · 2018-08-09T05:45:23Z

...yst/src/test/scala/org/apache/spark/sql/catalyst/expressions/HigherOrderFunctionsSuite.scala

+  test("TransformKeys") {
+    val ai0 = Literal.create(
+      Map(1 -> 1, 2 -> 2, 3 -> 3),
+      MapType(IntegerType, IntegerType))


Can you add valueContainsNull explicitly?

ueshin · 2018-08-09T05:46:38Z

...yst/src/test/scala/org/apache/spark/sql/catalyst/expressions/HigherOrderFunctionsSuite.scala

+      MapType(IntegerType, IntegerType))
+    val ai2 = Literal.create(
+      Map(1 -> 1, 2 -> null, 3 -> 3),
+      MapType(IntegerType, IntegerType))


Can you add tests for Literal.create(null, MapType(IntegerType, IntegerType))?

ueshin · 2018-08-09T06:08:42Z

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

@@ -2117,6 +2117,198 @@ class DataFrameFunctionsSuite extends QueryTest with SharedSQLContext {
    assert(ex4.getMessage.contains("data type mismatch: argument 3 requires int type"))
  }

+  test("transform keys function - test various primitive data types combinations") {


We don't need so many cases here. We only need to verify the api works end to end.
Evaluation checks of the function should be in HigherOrderFunctionsSuite.

SparkQA · 2018-08-09T22:47:06Z

Test build #94518 has finished for PR 22013 at commit 1cbaf0c.

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

codeatri · 2018-08-09T23:35:45Z

Thanks for the review @ueshin! I have addressed all your comments.

SparkQA · 2018-08-14T17:55:55Z

Test build #94758 has finished for PR 22013 at commit bb52630.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-08-14T19:15:11Z

Test build #94765 has finished for PR 22013 at commit 621213d.

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

SparkQA · 2018-08-15T03:20:16Z

Test build #94775 has finished for PR 22013 at commit 5db526b.

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

ueshin · 2018-08-15T02:54:41Z

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

+      keyVar.value.set(map.keyArray().get(i, keyVar.dataType))
+      valueVar.value.set(map.valueArray().get(i, valueVar.dataType))
+      val result = f.eval(inputRow)
+      if (result ==  null) {


nit: extra space between == and null.

ueshin · 2018-08-15T03:00:23Z

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

+    MapType(function.dataType, map.valueType, map.valueContainsNull)
+  }
+
+  @transient val MapType(keyType, valueType, valueContainsNull) = argument.dataType


lazy val?
Could you add a test when argument is not a map in invalid cases of DataFrameFunctionsSuite?

ueshin · 2018-08-15T03:01:58Z

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

+
+  override def dataType: DataType = {
+    val map = argument.dataType.asInstanceOf[MapType]
+    MapType(function.dataType, map.valueType, map.valueContainsNull)


We can use valueType and valueContainsNull from the following val?

What about this?

ueshin · 2018-08-15T03:05:56Z

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

+  }
+
+  override def prettyName: String = "transform_keys"
+  }


nit: indent

ueshin · 2018-08-15T03:11:57Z

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

+
+    def testInvalidLambdaFunctions(): Unit = {
+      val ex1 = intercept[AnalysisException] {
+        dfExample1.selectExpr("transform_keys(i, k -> k )")


nit: extra space after k -> k.

ueshin · 2018-08-15T03:14:06Z

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

+        dfExample2.selectExpr("transform_keys(j, (k, v, x) -> k + 1)")
+      }
+      assert(ex2.getMessage.contains(
+      "The number of lambda function arguments '3' does not match"))


nit: indent

ueshin · 2018-08-15T03:16:17Z

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

+    ).toDF("i")
+
+    val dfExample2 = Seq(
+      Map[Int, Double](1 -> 1.0E0, 2 -> 1.4E0, 3 -> 1.7E0)


Do we need E0?

ueshin · 2018-08-15T03:22:28Z

sql/core/src/test/resources/sql-tests/inputs/higher-order-functions.sql

+
+create or replace temporary view nested as values
+  (1, map(1,1,2,2,3,3)),
+  (2, map(4,4,5,5,6,6))


nit:

(1, map(1, 1, 2, 2, 3, 3)), (2, map(4, 4, 5, 5, 6, 6))

ueshin · 2018-08-15T03:25:52Z

...yst/src/test/scala/org/apache/spark/sql/catalyst/expressions/HigherOrderFunctionsSuite.scala

+      MapType(StringType, StringType, valueContainsNull = true))
+    val as2 = Literal.create(null,
+      MapType(StringType, StringType, valueContainsNull = false))
+    val asn = Literal.create(Map.empty[StringType, StringType],


ueshin · 2018-08-15T03:32:11Z

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

+    testInvalidLambdaFunctions()
+    dfExample1.cache()
+    dfExample2.cache()
+    testInvalidLambdaFunctions()


We need dfExample3.cache() as well?

@ueshin I would like to ask you a generic question regarding higher-order functions. Is it necessary to perform checks with codegen paths if all the newly added functions extends from CodegenFallback? Eventually, is there a plan to add coden for these functions in future?

ueshin · 2018-08-15T03:36:18Z

Btw, we need one more right parenthesis after the second array(1, 2, 3) and a space at (k,v) in the description?

ueshin · 2018-08-15T03:52:09Z

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

+    val LambdaFunction(
+    _, (keyVar: NamedLambdaVariable) :: (valueVar: NamedLambdaVariable) :: Nil, _) = function
+    (keyVar, valueVar)
+  }


nit: how about:

@transient lazy val LambdaFunction(_, (keyVar: NamedLambdaVariable) :: (valueVar: NamedLambdaVariable) :: Nil, _) = function

Sorry, I meant we don't need to surround by:

@transient lazy val (keyVar, valueVar) = { ... (keyVar, valueVar) }

just

@transient lazy val LambdaFunction(_, (keyVar: NamedLambdaVariable) :: (valueVar: NamedLambdaVariable) :: Nil, _) = function

should work.

SparkQA · 2018-08-15T07:05:01Z

Test build #94788 has finished for PR 22013 at commit e5d9b05.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

mn-mikke · 2018-08-15T18:36:13Z

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

+  @transient lazy val MapType(keyType, valueType, valueContainsNull) = argument.dataType
+
+  override def dataType: DataType = {
+    MapType(function.dataType, valueType, valueContainsNull)


nit: just in one line?

mn-mikke · 2018-08-15T18:44:40Z

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

+
+  override def nullSafeEval(inputRow: InternalRow, argumentValue: Any): Any = {
+    val map = argumentValue.asInstanceOf[MapData]
+    val f = functionForEval


Can't we use functionForEval directly?

SparkQA · 2018-08-15T21:31:04Z

Test build #94811 has finished for PR 22013 at commit fb885f4.

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

SparkQA · 2018-08-15T22:56:19Z

Test build #94817 has finished for PR 22013 at commit 58b60b2.

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

SparkQA · 2018-08-15T23:40:59Z

Test build #94819 has finished for PR 22013 at commit 2f4943f.

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

ueshin · 2018-08-16T03:08:05Z

LGTM.
@mn-mikke @mgaido91 Do you have any other comments on this?

ueshin · 2018-08-16T08:06:33Z

I'd merge this now.
@mn-mikke @mgaido91 If you have any other comments, let's have a follow-up pr.

ueshin · 2018-08-16T08:06:49Z

Thanks! merging to master.

mgaido91

LGTM, just one very minor comment, thanks

mgaido91 · 2018-08-16T09:00:59Z

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

+    function: Expression)
+  extends MapBasedSimpleHigherOrderFunction with CodegenFallback {
+
+  override def nullable: Boolean = argument.nullable


I think this can be moved to SimpleHigherOrderFunction

makes sense.
Let's have wrap-up prs for higher-order functions after the remaining 2 prs are merged.

## What changes were proposed in this pull request? - Revert [SPARK-23935][SQL] Adding map_entries function: #21236 - Revert [SPARK-23937][SQL] Add map_filter SQL function: #21986 - Revert [SPARK-23940][SQL] Add transform_values SQL function: #22045 - Revert [SPARK-23939][SQL] Add transform_keys function: #22013 - Revert [SPARK-23938][SQL] Add map_zip_with function: #22017 - Revert the changes of map_entries in [SPARK-24331][SPARKR][SQL] Adding arrays_overlap, array_repeat, map_entries to SparkR: #21434 ## How was this patch tested? The existing tests. Closes #22827 from gatorsmile/revertMap2.4. Authored-by: gatorsmile <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

Added Support for transform_keys function

0a19cc4

hvanhovell reviewed Aug 7, 2018

View reviewed changes

mgaido91 reviewed Aug 7, 2018

View reviewed changes

mn-mikke reviewed Aug 7, 2018

View reviewed changes

codeatri added 3 commits August 7, 2018 11:34

Merge branch 'master' into SPARK-23939

12e0259

Addressed Review Commenst

5806ac4

Additional Tests containing nulls

150a6a5

codeatri added 3 commits August 8, 2018 11:57

nit: style

9f6a8ab

added additional unit test

6526630

nit:space

f7fd231

ueshin reviewed Aug 9, 2018

View reviewed changes

addressed review comments

1cbaf0c

Merge branch 'master' into SPARK-23939

bb52630

fix style

621213d

Merge master Refactoring changes

5db526b

ueshin reviewed Aug 15, 2018

View reviewed changes

review comments

e5d9b05

review comments

fb885f4

mn-mikke reviewed Aug 15, 2018

View reviewed changes

codeatri added 2 commits August 15, 2018 12:43

fix unit test

58b60b2

review

2f4943f

asfgit closed this in 5b4a38d Aug 16, 2018

mgaido91 reviewed Aug 16, 2018

View reviewed changes

gatorsmile mentioned this pull request Oct 25, 2018

[SPARK-25832][SQL][BRANCH-2.4] Revert newly added map related functions #22827

Closed

[SPARK-23939][SQL] Add transform_keys function #22013

[SPARK-23939][SQL] Add transform_keys function #22013

Conversation

codeatri commented Aug 6, 2018 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

hvanhovell commented Aug 6, 2018

SparkQA commented Aug 7, 2018

hvanhovell Aug 7, 2018 • 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

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

SparkQA commented Aug 8, 2018

SparkQA commented Aug 8, 2018

SparkQA commented Aug 8, 2018

SparkQA commented Aug 9, 2018

SparkQA commented Aug 9, 2018

codeatri commented Aug 9, 2018

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

SparkQA commented Aug 9, 2018

codeatri commented Aug 9, 2018

SparkQA commented Aug 14, 2018

SparkQA commented Aug 14, 2018

SparkQA commented Aug 15, 2018

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

ueshin commented Aug 15, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Aug 15, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Aug 15, 2018

SparkQA commented Aug 15, 2018

SparkQA commented Aug 15, 2018

ueshin commented Aug 16, 2018

ueshin commented Aug 16, 2018

ueshin commented Aug 16, 2018

mgaido91 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codeatri commented Aug 6, 2018 •

edited

Loading

hvanhovell Aug 7, 2018 •

edited

Loading