This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new c2d8ee9 [SPARK-28878][SQL][FOLLOWUP] Remove extra project for DSv2
streaming scan
c2d8ee9 is described below
commit c2d8ee9c54adf4a425ce41d8743e24dd8be864c3
Author: Wenchen Fan <[email protected]>
AuthorDate: Tue Sep 10 11:01:57 2019 +0800
[SPARK-28878][SQL][FOLLOWUP] Remove extra project for DSv2 streaming scan
### What changes were proposed in this pull request?
Remove the project node if the streaming scan is columnar
### Why are the changes needed?
This is a followup of https://github.com/apache/spark/pull/25586. Batch and
streaming share the same DS v2 read API so both can support columnar reads. We
should apply #25586 to streaming scan as well.
### Does this PR introduce any user-facing change?
no
### How was this patch tested?
existing tests
Closes #25727 from cloud-fan/follow.
Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../datasources/v2/DataSourceV2Strategy.scala | 29 ++++++++++++++++------
1 file changed, 21 insertions(+), 8 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
index 7cad305..f629f36 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
@@ -155,17 +155,30 @@ object DataSourceV2Strategy extends Strategy with
PredicateHelper {
case r: StreamingDataSourceV2Relation if r.startOffset.isDefined &&
r.endOffset.isDefined =>
val microBatchStream = r.stream.asInstanceOf[MicroBatchStream]
- // ensure there is a projection, which will produce unsafe rows required
by some operators
- ProjectExec(r.output,
- MicroBatchScanExec(
- r.output, r.scan, microBatchStream, r.startOffset.get,
r.endOffset.get)) :: Nil
+ val scanExec = MicroBatchScanExec(
+ r.output, r.scan, microBatchStream, r.startOffset.get, r.endOffset.get)
+
+ val withProjection = if (scanExec.supportsColumnar) {
+ scanExec
+ } else {
+ // Add a Project here to make sure we produce unsafe rows.
+ ProjectExec(r.output, scanExec)
+ }
+
+ withProjection :: Nil
case r: StreamingDataSourceV2Relation if r.startOffset.isDefined &&
r.endOffset.isEmpty =>
val continuousStream = r.stream.asInstanceOf[ContinuousStream]
- // ensure there is a projection, which will produce unsafe rows required
by some operators
- ProjectExec(r.output,
- ContinuousScanExec(
- r.output, r.scan, continuousStream, r.startOffset.get)) :: Nil
+ val scanExec = ContinuousScanExec(r.output, r.scan, continuousStream,
r.startOffset.get)
+
+ val withProjection = if (scanExec.supportsColumnar) {
+ scanExec
+ } else {
+ // Add a Project here to make sure we produce unsafe rows.
+ ProjectExec(r.output, scanExec)
+ }
+
+ withProjection :: Nil
case WriteToDataSourceV2(writer, query) =>
WriteToDataSourceV2Exec(writer, planLater(query)) :: Nil
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
