[SPARK-23821][SQL] Collection function: flatten #20938

mn-mikke · 2018-03-29T17:19:39Z

What changes were proposed in this pull request?

This PR adds a new collection function that transforms an array of arrays into a single array. The PR comprises:

An expression for flattening array structure
Flatten function
A wrapper for PySpark

How was this patch tested?

New tests added into:

CollectionExpressionsSuite
DataFrameFunctionsSuite

Codegen examples

Primitive type

val df = Seq(
  Seq(Seq(1, 2), Seq(4, 5)),
  Seq(null, Seq(1))
).toDF("i")
df.filter($"i".isNotNull || $"i".isNull).select(flatten($"i")).debugCodegen

Result:

/* 033 */         boolean inputadapter_isNull = inputadapter_row.isNullAt(0);
/* 034 */         ArrayData inputadapter_value = inputadapter_isNull ?
/* 035 */         null : (inputadapter_row.getArray(0));
/* 036 */
/* 037 */         boolean filter_value = true;
/* 038 */
/* 039 */         if (!(!inputadapter_isNull)) {
/* 040 */           filter_value = inputadapter_isNull;
/* 041 */         }
/* 042 */         if (!filter_value) continue;
/* 043 */
/* 044 */         ((org.apache.spark.sql.execution.metric.SQLMetric) references[0] /* numOutputRows */).add(1);
/* 045 */
/* 046 */         boolean project_isNull = inputadapter_isNull;
/* 047 */         ArrayData project_value = null;
/* 048 */
/* 049 */         if (!inputadapter_isNull) {
/* 050 */           for (int z = 0; !project_isNull && z < inputadapter_value.numElements(); z++) {
/* 051 */             project_isNull |= inputadapter_value.isNullAt(z);
/* 052 */           }
/* 053 */           if (!project_isNull) {
/* 054 */             long project_numElements = 0;
/* 055 */             for (int z = 0; z < inputadapter_value.numElements(); z++) {
/* 056 */               project_numElements += inputadapter_value.getArray(z).numElements();
/* 057 */             }
/* 058 */             if (project_numElements > 2147483632) {
/* 059 */               throw new RuntimeException("Unsuccessful try to flatten an array of arrays with " +
/* 060 */                 project_numElements + " elements due to exceeding the array size limit 2147483632.");
/* 061 */             }
/* 062 */
/* 063 */             long project_size = UnsafeArrayData.calculateSizeOfUnderlyingByteArray(
/* 064 */               project_numElements,
/* 065 */               4);
/* 066 */             if (project_size > 2147483632) {
/* 067 */               throw new RuntimeException("Unsuccessful try to flatten an array of arrays with " +
/* 068 */                 project_size + " bytes of data due to exceeding the limit 2147483632" +
/* 069 */                 " bytes for UnsafeArrayData.");
/* 070 */             }
/* 071 */
/* 072 */             byte[] project_array = new byte[(int)project_size];
/* 073 */             UnsafeArrayData project_tempArrayData = new UnsafeArrayData();
/* 074 */             Platform.putLong(project_array, 16, project_numElements);
/* 075 */             project_tempArrayData.pointTo(project_array, 16, (int)project_size);
/* 076 */             int project_counter = 0;
/* 077 */             for (int k = 0; k < inputadapter_value.numElements(); k++) {
/* 078 */               ArrayData arr = inputadapter_value.getArray(k);
/* 079 */               for (int l = 0; l < arr.numElements(); l++) {
/* 080 */                 if (arr.isNullAt(l)) {
/* 081 */                   project_tempArrayData.setNullAt(project_counter);
/* 082 */                 } else {
/* 083 */                   project_tempArrayData.setInt(
/* 084 */                     project_counter,
/* 085 */                     arr.getInt(l)
/* 086 */                   );
/* 087 */                 }
/* 088 */                 project_counter++;
/* 089 */               }
/* 090 */             }
/* 091 */             project_value = project_tempArrayData;
/* 092 */
/* 093 */           }
/* 094 */
/* 095 */         }

Non-primitive type

val df = Seq(
  Seq(Seq("a", "b"), Seq(null, "d")),
  Seq(null, Seq("a"))
).toDF("s")
df.filter($"s".isNotNull || $"s".isNull).select(flatten($"s")).debugCodegen

Result:

/* 033 */         boolean inputadapter_isNull = inputadapter_row.isNullAt(0);
/* 034 */         ArrayData inputadapter_value = inputadapter_isNull ?
/* 035 */         null : (inputadapter_row.getArray(0));
/* 036 */
/* 037 */         boolean filter_value = true;
/* 038 */
/* 039 */         if (!(!inputadapter_isNull)) {
/* 040 */           filter_value = inputadapter_isNull;
/* 041 */         }
/* 042 */         if (!filter_value) continue;
/* 043 */
/* 044 */         ((org.apache.spark.sql.execution.metric.SQLMetric) references[0] /* numOutputRows */).add(1);
/* 045 */
/* 046 */         boolean project_isNull = inputadapter_isNull;
/* 047 */         ArrayData project_value = null;
/* 048 */
/* 049 */         if (!inputadapter_isNull) {
/* 050 */           for (int z = 0; !project_isNull && z < inputadapter_value.numElements(); z++) {
/* 051 */             project_isNull |= inputadapter_value.isNullAt(z);
/* 052 */           }
/* 053 */           if (!project_isNull) {
/* 054 */             long project_numElements = 0;
/* 055 */             for (int z = 0; z < inputadapter_value.numElements(); z++) {
/* 056 */               project_numElements += inputadapter_value.getArray(z).numElements();
/* 057 */             }
/* 058 */             if (project_numElements > 2147483632) {
/* 059 */               throw new RuntimeException("Unsuccessful try to flatten an array of arrays with " +
/* 060 */                 project_numElements + " elements due to exceeding the array size limit 2147483632.");
/* 061 */             }
/* 062 */
/* 063 */             Object[] project_arrayObject = new Object[(int)project_numElements];
/* 064 */             int project_counter = 0;
/* 065 */             for (int k = 0; k < inputadapter_value.numElements(); k++) {
/* 066 */               ArrayData arr = inputadapter_value.getArray(k);
/* 067 */               for (int l = 0; l < arr.numElements(); l++) {
/* 068 */                 project_arrayObject[project_counter] = arr.getUTF8String(l);
/* 069 */                 project_counter++;
/* 070 */               }
/* 071 */             }
/* 072 */             project_value = new org.apache.spark.sql.catalyst.util.GenericArrayData(project_arrayObject);
/* 073 */
/* 074 */           }
/* 075 */
/* 076 */         }

HyukjinKwon · 2018-03-30T07:01:12Z

python/pyspark/sql/functions.py

+
+    :param col: name of column or expression
+
+    >>> df = spark.createDataFrame([([[1, 2, 3], [4, 5], [6]],),([None, [4, 5]],)], ['data'])


Quick note: ,),( -> ,), (

gatorsmile · 2018-03-30T16:44:56Z

Thanks for your contribution! Try to improve your test cases by reading the other open source code (e.g., this)?

gatorsmile · 2018-03-30T16:45:02Z

ok to test

SparkQA · 2018-03-30T18:24:54Z

Test build #88758 has finished for PR 20938 at commit 755d6db.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class Flatten(child: Expression) extends UnaryExpression

mn-mikke · 2018-04-02T22:17:48Z

Rewrote test cases. @gatorsmile Please let me know if it's OK.

SparkQA · 2018-04-03T03:18:08Z

Test build #88836 has finished for PR 20938 at commit ad46962.

This patch fails from timeout after a configured wait of `300m`.
This patch merges cleanly.
This patch adds no public classes.

mn-mikke · 2018-04-03T07:50:27Z

retest this please

SparkQA · 2018-04-03T19:00:35Z

Test build #88842 has finished for PR 20938 at commit eeab727.

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

mn-mikke · 2018-04-08T18:34:28Z

Any other comments?

HyukjinKwon · 2018-04-09T01:08:47Z

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

+    if (
+      ArrayType.acceptsType(child.dataType) &&
+      ArrayType.acceptsType(child.dataType.asInstanceOf[ArrayType].elementType)
+    ) {


How about this?

child.dataType match { case _: ArrayType(_: ArrayType, _) => TypeCheckResult.TypeCheckSuccess case _: => TypeCheckResult.TypeCheckFailure( "The argument should be an array of arrays, " + s"but '${child.sql}' is of ${child.dataType.simpleString} type.") }

HyukjinKwon · 2018-04-09T01:11:00Z

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

+  override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    nullSafeCodeGen(ctx, ev, c => {
+      val code =
+        if (CodeGenerator.isPrimitiveType(dataType.elementType)) {


very tiny nit: shall we move this line up?

val code = if (Code...

HyukjinKwon · 2018-04-09T01:11:23Z

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

+       |if(!${ev.isNull}) {
+       |  $coreLogic
+       |}
+       """.stripMargin


indentation

HyukjinKwon · 2018-04-09T01:12:34Z

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

+   * @group collection_funcs
+   * @since 2.4.0
+   */
+  def flatten(e: Column): Column = withExpr{ Flatten(e.expr) }


r{ -> r {

HyukjinKwon · 2018-04-09T01:13:12Z

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

+    checkAnswer(
+      oneRowDF.selectExpr("flatten(array(arr, array(null, 5), array(6, null)))"),
+      Seq(Row(Seq(1, 2, 3, null, 5, 6, null)))
+    )


Let's move it up.

HyukjinKwon · 2018-04-09T01:14:21Z

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

+    Examples:
+      > SELECT _FUNC_(array(array(1, 2), array(3, 4));
+       [1,2,3,4]
+  """)


""", since = "2.4.0")

HyukjinKwon · 2018-04-09T01:18:42Z

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

+    val elements = array.asInstanceOf[ArrayData].toObjectArray(dataType)
+
+    if (elements.contains(null)) {
+      null


does this mean if input array has null in the elements, return null ignoring other elements when we are not in codegen?

Yes, you are right. The function also behaves the same way when codegen is applied. See test cases with a null array in CollectionExpressionsSuite.

We can discuss whether the function should behave the same way as in Presto and just ignore null elements... But I think that the current approach fits more into the semantics of Spark functions.

concat("a",null,"c") => null 1 + null => null ...

SparkQA · 2018-04-09T14:07:05Z

Test build #89055 has finished for PR 20938 at commit a50d42e.

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

mn-mikke · 2018-04-09T14:49:33Z

Can't reproduce it locally and seems to unrelated...

mn-mikke · 2018-04-09T14:50:02Z

retest this please

SparkQA · 2018-04-10T09:39:59Z

Test build #89093 has finished for PR 20938 at commit e213341.

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

HyukjinKwon · 2018-04-10T09:54:10Z

retest this please

SparkQA · 2018-04-10T12:37:10Z

Test build #89104 has finished for PR 20938 at commit e213341.

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

mn-mikke · 2018-04-10T12:51:29Z

Any idea why those tests are failing?

gatorsmile · 2018-04-10T15:36:16Z

cc @ueshin

SparkQA · 2018-04-10T16:47:29Z

Test build #89119 has finished for PR 20938 at commit b9d99f7.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
sealed trait Node extends Serializable
sealed trait ClassificationNode extends Node
sealed trait RegressionNode extends Node
sealed trait LeafNode extends Node
sealed trait InternalNode extends Node
case class ExprCode(var code: String, var isNull: ExprValue, var value: ExprValue)
case class SubExprEliminationState(isNull: ExprValue, value: ExprValue)
abstract class ExprValue
class LiteralValue(val value: String, val javaType: String) extends ExprValue
case class VariableValue(
case class StatementValue(
case class GlobalValue(val value: String, val javaType: String) extends ExprValue

…atten-to-master

SparkQA · 2018-04-13T00:52:31Z

Test build #89303 has finished for PR 20938 at commit 0e0def4.

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

ueshin · 2018-04-13T09:28:55Z

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

+      coreLogic: String): String = {
+    s"""
+    |for(int z=0; z < $childVariableName.numElements(); z++) {
+    |  ${ev.isNull} |= $childVariableName.isNullAt(z);


How about breaking when null is found?

ueshin · 2018-04-13T10:03:49Z

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

+  since = "2.4.0")
+case class Flatten(child: Expression) extends UnaryExpression {
+
+  override def nullable: Boolean = child.nullable || dataType.containsNull


child.nullable || child.dataType.asInstanceOf[ArrayType].containsNull?

ueshin · 2018-04-13T10:17:51Z

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

+      val code = if (CodeGenerator.isPrimitiveType(dataType.elementType)) {
+          genCodeForConcatOfPrimitiveElements(ctx, c, ev.value)
+        } else {
+          genCodeForConcatOfComplexElements(ctx, c, ev.value)


I'm wondering if we say "complex" for non-primitive types?

ueshin · 2018-04-13T10:24:59Z

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

+  override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    nullSafeCodeGen(ctx, ev, c => {
+      val code = if (CodeGenerator.isPrimitiveType(dataType.elementType)) {
+          genCodeForConcatOfPrimitiveElements(ctx, c, ev.value)


nit: indent.

kiszk · 2018-04-13T10:43:49Z

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

+    if (elements.contains(null)) {
+      null
+    } else {
+      val flattened = elements.flatMap(


Do we need size check whether the total number of array elements is less than or equal to possible max array size?

I agree, especially in combination with this comment, provided that the resulted array length is known in advance. flatMap can then be replaced with a simple loop copying chanks of data into a preallocated array.

I've been searching for a well-defined constant indicating the VM limit for array size. It seems that the limit is platform-dependent... Any idea how to get the limit for a given platform?

kiszk · 2018-04-13T10:44:22Z

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

+    s"""
+    |$numElemCode
+    |$unsafeArraySizeInBytes
+    |byte[] $arrayName = new byte[$arraySizeName];


Do we need size check whether the total number of array elements is less than or equal to possible max array size?
If we could use long[], we can accept more array elements.

You made a really good point about checking the total number of array elements!

Re: long[] - It seems that UnsafeArrayData is not currently ready for that. It would require a bigger refactoring... In theory, we could push limits even further. If implemented UnsafeArrayData in a similar way like Scala Vectors but with leaves represented as byte[MAX_SIZE], the only limits would be the heap size and computing power. But is there any real case scenario where we needed to store more than 2GB into one record?

Sorry for late comment. I think that it is fine to use byte[] for now. It is just a possibly choice to use long[].

SparkQA · 2018-04-18T20:08:46Z

Test build #89518 has finished for PR 20938 at commit 9081291.

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

mn-mikke · 2018-04-18T20:36:02Z

retest this please

SparkQA · 2018-04-19T13:07:58Z

Test build #89559 has finished for PR 20938 at commit f11aa7b.

This patch fails from timeout after a configured wait of `300m`.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ArrayPosition(left: Expression, right: Expression)

ueshin · 2018-04-19T16:40:43Z

Jenkins, retest this please.

SparkQA · 2018-04-19T17:48:07Z

Test build #89574 has finished for PR 20938 at commit 88c4971.

This patch fails from timeout after a configured wait of `300m`.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ElementAt(left: Expression, right: Expression) extends GetMapValueUtil
abstract class GetMapValueUtil extends BinaryExpression with ImplicitCastInputTypes
case class GetMapValue(child: Expression, key: Expression)

SparkQA · 2018-04-19T17:53:00Z

Test build #89575 has finished for PR 20938 at commit 37b68cd.

This patch fails from timeout after a configured wait of `300m`.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ElementAt(left: Expression, right: Expression) extends GetMapValueUtil

SparkQA · 2018-04-19T20:50:39Z

Test build #89582 has finished for PR 20938 at commit 37b68cd.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ElementAt(left: Expression, right: Expression) extends GetMapValueUtil

ueshin · 2018-04-20T08:58:20Z

LGTM pending Jenkins.

SparkQA · 2018-04-20T13:22:59Z

Test build #89620 has finished for PR 20938 at commit 508fee0.

This patch fails from timeout after a configured wait of `300m`.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class Concat(children: Seq[Expression]) extends Expression

ueshin · 2018-04-20T13:27:43Z

Jenkins, retest this please.

…atten-to-master

SparkQA · 2018-04-20T18:33:06Z

Test build #89631 has finished for PR 20938 at commit 939fc23.

This patch fails from timeout after a configured wait of `300m`.
This patch merges cleanly.
This patch adds no public classes.

kiszk · 2018-04-20T19:18:35Z

retest this please

SparkQA · 2018-04-20T23:04:02Z

Test build #89658 has finished for PR 20938 at commit 939fc23.

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

maropu · 2018-04-24T06:24:07Z

@ueshin you forget to merge this? ;)

HyukjinKwon · 2018-04-24T08:33:26Z

retest this please

SparkQA · 2018-04-24T11:16:56Z

Test build #89775 has finished for PR 20938 at commit 939fc23.

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

maropu · 2018-04-24T11:39:06Z

retest this please

SparkQA · 2018-04-24T13:39:03Z

Test build #89783 has finished for PR 20938 at commit 939fc23.

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

HyukjinKwon · 2018-04-24T15:10:30Z

retest this please

SparkQA · 2018-04-24T19:02:54Z

Test build #89788 has finished for PR 20938 at commit 939fc23.

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

ueshin · 2018-04-25T02:18:29Z

I'm sorry for the delay.
Thanks! merging to master.

[SPARK-23821][SQL] Collection function: flatten

755d6db

HyukjinKwon reviewed Mar 30, 2018

View reviewed changes

[SPARK-23821][SQL] Improving test cases

ad46962

[SPARK-23821][SQL] Merging the current master into the feature branch.

eeab727

HyukjinKwon reviewed Apr 9, 2018

View reviewed changes

[SPARK-23821][SQL] Code-styling improvements

a50d42e

[SPARK-23821][SQL] Merging current master to the feature branch

e213341

[SPARK-23821][SQL] Merging current master to the feature branch

b9d99f7

Merge remote-tracking branch 'spark/master' into feature/array-api-fl…

0e0def4

…atten-to-master

ueshin reviewed Apr 13, 2018

View reviewed changes

kiszk reviewed Apr 13, 2018

View reviewed changes

[SPARK-23821][SQL] Merging current master to the feature branch.

9081291

mn-mikke added 3 commits April 19, 2018 09:59

[SPARK-23821][SQL] Merging current master to the feature branch.

f11aa7b

[SPARK-23821][SQL] Merging current master to the feature branch.

88c4971

[SPARK-23821][SQL] Small refactoring

37b68cd

[SPARK-23821][SQL] Merging current master to the feature branch.

508fee0

Merge remote-tracking branch 'spark/master' into feature/array-api-fl…

939fc23

…atten-to-master

asfgit closed this in 5fea17b Apr 25, 2018


		:param col: name of column or expression

		>>> df = spark.createDataFrame([([[1, 2, 3], [4, 5], [6]],),([None, [4, 5]],)], ['data'])

[SPARK-23821][SQL] Collection function: flatten #20938

[SPARK-23821][SQL] Collection function: flatten #20938

Uh oh!

Conversation

mn-mikke commented Mar 29, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Codegen examples

Primitive type

Non-primitive type

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gatorsmile commented Mar 30, 2018

Uh oh!

gatorsmile commented Mar 30, 2018

Uh oh!

SparkQA commented Mar 30, 2018

Uh oh!

mn-mikke commented Apr 2, 2018

Uh oh!

SparkQA commented Apr 3, 2018

Uh oh!

mn-mikke commented Apr 3, 2018

Uh oh!

SparkQA commented Apr 3, 2018

Uh oh!

mn-mikke commented Apr 8, 2018

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 Apr 9, 2018

Uh oh!

mn-mikke commented Apr 9, 2018

Uh oh!

mn-mikke commented Apr 9, 2018

Uh oh!

SparkQA commented Apr 10, 2018

Uh oh!

HyukjinKwon commented Apr 10, 2018

Uh oh!

SparkQA commented Apr 10, 2018

Uh oh!

mn-mikke commented Apr 10, 2018

Uh oh!

gatorsmile commented Apr 10, 2018

Uh oh!

SparkQA commented Apr 10, 2018

Uh oh!

SparkQA commented Apr 13, 2018

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!

kiszk Apr 13, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

mn-mikke commented Mar 29, 2018 •

edited

Loading

kiszk Apr 13, 2018 •

edited

Loading

wajda Apr 13, 2018 •

edited

Loading

kiszk Apr 13, 2018 •

edited

Loading

mn-mikke Apr 13, 2018 •

edited

Loading

kiszk Apr 18, 2018 •

edited

Loading

maropu commented Apr 24, 2018 •

edited

Loading