Native Iceberg writer raises a different exception than iceberg-java for unclustered input to a clustered writer

[andygrove]

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:

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.