test: sqlness test for insert default (#873)

* test: sqlness test for insert default

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* empty line

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* add more sqls

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* fix typos

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* update test according to typo fix

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

.github/workflows/develop.yml

@@ -110,6 +110,24 @@ jobs:
   #         GT_S3_ACCESS_KEY: ${{ secrets.S3_ACCESS_KEY }}
   #         UNITTEST_LOG_DIR: "__unittest_logs"
 
+  sqlness:
+    name: Sqlness Test
+    if: github.event.pull_request.draft == false
+    runs-on: ubuntu-latest-8-cores
+    timeout-minutes: 60
+    steps:
+      - uses: actions/checkout@v3
+      - uses: arduino/setup-protoc@v1
+        with:
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+      - uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: ${{ env.RUST_TOOLCHAIN }}
+      - name: Rust Cache
+        uses: Swatinem/rust-cache@v2
+      - name: Run sqlness
+        run: cargo run --bin sqlness-runner
+
   fmt:
     name: Rustfmt
     if: github.event.pull_request.draft == false

.github/workflows/release.yml

@@ -18,6 +18,8 @@ env:
   # In the future, we can change SCHEDULED_PERIOD to nightly.
   SCHEDULED_PERIOD: weekly
 
+  CARGO_PROFILE: weekly
+
 jobs:
   build:
     name: Build binary
@@ -82,13 +84,16 @@ jobs:
       - name: Output package versions
         run: protoc --version ; cargo version ; rustc --version ; gcc --version ; g++ --version
 
+      - name: Run tests
+        run: make unit-test integration-test sqlness-test
+
       - name: Run cargo build
-        run: cargo build ${{ matrix.opts }} --release --locked --target ${{ matrix.arch }}
+        run: cargo build ${{ matrix.opts }} --profile ${{ env.CARGO_PROFILE }} --locked --target ${{ matrix.arch }}
 
       - name: Calculate checksum and rename binary
         shell: bash
         run: |
-          cd target/${{ matrix.arch }}/release
+          cd target/${{ matrix.arch }}/${{ env.CARGO_PROFILE }}
           chmod +x greptime
           tar -zcvf ${{ matrix.file }}.tgz greptime
           echo $(shasum -a 256 ${{ matrix.file }}.tgz | cut -f1 -d' ') > ${{ matrix.file }}.sha256sum
@@ -97,13 +102,13 @@ jobs:
         uses: actions/upload-artifact@v3
         with:
           name: ${{ matrix.file }}
-          path: target/${{ matrix.arch }}/release/${{ matrix.file }}.tgz
+          path: target/${{ matrix.arch }}/${{ env.CARGO_PROFILE }}/${{ matrix.file }}.tgz
 
       - name: Upload checksum of artifacts
         uses: actions/upload-artifact@v3
         with:
           name: ${{ matrix.file }}.sha256sum
-          path: target/${{ matrix.arch }}/release/${{ matrix.file }}.sha256sum
+          path: target/${{ matrix.arch }}/${{ env.CARGO_PROFILE }}/${{ matrix.file }}.sha256sum
   release:
     name: Release artifacts
     needs: [build]

.pre-commit-config.yaml

@@ -5,11 +5,11 @@ repos:
     -    id: conventional-pre-commit
          stages: [commit-msg]
 
--   repo: https://github.com/DevinR528/cargo-sort
-    rev: e6a795bc6b2c0958f9ef52af4863bbd7cc17238f
-    hooks:
-    -    id: cargo-sort
-         args: ["--workspace"]
+# -   repo: https://github.com/DevinR528/cargo-sort
+#     rev: e6a795bc6b2c0958f9ef52af4863bbd7cc17238f
+#     hooks:
+#     -    id: cargo-sort
+#          args: ["--workspace"]
 
 -   repo: https://github.com/doublify/pre-commit-rust
     rev: v1.0

Cargo.toml

@@ -46,7 +46,10 @@ license = "Apache-2.0"
 
 [workspace.dependencies]
 arrow = "29.0"
+arrow-flight = "29.0"
 arrow-schema = { version = "29.0", features = ["serde"] }
+async-stream = "0.3"
+async-trait = "0.1"
 # TODO(LFC): Use released Datafusion when it officially dpendent on Arrow 29.0
 datafusion = { git = "https://github.com/apache/arrow-datafusion.git", rev = "4917235a398ae20145c87d20984e6367dc1a0c1e" }
 datafusion-common = { git = "https://github.com/apache/arrow-datafusion.git", rev = "4917235a398ae20145c87d20984e6367dc1a0c1e" }
@@ -54,8 +57,23 @@ datafusion-expr = { git = "https://github.com/apache/arrow-datafusion.git", rev
 datafusion-optimizer = { git = "https://github.com/apache/arrow-datafusion.git", rev = "4917235a398ae20145c87d20984e6367dc1a0c1e" }
 datafusion-physical-expr = { git = "https://github.com/apache/arrow-datafusion.git", rev = "4917235a398ae20145c87d20984e6367dc1a0c1e" }
 datafusion-sql = { git = "https://github.com/apache/arrow-datafusion.git", rev = "4917235a398ae20145c87d20984e6367dc1a0c1e" }
+futures = "0.3"
+futures-util = "0.3"
 parquet = "29.0"
+paste = "1.0"
+prost = "0.11"
+serde = { version = "1.0", features = ["derive"] }
+snafu = { version = "0.7", features = ["backtraces"] }
 sqlparser = "0.28"
+tokio = { version = "1", features = ["full"] }
+tonic = "0.8"
 
 [profile.release]
 debug = true
+
+[profile.weekly]
+inherits = "release"
+strip = true
+lto = true
+debug = false
+incremental = false

benchmarks/Cargo.toml

@@ -11,4 +11,4 @@ client = { path = "../src/client" }
 indicatif = "0.17.1"
 itertools = "0.10.5"
 parquet.workspace = true
-tokio = { version = "1.21", features = ["full"] }
+tokio.workspace = true

benchmarks/src/bin/nyc-taxi.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -25,10 +25,9 @@ use arrow::array::{ArrayRef, PrimitiveArray, StringArray, TimestampNanosecondArr
 use arrow::datatypes::{DataType, Float64Type, Int64Type};
 use arrow::record_batch::RecordBatch;
 use clap::Parser;
-use client::admin::Admin;
 use client::api::v1::column::Values;
-use client::api::v1::{Column, ColumnDataType, ColumnDef, CreateTableExpr, InsertExpr, TableId};
-use client::{Client, Database, Select};
+use client::api::v1::{Column, ColumnDataType, ColumnDef, CreateTableExpr, InsertRequest, TableId};
+use client::{Client, Database};
 use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
 use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
 use tokio::task::JoinSet;
@@ -100,7 +99,7 @@ async fn write_data(
     for record_batch in record_batch_reader {
         let record_batch = record_batch.unwrap();
         let (columns, row_count) = convert_record_batch(record_batch);
-        let insert_expr = InsertExpr {
+        let request = InsertRequest {
             schema_name: "public".to_string(),
             table_name: TABLE_NAME.to_string(),
             region_number: 0,
@@ -108,7 +107,7 @@ async fn write_data(
             row_count,
         };
         let now = Instant::now();
-        db.insert(insert_expr).await.unwrap();
+        db.insert(request).await.unwrap();
         let elapsed = now.elapsed();
         total_rpc_elapsed_ms += elapsed.as_millis();
         progress_bar.inc(row_count as _);
@@ -362,13 +361,11 @@ fn query_set() -> HashMap<String, String> {
     ret
 }
 
-async fn do_write(args: &Args, client: &Client) {
-    let admin = Admin::new("admin", client.clone());
-
+async fn do_write(args: &Args, db: &Database) {
     let mut file_list = get_file_list(args.path.clone().expect("Specify data path in argument"));
     let mut write_jobs = JoinSet::new();
 
-    let create_table_result = admin.create(create_table_expr()).await;
+    let create_table_result = db.create(create_table_expr()).await;
     println!("Create table result: {create_table_result:?}");
 
     let progress_bar_style = ProgressStyle::with_template(
@@ -383,7 +380,7 @@ async fn do_write(args: &Args, client: &Client) {
     let batch_size = args.batch_size;
     for _ in 0..args.thread_num {
         if let Some(path) = file_list.pop() {
-            let db = Database::new(DATABASE_NAME, client.clone());
+            let db = db.clone();
             let mpb = multi_progress_bar.clone();
             let pb_style = progress_bar_style.clone();
             write_jobs.spawn(async move { write_data(batch_size, &db, path, mpb, pb_style).await });
@@ -392,7 +389,7 @@ async fn do_write(args: &Args, client: &Client) {
     while write_jobs.join_next().await.is_some() {
         file_progress.inc(1);
         if let Some(path) = file_list.pop() {
-            let db = Database::new(DATABASE_NAME, client.clone());
+            let db = db.clone();
             let mpb = multi_progress_bar.clone();
             let pb_style = progress_bar_style.clone();
             write_jobs.spawn(async move { write_data(batch_size, &db, path, mpb, pb_style).await });
@@ -405,7 +402,7 @@ async fn do_query(num_iter: usize, db: &Database) {
         println!("Running query: {query}");
         for i in 0..num_iter {
             let now = Instant::now();
-            let _res = db.select(Select::Sql(query.clone())).await.unwrap();
+            let _res = db.sql(&query).await.unwrap();
             let elapsed = now.elapsed();
             println!(
                 "query {}, iteration {}: {}ms",
@@ -427,13 +424,13 @@ fn main() {
         .unwrap()
         .block_on(async {
             let client = Client::with_urls(vec![&args.endpoint]);
+            let db = Database::new(DATABASE_NAME, client);
 
             if !args.skip_write {
-                do_write(&args, &client).await;
+                do_write(&args, &db).await;
             }
 
             if !args.skip_read {
-                let db = Database::new(DATABASE_NAME, client.clone());
                 do_query(args.iter_num, &db).await;
             }
         })

config/datanode.example.toml

@@ -1,12 +1,19 @@
 node_id = 42
 mode = 'distributed'
 rpc_addr = '127.0.0.1:3001'
-wal_dir = '/tmp/greptimedb/wal'
 rpc_runtime_size = 8
 mysql_addr = '127.0.0.1:4406'
 mysql_runtime_size = 4
 enable_memory_catalog = false
 
+[wal]
+dir = "/tmp/greptimedb/wal"
+file_size = '1GB'
+purge_interval = '10m'
+purge_threshold = '50GB'
+read_batch_size = 128
+sync_write = false
+
 [storage]
 type = 'File'
 data_dir = '/tmp/greptimedb/data/'

config/standalone.example.toml

@@ -1,12 +1,20 @@
 node_id = 0
 mode = 'standalone'
-wal_dir = '/tmp/greptimedb/wal/'
 enable_memory_catalog = false
 
 [http_options]
 addr = '127.0.0.1:4000'
 timeout = "30s"
 
+[wal]
+dir = "/tmp/greptimedb/wal"
+file_size = '1GB'
+purge_interval = '10m'
+purge_threshold = '50GB'
+read_batch_size = 128
+sync_write = false
+
+
 [storage]
 type = 'File'
 data_dir = '/tmp/greptimedb/data/'

docker/Dockerfile

@@ -24,6 +24,8 @@ RUN cargo build --release
 # TODO(zyy17): Maybe should use the more secure container image.
 FROM ubuntu:22.04 as base
 
+RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get -y install ca-certificates
+
 WORKDIR /greptime
 COPY --from=builder /greptimedb/target/release/greptime /greptime/bin/
 ENV PATH /greptime/bin/:$PATH

docker/ci/Dockerfile

@@ -1,5 +1,7 @@
 FROM ubuntu:22.04
 
+RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get -y install ca-certificates
+
 ARG TARGETARCH
 
 ADD $TARGETARCH/greptime /greptime/bin/

src/api/Cargo.toml

@@ -5,13 +5,14 @@ edition.workspace = true
 license.workspace = true
 
 [dependencies]
+arrow-flight.workspace = true
 common-base = { path = "../common/base" }
 common-error = { path = "../common/error" }
 common-time = { path = "../common/time" }
 datatypes = { path = "../datatypes" }
-prost = "0.11"
+prost.workspace = true
 snafu = { version = "0.7", features = ["backtraces"] }
-tonic = "0.8"
+tonic.workspace = true
 
 [build-dependencies]
 tonic-build = "0.8"

src/api/build.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -12,16 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::path::PathBuf;
-
 fn main() {
-    let default_out_dir = PathBuf::from(std::env::var("OUT_DIR").unwrap());
     tonic_build::configure()
-        .file_descriptor_set_path(default_out_dir.join("greptime_fd.bin"))
         .compile(
             &[
-                "greptime/v1/select.proto",
-                "greptime/v1/greptime.proto",
+                "greptime/v1/database.proto",
                 "greptime/v1/meta/common.proto",
                 "greptime/v1/meta/heartbeat.proto",
                 "greptime/v1/meta/route.proto",

src/api/greptime/v1/common.proto

@@ -1,22 +0,0 @@
-syntax = "proto3";
-
-package greptime.v1;
-
-message RequestHeader {
-  string tenant = 1;
-}
-
-message ExprHeader {
-  uint32 version = 1;
-}
-
-message ResultHeader {
-  uint32 version = 1;
-  uint32 code = 2;
-  string err_msg = 3;
-}
-
-message MutateResult {
-  uint32 success = 1;
-  uint32 failure = 2;
-}

src/api/greptime/v1/database.proto

@@ -2,37 +2,25 @@ syntax = "proto3";
 
 package greptime.v1;
 
+import "greptime/v1/ddl.proto";
 import "greptime/v1/column.proto";
-import "greptime/v1/common.proto";
 
-message DatabaseRequest {
-  string name = 1;
-  repeated ObjectExpr exprs = 2;
-}
-
-message DatabaseResponse {
-  repeated ObjectResult results = 1;
-}
-
-message ObjectExpr {
-  ExprHeader header = 1;
-  oneof expr {
-    InsertExpr insert = 2;
-    SelectExpr select = 3;
-    UpdateExpr update = 4;
-    DeleteExpr delete = 5;
+message GreptimeRequest {
+  oneof request {
+    InsertRequest insert = 1;
+    QueryRequest query = 2;
+    DdlRequest ddl = 3;
   }
 }
 
-// TODO(fys): Only support sql now, and will support promql etc in the future
-message SelectExpr {
-  oneof expr {
+message QueryRequest {
+  oneof query {
     string sql = 1;
     bytes logical_plan = 2;
   }
 }
 
-message InsertExpr {
+message InsertRequest {
   string schema_name = 1;
   string table_name = 2;
 
@@ -41,26 +29,13 @@ message InsertExpr {
 
   // The row_count of all columns, which include null and non-null values.
   //
-  // Note: the row_count of all columns in a InsertExpr must be same.
+  // Note: the row_count of all columns in a InsertRequest must be same.
   uint32 row_count = 4;
 
   // The region number of current insert request.
   uint32 region_number = 5;
 }
 
-// TODO(jiachun)
-message UpdateExpr {}
-// TODO(jiachun)
-message DeleteExpr {}
-
-message ObjectResult {
-  ResultHeader header = 1;
-  oneof result {
-    SelectResult select = 2;
-    MutateResult mutate = 3;
-  }
-}
-
-message SelectResult {
-  bytes raw_data = 1;
+message FlightDataExt {
+  uint32 affected_rows = 1;
 }

src/api/greptime/v1/ddl.proto

@@ -3,31 +3,16 @@ syntax = "proto3";
 package greptime.v1;
 
 import "greptime/v1/column.proto";
-import "greptime/v1/common.proto";
 
-message AdminRequest {
-  string name = 1;
-  repeated AdminExpr exprs = 2;
-}
-
-message AdminResponse {
-  repeated AdminResult results = 1;
-}
-
-message AdminExpr {
-  ExprHeader header = 1;
+// "Data Definition Language" requests, that create, modify or delete the database structures but not the data.
+// `DdlRequest` could carry more information than plain SQL, for example, the "table_id" in `CreateTableExpr`.
+// So create a new DDL expr if you need it.
+message DdlRequest {
   oneof expr {
+    CreateDatabaseExpr create_database = 1;
     CreateTableExpr create_table = 2;
     AlterExpr alter = 3;
-    CreateDatabaseExpr create_database = 4;
-    DropTableExpr drop_table = 5;
-  }
-}
-
-message AdminResult {
-  ResultHeader header = 1;
-  oneof result {
-    MutateResult mutate = 2;
+    DropTableExpr drop_table = 4;
   }
 }
 
@@ -52,6 +37,7 @@ message AlterExpr {
   oneof kind {
     AddColumns add_columns = 4;
     DropColumns drop_columns = 5;
+    RenameTable rename_table = 6;
   }
 }
 
@@ -64,6 +50,7 @@ message DropTableExpr {
 message CreateDatabaseExpr {
   //TODO(hl): maybe rename to schema_name?
   string database_name = 1;
+  bool create_if_not_exists = 2;
 }
 
 message AddColumns {
@@ -74,6 +61,10 @@ message DropColumns {
   repeated DropColumn drop_columns = 1;
 }
 
+message RenameTable {
+  string new_table_name = 1;
+}
+
 message AddColumn {
   ColumnDef column_def = 1;
   bool is_key = 2;

src/api/greptime/v1/greptime.proto

@@ -1,22 +0,0 @@
-syntax = "proto3";
-
-package greptime.v1;
-
-import "greptime/v1/admin.proto";
-import "greptime/v1/common.proto";
-import "greptime/v1/database.proto";
-
-service Greptime {
-  rpc Batch(BatchRequest) returns (BatchResponse) {}
-}
-
-message BatchRequest {
-  RequestHeader header = 1;
-  repeated AdminRequest admins = 2;
-  repeated DatabaseRequest databases = 3;
-}
-
-message BatchResponse {
-  repeated AdminResponse admins = 1;
-  repeated DatabaseResponse databases = 2;
-}

src/api/greptime/v1/meta/heartbeat.proto

@@ -26,7 +26,7 @@ message HeartbeatRequest {
   TimeInterval report_interval = 4;
   // Node stat
   NodeStat node_stat = 5;
-  // Region stats in this node
+  // Region stats on this node
   repeated RegionStat region_stats = 6;
   // Follower nodes and stats, empty on follower nodes
   repeated ReplicaStat replica_stats = 7;
@@ -34,19 +34,19 @@ message HeartbeatRequest {
 
 message NodeStat {
   // The read capacity units during this period
-  uint64 rcus = 1;
+  int64 rcus = 1;
   // The write capacity units during this period
-  uint64 wcus = 2;
-  // Table number in this node
-  uint64 table_num = 3;
-  // Region number in this node
-  uint64 region_num = 4;
+  int64 wcus = 2;
+  // How many tables on this node
+  int64 table_num = 3;
+  // How many regions on this node
+  int64 region_num = 4;
 
   double cpu_usage = 5;
   double load = 6;
-  // Read disk I/O in the node
+  // Read disk IO on this node
   double read_io_rate = 7;
-  // Write disk I/O in the node
+  // Write disk IO on this node
   double write_io_rate = 8;
 
   // Others
@@ -57,13 +57,13 @@ message RegionStat {
   uint64 region_id = 1;
   TableName table_name = 2;
   // The read capacity units during this period
-  uint64 rcus = 3;
+  int64 rcus = 3;
   // The write capacity units during this period
-  uint64 wcus = 4;
-  // Approximate region size
-  uint64 approximate_size = 5;
-  // Approximate number of rows
-  uint64 approximate_rows = 6;
+  int64 wcus = 4;
+  // Approximate bytes of this region
+  int64 approximate_bytes = 5;
+  // Approximate number of rows in this region
+  int64 approximate_rows = 6;
 
   // Others
   map<string, string> attrs = 100;

src/api/greptime/v1/select.proto

@@ -1,10 +0,0 @@
-syntax = "proto3";
-
-package greptime.v1.codec;
-
-import "greptime/v1/column.proto";
-
-message SelectResult {
-  repeated Column columns = 1;
-  uint32 row_count = 2;
-}

src/api/src/error.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/api/src/helper.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/api/src/lib.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -15,7 +15,6 @@
 pub mod error;
 pub mod helper;
 pub mod prometheus;
-pub mod result;
 pub mod serde;
 pub mod v1;
 

src/api/src/prometheus.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/api/src/result.rs

@@ -1,203 +0,0 @@
-// Copyright 2022 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use common_error::prelude::ErrorExt;
-
-use crate::v1::codec::SelectResult;
-use crate::v1::{
-    admin_result, object_result, AdminResult, MutateResult, ObjectResult, ResultHeader,
-    SelectResult as SelectResultRaw,
-};
-
-pub const PROTOCOL_VERSION: u32 = 1;
-
-pub type Success = u32;
-pub type Failure = u32;
-
-#[derive(Default)]
-pub struct ObjectResultBuilder {
-    version: u32,
-    code: u32,
-    err_msg: Option<String>,
-    result: Option<Body>,
-}
-
-pub enum Body {
-    Mutate((Success, Failure)),
-    Select(SelectResult),
-}
-
-impl ObjectResultBuilder {
-    pub fn new() -> Self {
-        Self {
-            version: PROTOCOL_VERSION,
-            ..Default::default()
-        }
-    }
-
-    #[allow(dead_code)]
-    pub fn version(mut self, version: u32) -> Self {
-        self.version = version;
-        self
-    }
-
-    pub fn status_code(mut self, code: u32) -> Self {
-        self.code = code;
-        self
-    }
-
-    pub fn err_msg(mut self, err_msg: String) -> Self {
-        self.err_msg = Some(err_msg);
-        self
-    }
-
-    pub fn mutate_result(mut self, success: u32, failure: u32) -> Self {
-        self.result = Some(Body::Mutate((success, failure)));
-        self
-    }
-
-    pub fn select_result(mut self, select_result: SelectResult) -> Self {
-        self.result = Some(Body::Select(select_result));
-        self
-    }
-
-    pub fn build(self) -> ObjectResult {
-        let header = Some(ResultHeader {
-            version: self.version,
-            code: self.code,
-            err_msg: self.err_msg.unwrap_or_default(),
-        });
-
-        let result = match self.result {
-            Some(Body::Mutate((success, failure))) => {
-                Some(object_result::Result::Mutate(MutateResult {
-                    success,
-                    failure,
-                }))
-            }
-            Some(Body::Select(select)) => Some(object_result::Result::Select(SelectResultRaw {
-                raw_data: select.into(),
-            })),
-            None => None,
-        };
-
-        ObjectResult { header, result }
-    }
-}
-
-pub fn build_err_result(err: &impl ErrorExt) -> ObjectResult {
-    ObjectResultBuilder::new()
-        .status_code(err.status_code() as u32)
-        .err_msg(err.to_string())
-        .build()
-}
-
-#[derive(Debug)]
-pub struct AdminResultBuilder {
-    version: u32,
-    code: u32,
-    err_msg: Option<String>,
-    mutate: Option<(Success, Failure)>,
-}
-
-impl AdminResultBuilder {
-    pub fn status_code(mut self, code: u32) -> Self {
-        self.code = code;
-        self
-    }
-
-    pub fn err_msg(mut self, err_msg: String) -> Self {
-        self.err_msg = Some(err_msg);
-        self
-    }
-
-    pub fn mutate_result(mut self, success: u32, failure: u32) -> Self {
-        self.mutate = Some((success, failure));
-        self
-    }
-
-    pub fn build(self) -> AdminResult {
-        let header = Some(ResultHeader {
-            version: self.version,
-            code: self.code,
-            err_msg: self.err_msg.unwrap_or_default(),
-        });
-
-        let result = if let Some((success, failure)) = self.mutate {
-            Some(admin_result::Result::Mutate(MutateResult {
-                success,
-                failure,
-            }))
-        } else {
-            None
-        };
-
-        AdminResult { header, result }
-    }
-}
-
-impl Default for AdminResultBuilder {
-    fn default() -> Self {
-        Self {
-            version: PROTOCOL_VERSION,
-            code: 0,
-            err_msg: None,
-            mutate: None,
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use common_error::status_code::StatusCode;
-
-    use super::*;
-    use crate::error::UnknownColumnDataTypeSnafu;
-    use crate::v1::{object_result, MutateResult};
-
-    #[test]
-    fn test_object_result_builder() {
-        let obj_result = ObjectResultBuilder::new()
-            .version(101)
-            .status_code(500)
-            .err_msg("Failed to read this file!".to_string())
-            .mutate_result(100, 20)
-            .build();
-        let header = obj_result.header.unwrap();
-        assert_eq!(101, header.version);
-        assert_eq!(500, header.code);
-        assert_eq!("Failed to read this file!", header.err_msg);
-
-        let result = obj_result.result.unwrap();
-        assert_eq!(
-            object_result::Result::Mutate(MutateResult {
-                success: 100,
-                failure: 20,
-            }),
-            result
-        );
-    }
-
-    #[test]
-    fn test_build_err_result() {
-        let err = UnknownColumnDataTypeSnafu { datatype: 1 }.build();
-        let err_result = build_err_result(&err);
-        let header = err_result.header.unwrap();
-        let result = err_result.result;
-
-        assert_eq!(PROTOCOL_VERSION, header.version);
-        assert_eq!(StatusCode::InvalidArguments as u32, header.code);
-        assert!(result.is_none());
-    }
-}

src/api/src/serde.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -15,7 +15,6 @@
 pub use prost::DecodeError;
 use prost::Message;
 
-use crate::v1::codec::SelectResult;
 use crate::v1::meta::TableRouteValue;
 
 macro_rules! impl_convert_with_bytes {
@@ -36,80 +35,4 @@ macro_rules! impl_convert_with_bytes {
     };
 }
 
-impl_convert_with_bytes!(SelectResult);
 impl_convert_with_bytes!(TableRouteValue);
-
-#[cfg(test)]
-mod tests {
-    use std::ops::Deref;
-
-    use crate::v1::codec::*;
-    use crate::v1::{column, Column};
-
-    const SEMANTIC_TAG: i32 = 0;
-
-    #[test]
-    fn test_convert_select_result() {
-        let select_result = mock_select_result();
-
-        let bytes: Vec<u8> = select_result.into();
-        let result: SelectResult = bytes.deref().try_into().unwrap();
-
-        assert_eq!(8, result.row_count);
-        assert_eq!(1, result.columns.len());
-
-        let column = &result.columns[0];
-        assert_eq!("foo", column.column_name);
-        assert_eq!(SEMANTIC_TAG, column.semantic_type);
-        assert_eq!(vec![1], column.null_mask);
-        assert_eq!(
-            vec![2, 3, 4, 5, 6, 7, 8],
-            column.values.as_ref().unwrap().i32_values
-        );
-    }
-
-    #[should_panic]
-    #[test]
-    fn test_convert_select_result_wrong() {
-        let select_result = mock_select_result();
-
-        let mut bytes: Vec<u8> = select_result.into();
-
-        // modify some bytes
-        bytes[0] = 0b1;
-        bytes[1] = 0b1;
-
-        let result: SelectResult = bytes.deref().try_into().unwrap();
-
-        assert_eq!(8, result.row_count);
-        assert_eq!(1, result.columns.len());
-
-        let column = &result.columns[0];
-        assert_eq!("foo", column.column_name);
-        assert_eq!(SEMANTIC_TAG, column.semantic_type);
-        assert_eq!(vec![1], column.null_mask);
-        assert_eq!(
-            vec![2, 3, 4, 5, 6, 7, 8],
-            column.values.as_ref().unwrap().i32_values
-        );
-    }
-
-    fn mock_select_result() -> SelectResult {
-        let values = column::Values {
-            i32_values: vec![2, 3, 4, 5, 6, 7, 8],
-            ..Default::default()
-        };
-        let null_mask = vec![1];
-        let column = Column {
-            column_name: "foo".to_string(),
-            semantic_type: SEMANTIC_TAG,
-            values: Some(values),
-            null_mask,
-            ..Default::default()
-        };
-        SelectResult {
-            columns: vec![column],
-            row_count: 8,
-        }
-    }
-}

src/api/src/v1.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -15,11 +15,5 @@
 #![allow(clippy::derive_partial_eq_without_eq)]
 tonic::include_proto!("greptime.v1");
 
-pub const GREPTIME_FD_SET: &[u8] = tonic::include_file_descriptor_set!("greptime_fd");
-
-pub mod codec {
-    tonic::include_proto!("greptime.v1.codec");
-}
-
 mod column_def;
 pub mod meta;

src/api/src/v1/column_def.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/api/src/v1/meta.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/catalog/Cargo.toml

@@ -7,7 +7,7 @@ license.workspace = true
 [dependencies]
 api = { path = "../api" }
 arc-swap = "1.0"
-async-stream = "0.3"
+async-stream.workspace = true
 async-trait = "0.1"
 backoff = { version = "0.4", features = ["tokio"] }
 common-catalog = { path = "../common/catalog" }
@@ -21,7 +21,7 @@ common-time = { path = "../common/time" }
 datafusion.workspace = true
 datatypes = { path = "../datatypes" }
 futures = "0.3"
-futures-util = "0.3"
+futures-util.workspace = true
 lazy_static = "1.4"
 meta-client = { path = "../meta-client" }
 regex = "1.6"
@@ -30,7 +30,7 @@ serde_json = "1.0"
 snafu = { version = "0.7", features = ["backtraces"] }
 storage = { path = "../storage" }
 table = { path = "../table" }
-tokio = { version = "1.18", features = ["full"] }
+tokio.workspace = true
 
 [dev-dependencies]
 chrono = "0.4"
@@ -39,4 +39,4 @@ mito = { path = "../mito", features = ["test"] }
 object-store = { path = "../object-store" }
 storage = { path = "../storage" }
 tempdir = "0.3"
-tokio = { version = "1.0", features = ["full"] }
+tokio.workspace = true

src/catalog/src/error.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -86,9 +86,10 @@ pub enum Error {
         backtrace: Backtrace,
     },
 
-    #[snafu(display("Cannot find schema, schema info: {}", schema_info))]
+    #[snafu(display("Cannot find schema {} in catalog {}", schema, catalog))]
     SchemaNotFound {
-        schema_info: String,
+        catalog: String,
+        schema: String,
         backtrace: Backtrace,
     },
 
@@ -163,6 +164,9 @@ pub enum Error {
         source: datatypes::error::Error,
     },
 
+    #[snafu(display("Failure during SchemaProvider operation, source: {}", source))]
+    SchemaProviderOperation { source: BoxedError },
+
     #[snafu(display("Failed to execute system catalog table scan, source: {}", source))]
     SystemCatalogTableScanExec {
         #[snafu(backtrace)]
@@ -240,7 +244,9 @@ impl ErrorExt for Error {
             Error::SystemCatalogTableScanExec { source } => source.status_code(),
             Error::InvalidTableSchema { source, .. } => source.status_code(),
             Error::InvalidTableInfoInCatalog { .. } => StatusCode::Unexpected,
-            Error::Internal { source, .. } => source.status_code(),
+            Error::Internal { source, .. } | Error::SchemaProviderOperation { source } => {
+                source.status_code()
+            }
 
             Error::Unimplemented { .. } => StatusCode::Unsupported,
         }
@@ -263,7 +269,6 @@ impl From<Error> for DataFusionError {
 
 #[cfg(test)]
 mod tests {
-    use common_error::mock::MockError;
     use snafu::GenerateImplicitData;
 
     use super::*;
@@ -284,22 +289,6 @@ mod tests {
             InvalidKeySnafu { key: None }.build().status_code()
         );
 
-        assert_eq!(
-            StatusCode::StorageUnavailable,
-            Error::OpenSystemCatalog {
-                source: table::error::Error::new(MockError::new(StatusCode::StorageUnavailable))
-            }
-            .status_code()
-        );
-
-        assert_eq!(
-            StatusCode::StorageUnavailable,
-            Error::CreateSystemCatalog {
-                source: table::error::Error::new(MockError::new(StatusCode::StorageUnavailable))
-            }
-            .status_code()
-        );
-
         assert_eq!(
             StatusCode::StorageUnavailable,
             Error::SystemCatalog {

src/catalog/src/helper.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -91,6 +91,7 @@ pub fn build_table_regional_prefix(
 }
 
 /// Table global info has only one key across all datanodes so it does not have `node_id` field.
+#[derive(Clone)]
 pub struct TableGlobalKey {
     pub catalog_name: String,
     pub schema_name: String,
@@ -131,7 +132,6 @@ impl TableGlobalKey {
 pub struct TableGlobalValue {
     /// Id of datanode that created the global table info kv. only for debugging.
     pub node_id: u64,
-    // TODO(LFC): Maybe remove it?
     /// Allocation of region ids across all datanodes.
     pub regions_id_map: HashMap<u64, Vec<u32>>,
     pub table_info: RawTableInfo,

src/catalog/src/lib.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -19,7 +19,7 @@ use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
 use common_telemetry::info;
-use snafu::ResultExt;
+use snafu::{OptionExt, ResultExt};
 use table::engine::{EngineContext, TableEngineRef};
 use table::metadata::TableId;
 use table::requests::CreateTableRequest;
@@ -97,6 +97,9 @@ pub trait CatalogManager: CatalogList {
     /// schema registered.
     async fn register_schema(&self, request: RegisterSchemaRequest) -> Result<bool>;
 
+    /// Rename a table to [RenameTableRequest::new_table_name], returns whether the table is renamed.
+    async fn rename_table(&self, request: RenameTableRequest) -> Result<bool>;
+
     /// Register a system table, should be called before starting the manager.
     async fn register_system_table(&self, request: RegisterSystemTableRequest)
         -> error::Result<()>;
@@ -142,6 +145,15 @@ impl Debug for RegisterTableRequest {
     }
 }
 
+#[derive(Debug, Clone)]
+pub struct RenameTableRequest {
+    pub catalog: String,
+    pub schema: String,
+    pub table_name: String,
+    pub new_table_name: String,
+    pub table_id: TableId,
+}
+
 #[derive(Clone)]
 pub struct DeregisterTableRequest {
     pub catalog: String,
@@ -208,3 +220,38 @@ pub(crate) async fn handle_system_table_request<'a, M: CatalogManager>(
     }
     Ok(())
 }
+
+/// The number of regions in the datanode node.
+pub fn region_number(catalog_manager: &CatalogManagerRef) -> Result<u64> {
+    let mut region_number: u64 = 0;
+
+    for catalog_name in catalog_manager.catalog_names()? {
+        let catalog =
+            catalog_manager
+                .catalog(&catalog_name)?
+                .context(error::CatalogNotFoundSnafu {
+                    catalog_name: &catalog_name,
+                })?;
+
+        for schema_name in catalog.schema_names()? {
+            let schema = catalog
+                .schema(&schema_name)?
+                .context(error::SchemaNotFoundSnafu {
+                    catalog: &catalog_name,
+                    schema: &schema_name,
+                })?;
+
+            for table_name in schema.table_names()? {
+                let table = schema
+                    .table(&table_name)?
+                    .context(error::TableNotFoundSnafu {
+                        table_info: &table_name,
+                    })?;
+
+                let region_numbers = &table.table_info().meta.region_numbers;
+                region_number += region_numbers.len() as u64;
+            }
+        }
+    }
+    Ok(region_number)
+}

src/catalog/src/local.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/catalog/src/local/manager.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -47,7 +47,8 @@ use crate::tables::SystemCatalog;
 use crate::{
     format_full_table_name, handle_system_table_request, CatalogList, CatalogManager,
     CatalogProvider, CatalogProviderRef, DeregisterTableRequest, RegisterSchemaRequest,
-    RegisterSystemTableRequest, RegisterTableRequest, SchemaProvider, SchemaProviderRef,
+    RegisterSystemTableRequest, RegisterTableRequest, RenameTableRequest, SchemaProvider,
+    SchemaProviderRef,
 };
 
 /// A `CatalogManager` consists of a system catalog and a bunch of user catalogs.
@@ -241,7 +242,8 @@ impl LocalCatalogManager {
         let schema = catalog
             .schema(&t.schema_name)?
             .context(SchemaNotFoundSnafu {
-                schema_info: format!("{}.{}", &t.catalog_name, &t.schema_name),
+                catalog: &t.catalog_name,
+                schema: &t.schema_name,
             })?;
 
         let context = EngineContext {};
@@ -338,7 +340,8 @@ impl CatalogManager for LocalCatalogManager {
         let schema = catalog
             .schema(schema_name)?
             .with_context(|| SchemaNotFoundSnafu {
-                schema_info: format!("{catalog_name}.{schema_name}"),
+                catalog: catalog_name,
+                schema: schema_name,
             })?;
 
         {
@@ -377,6 +380,45 @@ impl CatalogManager for LocalCatalogManager {
         }
     }
 
+    async fn rename_table(&self, request: RenameTableRequest) -> Result<bool> {
+        let started = self.init_lock.lock().await;
+
+        ensure!(
+            *started,
+            IllegalManagerStateSnafu {
+                msg: "Catalog manager not started",
+            }
+        );
+
+        let catalog_name = &request.catalog;
+        let schema_name = &request.schema;
+
+        let catalog = self
+            .catalogs
+            .catalog(catalog_name)?
+            .context(CatalogNotFoundSnafu { catalog_name })?;
+
+        let schema = catalog
+            .schema(schema_name)?
+            .with_context(|| SchemaNotFoundSnafu {
+                catalog: catalog_name,
+                schema: schema_name,
+            })?;
+
+        // rename table in system catalog
+        self.system
+            .register_table(
+                catalog_name.clone(),
+                schema_name.clone(),
+                request.new_table_name.clone(),
+                request.table_id,
+            )
+            .await?;
+        Ok(schema
+            .rename_table(&request.table_name, request.new_table_name)
+            .is_ok())
+    }
+
     async fn deregister_table(&self, _request: DeregisterTableRequest) -> Result<bool> {
         UnimplementedSnafu {
             operation: "deregister table",
@@ -452,7 +494,8 @@ impl CatalogManager for LocalCatalogManager {
         let schema = catalog
             .schema(schema_name)?
             .with_context(|| SchemaNotFoundSnafu {
-                schema_info: format!("{catalog_name}.{schema_name}"),
+                catalog: catalog_name,
+                schema: schema_name,
             })?;
         schema.table(table_name)
     }

src/catalog/src/local/memory.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -25,11 +25,14 @@ use table::metadata::TableId;
 use table::table::TableIdProvider;
 use table::TableRef;
 
-use crate::error::{CatalogNotFoundSnafu, Result, SchemaNotFoundSnafu, TableExistsSnafu};
+use crate::error::{
+    CatalogNotFoundSnafu, Result, SchemaNotFoundSnafu, TableExistsSnafu, TableNotFoundSnafu,
+};
 use crate::schema::SchemaProvider;
 use crate::{
     CatalogList, CatalogManager, CatalogProvider, CatalogProviderRef, DeregisterTableRequest,
-    RegisterSchemaRequest, RegisterSystemTableRequest, RegisterTableRequest, SchemaProviderRef,
+    RegisterSchemaRequest, RegisterSystemTableRequest, RegisterTableRequest, RenameTableRequest,
+    SchemaProviderRef,
 };
 
 /// Simple in-memory list of catalogs
@@ -81,13 +84,33 @@ impl CatalogManager for MemoryCatalogManager {
         let schema = catalog
             .schema(&request.schema)?
             .with_context(|| SchemaNotFoundSnafu {
-                schema_info: format!("{}.{}", &request.catalog, &request.schema),
+                catalog: &request.catalog,
+                schema: &request.schema,
             })?;
         schema
             .register_table(request.table_name, request.table)
             .map(|v| v.is_none())
     }
 
+    async fn rename_table(&self, request: RenameTableRequest) -> Result<bool> {
+        let catalogs = self.catalogs.write().unwrap();
+        let catalog = catalogs
+            .get(&request.catalog)
+            .context(CatalogNotFoundSnafu {
+                catalog_name: &request.catalog,
+            })?
+            .clone();
+        let schema = catalog
+            .schema(&request.schema)?
+            .with_context(|| SchemaNotFoundSnafu {
+                catalog: &request.catalog,
+                schema: &request.schema,
+            })?;
+        Ok(schema
+            .rename_table(&request.table_name, request.new_table_name)
+            .is_ok())
+    }
+
     async fn deregister_table(&self, request: DeregisterTableRequest) -> Result<bool> {
         let catalogs = self.catalogs.write().unwrap();
         let catalog = catalogs
@@ -99,7 +122,8 @@ impl CatalogManager for MemoryCatalogManager {
         let schema = catalog
             .schema(&request.schema)?
             .with_context(|| SchemaNotFoundSnafu {
-                schema_info: format!("{}.{}", &request.catalog, &request.schema),
+                catalog: &request.catalog,
+                schema: &request.schema,
             })?;
         schema
             .deregister_table(&request.table_name)
@@ -288,6 +312,20 @@ impl SchemaProvider for MemorySchemaProvider {
         }
     }
 
+    fn rename_table(&self, name: &str, new_name: String) -> Result<TableRef> {
+        let mut tables = self.tables.write().unwrap();
+        if tables.get(name).is_some() {
+            let table = tables.remove(name).unwrap();
+            tables.insert(new_name, table.clone());
+            Ok(table)
+        } else {
+            TableNotFoundSnafu {
+                table_info: name.to_string(),
+            }
+            .fail()?
+        }
+    }
+
     fn deregister_table(&self, name: &str) -> Result<Option<TableRef>> {
         let mut tables = self.tables.write().unwrap();
         Ok(tables.remove(name))
@@ -352,6 +390,85 @@ mod tests {
         assert_eq!(StatusCode::TableAlreadyExists, err.status_code());
     }
 
+    #[tokio::test]
+    async fn test_mem_provider_rename_table() {
+        let provider = MemorySchemaProvider::new();
+        let table_name = "num";
+        assert!(!provider.table_exist(table_name).unwrap());
+        let test_table: TableRef = Arc::new(NumbersTable::default());
+        // register test table
+        assert!(provider
+            .register_table(table_name.to_string(), test_table.clone())
+            .unwrap()
+            .is_none());
+        assert!(provider.table_exist(table_name).unwrap());
+
+        // rename test table
+        let new_table_name = "numbers";
+        provider
+            .rename_table(table_name, new_table_name.to_string())
+            .unwrap();
+
+        // test old table name not exist
+        assert!(!provider.table_exist(table_name).unwrap());
+        assert!(provider.deregister_table(table_name).unwrap().is_none());
+
+        // test new table name exists
+        assert!(provider.table_exist(new_table_name).unwrap());
+        let registered_table = provider.table(new_table_name).unwrap().unwrap();
+        assert_eq!(
+            registered_table.table_info().ident.table_id,
+            test_table.table_info().ident.table_id
+        );
+
+        let other_table = Arc::new(NumbersTable::new(2));
+        let result = provider.register_table(new_table_name.to_string(), other_table);
+        let err = result.err().unwrap();
+        assert_eq!(StatusCode::TableAlreadyExists, err.status_code());
+    }
+
+    #[tokio::test]
+    async fn test_catalog_rename_table() {
+        let catalog = MemoryCatalogManager::default();
+        let schema = catalog
+            .schema(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME)
+            .unwrap()
+            .unwrap();
+
+        // register table
+        let table_name = "num";
+        let table_id = 2333;
+        let table: TableRef = Arc::new(NumbersTable::new(table_id));
+        let register_table_req = RegisterTableRequest {
+            catalog: DEFAULT_CATALOG_NAME.to_string(),
+            schema: DEFAULT_SCHEMA_NAME.to_string(),
+            table_name: table_name.to_string(),
+            table_id,
+            table,
+        };
+        assert!(catalog.register_table(register_table_req).await.unwrap());
+        assert!(schema.table_exist(table_name).unwrap());
+
+        // rename table
+        let new_table_name = "numbers";
+        let rename_table_req = RenameTableRequest {
+            catalog: DEFAULT_CATALOG_NAME.to_string(),
+            schema: DEFAULT_SCHEMA_NAME.to_string(),
+            table_name: table_name.to_string(),
+            new_table_name: new_table_name.to_string(),
+            table_id,
+        };
+        assert!(catalog.rename_table(rename_table_req).await.unwrap());
+        assert!(!schema.table_exist(table_name).unwrap());
+        assert!(schema.table_exist(new_table_name).unwrap());
+
+        let registered_table = catalog
+            .table(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, new_table_name)
+            .unwrap()
+            .unwrap();
+        assert_eq!(registered_table.table_info().ident.table_id, table_id);
+    }
+
     #[test]
     pub fn test_register_if_absent() {
         let list = MemoryCatalogManager::default();

src/catalog/src/remote.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/catalog/src/remote/client.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/catalog/src/remote/manager.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -43,7 +43,7 @@ use crate::remote::{Kv, KvBackendRef};
 use crate::{
     handle_system_table_request, CatalogList, CatalogManager, CatalogProvider, CatalogProviderRef,
     DeregisterTableRequest, RegisterSchemaRequest, RegisterSystemTableRequest,
-    RegisterTableRequest, SchemaProvider, SchemaProviderRef,
+    RegisterTableRequest, RenameTableRequest, SchemaProvider, SchemaProviderRef,
 };
 
 /// Catalog manager based on metasrv.
@@ -418,7 +418,8 @@ impl CatalogManager for RemoteCatalogManager {
             catalog_provider
                 .schema(&schema_name)?
                 .with_context(|| SchemaNotFoundSnafu {
-                    schema_info: format!("{}.{}", &catalog_name, &schema_name),
+                    catalog: &catalog_name,
+                    schema: &schema_name,
                 })?;
         if schema_provider.table_exist(&request.table_name)? {
             return TableExistsSnafu {
@@ -448,6 +449,13 @@ impl CatalogManager for RemoteCatalogManager {
         Ok(true)
     }
 
+    async fn rename_table(&self, _request: RenameTableRequest) -> Result<bool> {
+        UnimplementedSnafu {
+            operation: "rename table",
+        }
+        .fail()
+    }
+
     async fn register_system_table(&self, request: RegisterSystemTableRequest) -> Result<()> {
         let mut requests = self.system_table_requests.lock().await;
         requests.push(request);
@@ -474,7 +482,8 @@ impl CatalogManager for RemoteCatalogManager {
         let schema = catalog
             .schema(schema_name)?
             .with_context(|| SchemaNotFoundSnafu {
-                schema_info: format!("{catalog_name}.{schema_name}"),
+                catalog: catalog_name,
+                schema: schema_name,
             })?;
         schema.table(table_name)
     }
@@ -737,6 +746,13 @@ impl SchemaProvider for RemoteSchemaProvider {
         prev
     }
 
+    fn rename_table(&self, _name: &str, _new_name: String) -> Result<TableRef> {
+        UnimplementedSnafu {
+            operation: "rename table",
+        }
+        .fail()
+    }
+
     fn deregister_table(&self, name: &str) -> Result<Option<TableRef>> {
         let table_name = name.to_string();
         let table_key = self.build_regional_table_key(&table_name).to_string();

src/catalog/src/schema.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -35,6 +35,10 @@ pub trait SchemaProvider: Sync + Send {
     /// If a table of the same name existed before, it returns "Table already exists" error.
     fn register_table(&self, name: String, table: TableRef) -> Result<Option<TableRef>>;
 
+    /// If supported by the implementation, renames an existing table from this schema and returns it.
+    /// If no table of that name exists, returns "Table not found" error.
+    fn rename_table(&self, name: &str, new_name: String) -> Result<TableRef>;
+
     /// If supported by the implementation, removes an existing table from this schema and returns it.
     /// If no table of that name exists, returns Ok(None).
     fn deregister_table(&self, name: &str) -> Result<Option<TableRef>>;

src/catalog/src/system.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -186,11 +186,23 @@ fn build_system_catalog_schema() -> Schema {
     SchemaBuilder::try_from(cols).unwrap().build().unwrap()
 }
 
-pub fn build_table_insert_request(full_table_name: String, table_id: TableId) -> InsertRequest {
+/// Formats key string for table entry in system catalog
+#[inline]
+pub fn format_table_entry_key(catalog: &str, schema: &str, table_id: TableId) -> String {
+    format!("{catalog}.{schema}.{table_id}")
+}
+
+pub fn build_table_insert_request(
+    catalog: String,
+    schema: String,
+    table_name: String,
+    table_id: TableId,
+) -> InsertRequest {
+    let entry_key = format_table_entry_key(&catalog, &schema, table_id);
     build_insert_request(
         EntryType::Table,
-        full_table_name.as_bytes(),
-        serde_json::to_string(&TableEntryValue { table_id })
+        entry_key.as_bytes(),
+        serde_json::to_string(&TableEntryValue { table_name })
             .unwrap()
             .as_bytes(),
     )
@@ -285,8 +297,8 @@ pub fn decode_system_catalog(
         }
 
         EntryType::Table => {
-            // As for table entry, the key is a string with format: `<catalog_name>.<schema_name>.<table_name>`
-            // and the value is a JSON string with format: `{"table_id": <table_id>}`
+            // As for table entry, the key is a string with format: `<catalog_name>.<schema_name>.<table_id>`
+            // and the value is a JSON string with format: `{"table_name": <table_name>}`
             let table_parts = key.split('.').collect::<Vec<_>>();
             ensure!(
                 table_parts.len() >= 3,
@@ -298,11 +310,12 @@ pub fn decode_system_catalog(
             debug!("Table meta value: {}", String::from_utf8_lossy(value));
             let table_meta: TableEntryValue =
                 serde_json::from_slice(value).context(ValueDeserializeSnafu)?;
+            let table_id = table_parts[2].parse::<TableId>().unwrap();
             Ok(Entry::Table(TableEntry {
                 catalog_name: table_parts[0].to_string(),
                 schema_name: table_parts[1].to_string(),
-                table_name: table_parts[2].to_string(),
-                table_id: table_meta.table_id,
+                table_name: table_meta.table_name,
+                table_id,
             }))
         }
     }
@@ -362,12 +375,12 @@ pub struct TableEntry {
 
 #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
 pub struct TableEntryValue {
-    pub table_id: TableId,
+    pub table_name: String,
 }
 
 #[cfg(test)]
 mod tests {
-    use log_store::fs::noop::NoopLogStore;
+    use log_store::NoopLogStore;
     use mito::config::EngineConfig;
     use mito::engine::MitoEngine;
     use object_store::ObjectStore;
@@ -415,8 +428,8 @@ mod tests {
     pub fn test_decode_table() {
         let entry = decode_system_catalog(
             Some(EntryType::Table as u8),
-            Some("some_catalog.some_schema.some_table".as_bytes()),
-            Some("{\"table_id\":42}".as_bytes()),
+            Some("some_catalog.some_schema.42".as_bytes()),
+            Some("{\"table_name\":\"some_table\"}".as_bytes()),
         )
         .unwrap();
 
@@ -435,7 +448,7 @@ mod tests {
     pub fn test_decode_mismatch() {
         decode_system_catalog(
             Some(EntryType::Table as u8),
-            Some("some_catalog.some_schema.some_table".as_bytes()),
+            Some("some_catalog.some_schema.42".as_bytes()),
             None,
         )
         .unwrap();

src/catalog/src/tables.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -40,9 +40,7 @@ use table::{Table, TableRef};
 
 use crate::error::{Error, InsertCatalogRecordSnafu};
 use crate::system::{build_schema_insert_request, build_table_insert_request, SystemCatalogTable};
-use crate::{
-    format_full_table_name, CatalogListRef, CatalogProvider, SchemaProvider, SchemaProviderRef,
-};
+use crate::{CatalogListRef, CatalogProvider, SchemaProvider, SchemaProviderRef};
 
 /// Tables holds all tables created by user.
 pub struct Tables {
@@ -233,6 +231,10 @@ impl SchemaProvider for InformationSchema {
         panic!("System catalog & schema does not support register table")
     }
 
+    fn rename_table(&self, _name: &str, _new_name: String) -> crate::error::Result<TableRef> {
+        unimplemented!("System catalog & schema does not support rename table")
+    }
+
     fn deregister_table(&self, _name: &str) -> crate::error::Result<Option<TableRef>> {
         panic!("System catalog & schema does not support deregister table")
     }
@@ -269,8 +271,7 @@ impl SystemCatalog {
         table_name: String,
         table_id: TableId,
     ) -> crate::error::Result<usize> {
-        let full_table_name = format_full_table_name(&catalog, &schema, &table_name);
-        let request = build_table_insert_request(full_table_name, table_id);
+        let request = build_table_insert_request(catalog, schema, table_name, table_id);
         self.information_schema
             .system
             .insert(request)

src/catalog/tests/local_catalog_tests.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -17,7 +17,7 @@ mod tests {
     use std::sync::Arc;
 
     use catalog::local::LocalCatalogManager;
-    use catalog::{CatalogManager, RegisterTableRequest};
+    use catalog::{CatalogManager, RegisterTableRequest, RenameTableRequest};
     use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
     use common_telemetry::{error, info};
     use mito::config::EngineConfig;
@@ -38,6 +38,44 @@ mod tests {
         Ok(catalog_manager)
     }
 
+    #[tokio::test]
+    async fn test_rename_table() {
+        common_telemetry::init_default_ut_logging();
+        let catalog_manager = create_local_catalog_manager().await.unwrap();
+        // register table
+        let table_name = "test_table";
+        let table_id = 42;
+        let table = Arc::new(NumbersTable::new(table_id));
+        let request = RegisterTableRequest {
+            catalog: DEFAULT_CATALOG_NAME.to_string(),
+            schema: DEFAULT_SCHEMA_NAME.to_string(),
+            table_name: table_name.to_string(),
+            table_id,
+            table: table.clone(),
+        };
+        assert!(catalog_manager.register_table(request).await.unwrap());
+
+        // rename table
+        let new_table_name = "table_t";
+        let rename_table_req = RenameTableRequest {
+            catalog: DEFAULT_CATALOG_NAME.to_string(),
+            schema: DEFAULT_SCHEMA_NAME.to_string(),
+            table_name: table_name.to_string(),
+            new_table_name: new_table_name.to_string(),
+            table_id,
+        };
+        assert!(catalog_manager
+            .rename_table(rename_table_req)
+            .await
+            .unwrap());
+
+        let registered_table = catalog_manager
+            .table(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, new_table_name)
+            .unwrap()
+            .unwrap();
+        assert_eq!(registered_table.table_info().ident.table_id, table_id);
+    }
+
     #[tokio::test]
     async fn test_duplicate_register() {
         let catalog_manager = create_local_catalog_manager().await.unwrap();

src/catalog/tests/mock.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/catalog/tests/remote_catalog_tests.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/client/Cargo.toml

@@ -6,7 +6,8 @@ license.workspace = true
 
 [dependencies]
 api = { path = "../api" }
-async-stream = "0.3"
+arrow-flight.workspace = true
+async-stream.workspace = true
 common-base = { path = "../common/base" }
 common-error = { path = "../common/error" }
 common-grpc = { path = "../common/grpc" }
@@ -17,15 +18,17 @@ common-time = { path = "../common/time" }
 datafusion.workspace = true
 datatypes = { path = "../datatypes" }
 enum_dispatch = "0.3"
+futures-util.workspace = true
 parking_lot = "0.12"
+prost.workspace = true
 rand = "0.8"
-snafu = { version = "0.7", features = ["backtraces"] }
-tonic = "0.8"
+snafu.workspace = true
+tonic.workspace = true
 
 [dev-dependencies]
 datanode = { path = "../datanode" }
 substrait = { path = "../common/substrait" }
-tokio = { version = "1.0", features = ["full"] }
+tokio.workspace = true
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 

src/client/examples/insert.rs

@@ -1,106 +0,0 @@
-// Copyright 2022 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use api::v1::*;
-use client::{Client, Database};
-
-fn main() {
-    tracing::subscriber::set_global_default(tracing_subscriber::FmtSubscriber::builder().finish())
-        .unwrap();
-
-    run();
-}
-
-#[tokio::main]
-async fn run() {
-    let client = Client::with_urls(vec!["127.0.0.1:3001"]);
-    let db = Database::new("greptime", client);
-
-    let (columns, row_count) = insert_data();
-
-    let expr = InsertExpr {
-        schema_name: "public".to_string(),
-        table_name: "demo".to_string(),
-        region_number: 0,
-        columns,
-        row_count,
-    };
-    db.insert(expr).await.unwrap();
-}
-
-fn insert_data() -> (Vec<Column>, u32) {
-    const SEMANTIC_TAG: i32 = 0;
-    const SEMANTIC_FIELD: i32 = 1;
-    const SEMANTIC_TS: i32 = 2;
-
-    let row_count = 4;
-
-    let host_vals = column::Values {
-        string_values: vec![
-            "host1".to_string(),
-            "host2".to_string(),
-            "host3".to_string(),
-            "host4".to_string(),
-        ],
-        ..Default::default()
-    };
-    let host_column = Column {
-        column_name: "host".to_string(),
-        semantic_type: SEMANTIC_TAG,
-        values: Some(host_vals),
-        null_mask: vec![0],
-        ..Default::default()
-    };
-
-    let cpu_vals = column::Values {
-        f64_values: vec![0.31, 0.41, 0.2],
-        ..Default::default()
-    };
-    let cpu_column = Column {
-        column_name: "cpu".to_string(),
-        semantic_type: SEMANTIC_FIELD,
-        values: Some(cpu_vals),
-        null_mask: vec![2],
-        ..Default::default()
-    };
-
-    let mem_vals = column::Values {
-        f64_values: vec![0.1, 0.2, 0.3],
-        ..Default::default()
-    };
-    let mem_column = Column {
-        column_name: "memory".to_string(),
-        semantic_type: SEMANTIC_FIELD,
-        values: Some(mem_vals),
-        null_mask: vec![4],
-        ..Default::default()
-    };
-
-    let ts_vals = column::Values {
-        i64_values: vec![100, 101, 102, 103],
-        ..Default::default()
-    };
-    let ts_column = Column {
-        column_name: "ts".to_string(),
-        semantic_type: SEMANTIC_TS,
-        values: Some(ts_vals),
-        null_mask: vec![0],
-        ..Default::default()
-    };
-
-    (
-        vec![host_column, cpu_column, mem_column, ts_column],
-        row_count,
-    )
-}

src/client/examples/logical.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -13,7 +13,6 @@
 // limitations under the License.
 
 use api::v1::{ColumnDataType, ColumnDef, CreateTableExpr, TableId};
-use client::admin::Admin;
 use client::{Client, Database};
 use prost_09::Message;
 use substrait_proto::protobuf::plan_rel::RelType as PlanRelType;
@@ -66,8 +65,8 @@ async fn run() {
         region_ids: vec![0],
     };
 
-    let admin = Admin::new("create table", client.clone());
-    let result = admin.create(create_table_expr).await.unwrap();
+    let db = Database::new("create table", client.clone());
+    let result = db.create(create_table_expr).await.unwrap();
     event!(Level::INFO, "create table result: {:#?}", result);
 
     let logical = mock_logical_plan();

src/client/examples/select.rs

@@ -1,34 +0,0 @@
-// Copyright 2022 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use client::{Client, Database, Select};
-use tracing::{event, Level};
-
-fn main() {
-    tracing::subscriber::set_global_default(tracing_subscriber::FmtSubscriber::builder().finish())
-        .unwrap();
-
-    run();
-}
-
-#[tokio::main]
-async fn run() {
-    let client = Client::with_urls(vec!["127.0.0.1:3001"]);
-    let db = Database::new("greptime", client);
-
-    let sql = Select::Sql("select * from demo".to_string());
-    let result = db.select(sql).await.unwrap();
-
-    event!(Level::INFO, "result: {:#?}", result);
-}

src/client/src/admin.rs

@@ -1,137 +0,0 @@
-// Copyright 2022 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use api::v1::*;
-use common_error::prelude::StatusCode;
-use common_query::Output;
-use snafu::prelude::*;
-
-use crate::database::PROTOCOL_VERSION;
-use crate::{error, Client, Result};
-
-#[derive(Clone, Debug)]
-pub struct Admin {
-    name: String,
-    client: Client,
-}
-
-impl Admin {
-    pub fn new(name: impl Into<String>, client: Client) -> Self {
-        Self {
-            name: name.into(),
-            client,
-        }
-    }
-
-    pub async fn create(&self, expr: CreateTableExpr) -> Result<AdminResult> {
-        let header = ExprHeader {
-            version: PROTOCOL_VERSION,
-        };
-        let expr = AdminExpr {
-            header: Some(header),
-            expr: Some(admin_expr::Expr::CreateTable(expr)),
-        };
-        self.do_request(expr).await
-    }
-
-    pub async fn do_request(&self, expr: AdminExpr) -> Result<AdminResult> {
-        // `remove(0)` is safe because of `do_requests`'s invariants.
-        Ok(self.do_requests(vec![expr]).await?.remove(0))
-    }
-
-    pub async fn alter(&self, expr: AlterExpr) -> Result<AdminResult> {
-        let header = ExprHeader {
-            version: PROTOCOL_VERSION,
-        };
-        let expr = AdminExpr {
-            header: Some(header),
-            expr: Some(admin_expr::Expr::Alter(expr)),
-        };
-        self.do_request(expr).await
-    }
-
-    pub async fn drop_table(&self, expr: DropTableExpr) -> Result<AdminResult> {
-        let header = ExprHeader {
-            version: PROTOCOL_VERSION,
-        };
-        let expr = AdminExpr {
-            header: Some(header),
-            expr: Some(admin_expr::Expr::DropTable(expr)),
-        };
-
-        self.do_request(expr).await
-    }
-
-    /// Invariants: the lengths of input vec (`Vec<AdminExpr>`) and output vec (`Vec<AdminResult>`) are equal.
-    async fn do_requests(&self, exprs: Vec<AdminExpr>) -> Result<Vec<AdminResult>> {
-        let expr_count = exprs.len();
-        let req = AdminRequest {
-            name: self.name.clone(),
-            exprs,
-        };
-
-        let resp = self.client.admin(req).await?;
-
-        let results = resp.results;
-        ensure!(
-            results.len() == expr_count,
-            error::MissingResultSnafu {
-                name: "admin_results",
-                expected: expr_count,
-                actual: results.len(),
-            }
-        );
-        Ok(results)
-    }
-
-    pub async fn create_database(&self, expr: CreateDatabaseExpr) -> Result<AdminResult> {
-        let header = ExprHeader {
-            version: PROTOCOL_VERSION,
-        };
-        let expr = AdminExpr {
-            header: Some(header),
-            expr: Some(admin_expr::Expr::CreateDatabase(expr)),
-        };
-        Ok(self.do_requests(vec![expr]).await?.remove(0))
-    }
-}
-
-pub fn admin_result_to_output(admin_result: AdminResult) -> Result<Output> {
-    let header = admin_result.header.context(error::MissingHeaderSnafu)?;
-    if !StatusCode::is_success(header.code) {
-        return error::DatanodeSnafu {
-            code: header.code,
-            msg: header.err_msg,
-        }
-        .fail();
-    }
-
-    let result = admin_result.result.context(error::MissingResultSnafu {
-        name: "result".to_string(),
-        expected: 1_usize,
-        actual: 0_usize,
-    })?;
-    let output = match result {
-        admin_result::Result::Mutate(mutate) => {
-            if mutate.failure != 0 {
-                return error::MutateFailureSnafu {
-                    failure: mutate.failure,
-                }
-                .fail();
-            }
-            Output::AffectedRows(mutate.success as usize)
-        }
-    };
-    Ok(output)
-}

src/client/src/client.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -14,8 +14,7 @@
 
 use std::sync::Arc;
 
-use api::v1::greptime_client::GreptimeClient;
-use api::v1::*;
+use arrow_flight::flight_service_client::FlightServiceClient;
 use common_grpc::channel_manager::ChannelManager;
 use parking_lot::RwLock;
 use snafu::{OptionExt, ResultExt};
@@ -24,6 +23,21 @@ use tonic::transport::Channel;
 use crate::load_balance::{LoadBalance, Loadbalancer};
 use crate::{error, Result};
 
+pub(crate) struct FlightClient {
+    addr: String,
+    client: FlightServiceClient<Channel>,
+}
+
+impl FlightClient {
+    pub(crate) fn addr(&self) -> &str {
+        &self.addr
+    }
+
+    pub(crate) fn mut_inner(&mut self) -> &mut FlightServiceClient<Channel> {
+        &mut self.client
+    }
+}
+
 #[derive(Clone, Debug, Default)]
 pub struct Client {
     inner: Arc<Inner>,
@@ -104,57 +118,23 @@ impl Client {
         self.inner.set_peers(urls);
     }
 
-    pub async fn admin(&self, req: AdminRequest) -> Result<AdminResponse> {
-        let req = BatchRequest {
-            admins: vec![req],
-            ..Default::default()
-        };
-
-        let mut res = self.batch(req).await?;
-        res.admins.pop().context(error::MissingResultSnafu {
-            name: "admins",
-            expected: 1_usize,
-            actual: 0_usize,
-        })
-    }
-
-    pub async fn database(&self, req: DatabaseRequest) -> Result<DatabaseResponse> {
-        let req = BatchRequest {
-            databases: vec![req],
-            ..Default::default()
-        };
-
-        let mut res = self.batch(req).await?;
-        res.databases.pop().context(error::MissingResultSnafu {
-            name: "database",
-            expected: 1_usize,
-            actual: 0_usize,
-        })
-    }
-
-    pub async fn batch(&self, req: BatchRequest) -> Result<BatchResponse> {
-        let peer = self
+    pub(crate) fn make_client(&self) -> Result<FlightClient> {
+        let addr = self
             .inner
             .get_peer()
             .context(error::IllegalGrpcClientStateSnafu {
                 err_msg: "No available peer found",
             })?;
-        let mut client = self.make_client(&peer)?;
-        let result = client
-            .batch(req)
-            .await
-            .context(error::TonicStatusSnafu { addr: peer })?;
-        Ok(result.into_inner())
-    }
 
-    fn make_client(&self, addr: impl AsRef<str>) -> Result<GreptimeClient<Channel>> {
-        let addr = addr.as_ref();
         let channel = self
             .inner
             .channel_manager
-            .get(addr)
-            .context(error::CreateChannelSnafu { addr })?;
-        Ok(GreptimeClient::new(channel))
+            .get(&addr)
+            .context(error::CreateChannelSnafu { addr: &addr })?;
+        Ok(FlightClient {
+            addr,
+            client: FlightServiceClient::new(channel),
+        })
     }
 }
 

src/client/src/database.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -12,27 +12,26 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
+use std::str::FromStr;
 
-use api::v1::codec::SelectResult as GrpcSelectResult;
-use api::v1::column::SemanticType;
+use api::v1::ddl_request::Expr as DdlExpr;
+use api::v1::greptime_request::Request;
+use api::v1::query_request::Query;
 use api::v1::{
-    object_expr, object_result, select_expr, DatabaseRequest, ExprHeader, InsertExpr,
-    MutateResult as GrpcMutateResult, ObjectExpr, ObjectResult as GrpcObjectResult, SelectExpr,
+    AlterExpr, CreateTableExpr, DdlRequest, DropTableExpr, GreptimeRequest, InsertRequest,
+    QueryRequest,
 };
-use common_error::status_code::StatusCode;
-use common_grpc_expr::column_to_vector;
+use arrow_flight::{FlightData, Ticket};
+use common_error::prelude::*;
+use common_grpc::flight::{flight_messages_to_recordbatches, FlightDecoder, FlightMessage};
 use common_query::Output;
-use common_recordbatch::{RecordBatch, RecordBatches};
-use datatypes::prelude::*;
-use datatypes::schema::{ColumnSchema, Schema};
-use snafu::{ensure, OptionExt, ResultExt};
+use futures_util::{TryFutureExt, TryStreamExt};
+use prost::Message;
+use snafu::{ensure, ResultExt};
 
-use crate::error::{ColumnToVectorSnafu, ConvertSchemaSnafu, DatanodeSnafu, DecodeSelectSnafu};
+use crate::error::{ConvertFlightDataSnafu, IllegalFlightMessagesSnafu};
 use crate::{error, Client, Result};
 
-pub const PROTOCOL_VERSION: u32 = 1;
-
 #[derive(Clone, Debug)]
 pub struct Database {
     name: String,
@@ -51,198 +50,128 @@ impl Database {
         &self.name
     }
 
-    pub async fn insert(&self, insert: InsertExpr) -> Result<ObjectResult> {
-        let header = ExprHeader {
-            version: PROTOCOL_VERSION,
-        };
-        let expr = ObjectExpr {
-            header: Some(header),
-            expr: Some(object_expr::Expr::Insert(insert)),
-        };
-        self.object(expr).await?.try_into()
-    }
-
-    pub async fn batch_insert(&self, insert_exprs: Vec<InsertExpr>) -> Result<Vec<ObjectResult>> {
-        let header = ExprHeader {
-            version: PROTOCOL_VERSION,
-        };
-        let obj_exprs = insert_exprs
-            .into_iter()
-            .map(|expr| ObjectExpr {
-                header: Some(header.clone()),
-                expr: Some(object_expr::Expr::Insert(expr)),
-            })
-            .collect();
-        self.objects(obj_exprs)
-            .await?
-            .into_iter()
-            .map(|result| result.try_into())
-            .collect()
-    }
-
-    pub async fn select(&self, expr: Select) -> Result<ObjectResult> {
-        let select_expr = match expr {
-            Select::Sql(sql) => SelectExpr {
-                expr: Some(select_expr::Expr::Sql(sql)),
-            },
-        };
-        self.do_select(select_expr).await
-    }
-
-    pub async fn logical_plan(&self, logical_plan: Vec<u8>) -> Result<ObjectResult> {
-        let select_expr = SelectExpr {
-            expr: Some(select_expr::Expr::LogicalPlan(logical_plan)),
-        };
-        self.do_select(select_expr).await
-    }
-
-    async fn do_select(&self, select_expr: SelectExpr) -> Result<ObjectResult> {
-        let header = ExprHeader {
-            version: PROTOCOL_VERSION,
-        };
-
-        let expr = ObjectExpr {
-            header: Some(header),
-            expr: Some(object_expr::Expr::Select(select_expr)),
-        };
-
-        let obj_result = self.object(expr).await?;
-        obj_result.try_into()
-    }
-
-    pub async fn object(&self, expr: ObjectExpr) -> Result<GrpcObjectResult> {
-        let res = self.objects(vec![expr]).await?.pop().unwrap();
-        Ok(res)
-    }
-
-    async fn objects(&self, exprs: Vec<ObjectExpr>) -> Result<Vec<GrpcObjectResult>> {
-        let expr_count = exprs.len();
-        let req = DatabaseRequest {
-            name: self.name.clone(),
-            exprs,
-        };
-
-        let res = self.client.database(req).await?;
-        let res = res.results;
-
-        ensure!(
-            res.len() == expr_count,
-            error::MissingResultSnafu {
-                name: "object_results",
-                expected: expr_count,
-                actual: res.len(),
-            }
-        );
-
-        Ok(res)
-    }
-}
-
-#[derive(Debug)]
-pub enum ObjectResult {
-    Select(GrpcSelectResult),
-    Mutate(GrpcMutateResult),
-}
-
-impl TryFrom<api::v1::ObjectResult> for ObjectResult {
-    type Error = error::Error;
-
-    fn try_from(object_result: api::v1::ObjectResult) -> std::result::Result<Self, Self::Error> {
-        let header = object_result.header.context(error::MissingHeaderSnafu)?;
-        if !StatusCode::is_success(header.code) {
-            return DatanodeSnafu {
-                code: header.code,
-                msg: header.err_msg,
-            }
-            .fail();
-        }
-
-        let obj_result = object_result.result.context(error::MissingResultSnafu {
-            name: "result".to_string(),
-            expected: 1_usize,
-            actual: 0_usize,
-        })?;
-        Ok(match obj_result {
-            object_result::Result::Select(select) => {
-                let result = (*select.raw_data).try_into().context(DecodeSelectSnafu)?;
-                ObjectResult::Select(result)
-            }
-            object_result::Result::Mutate(mutate) => ObjectResult::Mutate(mutate),
+    pub async fn insert(&self, request: InsertRequest) -> Result<Output> {
+        self.do_get(GreptimeRequest {
+            request: Some(Request::Insert(request)),
         })
+        .await
     }
-}
 
-pub enum Select {
-    Sql(String),
-}
+    pub async fn sql(&self, sql: &str) -> Result<Output> {
+        self.do_get(GreptimeRequest {
+            request: Some(Request::Query(QueryRequest {
+                query: Some(Query::Sql(sql.to_string())),
+            })),
+        })
+        .await
+    }
 
-impl TryFrom<ObjectResult> for Output {
-    type Error = error::Error;
+    pub async fn logical_plan(&self, logical_plan: Vec<u8>) -> Result<Output> {
+        self.do_get(GreptimeRequest {
+            request: Some(Request::Query(QueryRequest {
+                query: Some(Query::LogicalPlan(logical_plan)),
+            })),
+        })
+        .await
+    }
 
-    fn try_from(value: ObjectResult) -> Result<Self> {
-        let output = match value {
-            ObjectResult::Select(select) => {
-                let vectors = select
-                    .columns
-                    .iter()
-                    .map(|column| {
-                        column_to_vector(column, select.row_count).context(ColumnToVectorSnafu)
+    pub async fn create(&self, expr: CreateTableExpr) -> Result<Output> {
+        self.do_get(GreptimeRequest {
+            request: Some(Request::Ddl(DdlRequest {
+                expr: Some(DdlExpr::CreateTable(expr)),
+            })),
+        })
+        .await
+    }
+
+    pub async fn alter(&self, expr: AlterExpr) -> Result<Output> {
+        self.do_get(GreptimeRequest {
+            request: Some(Request::Ddl(DdlRequest {
+                expr: Some(DdlExpr::Alter(expr)),
+            })),
+        })
+        .await
+    }
+
+    pub async fn drop_table(&self, expr: DropTableExpr) -> Result<Output> {
+        self.do_get(GreptimeRequest {
+            request: Some(Request::Ddl(DdlRequest {
+                expr: Some(DdlExpr::DropTable(expr)),
+            })),
+        })
+        .await
+    }
+
+    async fn do_get(&self, request: GreptimeRequest) -> Result<Output> {
+        let mut client = self.client.make_client()?;
+
+        // TODO(LFC): Streaming get flight data.
+        let flight_data: Vec<FlightData> = client
+            .mut_inner()
+            .do_get(Ticket {
+                ticket: request.encode_to_vec(),
+            })
+            .and_then(|response| response.into_inner().try_collect())
+            .await
+            .map_err(|e| {
+                let code = get_metadata_value(&e, INNER_ERROR_CODE)
+                    .and_then(|s| StatusCode::from_str(&s).ok())
+                    .unwrap_or(StatusCode::Unknown);
+                let msg = get_metadata_value(&e, INNER_ERROR_MSG).unwrap_or(e.to_string());
+                error::ExternalSnafu { code, msg }
+                    .fail::<()>()
+                    .map_err(BoxedError::new)
+                    .context(error::FlightGetSnafu {
+                        tonic_code: e.code(),
+                        addr: client.addr(),
                     })
-                    .collect::<Result<Vec<VectorRef>>>()?;
+                    .unwrap_err()
+            })?;
 
-                let column_schemas = select
-                    .columns
-                    .iter()
-                    .zip(vectors.iter())
-                    .map(|(column, vector)| {
-                        let datatype = vector.data_type();
-                        // nullable or not, does not affect the output
-                        let mut column_schema =
-                            ColumnSchema::new(&column.column_name, datatype, true);
-                        if column.semantic_type == SemanticType::Timestamp as i32 {
-                            column_schema = column_schema.with_time_index(true);
-                        }
-                        column_schema
-                    })
-                    .collect::<Vec<ColumnSchema>>();
+        let decoder = &mut FlightDecoder::default();
+        let flight_messages = flight_data
+            .into_iter()
+            .map(|x| decoder.try_decode(x).context(ConvertFlightDataSnafu))
+            .collect::<Result<Vec<_>>>()?;
 
-                let schema = Arc::new(Schema::try_new(column_schemas).context(ConvertSchemaSnafu)?);
-                let recordbatches = if vectors.is_empty() {
-                    RecordBatches::try_new(schema, vec![])
-                } else {
-                    RecordBatch::new(schema, vectors)
-                        .and_then(|batch| RecordBatches::try_new(batch.schema.clone(), vec![batch]))
+        let output = if let Some(FlightMessage::AffectedRows(rows)) = flight_messages.get(0) {
+            ensure!(
+                flight_messages.len() == 1,
+                IllegalFlightMessagesSnafu {
+                    reason: "Expect 'AffectedRows' Flight messages to be one and only!"
                 }
-                .context(error::CreateRecordBatchesSnafu)?;
-                Output::RecordBatches(recordbatches)
-            }
-            ObjectResult::Mutate(mutate) => {
-                if mutate.failure != 0 {
-                    return error::MutateFailureSnafu {
-                        failure: mutate.failure,
-                    }
-                    .fail();
-                }
-                Output::AffectedRows(mutate.success as usize)
-            }
+            );
+            Output::AffectedRows(*rows)
+        } else {
+            let recordbatches = flight_messages_to_recordbatches(flight_messages)
+                .context(ConvertFlightDataSnafu)?;
+            Output::RecordBatches(recordbatches)
         };
         Ok(output)
     }
 }
 
+fn get_metadata_value(e: &tonic::Status, key: &str) -> Option<String> {
+    e.metadata()
+        .get(key)
+        .and_then(|v| String::from_utf8(v.as_bytes().to_vec()).ok())
+}
+
 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
+
     use api::helper::ColumnDataTypeWrapper;
     use api::v1::Column;
     use common_grpc::select::{null_mask, values};
+    use common_grpc_expr::column_to_vector;
+    use datatypes::prelude::{Vector, VectorRef};
     use datatypes::vectors::{
         BinaryVector, BooleanVector, DateTimeVector, DateVector, Float32Vector, Float64Vector,
         Int16Vector, Int32Vector, Int64Vector, Int8Vector, StringVector, UInt16Vector,
         UInt32Vector, UInt64Vector, UInt8Vector,
     };
 
-    use super::*;
-
     #[test]
     fn test_column_to_vector() {
         let mut column = create_test_column(Arc::new(BooleanVector::from(vec![true])));

src/client/src/error.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -13,67 +13,43 @@
 // limitations under the License.
 
 use std::any::Any;
-use std::sync::Arc;
 
-use api::serde::DecodeError;
 use common_error::prelude::*;
-use datafusion::physical_plan::ExecutionPlan;
+use tonic::Code;
 
 #[derive(Debug, Snafu)]
 #[snafu(visibility(pub))]
 pub enum Error {
-    #[snafu(display("Connect failed to {}, source: {}", url, source))]
-    ConnectFailed {
-        url: String,
-        source: tonic::transport::Error,
+    #[snafu(display("Illegal Flight messages, reason: {}", reason))]
+    IllegalFlightMessages {
+        reason: String,
         backtrace: Backtrace,
     },
 
-    #[snafu(display("Missing {}, expected {}, actual {}", name, expected, actual))]
-    MissingResult {
-        name: String,
-        expected: usize,
-        actual: usize,
-    },
-
-    #[snafu(display("Missing result header"))]
-    MissingHeader,
-
-    #[snafu(display("Tonic internal error, addr: {}, source: {}", addr, source))]
-    TonicStatus {
+    #[snafu(display(
+        "Failed to do Flight get, addr: {}, code: {}, source: {}",
+        addr,
+        tonic_code,
+        source
+    ))]
+    FlightGet {
         addr: String,
-        source: tonic::Status,
-        backtrace: Backtrace,
+        tonic_code: Code,
+        source: BoxedError,
     },
 
-    #[snafu(display("Fail to decode select result, source: {}", source))]
-    DecodeSelect { source: DecodeError },
-
-    #[snafu(display("Error occurred on the data node, code: {}, msg: {}", code, msg))]
-    Datanode { code: u32, msg: String },
-
-    #[snafu(display("Failed to encode physical plan: {:?}, source: {}", physical, source))]
-    EncodePhysical {
-        physical: Arc<dyn ExecutionPlan>,
+    #[snafu(display("Failed to convert FlightData, source: {}", source))]
+    ConvertFlightData {
         #[snafu(backtrace)]
         source: common_grpc::Error,
     },
 
-    #[snafu(display("Mutate result has failure {}", failure))]
-    MutateFailure { failure: u32, backtrace: Backtrace },
-
     #[snafu(display("Column datatype error, source: {}", source))]
     ColumnDataType {
         #[snafu(backtrace)]
         source: api::error::Error,
     },
 
-    #[snafu(display("Failed to create RecordBatches, source: {}", source))]
-    CreateRecordBatches {
-        #[snafu(backtrace)]
-        source: common_recordbatch::error::Error,
-    },
-
     #[snafu(display("Illegal GRPC client state: {}", err_msg))]
     IllegalGrpcClientState {
         err_msg: String,
@@ -83,12 +59,6 @@ pub enum Error {
     #[snafu(display("Missing required field in protobuf, field: {}", field))]
     MissingField { field: String, backtrace: Backtrace },
 
-    #[snafu(display("Failed to convert schema, source: {}", source))]
-    ConvertSchema {
-        #[snafu(backtrace)]
-        source: datatypes::error::Error,
-    },
-
     #[snafu(display(
         "Failed to create gRPC channel, peer address: {}, source: {}",
         addr,
@@ -100,11 +70,9 @@ pub enum Error {
         source: common_grpc::error::Error,
     },
 
-    #[snafu(display("Failed to convert column to vector, source: {}", source))]
-    ColumnToVector {
-        #[snafu(backtrace)]
-        source: common_grpc_expr::error::Error,
-    },
+    /// Error deserialized from gRPC metadata
+    #[snafu(display("{}", msg))]
+    ExternalError { code: StatusCode, msg: String },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -112,21 +80,15 @@ pub type Result<T> = std::result::Result<T, Error>;
 impl ErrorExt for Error {
     fn status_code(&self) -> StatusCode {
         match self {
-            Error::ConnectFailed { .. }
-            | Error::MissingResult { .. }
-            | Error::MissingHeader { .. }
-            | Error::TonicStatus { .. }
-            | Error::DecodeSelect { .. }
-            | Error::Datanode { .. }
-            | Error::EncodePhysical { .. }
-            | Error::MutateFailure { .. }
+            Error::IllegalFlightMessages { .. }
             | Error::ColumnDataType { .. }
             | Error::MissingField { .. } => StatusCode::Internal,
-            Error::ConvertSchema { source } => source.status_code(),
-            Error::CreateRecordBatches { source } => source.status_code(),
-            Error::CreateChannel { source, .. } => source.status_code(),
+            Error::FlightGet { source, .. } => source.status_code(),
+            Error::CreateChannel { source, .. } | Error::ConvertFlightData { source } => {
+                source.status_code()
+            }
             Error::IllegalGrpcClientState { .. } => StatusCode::Unexpected,
-            Error::ColumnToVector { source, .. } => source.status_code(),
+            Error::ExternalError { code, .. } => *code,
         }
     }
 

src/client/src/lib.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub mod admin;
 mod client;
 mod database;
 mod error;
@@ -21,5 +20,5 @@ pub mod load_balance;
 pub use api;
 
 pub use self::client::Client;
-pub use self::database::{Database, ObjectResult, Select};
+pub use self::database::Database;
 pub use self::error::{Error, Result};

src/client/src/load_balance.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/cmd/Cargo.toml

@@ -18,17 +18,17 @@ common-telemetry = { path = "../common/telemetry", features = [
 ] }
 datanode = { path = "../datanode" }
 frontend = { path = "../frontend" }
-futures = "0.3"
+futures.workspace = true
 meta-client = { path = "../meta-client" }
 meta-srv = { path = "../meta-srv" }
-serde = "1.0"
+serde.workspace = true
 servers = { path = "../servers" }
-snafu = { version = "0.7", features = ["backtraces"] }
+snafu.workspace = true
 tokio = { version = "1.18", features = ["full"] }
 toml = "0.5"
 
 [dev-dependencies]
-serde = "1.0"
+serde.workspace = true
 tempdir = "0.3"
 
 [build-dependencies]

src/cmd/build.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/cmd/src/bin/greptime.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/cmd/src/datanode.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -125,7 +125,7 @@ impl TryFrom<StartCommand> for DatanodeOptions {
         }
 
         if let Some(wal_dir) = cmd.wal_dir {
-            opts.wal_dir = wal_dir;
+            opts.wal.dir = wal_dir;
         }
         Ok(opts)
     }
@@ -134,6 +134,7 @@ impl TryFrom<StartCommand> for DatanodeOptions {
 #[cfg(test)]
 mod tests {
     use std::assert_matches::assert_matches;
+    use std::time::Duration;
 
     use datanode::datanode::ObjectStoreConfig;
     use servers::Mode;
@@ -151,7 +152,7 @@ mod tests {
         };
         let options: DatanodeOptions = cmd.try_into().unwrap();
         assert_eq!("127.0.0.1:3001".to_string(), options.rpc_addr);
-        assert_eq!("/tmp/greptimedb/wal".to_string(), options.wal_dir);
+        assert_eq!("/tmp/greptimedb/wal".to_string(), options.wal.dir);
         assert_eq!("127.0.0.1:4406".to_string(), options.mysql_addr);
         assert_eq!(4, options.mysql_runtime_size);
         let MetaClientOpts {
@@ -216,6 +217,11 @@ mod tests {
             ..Default::default()
         })
         .unwrap();
+        assert_eq!("/tmp/greptimedb/wal", dn_opts.wal.dir);
+        assert_eq!(Duration::from_secs(600), dn_opts.wal.purge_interval);
+        assert_eq!(1024 * 1024 * 1024, dn_opts.wal.file_size.0);
+        assert_eq!(1024 * 1024 * 1024 * 50, dn_opts.wal.purge_threshold.0);
+        assert!(!dn_opts.wal.sync_write);
         assert_eq!(Some(42), dn_opts.node_id);
         let MetaClientOpts {
             metasrv_addrs: metasrv_addr,

src/cmd/src/error.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/cmd/src/frontend.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/cmd/src/lib.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/cmd/src/metasrv.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/cmd/src/standalone.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -16,7 +16,7 @@ use std::sync::Arc;
 
 use clap::Parser;
 use common_telemetry::info;
-use datanode::datanode::{Datanode, DatanodeOptions, ObjectStoreConfig};
+use datanode::datanode::{Datanode, DatanodeOptions, ObjectStoreConfig, WalConfig};
 use datanode::instance::InstanceRef;
 use frontend::frontend::{Frontend, FrontendOptions};
 use frontend::grpc::GrpcOptions;
@@ -63,6 +63,7 @@ impl SubCommand {
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(default)]
 pub struct StandaloneOptions {
     pub http_options: Option<HttpOptions>,
     pub grpc_options: Option<GrpcOptions>,
@@ -72,7 +73,7 @@ pub struct StandaloneOptions {
     pub influxdb_options: Option<InfluxdbOptions>,
     pub prometheus_options: Option<PrometheusOptions>,
     pub mode: Mode,
-    pub wal_dir: String,
+    pub wal: WalConfig,
     pub storage: ObjectStoreConfig,
     pub enable_memory_catalog: bool,
 }
@@ -88,7 +89,7 @@ impl Default for StandaloneOptions {
             influxdb_options: Some(InfluxdbOptions::default()),
             prometheus_options: Some(PrometheusOptions::default()),
             mode: Mode::Standalone,
-            wal_dir: "/tmp/greptimedb/wal".to_string(),
+            wal: WalConfig::default(),
             storage: ObjectStoreConfig::default(),
             enable_memory_catalog: false,
         }
@@ -112,7 +113,7 @@ impl StandaloneOptions {
 
     fn datanode_options(self) -> DatanodeOptions {
         DatanodeOptions {
-            wal_dir: self.wal_dir,
+            wal: self.wal,
             storage: self.storage,
             enable_memory_catalog: self.enable_memory_catalog,
             ..Default::default()

src/cmd/src/toml_loader.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -37,12 +37,23 @@ mod tests {
     use crate::error::Result;
 
     #[derive(Clone, PartialEq, Debug, Deserialize, Serialize)]
+    #[serde(default)]
     struct MockConfig {
         path: String,
         port: u32,
         host: String,
     }
 
+    impl Default for MockConfig {
+        fn default() -> Self {
+            Self {
+                path: "test".to_string(),
+                port: 0,
+                host: "localhost".to_string(),
+            }
+        }
+    }
+
     #[test]
     fn test_from_file() -> Result<()> {
         let config = MockConfig {
@@ -63,6 +74,21 @@ mod tests {
         let loaded_config: MockConfig = from_file!(&test_file)?;
         assert_eq!(loaded_config, config);
 
+        // Only host in file
+        let mut file = File::create(&test_file).unwrap();
+        file.write_all("host='greptime.test'\n".as_bytes()).unwrap();
+
+        let loaded_config: MockConfig = from_file!(&test_file)?;
+        assert_eq!(loaded_config.host, "greptime.test");
+        assert_eq!(loaded_config.port, 0);
+        assert_eq!(loaded_config.path, "test");
+
+        // Truncate the file.
+        let file = File::create(&test_file).unwrap();
+        file.set_len(0).unwrap();
+        let loaded_config: MockConfig = from_file!(&test_file)?;
+        assert_eq!(loaded_config, MockConfig::default());
+
         Ok(())
     }
 }

src/common/base/Cargo.toml

@@ -10,4 +10,7 @@ bytes = { version = "1.1", features = ["serde"] }
 common-error = { path = "../error" }
 paste = "1.0"
 serde = { version = "1.0", features = ["derive"] }
-snafu = { version = "0.7", features = ["backtraces"] }
+snafu.workspace = true
+
+[dev-dependencies]
+toml = "0.5"

src/common/base/src/bit_vec.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/base/src/buffer.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/base/src/bytes.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/base/src/lib.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -15,5 +15,7 @@
 pub mod bit_vec;
 pub mod buffer;
 pub mod bytes;
+#[allow(clippy::all)]
+pub mod readable_size;
 
 pub use bit_vec::BitVec;

src/common/base/src/readable_size.rs

@@ -0,0 +1,321 @@
+// Copyright (c) 2017-present, PingCAP, Inc. Licensed under Apache-2.0.
+
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// This file is copied from https://github.com/tikv/raft-engine/blob/8dd2a39f359ff16f5295f35343f626e0c10132fa/src/util.rs without any modification.
+
+use std::fmt;
+use std::fmt::{Display, Write};
+use std::ops::{Div, Mul};
+use std::str::FromStr;
+
+use serde::de::{Unexpected, Visitor};
+use serde::{de, Deserialize, Deserializer, Serialize, Serializer};
+
+const UNIT: u64 = 1;
+
+const BINARY_DATA_MAGNITUDE: u64 = 1024;
+pub const B: u64 = UNIT;
+pub const KIB: u64 = B * BINARY_DATA_MAGNITUDE;
+pub const MIB: u64 = KIB * BINARY_DATA_MAGNITUDE;
+pub const GIB: u64 = MIB * BINARY_DATA_MAGNITUDE;
+pub const TIB: u64 = GIB * BINARY_DATA_MAGNITUDE;
+pub const PIB: u64 = TIB * BINARY_DATA_MAGNITUDE;
+
+#[derive(Clone, Debug, Copy, PartialEq, Eq, PartialOrd)]
+pub struct ReadableSize(pub u64);
+
+impl ReadableSize {
+    pub const fn kb(count: u64) -> ReadableSize {
+        ReadableSize(count * KIB)
+    }
+
+    pub const fn mb(count: u64) -> ReadableSize {
+        ReadableSize(count * MIB)
+    }
+
+    pub const fn gb(count: u64) -> ReadableSize {
+        ReadableSize(count * GIB)
+    }
+
+    pub const fn as_mb(self) -> u64 {
+        self.0 / MIB
+    }
+}
+
+impl Div<u64> for ReadableSize {
+    type Output = ReadableSize;
+
+    fn div(self, rhs: u64) -> ReadableSize {
+        ReadableSize(self.0 / rhs)
+    }
+}
+
+impl Div<ReadableSize> for ReadableSize {
+    type Output = u64;
+
+    fn div(self, rhs: ReadableSize) -> u64 {
+        self.0 / rhs.0
+    }
+}
+
+impl Mul<u64> for ReadableSize {
+    type Output = ReadableSize;
+
+    fn mul(self, rhs: u64) -> ReadableSize {
+        ReadableSize(self.0 * rhs)
+    }
+}
+
+impl Serialize for ReadableSize {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        let size = self.0;
+        let mut buffer = String::new();
+        if size == 0 {
+            write!(buffer, "{}KiB", size).unwrap();
+        } else if size % PIB == 0 {
+            write!(buffer, "{}PiB", size / PIB).unwrap();
+        } else if size % TIB == 0 {
+            write!(buffer, "{}TiB", size / TIB).unwrap();
+        } else if size % GIB as u64 == 0 {
+            write!(buffer, "{}GiB", size / GIB).unwrap();
+        } else if size % MIB as u64 == 0 {
+            write!(buffer, "{}MiB", size / MIB).unwrap();
+        } else if size % KIB as u64 == 0 {
+            write!(buffer, "{}KiB", size / KIB).unwrap();
+        } else {
+            return serializer.serialize_u64(size);
+        }
+        serializer.serialize_str(&buffer)
+    }
+}
+
+impl FromStr for ReadableSize {
+    type Err = String;
+
+    // This method parses value in binary unit.
+    fn from_str(s: &str) -> Result<ReadableSize, String> {
+        let size_str = s.trim();
+        if size_str.is_empty() {
+            return Err(format!("{:?} is not a valid size.", s));
+        }
+
+        if !size_str.is_ascii() {
+            return Err(format!("ASCII string is expected, but got {:?}", s));
+        }
+
+        // size: digits and '.' as decimal separator
+        let size_len = size_str
+            .to_string()
+            .chars()
+            .take_while(|c| char::is_ascii_digit(c) || ['.', 'e', 'E', '-', '+'].contains(c))
+            .count();
+
+        // unit: alphabetic characters
+        let (size, unit) = size_str.split_at(size_len);
+
+        let unit = match unit.trim() {
+            "K" | "KB" | "KiB" => KIB,
+            "M" | "MB" | "MiB" => MIB,
+            "G" | "GB" | "GiB" => GIB,
+            "T" | "TB" | "TiB" => TIB,
+            "P" | "PB" | "PiB" => PIB,
+            "B" | "" => B,
+            _ => {
+                return Err(format!(
+                    "only B, KB, KiB, MB, MiB, GB, GiB, TB, TiB, PB, and PiB are supported: {:?}",
+                    s
+                ));
+            }
+        };
+
+        match size.parse::<f64>() {
+            Ok(n) => Ok(ReadableSize((n * unit as f64) as u64)),
+            Err(_) => Err(format!("invalid size string: {:?}", s)),
+        }
+    }
+}
+
+impl Display for ReadableSize {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        if self.0 >= PIB {
+            write!(f, "{:.1}PiB", self.0 as f64 / PIB as f64)
+        } else if self.0 >= TIB {
+            write!(f, "{:.1}TiB", self.0 as f64 / TIB as f64)
+        } else if self.0 >= GIB {
+            write!(f, "{:.1}GiB", self.0 as f64 / GIB as f64)
+        } else if self.0 >= MIB {
+            write!(f, "{:.1}MiB", self.0 as f64 / MIB as f64)
+        } else if self.0 >= KIB {
+            write!(f, "{:.1}KiB", self.0 as f64 / KIB as f64)
+        } else {
+            write!(f, "{}B", self.0)
+        }
+    }
+}
+
+impl<'de> Deserialize<'de> for ReadableSize {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        struct SizeVisitor;
+
+        impl<'de> Visitor<'de> for SizeVisitor {
+            type Value = ReadableSize;
+
+            fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+                formatter.write_str("valid size")
+            }
+
+            fn visit_i64<E>(self, size: i64) -> Result<ReadableSize, E>
+            where
+                E: de::Error,
+            {
+                if size >= 0 {
+                    self.visit_u64(size as u64)
+                } else {
+                    Err(E::invalid_value(Unexpected::Signed(size), &self))
+                }
+            }
+
+            fn visit_u64<E>(self, size: u64) -> Result<ReadableSize, E>
+            where
+                E: de::Error,
+            {
+                Ok(ReadableSize(size))
+            }
+
+            fn visit_str<E>(self, size_str: &str) -> Result<ReadableSize, E>
+            where
+                E: de::Error,
+            {
+                size_str.parse().map_err(E::custom)
+            }
+        }
+
+        deserializer.deserialize_any(SizeVisitor)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_readable_size() {
+        let s = ReadableSize::kb(2);
+        assert_eq!(s.0, 2048);
+        assert_eq!(s.as_mb(), 0);
+        let s = ReadableSize::mb(2);
+        assert_eq!(s.0, 2 * 1024 * 1024);
+        assert_eq!(s.as_mb(), 2);
+        let s = ReadableSize::gb(2);
+        assert_eq!(s.0, 2 * 1024 * 1024 * 1024);
+        assert_eq!(s.as_mb(), 2048);
+
+        assert_eq!((ReadableSize::mb(2) / 2).0, MIB);
+        assert_eq!((ReadableSize::mb(1) / 2).0, 512 * KIB);
+        assert_eq!(ReadableSize::mb(2) / ReadableSize::kb(1), 2048);
+    }
+
+    #[test]
+    fn test_parse_readable_size() {
+        #[derive(Serialize, Deserialize)]
+        struct SizeHolder {
+            s: ReadableSize,
+        }
+
+        let legal_cases = vec![
+            (0, "0KiB"),
+            (2 * KIB, "2KiB"),
+            (4 * MIB, "4MiB"),
+            (5 * GIB, "5GiB"),
+            (7 * TIB, "7TiB"),
+            (11 * PIB, "11PiB"),
+        ];
+        for (size, exp) in legal_cases {
+            let c = SizeHolder {
+                s: ReadableSize(size),
+            };
+            let res_str = toml::to_string(&c).unwrap();
+            let exp_str = format!("s = {:?}\n", exp);
+            assert_eq!(res_str, exp_str);
+            let res_size: SizeHolder = toml::from_str(&exp_str).unwrap();
+            assert_eq!(res_size.s.0, size);
+        }
+
+        let c = SizeHolder {
+            s: ReadableSize(512),
+        };
+        let res_str = toml::to_string(&c).unwrap();
+        assert_eq!(res_str, "s = 512\n");
+        let res_size: SizeHolder = toml::from_str(&res_str).unwrap();
+        assert_eq!(res_size.s.0, c.s.0);
+
+        let decode_cases = vec![
+            (" 0.5 PB", PIB / 2),
+            ("0.5 TB", TIB / 2),
+            ("0.5GB ", GIB / 2),
+            ("0.5MB", MIB / 2),
+            ("0.5KB", KIB / 2),
+            ("0.5P", PIB / 2),
+            ("0.5T", TIB / 2),
+            ("0.5G", GIB / 2),
+            ("0.5M", MIB / 2),
+            ("0.5K", KIB / 2),
+            ("23", 23),
+            ("1", 1),
+            ("1024B", KIB),
+            // units with binary prefixes
+            (" 0.5 PiB", PIB / 2),
+            ("1PiB", PIB),
+            ("0.5 TiB", TIB / 2),
+            ("2 TiB", TIB * 2),
+            ("0.5GiB ", GIB / 2),
+            ("787GiB ", GIB * 787),
+            ("0.5MiB", MIB / 2),
+            ("3MiB", MIB * 3),
+            ("0.5KiB", KIB / 2),
+            ("1 KiB", KIB),
+            // scientific notation
+            ("0.5e6 B", B * 500000),
+            ("0.5E6 B", B * 500000),
+            ("1e6B", B * 1000000),
+            ("8E6B", B * 8000000),
+            ("8e7", B * 80000000),
+            ("1e-1MB", MIB / 10),
+            ("1e+1MB", MIB * 10),
+            ("0e+10MB", 0),
+        ];
+        for (src, exp) in decode_cases {
+            let src = format!("s = {:?}", src);
+            let res: SizeHolder = toml::from_str(&src).unwrap();
+            assert_eq!(res.s.0, exp);
+        }
+
+        let illegal_cases = vec![
+            "0.5kb", "0.5kB", "0.5Kb", "0.5k", "0.5g", "b", "gb", "1b", "B", "1K24B", " 5_KB",
+            "4B7", "5M_",
+        ];
+        for src in illegal_cases {
+            let src_str = format!("s = {:?}", src);
+            assert!(toml::from_str::<SizeHolder>(&src_str).is_err(), "{}", src);
+        }
+    }
+}

src/common/base/tests/buffer_tests.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/catalog/Cargo.toml

@@ -11,7 +11,7 @@ common-telemetry = { path = "../telemetry" }
 datatypes = { path = "../../datatypes" }
 lazy_static = "1.4"
 regex = "1.6"
-serde = "1.0"
+serde.workspace = true
 serde_json = "1.0"
 snafu = { version = "0.7", features = ["backtraces"] }
 

src/common/catalog/src/consts.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/catalog/src/error.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/catalog/src/lib.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/error/Cargo.toml

@@ -5,4 +5,5 @@ edition.workspace = true
 license.workspace = true
 
 [dependencies]
+strum = "0.24.1"
 snafu = { version = "0.7", features = ["backtraces"] }

src/common/error/src/ext.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -33,72 +33,60 @@ pub trait ErrorExt: std::error::Error {
     fn as_any(&self) -> &dyn Any;
 }
 
-/// A helper macro to define a opaque boxed error based on errors that implement [ErrorExt] trait.
-#[macro_export]
-macro_rules! define_opaque_error {
-    ($Error:ident) => {
-        /// An error behaves like `Box<dyn Error>`.
-        ///
-        /// Define this error as a new type instead of using `Box<dyn Error>` directly so we can implement
-        /// more methods or traits for it.
-        pub struct $Error {
-            inner: Box<dyn $crate::ext::ErrorExt + Send + Sync>,
-        }
-
-        impl $Error {
-            pub fn new<E: $crate::ext::ErrorExt + Send + Sync + 'static>(err: E) -> Self {
-                Self {
-                    inner: Box::new(err),
-                }
-            }
-        }
-
-        impl std::fmt::Debug for $Error {
-            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-                // Use the pretty debug format of inner error for opaque error.
-                let debug_format = $crate::format::DebugFormat::new(&*self.inner);
-                debug_format.fmt(f)
-            }
-        }
-
-        impl std::fmt::Display for $Error {
-            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-                write!(f, "{}", self.inner)
-            }
-        }
-
-        impl std::error::Error for $Error {
-            fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
-                self.inner.source()
-            }
-        }
-
-        impl $crate::ext::ErrorExt for $Error {
-            fn status_code(&self) -> $crate::status_code::StatusCode {
-                self.inner.status_code()
-            }
-
-            fn backtrace_opt(&self) -> Option<&$crate::snafu::Backtrace> {
-                self.inner.backtrace_opt()
-            }
-
-            fn as_any(&self) -> &dyn std::any::Any {
-                self.inner.as_any()
-            }
-        }
-
-        // Implement ErrorCompat for this opaque error so the backtrace is also available
-        // via `ErrorCompat::backtrace()`.
-        impl $crate::snafu::ErrorCompat for $Error {
-            fn backtrace(&self) -> Option<&$crate::snafu::Backtrace> {
-                self.inner.backtrace_opt()
-            }
-        }
-    };
+/// An opaque boxed error based on errors that implement [ErrorExt] trait.
+pub struct BoxedError {
+    inner: Box<dyn crate::ext::ErrorExt + Send + Sync>,
 }
 
-// Define a general boxed error.
-define_opaque_error!(BoxedError);
+impl BoxedError {
+    pub fn new<E: crate::ext::ErrorExt + Send + Sync + 'static>(err: E) -> Self {
+        Self {
+            inner: Box::new(err),
+        }
+    }
+}
+
+impl std::fmt::Debug for BoxedError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        // Use the pretty debug format of inner error for opaque error.
+        let debug_format = crate::format::DebugFormat::new(&*self.inner);
+        debug_format.fmt(f)
+    }
+}
+
+impl std::fmt::Display for BoxedError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.inner)
+    }
+}
+
+impl std::error::Error for BoxedError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        self.inner.source()
+    }
+}
+
+impl crate::ext::ErrorExt for BoxedError {
+    fn status_code(&self) -> crate::status_code::StatusCode {
+        self.inner.status_code()
+    }
+
+    fn backtrace_opt(&self) -> Option<&crate::snafu::Backtrace> {
+        self.inner.backtrace_opt()
+    }
+
+    fn as_any(&self) -> &dyn std::any::Any {
+        self.inner.as_any()
+    }
+}
+
+// Implement ErrorCompat for this opaque error so the backtrace is also available
+// via `ErrorCompat::backtrace()`.
+impl crate::snafu::ErrorCompat for BoxedError {
+    fn backtrace(&self) -> Option<&crate::snafu::Backtrace> {
+        self.inner.backtrace_opt()
+    }
+}
 
 #[cfg(test)]
 mod tests {

src/common/error/src/format.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/error/src/lib.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -24,6 +24,9 @@ pub mod prelude {
     pub use crate::ext::{BoxedError, ErrorExt};
     pub use crate::format::DebugFormat;
     pub use crate::status_code::StatusCode;
+
+    pub const INNER_ERROR_CODE: &str = "INNER_ERROR_CODE";
+    pub const INNER_ERROR_MSG: &str = "INNER_ERROR_MSG";
 }
 
 pub use snafu;

src/common/error/src/mock.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/error/src/status_code.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -14,8 +14,10 @@
 
 use std::fmt;
 
+use strum::EnumString;
+
 /// Common status code for public API.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, EnumString)]
 pub enum StatusCode {
     // ====== Begin of common status code ==============
     /// Success.

src/common/function-macro/Cargo.toml

@@ -15,5 +15,5 @@ syn = "1.0"
 arc-swap = "1.0"
 common-query = { path = "../query" }
 datatypes = { path = "../../datatypes" }
-snafu = { version = "0.7", features = ["backtraces"] }
+snafu.workspace = true
 static_assertions = "1.1.0"

src/common/function-macro/src/lib.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function-macro/tests/test_derive.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/Cargo.toml

@@ -18,7 +18,7 @@ num = "0.4"
 num-traits = "0.2"
 once_cell = "1.10"
 paste = "1.0"
-snafu = { version = "0.7", features = ["backtraces"] }
+snafu.workspace = true
 statrs = "0.15"
 
 [dev-dependencies]

src/common/function/src/lib.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/src/scalars.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/src/scalars/aggregate.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/src/scalars/aggregate/argmax.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/src/scalars/aggregate/argmin.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/src/scalars/aggregate/diff.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/src/scalars/aggregate/mean.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/src/scalars/aggregate/percentile.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/src/scalars/aggregate/polyval.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/src/scalars/aggregate/scipy_stats_norm_cdf.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/src/scalars/aggregate/scipy_stats_norm_pdf.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/src/scalars/expression.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/src/scalars/expression/binary.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/src/scalars/expression/ctx.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,

src/common/function/src/scalars/expression/unary.rs

@@ -1,10 +1,10 @@
-// Copyright 2022 Greptime Team
+// Copyright 2023 Greptime Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-// http://www.apache.org/licenses/LICENSE-2.0
+//     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,