feat: improved partitioned table tests powered by tsbs

.github/workflows/ci.yml

@@ -295,3 +295,45 @@ jobs:
           name: recovery-test-${{ github.sha }}
           path: |
             /tmp/ceresdb-stdout.log
+
+  dist-query-test:
+    name: dist-query-test
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          submodules: true
+      - name: Cache Rust Dependencies
+        uses: actions/cache@v3
+        with:
+          path: |
+            ~/.cargo
+            ./target
+          key: debug-${{ runner.os }}-${{ hashFiles('rust-toolchain') }}-${{ hashFiles('Cargo.lock') }}
+          restore-keys: |
+            debug-${{ runner.os }}-${{ hashFiles('rust-toolchain') }}-
+            debug-${{ runner.os }}-
+            debug-
+      - run: |
+          rustup set auto-self-update disable
+          rustup toolchain install ${RUST_VERSION} --profile minimal
+      - name: Release Disk Quota
+        run: |
+          sudo rm -rf /usr/local/lib/android # release about 10 GB
+          sudo rm -rf /usr/share/dotnet # release about 20GB
+      - name: Setup Build Environment
+        run: |
+          sudo apt update
+          sudo apt install --yes protobuf-compiler
+      - name: Run dist query tests
+        working-directory: integration_tests
+        run: |
+          make run-dist-query
+      - name: Upload Logs
+        if: always()
+        uses: actions/upload-artifact@v3
+        with:
+          name: dist-query-test-${{ github.sha }}
+          path: |
+            /tmp/ceresdb-stdout.log

.gitignore

@@ -3,3 +3,6 @@ target
 .idea/
 .vscode
 .dir-locals.el
+integration_tests/dist_query/dist-query-testing
+integration_tests/dist_query/tsbs
+integration_tests/dist_query/output

integration_tests/Makefile

@@ -8,14 +8,14 @@ CERESMETA_DATA_DIR = /tmp/ceresmeta
 export CERESDB_TEST_CASE_PATH ?= $(ROOT)/cases/env
 export CERESDB_TEST_BINARY ?= $(ROOT)/../target/$(MODE)/ceresdb-test
 
-# environment variables for standalone
+# Environment variables for standalone
 export CERESDB_SERVER_GRPC_ENDPOINT ?= 127.0.0.1:8831
 export CERESDB_SERVER_HTTP_ENDPOINT ?= 127.0.0.1:5440
 export CERESDB_BINARY_PATH ?= $(ROOT)/../target/$(MODE)/ceresdb-server
 export CERESDB_STDOUT_FILE ?= /tmp/ceresdb-stdout.log
 export CERESDB_CONFIG_FILE ?= $(ROOT)/../docs/minimal.toml
 
-# environment variables for cluster
+# Environment variables for cluster
 export CERESMETA_BINARY_PATH ?= $(ROOT)/ceresmeta/ceresmeta
 export CERESMETA_CONFIG_PATH ?= $(ROOT)/config/ceresmeta.toml
 export CERESMETA_STDOUT_FILE ?= /tmp/ceresmeta-stdout.log
@@ -25,11 +25,18 @@ export CLUSTER_CERESDB_STDOUT_FILE_0 ?= /tmp/ceresdb-stdout-0.log
 export CLUSTER_CERESDB_STDOUT_FILE_1 ?= /tmp/ceresdb-stdout-1.log
 export RUST_BACKTRACE=1
 
+# Whether update related repos
+export UPDATE_REPOS_TO_LATEST ?= true
+
+# Used in dist query test, we don't want to rebuild the binarie and data in sometimes(e.g. debugging),
+# and we can set it to true.
+export DIST_QUERY_TEST_NO_INIT ?= false
+
 clean:
 	rm -rf $(CERESDB_DATA_DIR) $(CERESDB_DATA_DIR_0) $(CERESDB_DATA_DIR_1) $(CERESMETA_DATA_DIR)
 
 build-meta:
-	sh ./build_meta.sh
+	./build_meta.sh
 
 build-ceresdb:
 	cd .. && cargo build --bin ceresdb-server
@@ -78,3 +85,7 @@ run-opentsdb:
 
 run-recovery: clean build-ceresdb kill-old-process
 	cd recovery && ./run.sh && ./run.sh shard_based
+
+run-dist-query: prepare build-meta
+	CERESDB_INTEGRATION_TEST_BIN_RUN_MODE=build_cluster $(CERESDB_TEST_BINARY)
+	cd dist_query && ./run.sh

integration_tests/build_meta.sh

@@ -1,15 +1,19 @@
-#!/usr/bin/env bash
+#!/bin/bash
 
 set -e
 
 SRC=/tmp/ceresmeta-src
 TARGET=$(pwd)/ceresmeta
 
-if [ -d $SRC ]; then
+if [[ -d $SRC ]] && [[ $UPDATE_REPOS_TO_LATEST == 'true' ]]; then
   echo "Remove old meta..."
   rm -rf $SRC
 fi
 
-git clone --depth 1 https://github.com/ceresdb/ceresmeta.git ${SRC}
+if [[ ! -d $SRC ]]; then
+  echo "Pull meta repo..."
+  git clone --depth 1 https://github.com/ceresdb/ceresmeta.git ${SRC}
+fi
+
 cd ${SRC}
 go build -o ${TARGET}/ceresmeta ./cmd/meta/...

integration_tests/dist_query/diff.py

@@ -0,0 +1,47 @@
+import argparse
+import difflib
+
+def get_args():
+    parser = argparse.ArgumentParser(description='cmd args')
+    parser.add_argument('--expected', '-e', type=str, help='expected queries result file')
+    parser.add_argument('--actual', '-a', type=str, help='actual queries result file')
+    args = vars(parser.parse_args())
+    return args
+
+def main():
+    args = get_args()
+
+    # Load queries results.
+    f_expected_path = args['expected']
+    f_actual_path = args['actual']
+
+    f_expected = open(f_expected_path, "r")
+    expecteds = f_expected.readlines()
+
+    f_actual = open(f_actual_path, "r")
+    actuals = f_actual.readlines()
+
+    # Diff them.
+    diffs = difflib.context_diff(expecteds, actuals)
+    diff_num = 0
+    for diff in diffs:
+        diff_num += 1
+        print(diff)
+
+    f_expected.close()
+    f_actual.close()
+
+    # If diff exists, write the actual to expected, we can use `git diff` to inspect the detail diffs.
+    if diff_num != 0:
+        f = open(f_expected_path, "w")
+        f.writelines(actuals)
+        f.close()
+        # Test failed, just panic
+        print("Test failed...")
+        assert(False)
+
+    # Haha, test passed!
+    print("Test passed...")
+
+if __name__ == '__main__':
+    main()

integration_tests/dist_query/run.sh

@@ -0,0 +1,82 @@
+#!/bin/bash
+
+# Get current dir
+export CURR_DIR=$(pwd)
+
+# Related components
+TSBS_REPO_PATH=${CURR_DIR}/tsbs
+DATA_REPO_PATH=${CURR_DIR}/dist-query-testing
+# Case contexts
+CASE_DIR=tsbs-cpu-only
+CASE_DATASOURCE=data.out
+CASE_QUERY=single-groupby-5-8-1-queries.gz
+CASE_QUERY_RESULT=queries.result
+
+# Test params
+export RESULT_FILE=${RESULT_FILE:-${DEFAULT_RESULT_FILE}}
+export OUTPUT_DIR=${OUTPUT_DIR:-${CURR_DIR}/output}
+export CERESDB_ADDR=${CERESDB_ADDR:-127.0.0.1:8831}
+export CERESDB_HTTP_ADDR=${CERESDB_HTTP_ADDR:-127.0.0.1:5440}
+export WRITE_WORKER_NUM=${WRITE_WORKER_NUM:-36}
+export WRITE_BATCH_SIZE=${WRITE_BATCH_SIZE:-500}
+## Where generated data stored
+export DATA_FILE=${DATA_FILE:-${CURR_DIR}/dist-query-testing/${CASE_DIR}/${CASE_DATASOURCE}}
+## How many values in host tag
+export HOST_NUM=${HOST_NUM:-10000}
+export BULK_DATA_DIR=${CURR_DIR}/dist-query-testing/${CASE_DIR}
+## Used for `generate_queries.sh` end.
+export QUERY_TYPES="\
+single-groupby-1-1-1 \
+single-groupby-1-8-1 \
+single-groupby-5-1-1 \
+single-groupby-5-8-1"
+## Where query results stored
+export QUERY_RESULTS_FILE=${CURR_DIR}/output/queries.reuslt.tmp
+export QUERY_EXPECTED_RESULTS_FILE=${QUERY_EXPECTED_RESULTS_FILE:-${CURR_DIR}/dist-query-testing/${CASE_DIR}/${CASE_QUERY_RESULT}}
+
+set -x
+
+mkdir -p ${OUTPUT_DIR}
+
+# Prepare components
+## Tsbs
+if [[ -d ${TSBS_REPO_PATH} ]] && [[ ${UPDATE_REPOS_TO_LATEST} == 'true' ]] && [[ ${DIST_QUERY_TEST_NO_INIT} == 'false' ]]; then
+  echo "Remove old tsbs..."
+  rm -rf ${TSBS_REPO_PATH}
+fi
+
+if [[ ! -d ${TSBS_REPO_PATH} ]] && [[ ${DIST_QUERY_TEST_NO_INIT} == 'false' ]]; then
+    echo "Pull tsbs repo..."
+    git clone -b feat-ceresdb --depth 1 --single-branch https://github.com/CeresDB/tsbs.git
+fi
+## Data
+if [[ -d ${DATA_REPO_PATH} ]] && [[ $UPDATE_REPOS_TO_LATEST == 'true' ]] && [[ ${DIST_QUERY_TEST_NO_INIT} == 'false' ]]; then
+  echo "Remove old dist query testing..."
+  rm -rf ${DATA_REPO_PATH}
+fi
+
+if [[ ! -d ${DATA_REPO_PATH} ]] && [[ ${DIST_QUERY_TEST_NO_INIT} == 'false' ]]; then
+    echo "Pull dist query testing repo..."
+    git clone -b main --depth 1 --single-branch https://github.com/CeresDB/dist-query-testing.git
+fi
+## Build tsbs bins
+if [[ ${DIST_QUERY_TEST_NO_INIT} == 'false' ]]; then 
+    cd tsbs
+    go build ./cmd/tsbs_generate_data
+    go build ./cmd/tsbs_load_ceresdb
+    go build ./cmd/tsbs_generate_queries
+    go build ./cmd/tsbs_run_queries_ceresdb
+fi
+
+# Clean old table if exist
+curl -XPOST "${CERESDB_HTTP_ADDR}/sql" -d 'DROP TABLE IF EXISTS `cpu`'
+
+# Write data to ceresdb
+${CURR_DIR}/tsbs/tsbs_load_ceresdb --ceresdb-addr=${CERESDB_ADDR} --file ${DATA_FILE} --batch-size ${WRITE_BATCH_SIZE} --workers ${WRITE_WORKER_NUM}  --access-mode proxy --partition-keys hostname --update-mode APPEND | tee ${OUTPUT_DIR}/${CASE_DIR}-${CASE_DATASOURCE}.log
+
+# Run queries against ceresdb
+# TODO: support more kinds of queries besides 5-8-1.
+cat ${BULK_DATA_DIR}/${CASE_QUERY} | gunzip | ${CURR_DIR}/tsbs/tsbs_run_queries_ceresdb --ceresdb-addr=${CERESDB_ADDR} --print-responses true --access-mode proxy --responses-file ${QUERY_RESULTS_FILE} | tee ${OUTPUT_DIR}/${CASE_DIR}-${CASE_QUERY}.log
+
+# Diff the results
+python3 ${CURR_DIR}/diff.py --expected ${QUERY_EXPECTED_RESULTS_FILE} --actual ${QUERY_RESULTS_FILE}

integration_tests/src/main.rs

@@ -27,6 +27,7 @@ mod database;
 
 const CASE_ROOT_PATH_ENV: &str = "CERESDB_TEST_CASE_PATH";
 const ENV_FILTER_ENV: &str = "CERESDB_ENV_FILTER";
+const RUN_MODE: &str = "CERESDB_INTEGRATION_TEST_BIN_RUN_MODE";
 
 struct CeresDBController;
 struct UntypedCeresDB {
@@ -75,16 +76,34 @@ impl EnvController for CeresDBController {
 
 #[tokio::main]
 async fn main() -> Result<()> {
-    let case_dir = env::var(CASE_ROOT_PATH_ENV)?;
-    let env_filter = env::var(ENV_FILTER_ENV).unwrap_or_else(|_| ".*".to_string());
     let controller = CeresDBController;
-    let config = sqlness::ConfigBuilder::default()
-        .case_dir(case_dir)
-        .env_filter(env_filter)
-        .follow_links(true)
-        .build()?;
-    let runner = Runner::new_with_config(config, controller).await?;
-    runner.run().await?;
+    let run_mode = env::var(RUN_MODE).unwrap_or_else(|_| "sql_test".to_string());
+
+    match run_mode.as_str() {
+        // Run sql tests powered by `sqlness`.
+        "sql_test" => {
+            let case_dir = env::var(CASE_ROOT_PATH_ENV)?;
+            let env_filter = env::var(ENV_FILTER_ENV).unwrap_or_else(|_| ".*".to_string());
+            let config = sqlness::ConfigBuilder::default()
+                .case_dir(case_dir)
+                .env_filter(env_filter)
+                .follow_links(true)
+                .build()?;
+            let runner = Runner::new_with_config(config, controller).await?;
+            runner.run().await?;
+        }
+        // Just build the cluster testing env.
+        "build_cluster" => {
+            let _ = controller.start("cluster", None).await;
+        }
+        // Just build the local testing env.
+        "build_local" => {
+            let _ = controller.start("local", None).await;
+        }
+        other => {
+            panic!("Unknown run mode:{other}")
+        }
+    }
 
     Ok(())
 }