(
&self,
partition: usize,
context: Arc<TaskContext>,
)
| 597 | } |
| 598 | |
| 599 | fn execute( |
| 600 | &self, |
| 601 | partition: usize, |
| 602 | context: Arc<TaskContext>, |
| 603 | ) -> Result<SendableRecordBatchStream> { |
| 604 | assert_eq_or_internal_err!( |
| 605 | self.left.output_partitioning().partition_count(), |
| 606 | 1, |
| 607 | "Invalid NestedLoopJoinExec, the output partition count of the left child must be 1,\ |
| 608 | consider using CoalescePartitionsExec or the EnforceDistribution rule" |
| 609 | ); |
| 610 | |
| 611 | let metrics = NestedLoopJoinMetrics::new(&self.metrics, partition); |
| 612 | let batch_size = context.session_config().batch_size(); |
| 613 | |
| 614 | // update column indices to reflect the projection |
| 615 | let column_indices_after_projection = match self.projection.as_ref() { |
| 616 | Some(projection) => projection |
| 617 | .iter() |
| 618 | .map(|i| self.column_indices[*i].clone()) |
| 619 | .collect(), |
| 620 | None => self.column_indices.clone(), |
| 621 | }; |
| 622 | |
| 623 | let right_partition_count = self.right().output_partitioning().partition_count(); |
| 624 | |
| 625 | // Always try to buffer all left data in memory via OnceFut. |
| 626 | // If that fails with OOM, the stream will fallback to memory-limited |
| 627 | // mode (if conditions allow). |
| 628 | let load_reservation = |
| 629 | MemoryConsumer::new(format!("NestedLoopJoinLoad[{partition}]")) |
| 630 | .register(context.memory_pool()); |
| 631 | |
| 632 | let build_side_data = self.build_side_data.try_once(|| { |
| 633 | let stream = self.left.execute(0, Arc::clone(&context))?; |
| 634 | |
| 635 | Ok(collect_left_input( |
| 636 | stream, |
| 637 | metrics.join_metrics.clone(), |
| 638 | load_reservation, |
| 639 | need_produce_result_in_final(self.join_type), |
| 640 | right_partition_count, |
| 641 | )) |
| 642 | })?; |
| 643 | |
| 644 | let probe_side_data = self.right.execute(partition, Arc::clone(&context))?; |
| 645 | |
| 646 | // Determine if OOM fallback to memory-limited mode is possible. |
| 647 | // Conditions: |
| 648 | // 1. Disk manager supports temp files (needed for spilling). |
| 649 | // 2. FULL join with multiple right partitions is not yet supported |
| 650 | // in the fallback path. FULL join needs to track BOTH left-side |
| 651 | // matches (for unmatched left rows) AND right-side matches (for |
| 652 | // unmatched right rows). The fallback path builds a per-partition |
| 653 | // `JoinLeftData` with `probe_threads_counter == 1`, so each |
| 654 | // partition emits unmatched left rows based only on its own |
| 655 | // right-side matches, producing incorrect duplicate output for |
| 656 | // left rows that match in another partition. Other join types |
no test coverage detected