Feat add sharding

pisa-proxy/app/config/Cargo.toml

@@ -13,6 +13,7 @@ api = { path = "../api" }
 clap = { version = "3.2.12", features = ["env"] }
 plugin = { path = "../../plugin" }
 proxy = { path = "../../proxy" }
+strategy = { path = "../../proxy/strategy" }
 reqwest = { version = "0.11.10", features = ["blocking", "json"] }
 runtime_mysql = { path = "../../runtime/mysql" }
 serde = { version = "1.0.133", default-features = false, features = ["derive"] }

pisa-proxy/app/config/src/config.rs

@@ -18,6 +18,7 @@ use std::{env, fs::File, io::prelude::*};
 use api::config::Admin;
 use clap::{value_parser, Arg, Command};
 use proxy::proxy::{MySQLNode, MySQLNodes, ProxiesConfig, ProxyConfig};
+use strategy::config::NodeGroup;
 use serde::{Deserialize, Serialize};
 use tracing::trace;
 
@@ -243,6 +244,7 @@ pub struct PisaProxyConfig {
     pub admin: Admin,
     pub proxy: Option<ProxiesConfig>,
     pub mysql: Option<MySQLNodes>,
+    pub nodegroup: Option<NodeGroup>,
     pub shardingsphere_proxy: Option<MySQLNodes>,
     pub version: Option<String>,
 }

pisa-proxy/app/server/src/server.rs

@@ -35,8 +35,8 @@ impl ProxyFactory for PisaProxyFactory {
         match kind {
             ProxyKind::MySQL => Box::new(runtime_mysql::mysql::MySQLProxy {
                 proxy_config: config,
-                // mysql_nodes: self.pisa_config.mysql_nodes.clone(),
                 mysql_nodes: self.pisa_config.get_mysql().to_vec(),
+                node_group: self.pisa_config.nodegroup.clone(),
                 pisa_version: self.pisa_config.get_version().to_string(),
                 // mysql_nodes: self.pisa_config.mysql.as_ref().unwrap().node.as_ref().unwrap().to_vec(),
                 // pisa_version: self.pisa_config.version.as_ref().unwrap().to_string(),

pisa-proxy/parser/mysql/src/ast/base.rs

@@ -895,8 +895,8 @@ pub enum Value {
 
     Num {
         span: Span,
-        // todo add parse i64 string
         value: String,
+        signed: bool,
     },
 
     HexNum {
@@ -907,8 +907,8 @@ pub enum Value {
 
     FloatNum {
         span: Span,
-        // todo add parse f64 string
         value: String,
+        signed: bool,
     },
 
     BinNum {
@@ -963,12 +963,17 @@ impl Value {
                 var_ident
             }
 
-            Self::Num { span: _, value } => (*value).to_string(),
+            Self::Num { span: _, signed, value } | Self::FloatNum { span:_, value, signed }=> {
+                let mut value = (*value).to_string();
+                if *signed {
+                    value.insert(0, '-');
+                }
+
+                value
+            }
 
             Self::HexNum { span: _, value } => (*value).to_string(),
 
-            Self::FloatNum { span: _, value } => (*value).to_string(),
-
             Self::BinNum { span: _, value } => (*value).to_string(),
 
             Self::True => "true".to_string(),

pisa-proxy/parser/mysql/src/ast/dml.rs

@@ -2714,21 +2714,15 @@ impl ValOrVals {
 pub struct InsertVals {
     pub span: Span,
     pub val_ident: ValOrVals,
-    pub values: Vec<Vec<Expr>>,
+    pub values: Vec<RowValue>,
 }
 
 impl InsertVals {
     pub fn format(&self) -> String {
         let mut values = Vec::with_capacity(3);
 
         for v in &self.values {
-            let row = vec![
-                "(".to_string(),
-                v.iter().map(|x| x.format()).collect::<Vec<String>>().join(","),
-                ")".to_string(),
-            ];
-
-            values.push(row.join(" "))
+            values.push(v.format())
         }
 
         vec![self.val_ident.format(), values.join(",")].join(" ")
@@ -2740,20 +2734,53 @@ impl Visitor for InsertVals {
         let mut new_values = Vec::with_capacity(self.values.len());
 
         for v in self.values.iter_mut() {
-            let mut sub_new_exprs = Vec::with_capacity(v.len());
-            for vv in v.iter_mut() {
-                let mut node = Node::Expr(vv);
-                tf.trans(&mut node);
+            let mut node = Node::RowValue(v);
+            tf.trans(&mut node);
+            let new_node = node.into_row_value().unwrap().visit(tf);
+            new_values.push(new_node);
+        }
 
-                let new_node = node.into_expr().unwrap().visit(tf);
-                sub_new_exprs.push(new_node);
-            }
+        self.values = new_values;
+
+        self.clone()
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct RowValue {
+    pub span: Span,
+    pub values: Vec<Expr>
+}
 
-            new_values.push(sub_new_exprs);
+impl RowValue {
+    fn format(&self) -> String {
+        let values = vec![ 
+            "(".to_string(),
+            self.values.iter().map(|x| x.format()).collect::<Vec<String>>().join(","),
+            ")".to_string(),
+        ];
+
+        values.join(" ")
+    }
+}
+
+
+impl Visitor for RowValue {
+    fn visit<T: Transformer>(&mut self, tf: &mut T) -> Self {
+        let mut new_values = Vec::with_capacity(self.values.len());
+
+        for v in self.values.iter_mut() {
+            let mut node = Node::Expr(v);
+            tf.trans(&mut node);
+
+            let new_node = node.into_expr().unwrap().visit(tf);
+            new_values.push(new_node);
         }
 
         self.values = new_values;
 
+        tf.complete(&mut Node::RowValue(self));
+
         self.clone()
     }
 }

pisa-proxy/parser/mysql/src/grammar.y

@@ -3704,10 +3704,13 @@ opt_equal -> Option<&'input str>:
   | equal       { Some($1) }
   ;
 
-row_value -> Vec<Expr>:
+row_value -> RowValue:
   '(' opt_values ')' 
   { 
-    $2
+    RowValue {
+      span: $span,
+      values: $2,
+    }
   }
   ;
 
@@ -4488,20 +4491,23 @@ NUM_literal -> Value:
       Value::Num {
         span: $span,
         value: $1,
+        signed: false,
       }
     }
   | 'DECIMAL_NUM'
     {
       Value::Num {
         span: $span,
         value: String::from($lexer.span_str($1.as_ref().unwrap().span())),
+        signed: false,
       }
     }
   | 'FLOAT_NUM'
     {
       Value::FloatNum {
         span: $span,
         value: String::from($lexer.span_str($1.as_ref().unwrap().span())),
+        signed: false,
       }
     }
   ;
@@ -4724,7 +4730,13 @@ signed_literal -> Value:
     }
   | '-' NUM_literal
     {
-      $2
+      match $2 {
+        Value::Num { span, value, signed:_ } => {
+          Value::Num { span, value, signed: true}
+        },
+
+        _ => unreachable!()
+      }
     }
   ;
 
@@ -5607,7 +5619,7 @@ value_or_values -> ValOrVals:
   | VALUES    { ValOrVals::Values }
   ;
 
-values_list -> Vec<Vec<Expr>>:
+values_list -> Vec<RowValue>:
     values_list ','  row_value
     {
       $1.push($3);

pisa-proxy/protocol/mysql/src/client/codec.rs

@@ -20,7 +20,7 @@ use std::{
 };
 
 use bytes::{Buf, BufMut, BytesMut};
-use futures::Stream;
+use futures::{Stream, stream::Fuse};
 use pin_project::pin_project;
 use protocol_codegen::mysql_codec_convert;
 use tokio::io::Interest;
@@ -98,6 +98,70 @@ impl ClientCodec {
     }
 }
 
+
+#[pin_project]
+pub struct MergeStream<S> 
+where 
+    S: Stream + std::marker::Unpin,
+{
+    pub inner: Vec<Fuse<S>>,
+    idx: usize,
+    buf: Vec<Option<S::Item>>,
+    length: usize
+}
+
+impl<S> MergeStream<S>
+where 
+    S: Stream + std::marker::Unpin,
+{
+   pub fn new(inner: Vec<Fuse<S>>, length: usize) -> Self {
+        MergeStream { 
+            inner, 
+            idx: 0,
+            buf: Vec::with_capacity(length),
+            length,
+        }
+   } 
+}
+
+impl<S> Stream for MergeStream<S> 
+where 
+    S: Stream + std::marker::Unpin,
+    <S as Stream>::Item: std::fmt::Debug
+{
+    type Item = Vec<Option<S::Item>>;
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let me = self.project();
+        loop {
+            let s = unsafe {
+                Pin::new_unchecked(me.inner.get_unchecked_mut(*me.idx))
+            };
+            match s.poll_next(cx) {
+                Poll::Ready(data) => {
+                    *me.idx += 1;
+                    me.buf.push(data);
+                },
+
+                Poll::Pending => return Poll::Pending,
+            };
+
+            if *me.idx == *me.length {
+                *me.idx = 0;
+                break
+            }
+
+        }
+
+        if me.buf.is_empty() || me.buf.iter().all(|x| x.is_none()) {
+            return Poll::Ready(None);
+        } else {
+            return  Poll::Ready(Some(std::mem::replace(me.buf, Vec::with_capacity(*me.length))))
+        }
+
+    }
+}
+
+
 #[derive(Debug)]
 #[pin_project]
 pub struct ResultsetStream<'a> {