andygrove commented on code in PR #5663:
URL: https://github.com/apache/datafusion-comet/pull/5663#discussion_r3969937961
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -1290,6 +1293,146 @@ class CometIcebergWriteActionSuite
}
}
+ // A one-byte target file size makes the rolling writer finalize a file per
batch, and a two-row
+ // Comet batch size means the task has handed several batches to the writer
before the UDF
+ // throws on id 7. iceberg-java's writer abort deletes such files; the
native path must too.
+ test("native acceleration: a failed task deletes the data files it already
finalized") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ val session = spark
+ import session.implicits._
+ (1 to 10)
+ .map(i => (i, s"r$i", i.toDouble))
+ .toDF("id", "region", "amount")
+ .coalesce(1)
+ .createOrReplaceTempView("cleanup_src")
+ spark.udf.register(
+ "boom_on_seven_cleanup",
+ (id: Int) => {
+ if (id == 7) throw new RuntimeException("boom")
+ id
+ })
+ val rollingProps = Some("'write.target-file-size-bytes'='1'")
+
+ withNativeEnabled(withSQLConf(CometConf.COMET_BATCH_SIZE.key -> "2") {
+ // Control: the same source and settings without the failure roll into
several files, so
+ // the failing run below really does have finalized files to clean up.
+ createTable(
+ warehouseDir,
+ "cleanup_control",
+ partitionSpec = "",
+ properties = rollingProps)
+ val controlPlans = capturePlans {
Review Comment:
Fixed in a34fda44c. All four were exactly as you described — this is the
merge hazard where the base moves a helper into the shared trait and the
branch's own new code still uses the old shape.
- `capturePlans(spark) { ... }` at the control-plan site, which was passing
the INSERT DataFrame into the `SparkSession` parameter.
- Restored `QueryExecution` (added to the existing
`org.apache.spark.sql.execution` import) and
`org.apache.spark.sql.util.QueryExecutionListener`, which `captureFailedPlans`
still needs even though `capturePlans` no longer does.
Verified by compiling both profiles rather than just the default:
`-Pspark-3.5` and the default 4.1 both now produce
`CometIcebergWriteActionSuite.class`. The cleanup tests themselves run green:
```
- native acceleration: a failed task deletes the data files it already
finalized
- a failed write job deletes the data files of tasks that completed
(native=true)
- a failed write job deletes the data files of tasks that completed
(native=false)
Tests: succeeded 3, failed 0
```
Separately: this branch now conflicts with `main` in
`native/core/src/execution/operators/iceberg_write.rs`. I have deliberately not
resolved that here, because it overlaps the abort-guard change in #5652 that
this is stacked behind and your open P2 there moves the disarm point. Resolving
it before that lands would mean merging a version of the guard that is about to
change. I'll take the merge once #5652 settles.
--
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]
