feat: agg-on-join end-to-end IN/NOT IN/EXISTS/NOT EXISTS with null-aware semantics - #5005

Adds `joinscan_notin_null_safety.sql` — a multi-table query with a
NULL-bearing inner column that exercises the JoinScan path's NOT IN
handling. SQL three-valued logic requires the query to return 0 rows
(any NULL in the inner key column poisons NOT IN under WHERE);
plain DataFusion `LeftAnti` would silently return non-zero (NOT
EXISTS semantics).

The pg_search-side null_aware code that makes this query return the
correct answer is supplied by PR #5005 (which lifts NOT IN SubPlans
to `JoinType::Anti` with `null_aware=true`). This PR adds the
end-to-end regress test that exercises that code path through
JoinScan's leader-worker IPC.

End-to-end correctness for this test additionally requires a patched
`datafusion-proto`: upstream Apache DataFusion 53.1.0 added the
`null_aware` field to `LogicalPlan::Join` (PR apache/datafusion#19635)
and threaded it through the optimizer + physical plan + physical
HashJoinExec proto, but missed adding it to the LogicalPlan Join
proto in `datafusion/proto/proto/datafusion.proto`. The encoder
destructures `LogicalPlan::Join` with `..` and silently drops the
flag.

Most DataFusion users don't notice because they execute logical
plans in-process. pg_search's JoinScan serializes its LogicalPlan
to bytes via `datafusion_proto::bytes::logical_plan_to_bytes_with_extension_codec`
for Postgres' parallel-execution leader→worker IPC (see
`pg_search/src/scan/codec.rs::serialize_logical_plan`), so the
proto round-trip drops the flag and reverts to the silent-wrong-
answer behavior.

Without the patched `datafusion-proto`, this test will return 89
rows (NOT EXISTS semantics) instead of the correct 0 rows. The
patch is a ~10-line follow-up to apache/datafusion#19635: add
`bool null_aware = 9` to `message JoinNode`, propagate in to_proto,
restore in from_proto via `Join::try_new(.., null_aware=true)`.

Note: the agg-on-join path (PR #5005) does NOT need the proto fork
because it executes in-process and never serializes its logical
plan. Only the JoinScan path is affected by the proto bug.

Dependencies (must land before this PR's CI passes):
  1. PR #5005 (Semi/Anti agg-on-join unlock + Path B v2 + null_aware
     code) — provides the pg_search-side null_aware plumbing.
  2. A patched `datafusion-proto` published as
     `paradedb/datafusion-proto` (or equivalent), pinned via
     `[patch.crates-io]` in the workspace `Cargo.toml`.

Once both land, this test exercises the full chain and asserts the
correct (0-row) result on JoinScan-on, with `LeftAnti` (not the
swapped `RightAnti`) visible in the EXPLAIN output.

Refs #4911 — JoinScan NOT IN portion.

#4988's MPP plumbing was not backported to 0.23.x, so #5005's diff
could not anchor on its context lines. Drop mpp_ctx / MppPlanContext;
keep expr_context / planstate plumbing; remove the 8 #4988-only
methods git dragged into the 456-line conflict block in mod.rs.

Adjust aggregate_join_semi_anti.out to match what 0.23.x's
explain_custom_scan emits (no DataFusion Physical Plan section).

cargo check / clippy clean; aggregate_join_* and join_semi_anti*
regress tests pass on pg18.

…are semantics (#5063)

# Description
Backport of #5005 to `0.23.x`.

---------

Co-authored-by: Mithun Chicklore Yogendra <[email protected]>

Adds `joinscan_notin_null_safety.sql` — a multi-table query with a
NULL-bearing inner column that exercises the JoinScan path's NOT IN
handling. SQL three-valued logic requires the query to return 0 rows
(any NULL in the inner key column poisons NOT IN under WHERE);
plain DataFusion `LeftAnti` would silently return non-zero (NOT
EXISTS semantics).

The pg_search-side null_aware code that makes this query return the
correct answer is supplied by PR #5005 (which lifts NOT IN SubPlans
to `JoinType::Anti` with `null_aware=true`). This PR adds the
end-to-end regress test that exercises that code path through
JoinScan's leader-worker IPC.

End-to-end correctness for this test additionally requires a patched
`datafusion-proto`: upstream Apache DataFusion 53.1.0 added the
`null_aware` field to `LogicalPlan::Join` (PR apache/datafusion#19635)
and threaded it through the optimizer + physical plan + physical
HashJoinExec proto, but missed adding it to the LogicalPlan Join
proto in `datafusion/proto/proto/datafusion.proto`. The encoder
destructures `LogicalPlan::Join` with `..` and silently drops the
flag.

Most DataFusion users don't notice because they execute logical
plans in-process. pg_search's JoinScan serializes its LogicalPlan
to bytes via `datafusion_proto::bytes::logical_plan_to_bytes_with_extension_codec`
for Postgres' parallel-execution leader→worker IPC (see
`pg_search/src/scan/codec.rs::serialize_logical_plan`), so the
proto round-trip drops the flag and reverts to the silent-wrong-
answer behavior.

Without the patched `datafusion-proto`, this test will return 89
rows (NOT EXISTS semantics) instead of the correct 0 rows. The
patch is a ~10-line follow-up to apache/datafusion#19635: add
`bool null_aware = 9` to `message JoinNode`, propagate in to_proto,
restore in from_proto via `Join::try_new(.., null_aware=true)`.

Note: the agg-on-join path (PR #5005) does NOT need the proto fork
because it executes in-process and never serializes its logical
plan. Only the JoinScan path is affected by the proto bug.

Dependencies (must land before this PR's CI passes):
  1. PR #5005 (Semi/Anti agg-on-join unlock + Path B v2 + null_aware
     code) — provides the pg_search-side null_aware plumbing.
  2. A patched `datafusion-proto` published as
     `paradedb/datafusion-proto` (or equivalent), pinned via
     `[patch.crates-io]` in the workspace `Cargo.toml`.

Once both land, this test exercises the full chain and asserts the
correct (0-row) result on JoinScan-on, with `LeftAnti` (not the
swapped `RightAnti`) visible in the EXPLAIN output.

Refs #4911 — JoinScan NOT IN portion.

…n-proto) (#5006)

## Summary

Adds `joinscan_notin_null_safety.sql` -- a multi-table regress test that
exercises the JoinScan path's `NOT IN` handling against a NULL-bearing
inner column. Asserts the result is 0 rows (matching SQL `NOT IN`
three-valued logic -- any NULL in the inner key column poisons NOT IN
under WHERE) and that the EXPLAIN does not show a swapped `RightAnti`
(which would lose null-aware semantics).

This PR is a regress test only. The pg_search-side null_aware code that
makes the test return the correct answer is supplied by sister PR #5005
-- that PR adds the `null_aware` field on `JoinNode`, lifts `NOT IN`
SubPlans to `JoinType::Anti` with `null_aware=true` in
`wrap_with_semi_anti`, and routes through
`LogicalPlanBuilder::join_detailed_with_options(.., null_aware=true)` in
the translator.

## Why this regress test needs a patched `datafusion-proto`

DataFusion fully implements null-aware NOT IN -- `LogicalPlan::Join` has
a `null_aware: bool` field, the optimizer routes NOT IN with nullable
join keys through it, the physical planner threads it down to
`HashJoinExec`, and `HashJoinExec` tracks `probe_side_has_null` to emit
zero rows when needed. All of this was added in
[apache/datafusion#19635](apache/datafusion#19635)
and ships in DataFusion 53.1.0.

But that PR missed one symmetric proto change: `null_aware` was added to
the **physical** `HashJoinExecNode` proto in `datafusion-proto` but
**not** to the **logical** `JoinNode` proto. The encoder destructures
`LogicalPlan::Join` with `..` and silently drops `null_aware` during
serialization.

```protobuf
// datafusion-proto-53.1.0/proto/datafusion.proto:244 (line is the JoinNode for the LogicalPlan)
message JoinNode {
  LogicalPlanNode left = 1;
  LogicalPlanNode right = 2;
  // ... 6 more fields ...
  // null_aware MISSING -- should be `bool null_aware = 9;`
}
```

```rust
// datafusion-proto-53.1.0/src/logical_plan/mod.rs:1316
LogicalPlan::Join(Join {
    left, right, on, filter, join_type, join_constraint, null_equality,
    ..   // <- `null_aware` falls into "the rest" and is silently dropped
}) => { /* serialize without null_aware */ }
```

**pg_search's JoinScan path serializes its `LogicalPlan` to bytes** for
Postgres' parallel-execution leader/worker IPC (see
`pg_search/src/scan/codec.rs::serialize_logical_plan`), so the
round-trip drops the flag and reverts to the silent-wrong-answer
behavior. The agg-on-join path (covered by #5005) does NOT serialize the
LogicalPlan -- it executes in-process -- and is unaffected.

Empirically demonstrated:

| Configuration | Rows returned |
|---|---|
| Pure PG (correct ground truth) | 0 |
| pg_search JoinScan ON, no proto patch | 89 (wrong -- `LeftAnti`
semantics, not `NOT IN`) |
| pg_search JoinScan ON, patched datafusion-proto | 0 (correct) |

## What the test covers

The query is intentionally minimal -- single outer table, single inner
list table with a NULL-bombed exclude set, single-column equi-key --
chosen to exercise:

1. JoinScan accepts the lifted `NOT IN` SubPlan (via #5005's
`wrap_with_semi_anti` change).
2. JoinScan's translator routes `null_aware=true` through
`LogicalPlanBuilder::join_detailed_with_options(.., null_aware=true)`.
3. JoinScan serializes the LogicalPlan via
`datafusion_proto::bytes::logical_plan_to_bytes_with_extension_codec`
for IPC.
4. The patched `datafusion-proto` preserves `null_aware` through
round-trip.
5. DataFusion's physical `HashJoinExec` runs with `null_aware=true`,
refuses the `LeftAnti`->`RightAnti` swap (guarded in `join_selection`),
and returns 0 rows when the probe side has NULL.

## Dependencies (must land before this PR's CI passes)

1. **#5005** -- adds the pg_search-side `null_aware` plumbing and
`wrap_with_semi_anti` lift.
2. **A published patched `datafusion-proto`** -- for example a
`paradedb/datafusion-proto` fork containing the missing `null_aware`
field on `LogicalPlan::JoinNode` plus the symmetric `to_proto` /
`from_proto` code. Pinned via `[patch.crates-io]` in the workspace
`Cargo.toml`.
3. **An upstream issue / PR at apache/datafusion** as a follow-up to
#19635 -- to add the missing field to the logical Join proto. Filed as
[apache/datafusion#22065](apache/datafusion#22065).
Once merged and released, the `[patch.crates-io]` entry can be dropped.

The proto patch itself is small (~10 lines): add `bool null_aware = 9`
to `message JoinNode`, propagate `null_aware` through `to_proto`'s
destructure, restore via `Join::try_new(.., null_aware: true)` in
`from_proto` when set. Local fork at `/tmp/datafusion-fork` works
empirically end-to-end; needs to be hosted somewhere reachable for CI.

## What CI will look like

| State | CI result |
|---|---|
| #5005 not merged, no proto patch | Test returns `89` (wrong); CI
fails. |
| #5005 merged, no proto patch | Test returns `89` (wrong); CI fails. |
| #5005 merged, proto patch in `Cargo.toml` `[patch.crates-io]` | Test
returns `0` (correct); CI passes. |

This PR is in the third state on the author's machine. To merge to main,
the `[patch.crates-io]` `Cargo.toml` change needs to ship as a follow-up
commit on this branch once the fork is published.

Refs #4911 -- JoinScan NOT IN portion. The agg-on-join NOT IN portion is
in #5005.

)

## What

In MPP, a semi, anti, outer or mark join over range co-partitioned
inputs now runs task-locally (`mode=Partitioned`) instead of
broadcasting or re-shuffling its inputs, as inner joins already do. A
null-aware anti join (`NOT IN`) keeps its broadcast.

Fixes the semi-join blocker from #5738 (repro:
`stuhood.semi-join-range`, included here as its first commit).

## Why

`RangePartitioningRule` stamps the same range split points on both sides
of every join, but `RangeCoPartitionedJoinRule` only flipped
`CollectLeft` **inner** joins to `Partitioned`:

```rust
if join.join_type() != &JoinType::Inner { return Ok(Transformed::no(node)); }
```

So a co-partitioned semi join had both sides split into ranges *and* the
build side broadcast to every task:

```
HashJoinExec: mode=CollectLeft, join_type=RightSemi
  [Stage 1] => NetworkBroadcastExec          ← users sent to every task
    t0: PgSearchScan: table=u, partition=user_id[-∞..34)
    t1: PgSearchScan: table=u, partition=user_id[34..∞)
```

## How

```rust
if join.null_aware {   // was: join.join_type() != &JoinType::Inner
    return Ok(Transformed::no(node));
}
```

**Why this is safe.** With identical split points on the join keys,
every matching pair has the same key and so falls in the same partition
on both sides. Whether a row matches (inner, semi, mark) or has no match
(outer, anti) is therefore decided within its own partition, and each
task's result is its share of the whole join.

| Join type | Task-local? |
|---|---|
| Inner, Semi, Mark | yes: all of a row's matches are in its partition |
| Left / Right / Full outer, Anti (`NOT EXISTS`) | yes: "no match here"
= "no match anywhere" |
| **Null-aware anti (`NOT IN`)** | **no**: one NULL key anywhere on the
build side must empty every task's result, and no task can see the other
partitions |

`null_aware` is only ever set on single-key `LeftAnti` joins
(`HashJoinExec` validation), which ParadeDB builds for `NOT IN`
(`joinscan/planning.rs`, #5005). DataFusion's own planner already
requires these to stay `CollectLeft` ("Null-aware joins must use
CollectLeft").

The final co-partitioning check is unchanged: after flipping, DataFusion
must confirm both inputs satisfy the join's distribution, otherwise the
`CollectLeft` join is kept.

## Tests

`mpp_range_partitioning`:
- **Scenario 7** (the repro): now `mode=Partitioned` with no broadcast
stage.
- **Scenario 8:** semi (`IN`), anti (`NOT EXISTS`), outer (`LEFT JOIN`),
`FULL JOIN` and mark (`IN` under `OR`) joins each run `Partitioned`;
their rows were checked against a serial run. The `FULL JOIN` sorts
unmatched rows from both sides first, so each appears exactly once.
- **Scenario 9:** `NOT IN` with a NULL `user_id` stays `CollectLeft` and
returns 0 rows. With the guard removed, MPP returned 67 rows against 0
serially.

Existing tests whose plans improve (rows unchanged): the `LEFT JOIN`s in
`mpp_joinscan` and `range_partition_null_keys` previously re-split both
range-partitioned inputs by hash over the network (2 shuffles each);
they now join the matching range slices task-locally with no shuffle
(`range_partition_null_keys` 2 → 0). `range_partition_null_keys`
includes posts with NULL owners, which still appear exactly once. The
`defer_column_fetch = off` case in `mpp_joinscan` needs a decode in a
stage with no scan of `f`, so it now joins on `p.id` (not the partition
key) to keep its shuffle (`mpp_joinscan` 6 → 4 shuffles in total).

pg_regress (PG18): 392 passed.

---------

Co-authored-by: Stu Hood <[email protected]>