andygrove opened a new issue, #5698:
URL: https://github.com/apache/datafusion-comet/issues/5698
### Describe the bug
When unsorted input is handed to a clustered (non-fanout) Iceberg writer,
iceberg-java fails the write with a specific, documented exception. Comet's
native writer rejects the same input, but with a different exception type and a
different message, so code that catches the Iceberg error no longer works.
iceberg-java raises, as the cause:
```
java.lang.IllegalStateException: Incoming records violate the writer
assumption that records are
clustered by spec and by partition within each spec. Either cluster the
incoming records or switch
to fanout writers.
```
Comet's native writer raises:
```
org.apache.comet.CometNativeException: Unexpected => The input is not
sorted! Cannot write to
partition that was previously closed: PartitionKey { spec: PartitionSpec {
spec_id: 0,
fields: [PartitionField { source_id: 3, field_id: 1000, name: "c3",
transform: Identity }] }, ... }
```
`TestRequiredDistributionAndOrdering.testDisabledDistributionAndOrdering`
asserts on the iceberg-java contract:
```java
assertThatThrownBy(
() -> inputDF.writeTo(tableName)
.option(SparkWriteOptions.USE_TABLE_DISTRIBUTION_AND_ORDERING,
"false")
.option(SparkWriteOptions.FANOUT_ENABLED, "false")
.append())
.cause()
.isInstanceOf(IllegalStateException.class)
.hasMessageStartingWith(
"Incoming records violate the writer assumption that records are
clustered by spec "
+ "and by partition within each spec. Either cluster the
incoming records or switch to fanout writers.");
```
and fails on the type check.
Both writers correctly reject the write, so this is an error-fidelity gap
rather than a data problem. It is still user visible: anyone catching
`IllegalStateException` or matching on that message gets a
`CometNativeException` instead once the native writer is in play.
### Steps to reproduce
Spark 4.1.3, Iceberg 1.11.0, `spark.comet.iceberg.write.enabled=true` plus
the Iceberg Spark SQL test setup from `dev/diffs/iceberg/1.11.0.diff`:
```
./gradlew -DsparkVersions=4.1 -DscalaVersion=2.13 -DflinkVersions=
-DkafkaVersions= \
:iceberg-spark:iceberg-spark-4.1_2.13:test \
--tests '*TestRequiredDistributionAndOrdering*' -Pquick=true -x javadoc
```
Fails for all four catalog configurations, on Iceberg 1.8.1, 1.9.1, 1.10.0
and 1.11.0.
### Expected behavior
The native writer surfaces the same exception type and message as
iceberg-java when it rejects unclustered input for a clustered writer, so
callers cannot tell the two writers apart by the error they raise.
### Additional context
Found by turning the two Iceberg write flags on by default in #5677.
Worth noting how it became visible. The test's input is
`ds.coalesce(1).sortWithinPartitions("c1")`, which gives the write a plain
Spark source and no exchange. Before `spark.comet.exec.localTableScan.enabled`
was set in the Iceberg test diffs,
`CometIcebergNativeWrite.requiresNativeChildren` declined that plan, the JVM
writer ran, and the test passed. It fails deterministically once the native
writer actually reaches the write, on every Iceberg version. That is a fair
illustration of how much of the Iceberg write surface was previously being
tested against Spark's writer rather than Comet's.
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]
