[FEATURE][PERF] JOIN queries don't scale past a few million records

Is your feature request related to a problem?

Consider two indices with some shared key, each having at least a million records. If we want to execute a query like the following:

SELECT left.key, left.var1, right.var2 FROM idx_1 AS left JOIN idx_2 AS right ON left.key = right.key LIMIT 5

The current HashJoin implementation is going to page through the entirety of idx_2 while collecting the hash table, before it starts collecting results by going through index 1. This means the algorithm must page through the entirety of the index to start computing results, even with a small limit.

Once this process hits the 60s timeout, joins will start failing with IllegalStateExceptions (tested on a custom 2.19 domain):

[2025-06-06T00:37:00,730][ERROR][o.o.s.l.e.ElasticHitsExecutor] [773d4d7f8150cbc0a33b58112dd9897b] Failed during join query run.
java.lang.IllegalStateException: Error happened during execution
    at org.opensearch.sql.legacy.query.planner.physical.PhysicalPlan.execute(PhysicalPlan.java:60)
    at org.opensearch.sql.legacy.query.planner.core.QueryPlanner.execute(QueryPlanner.java:67)
    at org.opensearch.sql.legacy.executor.join.QueryPlanElasticExecutor.innerRun(QueryPlanElasticExecutor.java:30)
    at org.opensearch.sql.legacy.executor.join.ElasticJoinExecutor.run(ElasticJoinExecutor.java:97)
    at org.opensearch.sql.legacy.executor.QueryActionElasticExecutor.executeJoinSearchAction(QueryActionElasticExecutor.java:47)
    at org.opensearch.sql.legacy.executor.QueryActionElasticExecutor.executeAnyAction(QueryActionElasticExecutor.java:104)
    at org.opensearch.sql.legacy.executor.format.PrettyFormatRestExecutor.execute(PrettyFormatRestExecutor.java:76)
    at org.opensearch.sql.legacy.executor.format.PrettyFormatRestExecutor.execute(PrettyFormatRestExecutor.java:50)
    at org.opensearch.sql.legacy.executor.AsyncRestExecutor.doExecuteWithTimeMeasured(AsyncRestExecutor.java:154)
    at org.opensearch.sql.legacy.executor.AsyncRestExecutor.lambda$async$1(AsyncRestExecutor.java:110)
    at org.opensearch.sql.common.utils.QueryContext.lambda$withCurrentContext$0(QueryContext.java:61)
    at org.opensearch.common.util.concurrent.ThreadContext$ContextPreservingRunnable.run(ThreadContext.java:964)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
    at java.base/java.lang.Thread.run(Thread.java:1583)
Caused by: java.lang.IllegalStateException: Failed to prefetch next batch
    at org.opensearch.sql.legacy.query.planner.physical.node.BatchPhysicalOperator.prefetchSafely(BatchPhysicalOperator.java:74)
    at org.opensearch.sql.legacy.query.planner.physical.node.BatchPhysicalOperator.hasNext(BatchPhysicalOperator.java:47)
    at org.opensearch.sql.legacy.query.planner.logical.node.Top.hasNext(Top.java:41)
    at org.opensearch.sql.legacy.query.planner.logical.node.Project.hasNext(Project.java:78)
    at org.opensearch.sql.legacy.query.planner.physical.PhysicalPlan.doExecutePlan(PhysicalPlan.java:76)
    at org.opensearch.sql.legacy.query.planner.physical.PhysicalPlan.execute(PhysicalPlan.java:56)
    ... 14 more
Caused by: java.lang.IllegalStateException: Exit due to time out
    at org.opensearch.sql.legacy.query.planner.physical.node.BatchPhysicalOperator.prefetchSafely(BatchPhysicalOperator.java:78)
    at org.opensearch.sql.legacy.query.planner.physical.node.BatchPhysicalOperator.hasNext(BatchPhysicalOperator.java:47)
    at org.opensearch.sql.legacy.query.planner.physical.node.join.BlockHashJoin.probe(BlockHashJoin.java:66)
    at org.opensearch.sql.legacy.query.planner.physical.node.join.JoinAlgorithm.probeMatchAndBookkeepMismatch(JoinAlgorithm.java:139)
    at org.opensearch.sql.legacy.query.planner.physical.node.join.JoinAlgorithm.prefetch(JoinAlgorithm.java:116)
    at org.opensearch.sql.legacy.query.planner.physical.node.BatchPhysicalOperator.prefetchSafely(BatchPhysicalOperator.java:72)
    ... 19 more

This puts a hard scalability limit on these types of join queries, at around 5 million records or so.

What solution would you like?

It's hard to apply a "proper" hash join, because we would definitely need to pay the pagination cost of going through an index to collect all the hashes. I see two other ways we could consider working around this:

1. Do away with hash join, only use merge join.

In the case where the key can be sorted, this does mean we need to pay the sorting cost, but I think this cost is cheaper than the full pagination. On the same domain I used to repro the timeouts, OpenSearch can return a sorted page in under a second. This allows us to execute that query by traversing the sorted pages, which can then push down the LIMIT query. If the keys are "roughly" evenly distributed, we might not even need to pull a second page:

left  right
---------------------
0
1
      2
3
4     4     -- Match!
      5
6     6     -- Match!
...

If we have a case where the IDs are skewed, we can call for the right page with search_after to skip directly to the next relevant page.

left  right
---------------------
      0     -- Low IDs skipped as we go through left, because...
      1
      ...
1000        -- We pass search_after 1000
1001
1002  1002  -- Match!
...

2. Go with HashJoin by blocks

This relies on more luck to work, but has lower theoretical complexity. We can try and find hash matches one block at a time, as we collect pages from the right index. This means we don't need to page through the whole index to find results if there's a lot of matches, but may end up scanning the full indices if matches are sparse. I think this would overall be less reliable than merge join, but much faster if we get those lucky hits. It would be most effective if we know ahead of time that the IDs are ordered in each index, and that the results are nearly-one-to-one.

This approach is discussed in more detail in https://cratedb.com/blog/lab-notes-how-we-made-joins-23-thousand-times-faster-part-three.

What alternatives have you considered?
N/A

Do you have any additional context?

• We have a benchmark for Hash Join from a while ago, but it hasn't been re-run in years. We should include a benchmark for these types of queries in OpenSearch Benchmark.
• This can potentially be done as part of the Calcite migration, up to how the merge queries are implemented physically.

Start with the legacy physical join path around HashJoin and BlockHashJoin, then read docs/dev/testing-hash-join-benchmark.md. Compare the proposed merge-join and block-based approaches and assess how Calcite migration affects physical implementation. Define benchmarks for limited joins at multi-million-record scale, including coverage in OpenSearch Benchmark, and verify result and timeout behavior.