andygrove opened a new issue, #5689:
URL: https://github.com/apache/datafusion-comet/issues/5689
### Describe the bug
With `spark.comet.iceberg.write.enabled=true` and adaptive query execution
disabled, every Iceberg copy-on-write row-level operation (DELETE, UPDATE,
MERGE) fails with:
```
java.lang.ClassCastException: class
org.apache.spark.sql.vectorized.ColumnarBatch
cannot be cast to class org.apache.spark.sql.catalyst.InternalRow
at scala.collection.Iterator$$anon$9.next(Iterator.scala:584)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext
at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
...WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.hasNext
at
org.apache.spark.sql.comet.execution.shuffle.CometBypassMergeSortShuffleWriter.write(CometBypassMergeSortShuffleWriter.java:158)
```
Converting `IcebergWriteExec` into `CometIcebergWriteExec` rewrites the
write subtree and loses the `ColumnarToRow` that Spark had inserted between
Iceberg's columnar `BatchScan (IcebergCopyOnWriteScan)` and the whole-stage
codegen stage above it. At runtime the codegen stage receives `ColumnarBatch`
and casts it to `InternalRow`.
Executed plan with `spark.comet.iceberg.write.enabled=false` (works):
```
IcebergWrite
+- *(2) CometColumnarToRow
+- CometSort
+- CometColumnarExchange hashpartitioning(dep, 10),
REBALANCE_PARTITIONS_BY_COL, CometColumnarShuffle
+- *(1) Project
+- *(1) Filter
+- *(1) BroadcastHashJoin
:- *(1) ColumnarToRow <-- present
: +- BatchScan IcebergCopyOnWriteScan
```
Same query with `spark.comet.iceberg.write.enabled=true` (fails):
```
CometIcebergWrite
+- CometSort
+- CometColumnarExchange hashpartitioning(dep, 10),
REBALANCE_PARTITIONS_BY_COL, CometColumnarShuffle
+- *(1) Project
+- *(1) Filter
+- *(1) BroadcastHashJoin
:- BatchScan IcebergCopyOnWriteScan <--
ColumnarToRow gone
```
With AQE enabled the failure disappears, because transitions are re-inserted
when each stage is materialised. Every existing Comet Iceberg suite runs with
AQE on, which is why this was never caught.
### Steps to reproduce
Spark 4.1.3, Iceberg 1.11.0. Session config:
```
spark.comet.scan.icebergNative.enabled = true
spark.comet.write.iceberg.splitOperator.enabled = true
spark.comet.iceberg.write.enabled = true
spark.sql.adaptive.enabled = false
```
```sql
CREATE TABLE cat.db.t (id INT, dep STRING, amount DOUBLE)
USING iceberg PARTITIONED BY (dep)
TBLPROPERTIES (
'format-version'='2',
'write.delete.mode'='copy-on-write',
'write.distribution-mode'='none',
'write.spark.fanout.enabled'='false');
INSERT INTO cat.db.t VALUES
(1,'hr',1.0),(2,'hr',2.0),(3,'eng',3.0),(4,'eng',4.0),(5,'ops',5.0);
-- src is any small view, e.g. SELECT explode(array(1,3)) AS id
DELETE FROM cat.db.t WHERE id IN (SELECT id FROM src);
```
DELETE, UPDATE and MERGE all fail the same way, on the table's main branch
and on a named branch.
Isolated with a config matrix on that DELETE:
| split operator | native write | localTableScan | result |
| -------------- | ------------ | -------------- | --------------- |
| on | on | on | ClassCastException |
| on | on | off | ClassCastException |
| on | off | on | ok |
| on | off | off | ok |
| off | off | off | ok |
So the split-operator plan alone does not trigger it and
`spark.comet.exec.localTableScan.enabled` is irrelevant. It needs the native
write flag plus AQE off.
### Expected behavior
The copy-on-write operation succeeds with the native writer regardless of
whether AQE is enabled, or the operator declines conversion rather than
producing a plan that violates the columnar contract.
### Additional context
Found by turning the two Iceberg write flags on by default in #5677 and
reading the Iceberg Spark SQL job logs. This is the dominant failure in
Iceberg's `spark-extensions` suites across Iceberg 1.8.1, 1.9.1, 1.10.0 and
1.11.0.
Note that Iceberg's `ExtensionsTestBase` sets `ADAPTIVE_EXECUTION_ENABLED`
to `RANDOM.nextBoolean()` per session, so the number of extension tests this
takes down swings a lot between runs of identical code (99 failures in one run
and 31 in the next, on the same commit's behaviour). That randomisation should
be kept in mind when reading those job results.
Part of #5649.
--
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]
