feat(frontend): e2e explain join and add test (#877)

* feat(frontend): e2e explain join and add test

Signed-off-by: TennyZhuang <[email protected]>

* rest

Signed-off-by: TennyZhuang <[email protected]>

* better format

Signed-off-by: TennyZhuang <[email protected]>

* fix clippy

Signed-off-by: TennyZhuang <[email protected]>

e2e_test/v2/join.slt

@@ -0,0 +1,19 @@
+statement ok
+create table t1 (v1 int not null, v2 int not null);
+
+statement ok
+create table t2 (v1 int not null, v2 int not null);
+
+query T
+explain select t1.v2, t2.v2 from t1 join t2 on t1.v1 = t2.v1;
+----
+BatchProject { exprs: [$1, $3], expr_alias: [None, None] }
+    BatchHashJoin(predicate: $1 = $3)
+        BatchScan { table: "t1", columns: ["v1", "v2"] }
+        BatchScan { table: "t1", columns: ["v1", "v2"] }
+
+statement ok
+drop table t1;
+
+statement ok
+drop table t2;

rust/frontend/src/expr/input_ref.rs

@@ -1,3 +1,5 @@
+use std::fmt;
+
 use risingwave_common::types::DataType;
 
 use super::Expr;
@@ -7,6 +9,13 @@ pub struct InputRef {
     index: usize,
     data_type: DataType,
 }
+
+impl fmt::Display for InputRef {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "${}", self.index + 1)
+    }
+}
+
 impl InputRef {
     pub fn new(index: usize, data_type: DataType) -> Self {
         InputRef { index, data_type }

rust/frontend/src/optimizer/plan_node/batch_hash_join.rs

@@ -43,8 +43,8 @@ impl BatchHashJoin {
 }
 
 impl fmt::Display for BatchHashJoin {
-    fn fmt(&self, _f: &mut fmt::Formatter) -> fmt::Result {
-        todo!()
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "BatchHashJoin(predicate: {})", self.eq_join_predicate())
     }
 }
 

rust/frontend/src/optimizer/plan_node/eq_join_predicate.rs

@@ -1,3 +1,5 @@
+use std::fmt;
+
 use fixedbitset::FixedBitSet;
 
 use crate::expr::{get_inputs_col_index, ExprImpl, ExprType, FunctionCall, InputRef};
@@ -15,7 +17,23 @@ pub struct EqJoinPredicate {
     eq_keys: Vec<(InputRef, InputRef)>,
 }
 
-#[allow(dead_code)]
+impl fmt::Display for EqJoinPredicate {
+    fn fmt(&self, f: &mut fmt::Formatter) -> std::fmt::Result {
+        let mut eq_keys = self.eq_keys().iter();
+        if let Some((k1, k2)) = eq_keys.next() {
+            write!(f, "{} = {}", k1, k2)?;
+        }
+        for (k1, k2) in eq_keys {
+            write!(f, "AND {} = {}", k1, k2)?;
+        }
+        if !self.other_cond.is_empty() {
+            write!(f, "AND {}", self.other_cond)?;
+        }
+
+        Ok(())
+    }
+}
+
 impl EqJoinPredicate {
     /// The new method for `JoinPredicate` without any analysis, check or rewrite.
     pub fn new(other_cond: Condition, eq_keys: Vec<(InputRef, InputRef)>) -> Self {

rust/frontend/src/utils/condition.rs

@@ -1,3 +1,5 @@
+use std::fmt;
+
 use fixedbitset::FixedBitSet;
 use risingwave_common::types::{DataType, ScalarImpl};
 
@@ -12,7 +14,24 @@ pub struct Condition {
     pub conjunctions: Vec<ExprImpl>,
 }
 
+impl fmt::Display for Condition {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let mut conjunctions = self.conjunctions.iter();
+        if let Some(expr) = conjunctions.next() {
+            write!(f, "{:?}", expr)?;
+        }
+        for expr in conjunctions {
+            write!(f, "AND {:?}", expr)?;
+        }
+        Ok(())
+    }
+}
+
 impl Condition {
+    pub fn is_empty(&self) -> bool {
+        self.conjunctions.is_empty()
+    }
+
     pub fn with_expr(expr: ExprImpl) -> Self {
         Self {
             conjunctions: to_conjunctions(expr),