Repository navigation
Native Iceberg writes into NOT NULL columns fall back on Spark 3.5+ because Comet has no native AssertNotNull #7049
Description
What is the problem the feature request solves?
On Spark 3.5 and later, when a nullable input feeds a NOT NULL column of a V2 table, Spark wraps the value in AssertNotNull, which checks each row at run time. That applies to top-level columns and to required fields inside a struct, where Spark rebuilds the struct as named_struct('a', assertnotnull(s.a), ...). Spark 3.4 rejects such writes at analysis instead (Cannot write nullable values to non-null column).
Comet has no native AssertNotNull, so the projection that carries it stays in Spark. When nothing separates that projection from the writer, CometIcebergWriteExec gets no Arrow input and the whole write falls back to iceberg-java:
Cannot perform native operation because input is not in Arrow format; assertnotnull is not supported; unsupported arguments for CreateNamedStruct
That is the case for unpartitioned tables and for write.distribution-mode=none. Iceberg requires identifier fields to be NOT NULL, so a table keyed for upserts hits it on every INSERT ... SELECT from a nullable source. When an exchange does sit between the projection and the writer, as with a partitioned table's default hash distribution or the rebalance in a copy-on-write UPDATE, the write stays native: the Spark projection checks the rows and the columnar shuffle hands Arrow batches to the writer.
Measured on main at c230bf4 with a Parquet source whose columns are nullable but hold no NULLs, on Spark 3.5.9 (Iceberg 1.8.1) and 4.1.3 (Iceberg 1.11.0). The bucketed tables ran on 4.1.3 only.
| Statement | id INT NOT NULL |
s STRUCT<a: INT NOT NULL, b: STRING> |
id INT NOT NULL, bucket(4, id) |
id INT NOT NULL, bucket(4, id), write.distribution-mode=none |
|---|---|---|---|---|
INSERT INTO t SELECT * FROM src |
falls back | falls back | native | falls back |
spark.table("src").writeTo("t").append() |
falls back | falls back | native | falls back |
INSERT INTO t VALUES (...) |
native | native | native | native |
INSERT INTO t SELECT coalesce(id, -1), ... |
native | native | native | native |
copy-on-write UPDATE |
native | native | native | native |
A NOT NULL struct column (s STRUCT<...> NOT NULL) behaves like the nested case. When the input does hold a NULL, the fallback raises Spark's error, as it should.
Describe the potential solution
A native AssertNotNull that passes its child through and fails on the first NULL with Spark's error, where <path> is the walked type path (for example s then a):
- Spark 4.x:
SparkRuntimeException[NOT_NULL_ASSERT_VIOLATION] NULL value appeared in non-nullable field: <path>. - Spark 3.5:
java.lang.NullPointerException: Null value appeared in non-nullable field: <path>.
The CreateNamedStruct reason in the nested case comes from the same expression, because its argument is the AssertNotNull.
Additional context
Found in a nested-type audit of the native writers. Native Iceberg writes are opt-in (spark.comet.write.iceberg.enabled, off by default since #6962), so this affects users who enable them. V1 file tables never have NOT NULL columns, so the native Parquet writer doesn't see AssertNotNull. Until there is a native one, writes stay native when the input is not nullable (for example through coalesce) or when an exchange sits in front of the writer.
Part of #5649.
Activity