andygrove opened a new pull request, #5779:
URL: https://github.com/apache/datafusion-comet/pull/5779
## Which issue does this PR close?
Closes #5698.
## Rationale for this change
iceberg-java's `ClusteredWriter` rejects unclustered input with a documented
exception:
```
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.
Encountered records that belong to already closed files:
partition 'c3=1' in spec [
1000: c3: identity(3)
]
```
Comet's native writer rejected the same input, but as a
`CometNativeException` carrying
iceberg-rust's differently worded `The input is not sorted! Cannot write to
partition that was
previously closed: PartitionKey { ... }`. Both writers correctly refuse the
write, so this was an
error-fidelity gap rather than a data problem — but a user-visible one:
anyone catching
`IllegalStateException` or matching on that message got something else once
the native writer was
in play. Iceberg's own
`TestRequiredDistributionAndOrdering.testDisabledDistributionAndOrdering`
asserts on both the type and the message and failed on the type check.
## What changes are included in this PR?
- `iceberg_write.rs`: the clustered write path now mirrors
`ClusteredWriter`'s closed-partition
bookkeeping (the partition currently open, plus the ones already finished)
and detects a
revisited partition itself, before iceberg-rust does. That keeps the error
message ours and
gives us the offending `PartitionKey`, so the context can be rendered as
`partition '<path>' in spec <spec>` — `PartitionKey::to_path()` is
iceberg-rust's equivalent of
`PartitionSpec.partitionToPath`, and the spec is rendered the way
iceberg-java's
`PartitionSpec.toString` / `PartitionField.toString` do. Only the
partition branch of
iceberg-java's check is reachable: a task writes through a single output
spec, so the spec
cannot change mid-stream.
- `errors.rs`: a new `CometError::IllegalState` classification reaches the
JVM as
`java.lang.IllegalStateException` with the message untouched. It travels as
`DataFusionError::External` and is recovered by a cause-chain walk, so
DataFusion's
`Context`/`Shared` wrappers cannot flatten it into a string on the way
out. That walk already
existed for shuffle capacity failures (`shuffle_size_limit_message`); it
is now a single
`typed_jvm_exception` that handles both. The classification is by error
variant only, never
inferred from error text.
## How are these changes tested?
- New end-to-end test in `CometIcebergWriteActionSuite` writes interleaved
partition values with
`use-table-distribution-and-ordering=false` and `fanout-enabled=false` to
two identical tables,
once through the JVM writer and once through the native writer, and
asserts the two are
indistinguishable: same cause-chain shape (so
`assertThatThrownBy(...).cause()` finds the
exception in the same place either way) and a byte-identical
`IllegalStateException` message.
Pinning against the JVM writer on the same runtime rather than against a
literal means an
iceberg-java wording change shows up here as a parity failure. It also
asserts the aborted write
really went through `CometIcebergWriteExec`, so the test cannot pass by
falling back, and that
neither table committed. Confirmed it fails without the fix with exactly
the reported symptom:
`List(SparkException, CometNativeException) did not equal
List(SparkException, IllegalStateException)`.
- New Rust test in `iceberg_write.rs` asserts the native error is a
`CometError::IllegalState`
whose message matches iceberg-java's byte for byte, plus one covering the
parameterised
transforms (`bucket[8]`, `truncate[4]`) in the spec rendering.
- New Rust test in `errors.rs` asserts `CometError::IllegalState` survives
`Shared`/`Context`/`External` wrapping and arrives as
`java/lang/IllegalStateException`.
- `CometIcebergWriteActionSuite` (57), `CometIcebergNativeSuite` (100),
`CometIcebergWriteDetectionSuite`,
`CometIcebergRewriteActionSuite`, `IcebergWriteProtoTranslationSuite` (73
combined) all pass;
`cargo test` for `datafusion-comet` (292), `datafusion-comet-jni-bridge`
(28) and
`datafusion-comet-shuffle` (125) pass; clippy and rustfmt clean;
`test-compile` verified against
spark-3.4/scala-2.12, spark-3.5, spark-4.0 and the default spark-4.1.
--
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]
