update

Author: forsaken628

src/common/base/src/runtime/perf/query_perf.rs

@@ -54,7 +54,7 @@ impl QueryPerf {
             .blocklist(&["libc", "libgcc", "pthread", "vdso"])
             .set_filter_func(filter_closure)
             .build()
-            .map_err(|_e| ErrorCode::Internal("Failed to create profiler"))?;
+            .map_err(|e| ErrorCode::Internal(format!("Failed to create profiler, {e}")))?;
         debug!("starting perf with frequency: {}", frequency);
         let mut payload = ThreadTracker::new_tracking_payload();
         payload.perf_enabled = true;

src/query/service/src/physical_plans/physical_hash_join.rs

@@ -1219,7 +1219,21 @@ impl PhysicalPlanBuilder {
         let mut predicates =
             Vec::with_capacity(join.equi_conditions.len() + join.non_equi_conditions.len());
 
+        let is_simple_expr = |expr| {
+            matches!(
+                expr,
+                ScalarExpr::BoundColumnRef(_)
+                    | ScalarExpr::ConstantExpr(_)
+                    | ScalarExpr::TypedConstantExpr(_, _)
+            )
+        };
+
         for condition in &join.equi_conditions {
+            if !is_simple_expr(&condition.left) || !is_simple_expr(&condition.right) {
+                // todo: Filtering after cross join cause expression to be evaluated multiple times
+                return Ok(None);
+            }
+
             let scalar = condition_to_expr(condition)?;
             match resolve_scalar(&scalar, &merged) {
                 Ok(expr) => predicates.push(expr),

src/query/service/src/pipelines/processors/transforms/new_hash_join/memory/nested_loop.rs

@@ -69,7 +69,12 @@ impl<T: Join> Join for NestedLoopJoin<T> {
             return self.inner.probe_block(data);
         };
 
-        let nested = Box::new(LoopJoinStream::new(data, build_blocks));
+        let nested: Box<dyn JoinStream + '_> = if data.num_rows() <= *self.basic_state.build_rows {
+            Box::new(ConstProbeLoopJoinStream::new(data, build_blocks))
+        } else {
+            Box::new(ConstBuildLoopJoinStream::new(data, build_blocks))
+        };
+
         Ok(InnerHashJoinFilterStream::create(
             nested,
             &mut self.desc.filter,
@@ -78,15 +83,15 @@ impl<T: Join> Join for NestedLoopJoin<T> {
     }
 }
 
-struct LoopJoinStream<'a> {
+struct ConstProbeLoopJoinStream<'a> {
     probe_rows: VecDeque<Vec<Scalar>>,
     probe_types: Vec<DataType>,
     build_blocks: &'a [DataBlock],
     build_index: usize,
 }
 
-impl<'a> LoopJoinStream<'a> {
-    pub fn new(probe: DataBlock, build_blocks: &'a [DataBlock]) -> Self {
+impl<'a> ConstProbeLoopJoinStream<'a> {
+    fn new(probe: DataBlock, build_blocks: &'a [DataBlock]) -> Self {
         let mut probe_rows = vec![Vec::new(); probe.num_rows()];
         for entry in probe.columns().iter() {
             match entry {
@@ -103,22 +108,22 @@ impl<'a> LoopJoinStream<'a> {
             }
         }
 
-        let left_types = probe
+        let probe_types = probe
             .columns()
             .iter()
             .map(|entry| entry.data_type())
             .collect();
 
-        LoopJoinStream {
+        ConstProbeLoopJoinStream {
             probe_rows: probe_rows.into(),
-            probe_types: left_types,
+            probe_types,
             build_blocks,
             build_index: 0,
         }
     }
 }
 
-impl<'a> JoinStream for LoopJoinStream<'a> {
+impl<'a> JoinStream for ConstProbeLoopJoinStream<'a> {
     fn next(&mut self) -> Result<Option<DataBlock>> {
         let Some(probe_entries) = self.probe_rows.front() else {
             return Ok(None);
@@ -144,3 +149,52 @@ impl<'a> JoinStream for LoopJoinStream<'a> {
         Ok(Some(DataBlock::new(entries, build_block.num_rows())))
     }
 }
+
+struct ConstBuildLoopJoinStream<'a> {
+    probe: DataBlock,
+    build_blocks: &'a [DataBlock],
+    block_index: usize,
+    row_index: usize,
+}
+
+impl<'a> ConstBuildLoopJoinStream<'a> {
+    fn new(probe: DataBlock, build_blocks: &'a [DataBlock]) -> Self {
+        ConstBuildLoopJoinStream {
+            probe,
+            build_blocks,
+            block_index: 0,
+            row_index: 0,
+        }
+    }
+}
+
+impl<'a> JoinStream for ConstBuildLoopJoinStream<'a> {
+    fn next(&mut self) -> Result<Option<DataBlock>> {
+        let Some(build_block) = self.build_blocks.get(self.block_index) else {
+            return Ok(None);
+        };
+
+        let num_rows = self.probe.num_rows();
+        let entries = self
+            .probe
+            .columns()
+            .iter()
+            .cloned()
+            .chain(build_block.columns().iter().map(|entry| {
+                BlockEntry::Const(
+                    entry.index(self.row_index).unwrap().to_owned(),
+                    entry.data_type(),
+                    num_rows,
+                )
+            }))
+            .collect();
+
+        self.row_index += 1;
+        if self.row_index >= build_block.num_rows() {
+            self.row_index = 0;
+            self.block_index += 1;
+        }
+
+        Ok(Some(DataBlock::new(entries, num_rows)))
+    }
+}