[multistage] Make join operator more resilient

[xiangfu0]

To protect the server from OOM, this PR introduces the limit for join operator to ensure the memory consumption is limited.

1. Add new query options for join operator right table size limit:

• joinOverflowMode(string) : Overflow mode, allowed value are: THROW/BREAK
  • Server config: pinot.query.join.overflow.mode=THROW
  • Session level: SET joinOverflowMode='THROW'
  • QueryHint level: join_overflow_mode='THROW'
• maxRowsInJoin(integer) : the number of rows allowed for right table cache in a join operator
  • Server config: pinot.query.join.max.rows=100000
  • Session level: SET maxRowsInJoin=1000000
  • QueryHint level: max_rows_in_join='1000000'

2. Add server partial result exception
3. Support continuous processing when joinOverflowMode = BREAK;

TODO: Support size based limit.

Examples:

SET joinOverflowMode='BREAK';
SET maxRowsInJoin=10;
SELECT a.playerName, a.teamID, b.teamName 
FROM baseballStats_OFFLINE AS a
JOIN dimBaseballTeams_OFFLINE AS b
ON a.teamID = b.teamID

SELECT /*+ joinOptions(join_overflow_mode='BREAK', max_rows_in_join='10') */ a.playerName, a.teamID, b.teamName 
FROM baseballStats_OFFLINE AS a
JOIN dimBaseballTeams_OFFLINE AS b
ON a.teamID = b.teamID

Sample screenshot for:
0. Normal run without limit hit: (Joined result is 47735)

1. Query option: Throw exception when row limit hit:

2. Query option: Continue when row limit hit: (Joined result is 29884)

3. JoinHints: Throw exception when row limit hit:

4. JoinHints: Continue when row limit hit: (Joined result is 29884)

More examples of difference between the threshold, and overflow mode:

[codecov-commenter]

Codecov Report

Merging #11401 (5c36a12) into master (32e47c7) will increase coverage by 0.02%.
The diff coverage is 63.51%.

@@             Coverage Diff              @@
##             master   #11401      +/-   ##
============================================
+ Coverage     62.89%   62.91%   +0.02%     
- Complexity     1079     1091      +12     
============================================
  Files          2301     2301              
  Lines        123699   123766      +67     
  Branches      18816    18833      +17     
============================================
+ Hits          77803    77873      +70     
+ Misses        40367    40355      -12     
- Partials       5529     5538       +9     

Flag	Coverage Δ
integration	<0.01% <0.00%> (ø)
integration1	<0.01% <0.00%> (ø)
integration2	0.00% <0.00%> (ø)
java-11	62.89% <63.51%> (+0.01%)	⬆️
java-17	62.76% <63.51%> (+0.02%)	⬆️
java-20	49.77% <63.51%> (+0.02%)	⬆️
temurin	62.91% <63.51%> (+0.02%)	⬆️
unittests	62.91% <63.51%> (+0.02%)	⬆️
unittests1	67.45% <63.51%> (+0.03%)	⬆️
unittests2	14.50% <0.00%> (-0.01%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files Changed	Coverage Δ
...pinot/query/planner/plannode/AbstractPlanNode.java	100.00% <ø> (ø)
...va/org/apache/pinot/query/runtime/QueryRunner.java	69.00% <31.57%> (-8.78%)	⬇️
...e/pinot/common/utils/config/QueryOptionsUtils.java	67.21% <66.66%> (-0.03%)	⬇️
...pinot/query/runtime/operator/HashJoinOperator.java	87.64% <71.11%> (-5.07%)	⬇️
.../apache/pinot/common/exception/QueryException.java	93.54% <100.00%> (+0.14%)	⬆️
.../query/planner/logical/RelToPlanNodeConverter.java	86.60% <100.00%> (ø)
.../apache/pinot/query/planner/plannode/JoinNode.java	92.30% <100.00%> (+0.64%)	⬆️
...va/org/apache/pinot/spi/utils/CommonConstants.java	22.09% <100.00%> (ø)

... and 7 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[Jackie-Jiang]

LGTM otherwise

[Jackie-Jiang]

Can you check if we read exception from the data block and EOS block? I feel they are just ignored

[xiangfu0]

they are ignored right now. Need to open up another thread to fix this.