feat: Add LimitedBatchCoalescer to HashJoinExec + remove `HashJoi…
#18858
Conversation
…nExec` from `CoalesceBatches` Optimizer Rule
github-actions
bot
added
optimizer
|
|
||
| /// Batch coalescer for coalescing batches from the probe side | ||
| batch_coalescer: LimitedBatchCoalescer, | ||
| /// Flag to track if we've handled empty output to avoid returning empty schema |
jonathanc-n
left a comment
jonathanc-n
left a comment
There was a problem hiding this comment.
#18295 should be a nice follow up to this.
|
Anybody know how to quickly remove all the |
|
| Self { | ||
| partition, | ||
| schema, | ||
| schema: Arc::clone(&schema), |
There was a problem hiding this comment.
Do you need to clone here ? Usually this happens at the call site
| )? | ||
| }; | ||
|
|
||
| self.batch_coalescer.push_batch(result)?; |
There was a problem hiding this comment.
Maybe add a comment that ignoring PushBatchStatus::LimitReached result is intentional. Same at line 762 below.
|
|
||
| self.state = HashJoinStreamState::FetchProbeBatch; | ||
|
|
||
| return Ok(StatefulStreamResult::Ready(Some(result))); |
There was a problem hiding this comment.
Doesn't this also need to be improved ?
Currently it returns the result without pushing it to the coalescer.
I think it should call self.batch_coalescer.push_batch(result)?; and then check for a completed batch:
if let Some(batch) = self.batch_coalescer.next_completed_batch() {
return Ok(StatefulStreamResult::Ready(Some(batch)));
}
return Ok(StatefulStreamResult::Continue);
Also for sqllogictests: |
3 participants
…nExec
fromCoalesceBatches` Optimizer RuleWhich issue does this PR close?
BatchCoalescerintoHashJoinExecand remove fromCoalesceBatchesoptimization rule #18781 .What changes are included in this PR?
LimitedBatchCoalescerin HashJoinExecAre these changes tested?
Existing tests