viirya commented on code in PR #38005:
URL: https://github.com/apache/spark/pull/38005#discussion_r1081796690
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala:
##########
@@ -477,6 +507,73 @@ object DataWritingSparkTask extends Logging {
}
}
+object DataWritingSparkTask extends WritingSparkTask[DataWriter[InternalRow]] {
+ override protected def write(writer: DataWriter[InternalRow], row:
InternalRow): Unit = {
+ writer.write(row)
+ }
+}
+
+case class DeltaWritingSparkTask(
+ projections: WriteDeltaProjections) extends
WritingSparkTask[DeltaWriter[InternalRow]] {
+
+ private lazy val rowProjection = projections.rowProjection.orNull
+ private lazy val rowIdProjection = projections.rowIdProjection
+
+ override protected def write(writer: DeltaWriter[InternalRow], row:
InternalRow): Unit = {
+ val operation = row.getInt(0)
+
+ operation match {
+ case DELETE_OPERATION =>
+ rowIdProjection.project(row)
+ writer.delete(null, rowIdProjection)
+
+ case UPDATE_OPERATION =>
+ rowProjection.project(row)
+ rowIdProjection.project(row)
+ writer.update(null, rowIdProjection, rowProjection)
+
+ case INSERT_OPERATION =>
+ rowProjection.project(row)
+ writer.insert(rowProjection)
Review Comment:
I can see `DELETE_OPERATION` is used in `buildWriteDeltaPlan`.
`UPDATE_OPERATION` and `INSERT_OPERATION` are defined together in this PR
too. But seems I don't find where they are used in query plan?
--
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.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
