x

Author: forsaken628

src/query/service/src/pipelines/processors/transforms/new_hash_join/mod.rs

@@ -26,6 +26,7 @@ mod transform_hash_join;
 pub use grace::GraceHashJoin;
 pub use hash_join_factory::HashJoinFactory;
 pub use join::Join;
+pub use join::JoinStream;
 pub use memory::BasicHashJoinState;
 pub use memory::InnerHashJoin;
 pub use runtime_filter::RuntimeFiltersDesc;

src/query/service/src/pipelines/processors/transforms/transform_loop_join.rs

@@ -18,6 +18,7 @@ use std::sync::Arc;
 use std::sync::Mutex;
 use std::sync::RwLock;
 
+use databend_common_base::base::ProgressValues;
 use databend_common_exception::Result;
 use databend_common_expression::BlockEntry;
 use databend_common_expression::DataBlock;
@@ -28,6 +29,8 @@ use databend_common_pipeline::core::Processor;
 use databend_common_pipeline::sinks::Sink;
 use databend_common_sql::plans::JoinType;
 
+use super::Join;
+use super::JoinStream;
 use crate::physical_plans::NestedLoopJoin;
 use crate::pipelines::executor::WatchNotify;
 use crate::sessions::QueryContext;
@@ -137,7 +140,6 @@ pub struct LoopJoinState {
 
     right_sinker_count: RwLock<usize>,
 
-    #[allow(dead_code)]
     join_type: JoinType,
 }
 
@@ -255,3 +257,49 @@ impl LoopJoinState {
         Ok(iter)
     }
 }
+
+impl Join for LoopJoinState {
+    fn add_block(&mut self, data: Option<DataBlock>) -> Result<()> {
+        let Some(right_block) = data else {
+            return Ok(());
+        };
+
+        let right = if matches!(self.join_type, JoinType::Left | JoinType::Full) {
+            let rows = right_block.num_rows();
+            let entries = right_block
+                .take_columns()
+                .into_iter()
+                .map(|entry| entry.into_nullable())
+                .collect::<Vec<_>>();
+            DataBlock::new(entries, rows)
+        } else {
+            right_block
+        };
+        self.right_table.write()?.push(right);
+        Ok(())
+    }
+
+    fn final_build(&mut self) -> Result<Option<ProgressValues>> {
+        let progress = self.right_table.read()?.iter().fold(
+            ProgressValues::default(),
+            |mut progress, block| {
+                progress.rows += block.num_rows();
+                progress.bytes += block.memory_size();
+                progress
+            },
+        );
+        Ok(Some(progress))
+    }
+
+    fn probe_block(&mut self, data: DataBlock) -> Result<Box<dyn JoinStream + '_>> {
+        todo!();
+    }
+}
+
+struct LoopJoinStream {}
+
+impl JoinStream for LoopJoinStream {
+    fn next(&mut self) -> Result<Option<DataBlock>> {
+        todo!()
+    }
+}