andygrove commented on code in PR #6243:
URL: https://github.com/apache/datafusion-comet/pull/6243#discussion_r4184291801
##########
spark/src/main/scala/org/apache/comet/CometExecIterator.scala:
##########
@@ -247,6 +257,12 @@ class CometExecIterator(
result
} catch {
+ // A JVM input threw while native pulled a batch from it. Native saw
only the text of that
+ // exception and failed with a CometNativeException built from it, so
rethrow the exception
+ // itself, which is what the task would have thrown without Comet.
+ case _: Throwable if inputFailure.isDefined =>
Review Comment:
Done in 5c54ecd46. The lookup is an `InputFailure` extractor now, so it runs
once per failure, and the comment at the catch site is a single line, since
`CometArrowStream.inputFailure` carries the explanation.
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -1494,6 +1494,62 @@ class CometIcebergWriteActionSuite
}
}
+ // https://github.com/apache/datafusion-comet/issues/6234. The native writer
pulls its input
+ // through an Arrow C stream, and Arrow Java hands native only the text of
an exception thrown
+ // while producing a batch, so the user got a CometNativeException instead
of Spark's exception.
+ // The first batch is read on the JVM to derive the stream's schema, which
is why the overflow
+ // has to land past it: row 90000 of a single 100000-row data file, so one
task reads it all.
+ test("native acceleration: an input error past the first batch fails like
the JVM writer") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ val (nativeTable, jvmTable) = ("overflow_native", "overflow_jvm")
+ Seq("overflow_src", nativeTable, jvmTable).foreach { table =>
+ spark.sql(s"CREATE TABLE $catalog.$ns.$table (v INT) USING iceberg")
+ }
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .range(0, 100000, 1, 1)
+ .selectExpr("CAST(id AS INT) AS v")
+ .writeTo(s"$catalog.$ns.overflow_src")
+ .append()
+ }
+ def insert(table: String): Unit =
+ spark.sql(
+ s"INSERT INTO $catalog.$ns.$table " +
+ s"SELECT v + ${Int.MaxValue - 90000} FROM
$catalog.$ns.overflow_src")
+
+ var nativeFailure: Throwable = null
Review Comment:
Added `failedWrites(nativeTable, jvmTable)(write)`, which does the
`CometIcebergWriteExec` check and returns `(nativeFailure, jvmFailure)`, and
moved the two clustered-writer tests onto it as well (5c54ecd46).
Yes, this one is mainly the end-to-end repro from #6234. The
`TakeOrderedAndProjectExec` test reaches the same stream, but this is the only
one that also checks that the failed write commits nothing.
--
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]
