[LinSimon-901101]

Which issue does this PR close?

Closes #7049.

Rationale for this change

On Spark 3.5+, writing nullable input to required DataSource V2 fields inserts AssertNotNull. Without native support, this projection can force otherwise eligible native Iceberg writes to fall back, including unpartitioned tables and writes with write.distribution-mode=none.

What changes are included in this PR?

• Add native AssertNotNull serialization, planning and evaluation. Preserve the child value and validate logical nulls, including dictionary arrays and outer struct validity.
• Preserve Spark's version-specific exception type, message and walked type path.
• Support the structural and conditional expressions used by V2 writes while preserving existing JVM codegen dispatch.
• Preserve short-circuit and partial-consumption semantics through conservative fallback, including assertions below LIMIT and in partially consumed TopK or aggregate result projections. Reuse the existing partial-reader tracking.
• Document the supported contexts and fallback restrictions. Native Iceberg writes remain opt-in.

How are these changes tested?

• Native unit tests cover values, slices, scalar and dictionary nulls, empty batches, struct validity and error paths.
• Expression regressions cover Spark error parity, conditional and short-circuit evaluation, codegen dispatch, and partial consumption with AQE enabled and disabled.
• Iceberg regressions cover SQL inserts and DataFrame appends into required top-level and nested fields, nullable parent structs, limited input, and rejection of null input without committing a snapshot.
• Local targeted validation passed 115 JVM test executions across Spark 3.4–4.2 and four native tests. Writer regressions ran on Spark 3.5, 4.0 and 4.1; local Spark 3.4 and 4.2 coverage is expression-only.
• Fork CI passed on commit 9737fc3: 137 successful jobs, 13 expected conditional skips, and no failed or canceled jobs. The complete configured matrices and critical test steps were verified.

[sunchao]

Reviewed the full diff from da2ff08134ba6ceeccade058aea09e7c21d9472b to 9737fc3c8f37a5fa332fc18ccb640f154559a5da, covering all three commits and 17 files. The PR is not a draft. The snapshot and live discussion contained no existing reviews, comments, or threads. One introduced P2 issue was identified.

Summary

• Prior state and problem: Standalone AssertNotNull lacked native support, forcing projections for required DataSource V2 fields to fall back and preventing otherwise eligible native Iceberg writes.
• Design approach: Add a native passthrough expression that validates logical nulls, serialize the walked field path, and preserve Spark errors through version-specific shims. Restrict enclosing expressions and partially consumed operators conservatively.
• Correctness: Checked Spark expression implementations and tests for 3.4.3, 3.5.9, 4.0.4, 4.1.3, and 4.2.0. Value preservation, dictionary nulls, struct validity, and error conversion are consistent with those sources. The partial-consumption protection misses early-exit semi/anti join conditions, as detailed below.
• Compatibility analysis: Spark 3.4 rejects nullable-source writes to required V2 fields during analysis. Spark 3.5 adds runtime assertions. The shims correctly distinguish 3.x NullPointerException from 4.x NOT_NULL_ASSERT_VIOLATION. Compared with branch-1.1 at 89c729a9697467e8f8d89e3eadf1797a75e2328c, native assertion execution is an intended expansion, but the join failure is an unintended consequence.
• Key design decisions: Keep Iceberg writes opt-in, preserve child types and arrays, check outer struct validity independently of nested fields, and leave assertions captured inside JVM dispatch on their existing path.
• Implementation sketch: Register protobuf field 76 and the expression builder, implement native evaluation, update error parameters and three Spark shims, extend planner guards, and add expression/write regressions plus documentation.
• Performance: Successful evaluation returns the original value without copying its buffers. Flat arrays use null-count metadata, while dictionary values require logical-null inspection. No benchmark establishes an end-to-end speedup, and no separate P1/P2 performance regression was identified.
• Design: The implementation is reasonably contained and reuses existing consumption tracking. Its safety boundary needs to include consumers that short-circuit internally, even without an external limit.
• Abstraction & complexity: The dedicated physical expression fits the passthrough contract. Shared planner tags and protobuf traversal avoid separate tracking systems and container-specific assertion searches. No additional abstraction issue met the reporting bar.
• Behavioral changes worth calling out: Eligible nullable-input V2 write projections can now remain native. Unsafe enclosing expressions and externally limited assertion projections fall back. Semi/anti join conditions currently escape that fallback and can fail where Spark succeeds.
• Suggested improvements: Reject native assertions in early-exit join conditions until candidate evaluation preserves Spark’s short-circuit behavior. Add semi/anti join regressions with a successful candidate preceding an assertion-failing candidate.

P2 finding: Preserve early-exit semantics in semi/anti join conditions

At spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:1203, assertion fallback requires a PARTIAL_READER tag. A standalone broadcast semi/anti join receives no such tag. With left IDs [0], broadcast right IDs [0, 1], and Catalyst condition IF(l.id = r.id, true, AssertNotNull(NULL::BOOLEAN)), Spark stops after the first matching candidate. The native join checks the later candidate and throws NOT_NULL_ASSERT_VIOLATION. This turns a successful query into a task failure. Before this PR, the assertion prevented native conversion.

Routed skills: review-comet-pr, review-comet-expression-pr, and review-comet-iceberg-write-pr.

Exact-head CI: Apache has only a successful label check attached. Fork run https://github.com/LinSimon-901101/datafusion-comet/actions/runs/39644062061 used the exact reviewed SHA and completed with 137 successful jobs and 13 skipped jobs, including Spark 3.4–4.2 and Iceberg matrices. Inspected logs confirm all 11 new expression cases and eight new write cases passed on Spark 4.1.

Local validation: All four added native unit tests passed. A bounded native physical-join reproduction demonstrated the semi/anti failure. Spark 4.1.3 interpreted and generated predicates confirmed that its consumer skips the failing candidate. Targeted optimizer checks retained the conditional assertion. Full JVM/Iceberg suites and a complete Comet SQL reproduction were not run locally. A full Spark-session probe was blocked by incomplete local runtime dependencies. The finding is supported by the bounded reproductions and the planner/consumer source paths. Project code remains unchanged.

[sunchao]

[P2] Preserve early-exit semantics in semi/anti join conditions. This guard only checks assertions when an external consumer supplied PARTIAL_READER, but a broadcast semi/anti join also stops evaluating candidates after its first match. For a BuildRight join with left IDs [0], right IDs [0, 1], and Catalyst condition IF(l.id = r.id, true, AssertNotNull(NULL::BOOLEAN)), Spark returns [0] for LEFT SEMI and no rows for LEFT ANTI without evaluating the assertion. The native join evaluates the second candidate and throws NOT_NULL_ASSERT_VIOLATION. If is admitted by the new expression guard, and the join receives no partial-reader tag, so the previously unsupported assertion now enables this failing native path. Please decline native assertions in these early-exit join conditions until their candidate-level short-circuit behavior is preserved.

Evidence: Ran the four exact-head native assertion unit tests successfully. A disposable Rust test used the exact-head assert_not_null.rs, locked DataFusion 55.2.0, batches [0] and [0,1], a lazy CASE equivalent to the IF condition, and NestedLoopJoinExec::try_new(...).swap_inputs() matching Comet's BuildRight planner. Both LeftSemi and LeftAnti returned NOT_NULL_ASSERT_VIOLATION. A Spark 4.1.3 JVM probe tested interpreted and generated predicates with buildRows.exists, the consumer used by BroadcastNestedLoopJoinExec.leftExistenceJoin: both stopped after one candidate and succeeded. Evaluating the skipped candidate separately raised the assertion. SimplifyConditionals, NullPropagation, ConstantFolding, and PushPredicateThroughJoin retained the conditional assertion. Sources: Spark v4.1.3 BroadcastNestedLoopJoinExec.scala:221–224, Comet predicates.scala:505–508, and native/core/src/execution/planner/join.rs:309–318. Reproduction files and logs are under /tmp/comet7155-review/probe/. A full Comet JVM query was not executed locally.