[SPARK-24991][SQL] use InternalRow in DataSourceWriter #21948

cloud-fan · 2018-08-01T19:10:37Z

What changes were proposed in this pull request?

A follow up of #21118

Since we use InternalRow in the read API of data source v2, we should do the same thing for the write API.

How was this patch tested?

existing tests.

holdensmagicalunicorn · 2018-08-01T19:10:41Z

@cloud-fan, thanks! I am a bot who has found some folks who might be able to help with the review:@gatorsmile, @zsxwing and @tdas

cloud-fan · 2018-08-01T19:12:15Z

...n/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousRateStreamSource.scala

@@ -89,8 +89,7 @@ class RateStreamContinuousReader(options: DataSourceOptions) extends ContinuousR
        start.runTimeMs,
        i,
        numPartitions,
-        perPartitionRate)
-        .asInstanceOf[InputPartition[InternalRow]]
+        perPartitionRate): InputPartition[InternalRow]


This is needed to make it compile, but at least we don't need to do cast.

Sounds good to me.

This should be in a separate commit. I didn't notice yesterday that this is for the writer until it was linked from the other issue. I think this change needs to get in, but it should not be mixed into changes for the write path.

cloud-fan · 2018-08-01T19:12:32Z

cc @rdblue @jose-torres @gatorsmile

jose-torres · 2018-08-01T21:03:24Z

lgtm

SparkQA · 2018-08-01T22:10:50Z

Test build #93898 has finished for PR 21948 at commit 852c6f3.

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

cloud-fan · 2018-08-02T03:49:09Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/writer/DataWriterFactory.java

+  /**
+   * When true, Spark will reuse the same data object instance when sending data to the data writer,
+   * for better performance. Data writers should carefully handle the data objects if it's reused,
+   * e.g. do not buffer the data objects in a list. By default it returns false for safety, data


@rdblue did you hit this problem in iceberg?

No, Iceberg assumes that data objects are reused.

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

Test build #93937 has finished for PR 21948 at commit a42e628.

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

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

Test build #93938 has finished for PR 21948 at commit 1e0cb90.

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

kiszk · 2018-08-02T07:26:16Z

retest this please

SparkQA · 2018-08-02T11:29:39Z

Test build #93974 has finished for PR 21948 at commit 1e0cb90.

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

gengliangwang · 2018-08-02T13:57:45Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2.scala


    // write the data and commit this writer.
    Utils.tryWithSafeFinallyAndFailureCallbacks(block = {
      while (iter.hasNext) {
-        dataWriter.write(iter.next())
+        // Internally Spark reuse the same UnsafeRow instance when producing output rows, here we


nit: reuse => reuses

gengliangwang · 2018-08-02T14:09:06Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/writer/DataWriterFactory.java

+
+  /**
+   * When true, Spark will reuse the same data object instance when sending data to the data writer,
+   * for better performance. Data writers should carefully handle the data objects if it's reused,


Nit: if it's reused the it here is ambiguous. Maybe change to if they are reused?

gengliangwang · 2018-08-02T14:10:17Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/writer/DataWriterFactory.java

+  /**
+   * When true, Spark will reuse the same data object instance when sending data to the data writer,
+   * for better performance. Data writers should carefully handle the data objects if it's reused,
+   * e.g. do not buffer the data objects in a list. By default it returns false for safety, data


nit: By default it returns false, data sources => By default the method returns false. Data sources

gengliangwang

LGTM otherwise.

rdblue · 2018-08-02T16:37:31Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/writer/DataWriterFactory.java

+   * sources can override it if their data writers immediately write the data object to somewhere
+   * else like a memory buffer or disk.
+   */
+  default boolean reuseDataObject() {


I don't think this should be added in this commit. This is to move to InternalRow and should not alter the API. If we want to add this, then let's discuss it in a PR for this as a feature. I'm fine documenting the default reuse behavior in this commit, though.

I think writers are responsible for defensive copies if necessary. This default is going to cause sources to be slower and I don't think it is necessary for implementations that aren't tests buffering data in memory.

In general, I think it's okay for us to have higher expectations of sources than of users. It's okay to simply document that rows are reused.

rdblue · 2018-08-02T16:43:44Z

I'm changing my +1 to -1 because read-side changes are mixed in and because copies are the responsibility of data sources if they buffer and hold references to earlier rows. At a minimum, changing that expectation should be done in a separate issue and PR.

cloud-fan · 2018-08-03T06:13:09Z

@rdblue I have documented the object reuse behavior and ask data source to handle it, please take a look, thanks!

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

Test build #94106 has finished for PR 21948 at commit 86817c7.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class MemoryWriterCommitMessage(partition: Int, data: Seq[Row])
case class MemoryWriterFactory(outputMode: OutputMode, schema: StructType)
class MemoryDataWriter(partition: Int, outputMode: OutputMode, schema: StructType)

cloud-fan · 2018-08-04T06:37:11Z

retest this please

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

Test build #94201 has finished for PR 21948 at commit 86817c7.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class MemoryWriterCommitMessage(partition: Int, data: Seq[Row])
case class MemoryWriterFactory(outputMode: OutputMode, schema: StructType)
class MemoryDataWriter(partition: Int, outputMode: OutputMode, schema: StructType)

gatorsmile · 2018-08-04T07:13:02Z

retest this please

rdblue · 2018-08-05T00:08:30Z

...n/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousRateStreamSource.scala

@@ -89,7 +89,8 @@ class RateStreamContinuousReader(options: DataSourceOptions) extends ContinuousR
        start.runTimeMs,
        i,
        numPartitions,
-        perPartitionRate): InputPartition[InternalRow]
+        perPartitionRate)
+        .asInstanceOf[InputPartition[InternalRow]]


Why is this cast necessary?

This is to address your comments: do not mix read-side changes. So I reverted it.

Okay, sorry about that. I should have looked at the whole diff.

rdblue · 2018-08-05T00:10:47Z

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/MemorySinkV2Suite.scala

-        MemoryWriterCommitMessage(0, Seq(InternalRow(1), InternalRow(2))),
-        MemoryWriterCommitMessage(1, Seq(InternalRow(3), InternalRow(4))),
-        MemoryWriterCommitMessage(2, Seq(InternalRow(6), InternalRow(7)))
+        MemoryWriterCommitMessage(0, Seq(Row(1), Row(2))),


Why was this changed back to Row?

Now the DataWriter needs to copy the input row before buffering it, which can be done by the RowEncoder when converting InternalRow to Row. Then the write message carries Rows to the driver side.

Why not use InternalRow.copy? I'd rather keep the update to InternalRow, but as long as the tests pass I wouldn't block this.

because the memory sink needs Rows at the end. Instead of collecting InternalRows via copy and then convert to Rows, I think it's more efficient to collect Rows directly.

Sounds good. Thanks for explaining your rationale.

rdblue · 2018-08-05T00:11:55Z

@cloud-fan, thanks for documenting the behavior and removing the default copy. I had a couple of questions, but I think it is close.

cloud-fan · 2018-08-05T03:03:09Z

retest this please

rdblue · 2018-08-05T04:56:10Z

+1 when tests are passing.

viirya · 2018-08-05T05:24:41Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/writer/DataWriterFactory.java

+   * Returns a data writer to do the actual writing work. Note that, Spark will reuse the same data
+   * object instance when sending data to the data writer, for better performance. Data writers
+   * are responsible for defensive copies if necessary, e.g. copy the data before buffer it in a
+   * list.


nit: the description about defensive copied in data writers, may be put in DataWriter.

I'll fix it in my next PR, thanks!

viirya · 2018-08-05T05:25:36Z

LGTM

SparkQA · 2018-08-05T06:05:05Z

Test build #94233 has finished for PR 21948 at commit 86817c7.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class MemoryWriterCommitMessage(partition: Int, data: Seq[Row])
case class MemoryWriterFactory(outputMode: OutputMode, schema: StructType)
class MemoryDataWriter(partition: Int, outputMode: OutputMode, schema: StructType)

cloud-fan · 2018-08-06T03:09:06Z

retest this please

SparkQA · 2018-08-06T06:48:02Z

Test build #94263 has finished for PR 21948 at commit 86817c7.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class MemoryWriterCommitMessage(partition: Int, data: Seq[Row])
case class MemoryWriterFactory(outputMode: OutputMode, schema: StructType)
class MemoryDataWriter(partition: Int, outputMode: OutputMode, schema: StructType)

cloud-fan · 2018-08-06T07:52:53Z

thanks, merging to master!

A follow up of apache#21118 Since we use `InternalRow` in the read API of data source v2, we should do the same thing for the write API. existing tests. Author: Wenchen Fan <wenchen@databricks.com> Closes apache#21948 from cloud-fan/row-write. (cherry picked from commit ac527b5) Conflicts: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2.scala sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousWriteRDD.scala sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/WriteToContinuousDataSourceExec.scala sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/ForeachWriterProvider.scala sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/PackedRowWriterFactory.scala sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/memoryV2.scala sql/core/src/test/scala/org/apache/spark/sql/sources/v2/DataSourceV2Suite.scala sql/core/src/test/scala/org/apache/spark/sql/sources/v2/SimpleWritableDataSource.scala

A follow up of apache#21118 Since we use `InternalRow` in the read API of data source v2, we should do the same thing for the write API. existing tests. Author: Wenchen Fan <wenchen@databricks.com> Closes apache#21948 from cloud-fan/row-write.

A follow up of apache#21118 Since we use `InternalRow` in the read API of data source v2, we should do the same thing for the write API. existing tests. Author: Wenchen Fan <wenchen@databricks.com> Closes apache#21948 from cloud-fan/row-write. (cherry picked from commit ac527b5) Conflicts: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2.scala sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousWriteRDD.scala sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/WriteToContinuousDataSourceExec.scala sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/ForeachWriterProvider.scala sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/PackedRowWriterFactory.scala sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/memoryV2.scala sql/core/src/test/scala/org/apache/spark/sql/sources/v2/DataSourceV2Suite.scala sql/core/src/test/scala/org/apache/spark/sql/sources/v2/SimpleWritableDataSource.scala

A follow up of apache#21118 Since we use `InternalRow` in the read API of data source v2, we should do the same thing for the write API. existing tests. Author: Wenchen Fan <wenchen@databricks.com> Closes apache#21948 from cloud-fan/row-write. Ref: LIHADOOP-48531 RB=1855948 G=superfriends-reviewers R=yezhou,latang,mshen,fli,zolin A=

use InternalRow in DataSourceWriter

852c6f3

cloud-fan commented Aug 1, 2018

View reviewed changes

cloud-fan mentioned this pull request Aug 1, 2018

[SPARK-24971][SQL] remove SupportsDeprecatedScanRow #21921

Closed

more work

1e0cb90

cloud-fan force-pushed the row-write branch from a42e628 to 1e0cb90 Compare August 2, 2018 03:46

cloud-fan commented Aug 2, 2018

View reviewed changes

gengliangwang reviewed Aug 2, 2018

View reviewed changes

gengliangwang approved these changes Aug 2, 2018

View reviewed changes

rdblue reviewed Aug 2, 2018

View reviewed changes

address comment

86817c7

rdblue reviewed Aug 5, 2018

View reviewed changes

viirya reviewed Aug 5, 2018

View reviewed changes

asfgit closed this in ac527b5 Aug 6, 2018

[SPARK-24991][SQL] use InternalRow in DataSourceWriter #21948

[SPARK-24991][SQL] use InternalRow in DataSourceWriter #21948

Uh oh!

Conversation

cloud-fan commented Aug 1, 2018

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

holdensmagicalunicorn commented Aug 1, 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!

cloud-fan commented Aug 1, 2018

Uh oh!

jose-torres commented Aug 1, 2018

Uh oh!

SparkQA commented Aug 1, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Aug 2, 2018

Uh oh!

SparkQA commented Aug 2, 2018

Uh oh!

kiszk commented Aug 2, 2018

Uh oh!

SparkQA commented Aug 2, 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!

gengliangwang left a comment

Choose a reason for hiding this comment

Uh oh!

rdblue Aug 2, 2018 • 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!

rdblue commented Aug 2, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cloud-fan commented Aug 3, 2018

Uh oh!

SparkQA commented Aug 3, 2018

Uh oh!

cloud-fan commented Aug 4, 2018

Uh oh!

SparkQA commented Aug 4, 2018

Uh oh!

gatorsmile commented Aug 4, 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!

rdblue commented Aug 5, 2018

Uh oh!

cloud-fan commented Aug 5, 2018

Uh oh!

rdblue Aug 2, 2018 •

edited

Loading

rdblue commented Aug 2, 2018 •

edited

Loading