jzhuge commented on a change in pull request #30806:
URL: https://github.com/apache/spark/pull/30806#discussion_r544832653
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
##########
@@ -188,15 +189,20 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
orCreate = orCreate) :: Nil
}
- case AppendData(r: DataSourceV2Relation, query, writeOptions, _) =>
+ case AppendData(r: DataSourceV2Relation, query, writeOptions, _, write) =>
r.table.asWritable match {
case v1 if v1.supports(TableCapability.V1_BATCH_WRITE) =>
- AppendDataExecV1(v1, writeOptions.asOptions, query, refreshCache(r))
:: Nil
+ AppendDataExecV1(
+ v1, writeOptions.asOptions, query,
+ refreshCache(r), write.get.asInstanceOf[V1Write]) :: Nil
Review comment:
Possible to avoid instance cast? See my suggestion above.
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
##########
@@ -188,15 +189,20 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
orCreate = orCreate) :: Nil
}
- case AppendData(r: DataSourceV2Relation, query, writeOptions, _) =>
+ case AppendData(r: DataSourceV2Relation, query, writeOptions, _, write) =>
Review comment:
Is `write` guaranteed not be `None`?
How about rewriting this case as follows?
```
case AppendData(r @ DataSourceV2Relation(v1: SupportsWrite, _, _, _, _),
query, writeOptions,
_, Some(v1Write: V1Write)) if
v1.supports(TableCapability.V1_BATCH_WRITE) =>
AppendDataExecV1(v1, writeOptions.asOptions, query, refreshCache(r),
v1Write) :: Nil
case AppendData(r @ DataSourceV2Relation(v2: SupportsWrite, _, _, _, _),
query, writeOptions, _, Some(write)) =>
AppendDataExec(v2, writeOptions.asOptions, planLater(query),
refreshCache(r), write) :: Nil
```
It is not exactly the same as the existing code. Some unmatched cases (not
sure how many or if any) will fall through. Exception will be thrown later,
instead of right here upon instance cast or Option.get.
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
##########
@@ -477,9 +450,10 @@ private[v2] trait TableWriteExecHelper extends
V2TableWriteExec with SupportsV1W
writeOptions)
val writeBuilder = table.newWriteBuilder(info)
- val writtenRows = writeBuilder match {
- case v1: V1WriteBuilder => writeWithV1(v1.buildForV1Write())
- case v2 => writeWithV2(v2.buildForBatch())
+ val write = writeBuilder.build()
+ val writtenRows = write match {
Review comment:
Nit: merge line 451-454 into:
```
val writtenRows = table.newWriteBuilder(info).build() match {
```
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V1FallbackWriters.scala
##########
@@ -115,14 +80,10 @@ sealed trait V1FallbackWriters extends V2CommandExec with
SupportsV1Write {
trait SupportsV1Write extends SparkPlan {
def plan: LogicalPlan
- protected def writeWithV1(
- relation: InsertableRelation,
- refreshCache: () => Unit = () => ()): Seq[InternalRow] = {
+ protected def writeWithV1(relation: InsertableRelation): Seq[InternalRow] = {
Review comment:
Nicely simplified
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala
##########
@@ -188,15 +189,20 @@ class DataSourceV2Strategy(session: SparkSession) extends
Strategy with Predicat
orCreate = orCreate) :: Nil
}
- case AppendData(r: DataSourceV2Relation, query, writeOptions, _) =>
+ case AppendData(r: DataSourceV2Relation, query, writeOptions, _, write) =>
r.table.asWritable match {
case v1 if v1.supports(TableCapability.V1_BATCH_WRITE) =>
- AppendDataExecV1(v1, writeOptions.asOptions, query, refreshCache(r))
:: Nil
+ AppendDataExecV1(
+ v1, writeOptions.asOptions, query,
+ refreshCache(r), write.get.asInstanceOf[V1Write]) :: Nil
case v2 =>
- AppendDataExec(v2, writeOptions.asOptions, planLater(query),
refreshCache(r)) :: Nil
+ AppendDataExec(
+ v2, writeOptions.asOptions, planLater(query),
+ refreshCache(r), write.get) :: Nil
Review comment:
Possible to avoid Option.get? See my suggestion above.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
