fix: Normalize CollectLeft Joins and similar

[barbarj]

Closes #563

(This is the plan-fix and tests part of #582)

The bug in 563 is actually a failure to enforce a plan shape invariant for multi-task stages: for every stage, executing its plan once per task over the per-task input assignment and unioning the outputs must be equivalent to executing it once over all the data.

This PR applies fixes for the problematic shapes (normalize_collect_joins.rs). Enforcing this invariant will be handled separately. (See discussion in #582).

There are 5 invalid plan shapes to fix. The following strategies are used:

• HashJoin CollectLeft (join types: Left/LeftSemi/LeftAnti/LeftMark/Full) → change
  CollectLeft->Partitioned (keys exist; no replication needed)
• HashJoinCollectLeft, null_aware (join type: LeftAnti) -> cap at 1 task
• NLJ (join types: Left/LeftSemi/LeftAnti/LeftMark) -> call swap_inputs(), then broadcast as usual. (makes the join shape safe for broadcast).
• NLJ (join type: Full) -> cap at 1 task
• CrossJoin (always safe with broadcast) -> cap at 1 task if broadcasts are disabled.

The CollectLeft->Partitioned change causes the rewriting of a bunch of plan shapes in the plan tests. The problematic stages run over small data, and thus collapse to a single task during
prepare_network_boundaries, avoiding the bug. Due to the shape normalization pass necessarily running before task counts are decided, we can't see that the task consolidation will happen and must modify the plan.

In-repo benchmarks show equivalent or better performance for all queries, save one: TPCH q22. It drops by about 40% (~17ms -> ~24ms on my machine, 8 workers, 2 threads). As far as I can tell, it's losing a dynamic filter due to being cut into more stages.

[jayshrivastava]

👋🏽 Left comments in the other PR which are relevant in this one #582

[barbarj]

👋🏽 Left comments in the other PR which are relevant in this one #582

Comments addressed. I'm now normalizing Full HashExec joins and ensuring the HashExecJoin's internal properties are maintained with a unit test and the use of HashJoinExecBuilder.

[gabotechs]

This is looking very good!

Just some minor comments, I'd like to have @jayshrivastava chime in as well as he provided very useful comments here.

[gabotechs]

Flushing a couple more comments as I go

[gabotechs]

Thanks @barbarj for this amazing piece of work 🙏! it was definitely a master lesson on how to do joins correctly.

Once we get the CI green, it should be fine to merge.

[barbarj]

Once we get the CI green, it should be fine to merge.

Thanks @gabotechs! CI should be green now. (I changed build_side_fetch_is_preserved_by_normalize to assert counts rather than row contents, since distributed execution of that fetch is non-deterministic in regards to which rows survive).

[gabotechs]

Github runners are not in the mood today…

[gabotechs]

Seems like the build_side_fetch_is_preserved_by_normalize test is failing because of some non-determinism in how parquet files are partitioned?

@barbarj, unless you have a better idea, I think we can omit asserting the plan snapshot there.

[barbarj]

Seems like the build_side_fetch_is_preserved_by_normalize test is failing because of some non-determinism in how parquet files are partitioned?

@barbarj, unless you have a better idea, I think we can omit asserting the plan snapshot there.

I've got nothing better. Removed!