test: run sqlness test in distributed mode (#916)

* test: run sqlness test in distributed mode

* chore: fix ci test

* chore: improve the ci yaml

* chore: improve the code

* chore: fix conflicts

tests/cases/distributed/parser/operator_precedence.result

@@ -0,0 +1,88 @@
+SELECT 2*3+1;
+
++--------------------------------+
+| Int64(2) * Int64(3) + Int64(1) |
++--------------------------------+
+| 7                              |
++--------------------------------+
+
+SELECT 1+2*3;
+
++--------------------------------+
+| Int64(1) + Int64(2) * Int64(3) |
++--------------------------------+
+| 7                              |
++--------------------------------+
+
+SELECT 2^2 + 1;
+
++--------------------------------+
+| Int64(2) # Int64(2) + Int64(1) |
++--------------------------------+
+| 1                              |
++--------------------------------+
+
+SELECT 1+2^2;
+
++--------------------------------+
+| Int64(1) + Int64(2) # Int64(2) |
++--------------------------------+
+| 1                              |
++--------------------------------+
+
+SELECT 2*4 / 2;
+
++--------------------------------+
+| Int64(2) * Int64(4) / Int64(2) |
++--------------------------------+
+| 4                              |
++--------------------------------+
+
+SELECT 2*(4 / 2);
+
++--------------------------------+
+| Int64(2) * Int64(4) / Int64(2) |
++--------------------------------+
+| 4                              |
++--------------------------------+
+
+SELECT 16/2*4;
+
++---------------------------------+
+| Int64(16) / Int64(2) * Int64(4) |
++---------------------------------+
+| 32                              |
++---------------------------------+
+
+SELECT (16/2)*4;
+
++---------------------------------+
+| Int64(16) / Int64(2) * Int64(4) |
++---------------------------------+
+| 32                              |
++---------------------------------+
+
+SELECT 2*3*2;
+
++--------------------------------+
+| Int64(2) * Int64(3) * Int64(2) |
++--------------------------------+
+| 12                             |
++--------------------------------+
+
+SELECT 2^3*2;
+
++--------------------------------+
+| Int64(2) # Int64(3) * Int64(2) |
++--------------------------------+
+| 4                              |
++--------------------------------+
+
+SELECT 2*3^2;
+
++--------------------------------+
+| Int64(2) * Int64(3) # Int64(2) |
++--------------------------------+
+| 4                              |
++--------------------------------+
+

tests/cases/distributed/parser/operator_precedence.sql

@@ -0,0 +1,21 @@
+SELECT 2*3+1;
+
+SELECT 1+2*3;
+
+SELECT 2^2 + 1;
+
+SELECT 1+2^2;
+
+SELECT 2*4 / 2;
+
+SELECT 2*(4 / 2);
+
+SELECT 16/2*4;
+
+SELECT (16/2)*4;
+
+SELECT 2*3*2;
+
+SELECT 2^3*2;
+
+SELECT 2*3^2;

tests/cases/distributed/select/dummy.result

@@ -0,0 +1,36 @@
+select 1;
+
++----------+
+| Int64(1) |
++----------+
+| 1        |
++----------+
+
+select 2 + 3;
+
++---------------------+
+| Int64(2) + Int64(3) |
++---------------------+
+| 5                   |
++---------------------+
+
+select 4 + 0.5;
+
++-------------------------+
+| Int64(4) + Float64(0.5) |
++-------------------------+
+| 4.5                     |
++-------------------------+
+
+select "a";
+
+Error: 3000(PlanQuery), Schema error: No field named 'a'. Valid fields are .
+
+select "A";
+
+Error: 3000(PlanQuery), Schema error: No field named 'A'. Valid fields are .
+
+select * where "a" = "A";
+
+Error: 3000(PlanQuery), Schema error: No field named 'a'. Valid fields are .
+

tests/cases/distributed/select/dummy.sql

@@ -0,0 +1,11 @@
+select 1;
+
+select 2 + 3;
+
+select 4 + 0.5;
+
+select "a";
+
+select "A";
+
+select * where "a" = "A";

tests/runner/src/env.rs

@@ -28,8 +28,13 @@ use tokio::process::{Child, Command};
 
 use crate::util;
 
+const DATANODE_ADDR: &str = "127.0.0.1:4100";
+const METASRV_ADDR: &str = "127.0.0.1:3002";
 const SERVER_ADDR: &str = "127.0.0.1:4001";
 const SERVER_LOG_FILE: &str = "/tmp/greptime-sqlness.log";
+const METASRV_LOG_FILE: &str = "/tmp/greptime-sqlness-metasrv.log";
+const FRONTEND_LOG_FILE: &str = "/tmp/greptime-sqlness-frontend.log";
+const DATANODE_LOG_FILE: &str = "/tmp/greptime-sqlness-datanode.log";
 
 pub struct Env {}
 
@@ -48,8 +53,14 @@ impl EnvController for Env {
     /// Stop one [`Database`].
     #[allow(clippy::print_stdout)]
     async fn stop(&self, _mode: &str, mut database: Self::DB) {
-        database.server_process.kill().await.unwrap();
-        let _ = database.server_process.wait().await;
+        let mut server = database.server_process;
+        Env::stop_server(&mut server).await;
+        if let Some(mut metasrv) = database.metasrv_process.take() {
+            Env::stop_server(&mut metasrv).await;
+        }
+        if let Some(mut datanode) = database.datanode_process.take() {
+            Env::stop_server(&mut datanode).await;
+        }
         println!("Stopped DB.");
     }
 }
@@ -98,18 +109,99 @@ impl Env {
 
         GreptimeDB {
             server_process,
+            metasrv_process: None,
+            datanode_process: None,
             client,
             db,
         }
     }
 
     pub async fn start_distributed() -> GreptimeDB {
-        todo!()
+        let cargo_build_result = Command::new("cargo")
+            .current_dir(util::get_workspace_root())
+            .args(["build", "--bin", "greptime"])
+            .stdout(Stdio::null())
+            .output()
+            .await
+            .expect("Failed to start GreptimeDB")
+            .status;
+        if !cargo_build_result.success() {
+            panic!("Failed to build GreptimeDB (`cargo build` fails)");
+        }
+
+        // start a distributed GreptimeDB
+        let mut meta_server = Env::start_server("metasrv");
+        // wait for election
+        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+        let mut frontend = Env::start_server("frontend");
+        let mut datanode = Env::start_server("datanode");
+
+        for addr in [DATANODE_ADDR, METASRV_ADDR, SERVER_ADDR].iter() {
+            let is_up = util::check_port(addr.parse().unwrap(), Duration::from_secs(10)).await;
+            if !is_up {
+                Env::stop_server(&mut meta_server).await;
+                Env::stop_server(&mut frontend).await;
+                Env::stop_server(&mut datanode).await;
+                panic!("Server {addr} doesn't up in 10 seconds, quit.")
+            }
+        }
+
+        let client = Client::with_urls(vec![SERVER_ADDR]);
+        let db = DB::new("greptime", client.clone());
+
+        GreptimeDB {
+            server_process: frontend,
+            metasrv_process: Some(meta_server),
+            datanode_process: Some(datanode),
+            client,
+            db,
+        }
+    }
+
+    async fn stop_server(process: &mut Child) {
+        process.kill().await.unwrap();
+        let _ = process.wait().await;
+    }
+
+    fn start_server(subcommand: &str) -> Child {
+        let log_file_name = match subcommand {
+            "datanode" => DATANODE_LOG_FILE,
+            "frontend" => FRONTEND_LOG_FILE,
+            "metasrv" => METASRV_LOG_FILE,
+            _ => panic!("Unexpected subcommand: {subcommand}"),
+        };
+        let log_file = OpenOptions::new()
+            .create(true)
+            .write(true)
+            .truncate(true)
+            .open(log_file_name)
+            .unwrap_or_else(|_| panic!("Cannot open log file at {log_file_name}"));
+
+        let mut args = vec![subcommand, "start"];
+        if subcommand == "frontend" {
+            args.push("--metasrv-addr=0.0.0.0:3002");
+        } else if subcommand == "datanode" {
+            args.push("--rpc-addr=0.0.0.0:4100");
+            args.push("--metasrv-addr=0.0.0.0:3002");
+            args.push("--node-id=1");
+            args.push("--data-dir=/tmp/greptimedb_node_1/data");
+            args.push("--wal-dir=/tmp/greptimedb_node_1/wal");
+        }
+
+        let process = Command::new("./greptime")
+            .current_dir(util::get_binary_dir("debug"))
+            .args(args)
+            .stdout(log_file)
+            .spawn()
+            .expect("Failed to start the DB");
+        process
     }
 }
 
 pub struct GreptimeDB {
     server_process: Child,
+    metasrv_process: Option<Child>,
+    datanode_process: Option<Child>,
     #[allow(dead_code)]
     client: Client,
     db: DB,