sunchao commented on code in PR #5779:
URL: https://github.com/apache/datafusion-comet/pull/5779#discussion_r3962187322
##########
native/core/src/execution/operators/iceberg_write.rs:
##########
@@ -452,6 +460,68 @@ fn iceberg_err(e: iceberg::Error) -> DataFusionError {
DataFusionError::External(Box::new(e))
}
+/// iceberg-java's `ClusteredWriter` preamble, copied verbatim from
+/// `core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java`.
Applications catch that
+/// writer's `IllegalStateException` and match on this text -- Iceberg's own
+/// `TestRequiredDistributionAndOrdering` does both -- so the native writer
reproduces it instead
+/// of surfacing iceberg-rust's differently worded error.
+const NOT_CLUSTERED_ROWS_ERROR_MSG_TEMPLATE: &str = "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.\n\
+ Encountered records that belong to already closed files:\n";
+
+/// The message iceberg-rust's `ClusteredWriter` raises for the same condition.
+const UNSORTED_INPUT_MESSAGE_PREFIX: &str = "The input is not sorted!";
+
+/// Restates iceberg-rust's unclustered-input failure as iceberg-java's,
passing every other
+/// failure through unchanged.
+///
+/// Leaving `ClusteredWriter` as the sole judge of whether the input is
clustered means coupling to
+/// its message text, which is the cheaper of the two couplings available:
re-deriving the
+/// condition here would need a copy of the writer's closed-partition
bookkeeping, and because that
+/// copy would fire first, no test could catch it drifting from the original.
A wording change
+/// upstream instead makes
`clustered_write_rejects_unclustered_input_like_iceberg_java` fail, since
+/// that test drives the real writer.
+fn clustered_write_err(e: iceberg::Error, key: &PartitionKey) ->
DataFusionError {
+ if e.kind() == ErrorKind::Unexpected &&
e.message().starts_with(UNSORTED_INPUT_MESSAGE_PREFIX) {
+ not_clustered_error(key)
+ } else {
+ iceberg_err(e)
+ }
+}
+
+/// The error iceberg-java's `ClusteredWriter.write` raises when a closed
partition is revisited,
+/// down to the `partition '<path>' in spec <spec>` context. Only the
partition branch of that
+/// check is reachable here: a task writes through a single output spec, so
the spec can never
+/// change mid-stream.
+fn not_clustered_error(key: &PartitionKey) -> DataFusionError {
+ DataFusionError::from(CometError::IllegalState(format!(
+ "{NOT_CLUSTERED_ROWS_ERROR_MSG_TEMPLATE}partition '{}' in spec {}",
+ key.to_path(),
Review Comment:
### Correctness
[P2] Use the Java-compatible partition formatter in this error
Could this share the [partition-path formatting on the current
base](https://github.com/apache/datafusion-comet/blob/b56268d149f19e8a13cf35ec62a9145988f6f5e5/native/core/src/execution/operators/iceberg_partition_path.rs#L80)
instead of `PartitionKey::to_path()`? With
`use-table-distribution-and-ordering=false` and `fanout-enabled=false`, an
identity-binary partition sequence `X'0001FF', X'00', X'0001FF'` reaches
`clustered_write_err`, but the native diagnostic contains `part=0001FF` while
iceberg-java reports `part=AAH%2F`. The prior Java writer probes and extracted
Rust formatter result remain applicable because the relevant dependency and
rendering source are unchanged.
This also bypasses the base's `CometLocationGenerator` protection for a
revisited negative fractional `timestamptz`. The locked raw formatter for
`-1500000` microseconds panics before `CometError::IllegalState` is
constructed. The base can write that first partition successfully, so the
recurrence reaches this new error-formatting call. This is established by
source analysis, not a native reproduction. Please share the Java-compatible
formatting logic and extend the parity test to interleaved binary and
negative-timestamp partitions so both the exception class and message match.
--
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]
