[SPARK-22825][SQL] Fix incorrect results of Casting Array to String #20024

maropu · 2017-12-19T16:57:42Z

What changes were proposed in this pull request?

This pr fixed the issue when casting arrays into strings;

scala> val df = spark.range(10).select('id.cast("integer")).agg(collect_list('id).as('ids))
scala> df.write.saveAsTable("t")
scala> sql("SELECT cast(ids as String) FROM t").show(false)
+------------------------------------------------------------------+
|ids                                                               |
+------------------------------------------------------------------+
|org.apache.spark.sql.catalyst.expressions.UnsafeArrayData@8bc285df|
+------------------------------------------------------------------+

This pr modified the result into;

+------------------------------+
|ids                           |
+------------------------------+
|[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]|
+------------------------------+

How was this patch tested?

Added tests in CastSuite and SQLQuerySuite.

SparkQA · 2017-12-19T17:04:21Z

Test build #85118 has finished for PR 20024 at commit ed7f885.

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

maropu · 2017-12-19T17:17:33Z

MapType has the same issue;

scala> Seq(Map(1 -> "a", 2 -> "b")).toDF("a").write.saveAsTable("t")
scala> sql("SELECT cast(a as String) FROM t").show(false)
+----------------------------------------------------------------+
|a                                                               |
+----------------------------------------------------------------+
|org.apache.spark.sql.catalyst.expressions.UnsafeMapData@38bdd75d|
+----------------------------------------------------------------+

After the approach to solve the issue is fixed, I'll also fix the MapType one in following activities.

SparkQA · 2017-12-19T20:10:26Z

Test build #85119 has finished for PR 20024 at commit 211e7f3.

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

dongjoon-hyun · 2017-12-19T23:11:11Z

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

+    val df1 = sql("SELECT CAST(ARRAY(1, 2, 3, 4) AS STRING)")
+    checkAnswer(df1, Row("[1, 2, 3, 4]"))
+    val df2 = sql("SELECT CAST(ARRAY(ARRAY(1, 2), ARRAY(3, 4, 5), ARRAY(6, 7)) AS STRING)")
+    checkAnswer(df2, Row("[WrappedArray(1, 2), WrappedArray(3, 4, 5), WrappedArray(6, 7)]"))


Hi, @maropu .
Could you put the result after this PR into PR description? So far, only before result is described.

ok, thanks!

maropu · 2017-12-20T00:31:06Z

Could you check? Thanks! @gatorsmile @viirya

maropu · 2017-12-22T05:43:08Z

ping

gatorsmile · 2017-12-22T07:27:59Z

cc @cloud-fan

cloud-fan · 2017-12-22T07:55:50Z

I feel it's not the most effcient way to cast array to string by deserializing the catalyst array to java array. Instead, I think we should have a schema-aware string casting function, i.e. using ArrayData.get to get each element, and recursively cast each element to string, and concat them with ,

maropu · 2017-12-22T08:13:57Z

ok, I'll brush up based on your suggestion.

maropu · 2018-01-03T15:26:28Z

@cloud-fan How about the current impl.? (not finished yet though)

SparkQA · 2018-01-03T15:27:55Z

Test build #85630 has finished for PR 20024 at commit edd5abf.

This patch fails to build.
This patch merges cleanly.
This patch adds the following public classes (experimental):
public class StringBuffer

maropu · 2018-01-03T15:30:16Z

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/StringBuffer.java

+ */
+public class StringBuffer {
+
+  private BufferHolder buffer;


Currently, I reused BufferHolder for the string buffer though, probably we'd better to make another buffer implementation for this purpose?

Or, just use java.lang.StringBuffer?

SparkQA · 2018-01-03T15:33:07Z

Test build #85631 has finished for PR 20024 at commit d7cf391.

This patch fails to build.
This patch merges cleanly.
This patch adds the following public classes (experimental):
public class StringBuffer

SparkQA · 2018-01-03T16:22:55Z

Test build #85632 has finished for PR 20024 at commit e77ea6b.

This patch fails to build.
This patch merges cleanly.
This patch adds the following public classes (experimental):
public class StringBuffer

SparkQA · 2018-01-03T16:37:33Z

Test build #85633 has finished for PR 20024 at commit 570f13b.

This patch fails to build.
This patch merges cleanly.
This patch adds the following public classes (experimental):
public class StringWriterBuffer

SparkQA · 2018-01-03T18:50:07Z

Test build #85634 has finished for PR 20024 at commit 91a60c3.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
public class StringWriterBuffer

SparkQA · 2018-01-03T20:39:59Z

Test build #85640 has finished for PR 20024 at commit 9e13bb9.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
public class StringWriterBuffer

cloud-fan · 2018-01-04T12:34:47Z

...alyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UTF8StringBuilder.java

+  }
+
+  public void append(String value) {
+    append(value.getBytes(StandardCharsets.UTF_8));


This should be append(UTF8String.fromString(value)), then we can remove append(byte[] value)

cloud-fan · 2018-01-04T12:34:59Z

...alyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UTF8StringBuilder.java

+    return UTF8String.fromBytes(bytes);
+  }
+
+  public int totalSize() {


This doesn't need to be public

cloud-fan · 2018-01-04T12:35:59Z

...alyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UTF8StringBuilder.java

+  public UTF8String toUTF8String() {
+    final int len = totalSize();
+    final byte[] bytes = new byte[len];
+    Platform.copyMemory(buffer, Platform.BYTE_ARRAY_OFFSET, bytes, Platform.BYTE_ARRAY_OFFSET, len);


why copy? we can do UTF8String.fromBytes(buffer, 0, totalSize)

cloud-fan · 2018-01-04T12:36:51Z

...alyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UTF8StringBuilder.java

+    cursor += value.length;
+  }
+
+  public UTF8String toUTF8String() {


nit: public UTF8String build()

cloud-fan · 2018-01-04T12:37:06Z

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

@@ -206,6 +206,23 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
    case DateType => buildCast[Int](_, d => UTF8String.fromString(DateTimeUtils.dateToString(d)))
    case TimestampType => buildCast[Long](_,
      t => UTF8String.fromString(DateTimeUtils.timestampToString(t, timeZone)))
+    case ar: ArrayType =>
+      buildCast[ArrayData](_, array => {
+        val res = new UTF8StringBuilder


nit: builder

cloud-fan · 2018-01-04T12:37:23Z

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

@@ -206,6 +206,23 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
    case DateType => buildCast[Int](_, d => UTF8String.fromString(DateTimeUtils.dateToString(d)))
    case TimestampType => buildCast[Long](_,
      t => UTF8String.fromString(DateTimeUtils.timestampToString(t, timeZone)))
+    case ar: ArrayType =>


nit: case ArrayType(et, _)

cloud-fan · 2018-01-04T13:00:16Z

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

+             |$bufferClass $bufferTerm = new $bufferClass();
+             |$writeArrayToBuffer($c, $bufferTerm);
+             |$evPrim = $bufferTerm.toUTF8String();
+           """.stripMargin


We can simplify this too

val elementToStringCode = castToStringCode(et, ctx) val funcName = ctx.freshName("elementToString") val elementToStringFunc = ctx.addNewFunction(funcName, s""" private UTF8String $funcName(${ctx.dataType(et)} element) { UTF8String elementStr = null; ${elementToStringCode("element", "elementStr", null /* resultIsNull won't be touched */)} return elementStr; } """) ... $bufferClass $bufferTerm = new $bufferClass(); $bufferTerm.append("["); if ($c.numElements > 0) { if (!$c.isNullAt(0)) { $buffer.append($elementToStringFunc(${ctx.getValue(array, et, "0")})) } for (int $loopIndex = 1; $loopIndex < $arTerm.numElements(); $loopIndex++) ... }

SparkQA · 2018-01-04T15:22:42Z

Test build #85676 has finished for PR 20024 at commit b0b3cd6.

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

cloud-fan · 2018-01-04T15:32:46Z

...alyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UTF8StringBuilder.java

+import org.apache.spark.unsafe.types.UTF8String;
+
+/**
+ * A helper class to write `UTF8String`, `String`, and `byte[]` data into an internal byte buffer


A helper class to write {@link UTF8String}s to an internal buffer and build the concatenated {@link UTF8String} at the end.

cloud-fan · 2018-01-04T15:35:57Z

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

@@ -597,6 +619,44 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
    """
  }

+  private[this] def codegenWriteArrayElemCode(et: DataType, ctx: CodegenContext): String = {


It returns a function to write the array elements, maybe a better name is: writeArrayToStringBuilderFunc

oh wait, the returned function is only called once, I think we don't need to make it a function, but just return the code, e.g.

def writeArrayToStringBuilder(ctx: CodegenContext, et: DataType, arr: String, builder: String): String

elementToString needs a function because it's called twice.

ok, I'll update soon.

cloud-fan · 2018-01-04T15:40:42Z

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

@@ -2775,4 +2773,53 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
      }
    }
  }
+
+  test("SPARK-22825 Cast array to string") {


I think the unit test is good enough, we don't need this end-to-end test.

cloud-fan · 2018-01-04T15:40:58Z

LGTM except a few minor comments

SparkQA · 2018-01-04T18:01:57Z

Test build #85681 has finished for PR 20024 at commit 09fd22e.

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

maropu · 2018-01-04T22:56:52Z

Thanks for the kindly checks in the year start and all the comments done.

SparkQA · 2018-01-05T01:35:33Z

Test build #85701 has finished for PR 20024 at commit 449e2c9.

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

maropu · 2018-01-05T01:40:07Z

retest this please

cloud-fan · 2018-01-05T02:18:41Z

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

@@ -608,6 +665,17 @@ case class Cast(child: Expression, dataType: DataType, timeZoneId: Option[String
        val tz = ctx.addReferenceObj("timeZone", timeZone)
        (c, evPrim, evNull) => s"""$evPrim = UTF8String.fromString(
          org.apache.spark.sql.catalyst.util.DateTimeUtils.timestampToString($c, $tz));"""
+      case ArrayType(et, _) =>
+        (c, evPrim, evNull) => {
+          val bufferTerm = ctx.freshName("bufferTerm")


super nit: In codegen we usually don't add a term postfix, just call it buffer, array, etc.

SparkQA · 2018-01-05T04:49:12Z

Test build #85710 has finished for PR 20024 at commit 449e2c9.

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

SparkQA · 2018-01-05T05:31:17Z

Test build #85714 has finished for PR 20024 at commit dc15b93.

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

cloud-fan · 2018-01-05T06:03:05Z

thanks, merging to master/2.3!

cloud-fan · 2018-01-05T06:03:29Z

as a follow-up, we should do the same thing for struct and map type too.

## What changes were proposed in this pull request? This pr fixed the issue when casting arrays into strings; ``` scala> val df = spark.range(10).select('id.cast("integer")).agg(collect_list('id).as('ids)) scala> df.write.saveAsTable("t") scala> sql("SELECT cast(ids as String) FROM t").show(false) +------------------------------------------------------------------+ |ids | +------------------------------------------------------------------+ |org.apache.spark.sql.catalyst.expressions.UnsafeArrayData8bc285df| +------------------------------------------------------------------+ ``` This pr modified the result into; ``` +------------------------------+ |ids | +------------------------------+ |[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]| +------------------------------+ ``` ## How was this patch tested? Added tests in `CastSuite` and `SQLQuerySuite`. Author: Takeshi Yamamuro <[email protected]> Closes #20024 from maropu/SPARK-22825. (cherry picked from commit 52fc5c1) Signed-off-by: Wenchen Fan <[email protected]>

maropu · 2018-01-05T06:40:21Z

Thanks! I'll do.

maropu force-pushed the SPARK-22825 branch from ed7f885 to 211e7f3 Compare December 19, 2017 17:08

dongjoon-hyun reviewed Dec 19, 2017

View reviewed changes

maropu force-pushed the SPARK-22825 branch from edd5abf to d7cf391 Compare January 3, 2018 15:25

maropu commented Jan 3, 2018

View reviewed changes

Fix a bug

d7fd8cd

maropu force-pushed the SPARK-22825 branch from d7cf391 to e77ea6b Compare January 3, 2018 16:12

maropu force-pushed the SPARK-22825 branch 2 times, most recently from 570f13b to 91a60c3 Compare January 3, 2018 16:35

maropu force-pushed the SPARK-22825 branch 2 times, most recently from ca6519e to 9e13bb9 Compare January 3, 2018 18:05

Fix

91df078

maropu force-pushed the SPARK-22825 branch 2 times, most recently from 3f075d8 to c2e6757 Compare January 4, 2018 02:06

Drop StringWriterBuffer

8705f84

maropu force-pushed the SPARK-22825 branch from c2e6757 to 8705f84 Compare January 4, 2018 02:11

cloud-fan reviewed Jan 4, 2018

View reviewed changes

Fix

09fd22e

cloud-fan reviewed Jan 4, 2018

View reviewed changes

Fix

449e2c9

maropu force-pushed the SPARK-22825 branch from ee0f1d3 to 449e2c9 Compare January 4, 2018 22:55

cloud-fan reviewed Jan 5, 2018

View reviewed changes

Drop Term

dc15b93

asfgit closed this in 52fc5c1 Jan 5, 2018

[SPARK-22825][SQL] Fix incorrect results of Casting Array to String #20024

[SPARK-22825][SQL] Fix incorrect results of Casting Array to String #20024

Conversation

maropu commented Dec 19, 2017 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Dec 19, 2017

maropu commented Dec 19, 2017

SparkQA commented Dec 19, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

maropu commented Dec 20, 2017 • edited Loading

maropu commented Dec 22, 2017

gatorsmile commented Dec 22, 2017

cloud-fan commented Dec 22, 2017

maropu commented Dec 22, 2017

maropu commented Jan 3, 2018

SparkQA commented Jan 3, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jan 3, 2018

SparkQA commented Jan 3, 2018

SparkQA commented Jan 3, 2018

SparkQA commented Jan 3, 2018

SparkQA commented Jan 3, 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

SparkQA commented Jan 4, 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

cloud-fan commented Jan 4, 2018

SparkQA commented Jan 4, 2018

maropu commented Jan 4, 2018

SparkQA commented Jan 5, 2018

maropu commented Jan 5, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jan 5, 2018

SparkQA commented Jan 5, 2018

cloud-fan commented Jan 5, 2018

cloud-fan commented Jan 5, 2018

maropu commented Jan 5, 2018

maropu commented Dec 19, 2017 •

edited

Loading

maropu commented Dec 20, 2017 •

edited

Loading