comphead commented on code in PR #7536:
URL: https://github.com/apache/datafusion-comet/pull/7536#discussion_r4532980821
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergWriteExec.scala:
##########
@@ -230,29 +231,41 @@ case class CometIcebergWriteExec(
override def doExecuteColumnar(): RDD[ColumnarBatch] = {
val schedulerProbe = IcebergSchedulerTestProbe.current
- val inputRDD = if (child.supportsColumnar) {
- child.executeColumnar()
- } else {
- throw new UnsupportedOperationException(
- "CometIcebergWriteExec requires a columnar (Comet native) child; got "
+
- child.getClass.getName)
+ // One Arrow stream per partition, which native consumes through the C
Stream Interface.
+ val streamRDD: RDD[ArrowArrayStream] = child match {
+ // Native reads the source's Arrow stream, as it does for a native
operator's input. The
+ // source reuses its vectors for the next batch, and wrapping its
`executeColumnar` batches
+ // below closes each batch once native has it; closing a struct vector
drops its children,
+ // so the next batch would lose them (#7239). The scheduler probe counts
rows as batches.
+ case source: CometNativeArrowSource if schedulerProbe.isEmpty =>
Review Comment:
As the description says, on `main` this failure no longer happens since
#7325, so the comment describes the release branches only. Could it say why the
stream route is still preferred on `main` (it skips the per-batch
`VectorSchemaRoot` re-wrap and unload/load in `readerBatchIter` and
`ColumnarBatchArrowReader`) and point to #7239 for the older branches?
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergWriteExec.scala:
##########
@@ -230,29 +231,41 @@ case class CometIcebergWriteExec(
override def doExecuteColumnar(): RDD[ColumnarBatch] = {
val schedulerProbe = IcebergSchedulerTestProbe.current
- val inputRDD = if (child.supportsColumnar) {
- child.executeColumnar()
- } else {
- throw new UnsupportedOperationException(
- "CometIcebergWriteExec requires a columnar (Comet native) child; got "
+
- child.getClass.getName)
+ // One Arrow stream per partition, which native consumes through the C
Stream Interface.
+ val streamRDD: RDD[ArrowArrayStream] = child match {
+ // Native reads the source's Arrow stream, as it does for a native
operator's input. The
+ // source reuses its vectors for the next batch, and wrapping its
`executeColumnar` batches
+ // below closes each batch once native has it; closing a struct vector
drops its children,
+ // so the next batch would lose them (#7239). The scheduler probe counts
rows as batches.
+ case source: CometNativeArrowSource if schedulerProbe.isEmpty =>
Review Comment:
With `schedulerProbe.isEmpty` in the match, every test that installs the
probe over a VALUES or RDD write (`CometLocalTableScanExec` and
`CometSparkToColumnarExec` are both `CometNativeArrowSource`) now takes the
wrapped route. So the speculation, executor-loss and scheduler failure tests no
longer exercise the route production uses for those children. Which suites
cover failure and cleanup on the stream route, for example the handoff
failpoint tests, which do not install the probe?
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -412,6 +413,45 @@ class CometIcebergWriteActionSuite
}
}
+ // https://github.com/apache/datafusion-comet/issues/7239. A conversion of
Spark rows writes
+ // every batch into the same vectors, and closing a struct vector drops its
children. The write
+ // reads the conversion's Arrow stream, as native shuffle does, rather than
batches it would
+ // close once native had them, which left the second batch of a partition
without them.
+ test("native write reads a struct column that a conversion writes over
several batches") {
Review Comment:
On `main` the old route already hands out independent vectors since #7325,
so I expect this test to pass there with and without the change (I have not run
it). It guards the release branches, so could the backport PRs confirm it fails
there without the fix? The #6689 test also ran a local table scan
(`CometLocalTableScanExec`), which reaches the same branch through the trait.
Looping over both, with more than one partition, would cover the new case fully.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
