test: JoinScan NOT IN null-safety regress (requires patched datafusion-proto) - #5006

…are semantics

Push down `<aggregate> over <Semi/Anti/IN/NOT IN/EXISTS/NOT EXISTS>`
queries into a single `Custom Scan (ParadeDB Aggregate Scan)` node,
including `NOT IN` against a NULL-bearing inner where SQL three-valued
logic must return zero rows.

Pre-PR the agg-on-join walker bailed on these shapes with one of:

  - `unexpected node type T_FromExpr in join tree` (post-pull-up shape)
  - `aggregate-on-join does not support Semi/Anti JOIN` (allow-list)
  - `Aggregate-on-join does not support Anti JOIN` (translator EquiOnly)

Plus, un-pulled-up `IN`/`NOT IN` SubPlans in `baserestrictinfo` were
silently dropped by the per-RI `extract_quals` loop, producing wrong
row counts on push-down.

## Coverage

| Query shape                          | Result                              |
|--------------------------------------|-------------------------------------|
| `IN (SELECT ...)` PG-pulled-up       | Pushed down (Semi)                  |
| `EXISTS / NOT EXISTS`                | Pushed down (Semi/Anti)             |
| Single-col `IN` un-pulled-up         | Lifted to Semi, pushed down         |
| Single-col `NOT IN`, no NULL inner   | Lifted to null-aware Anti           |
| Single-col `NOT IN`, NULL inner      | Lifted, returns 0 rows              |
| Multi-col `NOT IN` / `IN`            | Declines cleanly → PG fallback      |
| OR-nested SubPlan                    | Declines cleanly → PG fallback      |

## What changed

joinscan/build.rs: `JoinType::Anti` becomes `Anti { null_aware: bool }`.
Co-locating the flag with the variant makes `(Inner, null_aware: true)`
unrepresentable in the type system instead of relying on a runtime guard.
`JoinNode.null_aware: bool` field removed (now folded into the variant).

joinscan/planning.rs: `wrap_with_semi_anti` returns
`Result<RelNode, String>` with a site-specific reason at every former
silent-skip point (no inner final RelOptInfo, inner not pushdownable,
multi-col NOT IN, IN-with-undecodable-testexpr). Both callers — the new
agg caller and the existing JoinScan caller — propagate Err to a clean
decline. Closes a latent silent-drop in the JoinScan non-LIMIT path that
`is_limit_pushdown_safe` only caught for LIMIT queries.

aggregatescan/datafusion_build.rs: `build_relnode_from_node` handles
`T_FromExpr` (the post-pull-up shape PG produces). `build_join_node`
accept-list extends to Semi/Anti/RightSemi/RightAnti — these are
unconditionally safe for aggregate pushdown because they never project
the non-preserved side. `build_scan_node` classifies baserestrictinfo
into search / top-level SubPlans / OR-nested SubPlans, batches search
predicates into a single strict `extract_quals` call (no more silent
drop on un-translatable RIs), declines on OR-nested or unliftable
SubPlans, and lifts top-level SubPlans via `wrap_with_semi_anti`.
`populate_required_fields` switches from `(rti, attno)` lookup to
`(plan_position, attno)` — rti is only unique within a single
PlannerInfo, so SubPlan-derived sources can share an rti with the outer
scan and the old lookup matched the wrong source.

datafusion/translator.rs: `build_null_aware_anti_join` lowers
`Anti { null_aware: true }` lifts to `LogicalPlan::Join` with
`null_equality=NullEqualsNothing` and `null_aware=true`, which
DataFusion's HashJoinExec uses to emit zero rows when the probe (inner)
side has any NULL. Drops the dead `JoinTypeAllowList::EquiOnly` enum
variant; both callers always passed `All` after the agg path's allow-
list flip.

aggregatescan/datafusion_exec.rs + aggregatescan/mod.rs: plumb
`ExprContext` + `PlanState` from the executor's runtime through
`build_join_aggregate_plan` → `build_relnode_df` → `build_source_df`
into each per-relation `PgSearchTableProvider`, so HeapFilter queries
(runtime expressions like `=` on a `pdb.literal`-cast column) can
resolve. Skip the ExprContext allocation under `EXEC_FLAG_EXPLAIN_ONLY`
where no expressions execute.

## Tests

tests/pg_regress/sql/aggregate_join_semi_anti.sql:

  - Test 1: `IN (SELECT ...)` pulls up to Semi → AggregateScan
  - Test 2: `EXISTS / NOT EXISTS` → AggregateScan
  - Test 3: single-column `NOT IN` un-pulled-up → null-aware Anti lift,
    AggregateScan; no NULL inner so result equivalent to plain LeftAnti
  - Test 4: parity with `enable_aggregate_custom_scan = off` for Test 3
  - Test 5: multi-column `(a,b) NOT IN (SELECT x,y FROM t)` declines
    cleanly with a precise WARNING; PG plan runs; result matches PG
    custom-scan-OFF
  - Test 6: single-column `NOT IN` with a NULL-bearing inner →
    AggregateScan returns zero rows (SQL three-valued logic), parity
    with PG custom-scan-OFF, plus a sanity check that removing the
    NULL inner row makes the same query return non-zero rows

Note: DataFusion 53.1.0's null-aware mode is restricted to a single-
column equi-key (HashJoinExec runtime code only inspects `values[0]`);
multi-column NOT IN therefore can't ride the null-aware fast path and
is left for PG to handle. Sister PR #5006 covers the JoinScan-side
end-to-end path (which additionally needs a patched `datafusion-proto`
because upstream's logical Join proto schema is missing the
`null_aware` field).

Refs #4911.

…are semantics

Push down `<aggregate> over <Semi/Anti/IN/NOT IN/EXISTS/NOT EXISTS>`
queries into a single `Custom Scan (ParadeDB Aggregate Scan)` node,
including `NOT IN` against a NULL-bearing inner where SQL three-valued
logic must return zero rows.

Pre-PR the agg-on-join walker bailed on these shapes with one of:

  - `unexpected node type T_FromExpr in join tree` (post-pull-up shape)
  - `aggregate-on-join does not support Semi/Anti JOIN` (allow-list)
  - `Aggregate-on-join does not support Anti JOIN` (translator EquiOnly)

Plus, un-pulled-up `IN`/`NOT IN` SubPlans in `baserestrictinfo` were
silently dropped by the per-RI `extract_quals` loop, producing wrong
row counts on push-down.

## Coverage

| Query shape                          | Result                              |
|--------------------------------------|-------------------------------------|
| `IN (SELECT ...)` PG-pulled-up       | Pushed down (Semi)                  |
| `EXISTS / NOT EXISTS`                | Pushed down (Semi/Anti)             |
| Single-col `IN` un-pulled-up         | Lifted to Semi, pushed down         |
| Single-col `NOT IN`, no NULL inner   | Lifted to null-aware Anti           |
| Single-col `NOT IN`, NULL inner      | Lifted, returns 0 rows              |
| Multi-col `NOT IN` / `IN`            | Declines cleanly, PG fallback       |
| OR-nested SubPlan                    | Declines cleanly, PG fallback       |

## What changed

joinscan/build.rs: `JoinType::Anti` becomes `Anti { null_aware: bool }`.
Co-locating the flag with the variant makes `(Inner, null_aware: true)`
unrepresentable in the type system instead of relying on a runtime guard.
`JoinNode.null_aware: bool` field removed (now folded into the variant).

joinscan/planning.rs: `wrap_with_semi_anti` returns
`Result<RelNode, String>` with a site-specific reason at every former
silent-skip point (no inner final RelOptInfo, inner not pushdownable,
multi-col NOT IN, IN-with-undecodable-testexpr). Both callers (the new
agg caller and the existing JoinScan caller) propagate Err to a clean
decline. Closes a latent silent-drop in the JoinScan non-LIMIT path that
`is_limit_pushdown_safe` only caught for LIMIT queries.

aggregatescan/datafusion_build.rs: `build_relnode_from_node` handles
`T_FromExpr` (the post-pull-up shape PG produces). `build_join_node`
accept-list extends to Semi/Anti/RightSemi/RightAnti, which are
unconditionally safe for aggregate pushdown because they never project
the non-preserved side. `build_scan_node` classifies baserestrictinfo
into search / top-level SubPlans / OR-nested SubPlans, batches search
predicates into a single strict `extract_quals` call (no more silent
drop on un-translatable RIs), declines on OR-nested or unliftable
SubPlans, and lifts top-level SubPlans via `wrap_with_semi_anti`.
`populate_required_fields` switches from `(rti, attno)` lookup to
`(plan_position, attno)`. rti is only unique within a single
PlannerInfo, so SubPlan-derived sources can share an rti with the outer
scan and the old lookup matched the wrong source.

datafusion/translator.rs: `build_null_aware_anti_join` lowers
`Anti { null_aware: true }` lifts to `LogicalPlan::Join` with
`null_equality=NullEqualsNothing` and `null_aware=true`, which
DataFusion's HashJoinExec uses to emit zero rows when the probe (inner)
side has any NULL. Drops the dead `JoinTypeAllowList::EquiOnly` enum
variant; both callers always passed `All` after the agg path's allow-
list flip.

aggregatescan/datafusion_exec.rs + aggregatescan/mod.rs: plumb
`ExprContext` + `PlanState` from the executor's runtime through
`build_join_aggregate_plan` to `build_relnode_df` to `build_source_df`
into each per-relation `PgSearchTableProvider`, so HeapFilter queries
(runtime expressions like `=` on a `pdb.literal`-cast column) can
resolve. Skip the ExprContext allocation under `EXEC_FLAG_EXPLAIN_ONLY`
where no expressions execute.

## Tests

tests/pg_regress/sql/aggregate_join_semi_anti.sql:

  - Test 1: `IN (SELECT ...)` pulls up to Semi -> AggregateScan
  - Test 2: `EXISTS / NOT EXISTS` -> AggregateScan
  - Test 3: single-column `NOT IN` un-pulled-up -> null-aware Anti lift,
    AggregateScan; no NULL inner so result equivalent to plain LeftAnti
  - Test 4: parity with `enable_aggregate_custom_scan = off` for Test 3
  - Test 5: multi-column `(a,b) NOT IN (SELECT x,y FROM t)` declines
    cleanly with a precise WARNING; PG plan runs; result matches PG
    custom-scan-OFF
  - Test 6: single-column `NOT IN` with a NULL-bearing inner ->
    AggregateScan returns zero rows (SQL three-valued logic), parity
    with PG custom-scan-OFF, plus a sanity check that removing the
    NULL inner row makes the same query return non-zero rows

Note: DataFusion 53.1.0's null-aware mode is restricted to a single-
column equi-key (HashJoinExec runtime code only inspects `values[0]`);
multi-column NOT IN therefore can't ride the null-aware fast path and
is left for PG to handle. Sister PR #5006 covers the JoinScan-side
end-to-end path (which additionally needs a patched `datafusion-proto`
because upstream's logical Join proto schema is missing the
`null_aware` field).

Refs #4911.

…are semantics

Push down `<aggregate> over <Semi/Anti/IN/NOT IN/EXISTS/NOT EXISTS>`
queries into a single `Custom Scan (ParadeDB Aggregate Scan)` node,
including `NOT IN` against a NULL-bearing inner where SQL three-valued
logic must return zero rows.

Pre-PR the agg-on-join walker bailed on these shapes with one of:

  - `unexpected node type T_FromExpr in join tree` (post-pull-up shape)
  - `aggregate-on-join does not support Semi/Anti JOIN` (allow-list)
  - `Aggregate-on-join does not support Anti JOIN` (translator EquiOnly)

Plus, un-pulled-up `IN`/`NOT IN` SubPlans in `baserestrictinfo` were
silently dropped by the per-RI `extract_quals` loop, producing wrong
row counts on push-down.

## Coverage

| Query shape                          | Result                              |
|--------------------------------------|-------------------------------------|
| `IN (SELECT ...)` PG-pulled-up       | Pushed down (Semi)                  |
| `EXISTS / NOT EXISTS`                | Pushed down (Semi/Anti)             |
| Single-col `IN` un-pulled-up         | Lifted to Semi, pushed down         |
| Single-col `NOT IN`, no NULL inner   | Lifted to null-aware Anti           |
| Single-col `NOT IN`, NULL inner      | Lifted, returns 0 rows              |
| Multi-col `NOT IN` / `IN`            | Declines cleanly, PG fallback       |
| OR-nested SubPlan                    | Declines cleanly, PG fallback       |

## What changed

joinscan/build.rs: `JoinType::Anti` becomes `Anti { null_aware: bool }`.
Co-locating the flag with the variant makes `(Inner, null_aware: true)`
unrepresentable in the type system instead of relying on a runtime guard.
`JoinNode.null_aware: bool` field removed (now folded into the variant).

joinscan/planning.rs: `wrap_with_semi_anti` returns
`Result<RelNode, String>` with a site-specific reason at every former
silent-skip point (no inner final RelOptInfo, inner not pushdownable,
multi-col NOT IN, IN-with-undecodable-testexpr). Both callers (the new
agg caller and the existing JoinScan caller) propagate Err to a clean
decline. Closes a latent silent-drop in the JoinScan non-LIMIT path that
`is_limit_pushdown_safe` only caught for LIMIT queries.

aggregatescan/datafusion_build.rs: `build_relnode_from_node` handles
`T_FromExpr` (the post-pull-up shape PG produces). `build_join_node`
accept-list extends to Semi/Anti/RightSemi/RightAnti, which are
unconditionally safe for aggregate pushdown because they never project
the non-preserved side. `build_scan_node` classifies baserestrictinfo
into search / top-level SubPlans / OR-nested SubPlans, batches search
predicates into a single strict `extract_quals` call (no more silent
drop on un-translatable RIs), declines on OR-nested or unliftable
SubPlans, and lifts top-level SubPlans via `wrap_with_semi_anti`.
`populate_required_fields` switches from `(rti, attno)` lookup to
`(plan_position, attno)`. rti is only unique within a single
PlannerInfo, so SubPlan-derived sources can share an rti with the outer
scan and the old lookup matched the wrong source.

datafusion/translator.rs: `build_null_aware_anti_join` lowers
`Anti { null_aware: true }` lifts to `LogicalPlan::Join` with
`null_equality=NullEqualsNothing` and `null_aware=true`, which
DataFusion's HashJoinExec uses to emit zero rows when the probe (inner)
side has any NULL. Drops the dead `JoinTypeAllowList::EquiOnly` enum
variant; both callers always passed `All` after the agg path's allow-
list flip.

aggregatescan/datafusion_exec.rs + aggregatescan/mod.rs: plumb
`ExprContext` + `PlanState` from the executor's runtime through
`build_join_aggregate_plan` to `build_relnode_df` to `build_source_df`
into each per-relation `PgSearchTableProvider`, so HeapFilter queries
(runtime expressions like `=` on a `pdb.literal`-cast column) can
resolve. Skip the ExprContext allocation under `EXEC_FLAG_EXPLAIN_ONLY`
where no expressions execute.

## Tests

tests/pg_regress/sql/aggregate_join_semi_anti.sql:

  - Test 1: `IN (SELECT ...)` pulls up to Semi -> AggregateScan
  - Test 2: `EXISTS / NOT EXISTS` -> AggregateScan
  - Test 3: single-column `NOT IN` un-pulled-up -> null-aware Anti lift,
    AggregateScan; no NULL inner so result equivalent to plain LeftAnti
  - Test 4: parity with `enable_aggregate_custom_scan = off` for Test 3
  - Test 5: multi-column `(a,b) NOT IN (SELECT x,y FROM t)` declines
    cleanly with a precise WARNING; PG plan runs; result matches PG
    custom-scan-OFF
  - Test 6: single-column `NOT IN` with a NULL-bearing inner ->
    AggregateScan returns zero rows (SQL three-valued logic), parity
    with PG custom-scan-OFF, plus a sanity check that removing the
    NULL inner row makes the same query return non-zero rows

Note: DataFusion 53.1.0's null-aware mode is restricted to a single-
column equi-key (HashJoinExec runtime code only inspects `values[0]`);
multi-column NOT IN therefore can't ride the null-aware fast path and
is left for PG to handle. Sister PR #5006 covers the JoinScan-side
end-to-end path (which additionally needs a patched `datafusion-proto`
because upstream's logical Join proto schema is missing the
`null_aware` field).

Refs #4911.

…are semantics (#5005)

## Summary

Makes the agg-on-join path handle `IN (SELECT ...)`, `NOT IN (SELECT
...)`, `EXISTS (SELECT ...)`, and `NOT EXISTS (SELECT ...)` end-to-end,
including the `NOT IN` against a NULL-bearing inner case, which is the
hard one because of SQL's three-valued NULL logic.

Target query shape - aggregate over a join with IN/NOT IN sublinks plus
a BM25 search predicate:

```sql
SELECT contact_job_title, COUNT(*) AS doc_count
FROM contacts
WHERE contact_id IN     (SELECT ldf_id FROM contact_list WHERE list_id IN ('include_list'))
  AND contact_id NOT IN (SELECT ldf_id FROM contact_list WHERE list_id IN ('exclude_list'))
  AND contact_id @@@ paradedb.boolean(...)
GROUP BY contact_job_title
ORDER BY doc_count DESC LIMIT 10;
```

After this PR, this shape pushes down to a single `Custom Scan (ParadeDB
Aggregate Scan)` node and returns correct results in both NULL-bearing
and non-NULL inner cases.

## Coverage

| Query shape | Result |

|----------------------------------------------|-------------------------------------|
| `IN (SELECT ...)` PG-pulled-up | Pushed down (Semi) |
| `EXISTS / NOT EXISTS` | Pushed down (Semi/Anti) |
| Single-col `IN` un-pulled-up | Lifted to Semi, pushed down |
| Single-col `NOT IN`, no NULL inner | Lifted to null-aware Anti |
| Single-col `NOT IN`, NULL inner | Lifted, returns 0 rows |
| Multi-col `NOT IN` / `IN` | Declines cleanly, PG fallback |
| OR-nested SubPlan | Declines cleanly, PG fallback |

## What changed

Pre-PR the agg-on-join walker bailed on Semi/Anti shapes with one of:
`unexpected node type T_FromExpr in join tree`, `aggregate-on-join does
not support Semi/Anti JOIN`, or `Aggregate-on-join does not support Anti
JOIN`. Separately, un-pulled-up `IN`/`NOT IN` SubPlans in
`baserestrictinfo` were silently dropped by the per-RI `extract_quals`
loop, producing wrong row counts when push-down succeeded.

**Walker / accept-list.** `build_relnode_from_node` recognizes
`T_FromExpr` (the post-pull-up parse-tree shape PG produces) and
recurses into `build_relnode_from_fromexpr`. `build_join_node` extends
to `Semi`/`Anti`/`RightSemi`/`RightAnti`; all four are unconditionally
safe for aggregate pushdown because they never project the non-preserved
side. The translator's dead `JoinTypeAllowList::EquiOnly` enum is
dropped.

**SubPlan lifting.** `build_scan_node` classifies `baserestrictinfo`
into search predicates / top-level SubPlans / OR-nested SubPlans. Search
predicates batch into one strict `extract_quals` call (no silent drop).
OR-nested SubPlans decline upfront. Top-level SubPlans lift via shared
`wrap_with_semi_anti`, which now returns `Result<RelNode, String>`;
every former silent-skip path returns Err with a site-specific reason.
Both callers (new agg caller, existing JoinScan caller) propagate to a
clean decline. Side-effect: closes a latent silent-drop window in
JoinScan non-LIMIT queries that `is_limit_pushdown_safe` only caught for
LIMIT.

**Null-aware NOT IN.** `JoinType::Anti` becomes a struct variant `Anti {
null_aware: bool }`. The flag lives on the variant rather than as a
separate `JoinNode` field, so `(JoinType::Inner, null_aware: true)` is
unrepresentable in the type system. `wrap_with_semi_anti` constructs
`Anti { null_aware: is_anti }` for `NOT IN` lifts.
`build_null_aware_anti_join` lowers to `LogicalPlan::Join` with
`null_equality=NullEqualsNothing` and `null_aware=true`. DataFusion's
`HashJoinExec` then emits zero rows when the probe (inner) side has any
NULL, matching SQL three-valued logic.

**plan_position-stored targetlist refs.** Every agg-on-join targetlist
ref (`JoinGroupColumn`, `JoinAggColRef`, `AggOrderByEntry`,
`FilterExpr::ColumnRef`) carries a `plan_position` resolved once at
extraction time against the just-built `RelNode` tree; execution-time
column binding is a `plan_position` lookup. `rti` is only unique within
a single `PlannerInfo`, so post-lift trees that mix sources from
sub-PlannerInfos (e.g. SubPlans lifted by `wrap_with_semi_anti`) need a
`PlannerRootId` to disambiguate. Three new shared `RelNode` primitives
back this and unify with how JoinScan already addresses output columns:
`source_with(root_id, rti, attno)`, `plan_position(root_id, rti,
attno)`, `source_at_plan_position(plan_position)`. The FILTER build
context bundles `plan` + `outer_root_id` into
`Option<FilterPlanResolution>` so the two can't go out of sync.

**Executor plumbing.** `ExprContext` + `PlanState` are threaded from the
executor's runtime into each per-relation `PgSearchTableProvider`.
HeapFilter queries (runtime expressions like `=` on a `pdb.literal`-cast
column) need a live evaluation context. Skip the `ExecAssignExprContext`
allocation under `EXEC_FLAG_EXPLAIN_ONLY`.

## DataFusion null-aware single-column limitation

DataFusion 53.1.0's null-aware mode is restricted to a single-column
equi-key. The validation in `HashJoinExec::build` rejects multi-column
null-aware:

```rust
if exec.null_aware && on.len() != 1 {
    return plan_err!("null_aware anti join only supports single column join key, got {} columns", on.len());
}
```

The runtime stream code only inspects `state.values[0]` and
`left_data.values()[0]`. Multi-column `NOT IN` therefore can't ride the
null-aware fast path; this PR declines pushdown and lets PG's
`nodeSubplan.c::ExecHashSubPlan` handle them.

## Why this works without a `datafusion-proto` patch

Sister PR #5006 noted that `datafusion-proto 53.1.0` is missing
`null_aware` from the `LogicalPlan::Join` proto schema (oversight in
[apache/datafusion#19635](apache/datafusion#19635);
added everywhere except the logical Join proto). This bites consumers
that round-trip `LogicalPlan` through the proto codec.

**The agg-on-join path is unaffected.** The agg executor builds a
`LogicalPlan` in `build_join_aggregate_plan`, hands it to
`build_physical_plan` in the same Rust process, and runs the physical
plan via `physical_plan.execute(...)`. No proto serialization. The
`null_aware` flag travels purely through Rust struct fields from
construction to execution. The proto bug only matters for the JoinScan
path (which serializes its `LogicalPlan` for parallel leader/worker IPC)
and is tracked separately in #5006.

## Test plan

`aggregate_join_semi_anti.sql` - six tests covering the full feature
surface:

- **Test 1**: `IN (SELECT ...)` pulls up to Semi -> AggregateScan
- **Test 2**: `EXISTS / NOT EXISTS` -> AggregateScan
- **Test 3**: single-column `NOT IN` un-pulled-up -> null-aware Anti
lift, AggregateScan
- **Test 4**: parity with `enable_aggregate_custom_scan = off` for Test
3
- **Test 5**: multi-column `(a,b) NOT IN (SELECT x,y FROM t)` declines
cleanly with a precise WARNING; PG plan runs; result matches PG
custom-scan-OFF
- **Test 6**: single-column `NOT IN` with a NULL-bearing inner ->
AggregateScan returns zero rows (SQL three-valued logic), parity with PG
custom-scan-OFF, plus a sanity check that removing the NULL inner row
makes the query return non-zero rows (guards against trivially passing
with zero rows for the wrong reason)

All other `aggregate_join_*` and `join_*` regress tests pass on PG 18
(`cargo pgrx regress`); `cargo check` + `cargo clippy -- -D warnings`
clean.

Refs #4911. Sister PR #5006 covers the JoinScan-side end-to-end via the
proto fork (separate dependency).