[andygrove]

Which issue does this PR close?

Closes #6823.

Part of #6006.

Rationale for this change

Spark evaluates AND, OR and the candidates of IN lazily: a later operand runs only for the rows that the earlier ones have not decided. Comet evaluates every operand over the whole batch, so under ANSI mode an operand that can raise throws on a row Spark never evaluates. With v holding 0 to 3 on a Parquet table and ANSI on, each of these raised DIVIDE_BY_ZERO in Comet and returned rows in Spark:

• v != 0 AND 100 / v > 1
• v = 0 OR 100 / v > 1
• v IN (0, 100 / v)

CASE WHEN, IF and coalesce already match Spark.

What changes are included in this PR?

• LazyLogical (native/spark-expr/src/conditional_funcs/lazy_logical.rs) evaluates the right operand of AND and OR only for the rows the left operand has not decided, which are the rows where it is not false for AND and not true for OR. It shares the column projection and masked evaluation of NullShortCircuit, so skipping rows filters no other column. spark_comparison creates it for AND and OR.
• When more than half the rows still need the right operand, and it is built from deterministic expressions, it is first tried over the whole batch, which is cheaper than selecting rows. A failure there falls back to selecting the rows, here and for the batches after it. The result for a row the left operand decides is ignored, so this only matters if it fails there.
• It applies only when the right operand can fail, by the existing cannot_fail check. Otherwise the expression stays a plain BinaryExpr, so the common case has no extra cost.
• lazy_in_list rewrites v IN (c1, c2, ...) into a chain when a candidate after the first can fail. A run of candidates that cannot fail stays one IN, keeping its hash set, and each candidate that can fail is a lazy equality. NOT IN is the negation of that chain. This applies only when the value is a cheap expression that cannot fail, because the chain evaluates it once per candidate. Other shapes keep the existing whole-batch evaluation.

This does not cover the rest of #6006: aggregate FILTER, operator-level masks, and the compatibility guide section.

How are these changes tested?

• Rust unit tests in lazy_logical.rs cover AND and OR on rows the left side decides, a NULL left operand, a batch with no decided rows, an empty batch, a scalar left operand, a right operand that reads no column, an infallible right operand staying a BinaryExpr, and the IN and NOT IN rewrite, including floats. The datafusion-comet-spark-expr lib suite passes (1333 tests) and cargo clippy -D warnings is clean.
• New SQL file test expressions/conditional/logical_lazy_ansi.sql runs the cases above, chains, nesting under NOT, and IN with a failing cast. It also has controls where a row Spark does evaluate must still raise DIVIDE_BY_ZERO or CAST_INVALID_INPUT. It passes with CometSqlFileTestSuite on Spark 4.1, along with the existing conditional/ and predicates SQL files.
• native/spark-expr/benches/lazy_logical.rs compares the eager and lazy forms at 0 to 100% pending rows, and IN lists of 2, 8 and 64 candidates. Lazy evaluation is up to 3-4x faster when few rows are pending, and within noise of eager evaluation otherwise. A 64-candidate IN with one fallible candidate is 2.3x faster than DataFusion's.
• I did not run the Spark SQL suite or the other Spark profiles.

[sunchao]

Reviewed af39f64 across five independent scopes. The row masks, deterministic speculation, projection reuse, and SQL-test integration look sound. I found two issues in the IN rewrite: expensive values are evaluated repeatedly, and long dynamic lists become an unbounded recursive tree that can overflow the native thread stack. Details and focused reproduction evidence are inline.

The probes isolate the unchanged new module using cached dependencies and equivalent helpers for the tested expressions; they are not full Spark benchmarks or full exact-head builds. The exact-head native test run was blocked by unavailable locked versions in my configured dependency registry. Current Rust tests and the Spark 4.1 SQL validation are still queued/running, so those results remain outstanding.

[sunchao]

[P2] Restrict repeated IN values to cheap expressions

Could this use the existing cost-aware is_infallible(value, schema) check, or evaluate the value once and reuse it? cannot_fail explicitly accepts LEGACY/TRY string-to-integer parsing, so TRY_CAST(s AS INT) IN (CAST(n AS INT), CAST(n + 1 AS INT), ...) enters this rewrite under ANSI mode. Every term then evaluates the same parsing cast again. The previous InListExpr evaluated the value once. On 8,192 valid unmatched rows, an isolated probe of this module with cached dependencies measured roughly 3.2x overhead with 8 candidates and 4.7x with 64 candidates even for 12-byte strings. This is expression-level evidence, not end-to-end Spark timing. The current benchmark only uses a column as the value, so it misses this case. Please keep expensive values outside this path or materialize them once, and add this shape to the benchmark.

[sunchao]

[P2] Avoid an unbounded recursive tree for dynamic IN lists

Could this evaluate candidates iteratively or build a balanced tree while retaining their order? Appending each candidate to matched creates a left-deep tree, and LazyLogical::evaluate recursively evaluates its left child. A long flat IN list therefore acquires one stack frame per fallible candidate. In an isolated probe with the unchanged new module compiled at opt-level=3, a bare Int32 value column, 8,192 dynamic candidates of the form CAST(n + i AS INT) under ANSI mode, and one valid unmatched row, the previous eager InListExpr completed while the new expression aborted with a stack overflow on a 2 MiB thread. Comet leaves Tokio's default stack size unchanged. This used cached dependencies and establishes a native expression failure, not an end-to-end SQL threshold. Please bound the evaluation depth and cover a long dynamic list with a constrained-stack regression test.