andygrove opened a new issue, #7049:
URL: https://github.com/apache/datafusion-comet/issues/7049
### 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 c230bf4e98 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 on by default in 1.2.0 (#6664). 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.
--
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]
