perf: Support LEFT/RIGHT OUTER Spatial Joins for RangeJoinExec and DistanceJoinExec

[petern48]

We currently only support other join types for BroadcastIndexJoinExec. In #2561, a contributor added docs for a workaround for how they achieved using Sedona's RangeJoinExec for a LEFT JOIN. It involved rewriting the query with a CTE. While it works, we should support this behavior natively, so users don't have to resort to this workaround or, worse, move forward with the unoptimized behavior. Currently, we're falling back to BroadcastNestedLoopJoin from Spark if BroadcastIndexJoinExec isn't triggered.

Here's the workaround in the docs. We should also remove this from the docs once this is achieved.

WITH inner_join AS (
    SELECT
        dfA.a_id
        , dfB.b_id
    FROM dfA, dfB
    WHERE ST_INTERSECTS(dfA.geometry, dfB.geometry)
)

SELECT
    dfA.*,
    inner_join.b_id
FROM dfA
LEFT JOIN inner_join
   ON dfA.a_id = inner_join.a_id;

More context:

• Sedona LEFT JOIN using SPATIAL INDEX #2520 (reply in thread)
• Support for other join types for BroadcastIndexJoin (only works for smaller datasets): [SEDONA-191] Support join types other than inner in BroadcastIndexJoinExec #711

[petern48]

Hmm, ok. So we can get the matches easily with the traditional spatial join, but the challenge lies in finding all of the non-matches on the left side in a scalable way (i.e. we can't use the same approach implemented in BroadcastIndexJoinExec). Spark traditionally will try a SortMergeJoin if the join key is linearly sortable, but spatial data is not, so that doesn't work for us. It instead falls back on a BroadcastNestedLoopJoinExec.

I think at a theoretical level, there isn't a better way to extend support to outer joins aside from the workaround mentioned in the PR: post-filter the results after the spatial inner join. I think it would technicall possible to implement that natively in the query optimizer, but two complications come to mind:

1. We can't assume an id already exists in the table (which is needed to determine if there are no matches across all partitions. We could technically first generate a monotonically_increasing_id() column, though that would be expensive too (sorting required). Not sure how easy it would be to auto-detect an unique id column from code.

2. This would be a logical plan rewrite based on the physical plan, since we don't want to modify behavior for cases small enough to use BroadcastIndexJoinExec.

Generate PhysicalPlan from LogicalPlan (normal behavior)
If JoinExec is not BroadcastIndexJoinExec,
    Then perform the above logical plan rewrite.
    Then, go back and regenerate the physical plan from the new Logical Plan

Anyways, I think that implementation is more complicated than it is worth, for now. Didn't think about it too much when I first created the issue. Closing.