refactor: use inserter instead of grpc_handler in pipeline operator

.github/actions/setup-greptimedb-cluster/action.yml

@@ -57,6 +57,7 @@ runs:
         greptime/greptimedb-cluster \
         --create-namespace \
         -n my-greptimedb \
+        --values ./.github/actions/setup-greptimedb-cluster/values.yaml \
         --wait \
         --wait-for-jobs
   - name: Wait for GreptimeDB

.github/actions/setup-greptimedb-cluster/values.yaml

@@ -0,0 +1,18 @@
+meta:
+  config: |-
+    [runtime]
+    read_rt_size = 8
+    write_rt_size = 8
+    bg_rt_size = 8
+datanode:
+  config: |-
+    [runtime]
+    read_rt_size = 8
+    write_rt_size = 8
+    bg_rt_size = 8
+frontend:
+  config: |-
+    [runtime]
+    read_rt_size = 8
+    write_rt_size = 8
+    bg_rt_size = 8

.github/workflows/develop.yml

@@ -518,6 +518,9 @@ jobs:
       - name: Setup kafka server
         working-directory: tests-integration/fixtures/kafka
         run: docker compose -f docker-compose-standalone.yml up -d --wait
+      - name: Setup minio
+        working-directory: tests-integration/fixtures/minio
+        run: docker compose -f docker-compose-standalone.yml up -d --wait
       - name: Run nextest cases
         run: cargo llvm-cov nextest --workspace --lcov --output-path lcov.info -F pyo3_backend -F dashboard
         env:
@@ -528,6 +531,11 @@ jobs:
           GT_S3_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }}
           GT_S3_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }}
           GT_S3_REGION: ${{ vars.AWS_CI_TEST_BUCKET_REGION }}
+          GT_MINIO_BUCKET: greptime
+          GT_MINIO_ACCESS_KEY_ID: superpower_ci_user
+          GT_MINIO_ACCESS_KEY: superpower_password
+          GT_MINIO_REGION: us-west-2
+          GT_MINIO_ENDPOINT_URL: http://127.0.0.1:9000
           GT_ETCD_ENDPOINTS: http://127.0.0.1:2379
           GT_KAFKA_ENDPOINTS: 127.0.0.1:9092
           UNITTEST_LOG_DIR: "__unittest_logs"

Cargo.toml

@@ -154,6 +154,8 @@ reqwest = { version = "0.12", default-features = false, features = [
     "multipart",
 ] }
 rskafka = "0.5"
+rstest = "0.21"
+rstest_reuse = "0.7"
 rust_decimal = "1.33"
 schemars = "0.8"
 serde = { version = "1.0", features = ["derive"] }
@@ -257,3 +259,7 @@ strip = true
 [profile.dev.package.sqlness-runner]
 debug = false
 strip = true
+
+[profile.dev.package.tests-fuzz]
+debug = false
+strip = true

config/config.md

@@ -13,6 +13,10 @@
 | `mode` | String | `standalone` | The running mode of the datanode. It can be `standalone` or `distributed`. |
 | `enable_telemetry` | Bool | `true` | Enable telemetry to collect anonymous usage data. |
 | `default_timezone` | String | `None` | The default timezone of the server. |
+| `runtime` | -- | -- | The runtime options. |
+| `runtime.read_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
+| `runtime.write_rt_size` | Integer | `8` | The number of threads to execute the runtime for global write operations. |
+| `runtime.bg_rt_size` | Integer | `8` | The number of threads to execute the runtime for global background operations. |
 | `http` | -- | -- | The HTTP server options. |
 | `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
 | `http.timeout` | String | `30s` | HTTP request timeout. |
@@ -154,6 +158,10 @@
 | --- | -----| ------- | ----------- |
 | `mode` | String | `standalone` | The running mode of the datanode. It can be `standalone` or `distributed`. |
 | `default_timezone` | String | `None` | The default timezone of the server. |
+| `runtime` | -- | -- | The runtime options. |
+| `runtime.read_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
+| `runtime.write_rt_size` | Integer | `8` | The number of threads to execute the runtime for global write operations. |
+| `runtime.bg_rt_size` | Integer | `8` | The number of threads to execute the runtime for global background operations. |
 | `heartbeat` | -- | -- | The heartbeat options. |
 | `heartbeat.interval` | String | `18s` | Interval for sending heartbeat messages to the metasrv. |
 | `heartbeat.retry_interval` | String | `3s` | Interval for retrying to send heartbeat messages to the metasrv. |
@@ -240,6 +248,10 @@
 | `use_memory_store` | Bool | `false` | Store data in memory. |
 | `enable_telemetry` | Bool | `true` | Whether to enable greptimedb telemetry. |
 | `store_key_prefix` | String | `""` | If it's not empty, the metasrv will store all data with this key prefix. |
+| `runtime` | -- | -- | The runtime options. |
+| `runtime.read_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
+| `runtime.write_rt_size` | Integer | `8` | The number of threads to execute the runtime for global write operations. |
+| `runtime.bg_rt_size` | Integer | `8` | The number of threads to execute the runtime for global background operations. |
 | `procedure` | -- | -- | Procedure storage options. |
 | `procedure.max_retry_times` | Integer | `12` | Procedure max retry time. |
 | `procedure.retry_delay` | String | `500ms` | Initial retry delay of procedures, increases exponentially |
@@ -300,6 +312,10 @@
 | `rpc_max_recv_message_size` | String | `512MB` | The maximum receive message size for gRPC server. |
 | `rpc_max_send_message_size` | String | `512MB` | The maximum send message size for gRPC server. |
 | `enable_telemetry` | Bool | `true` | Enable telemetry to collect anonymous usage data. |
+| `runtime` | -- | -- | The runtime options. |
+| `runtime.read_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
+| `runtime.write_rt_size` | Integer | `8` | The number of threads to execute the runtime for global write operations. |
+| `runtime.bg_rt_size` | Integer | `8` | The number of threads to execute the runtime for global background operations. |
 | `heartbeat` | -- | -- | The heartbeat options. |
 | `heartbeat.interval` | String | `3s` | Interval for sending heartbeat messages to the metasrv. |
 | `heartbeat.retry_interval` | String | `3s` | Interval for retrying to send heartbeat messages to the metasrv. |

config/datanode.example.toml

@@ -32,6 +32,15 @@ rpc_max_send_message_size = "512MB"
 ## Enable telemetry to collect anonymous usage data.
 enable_telemetry = true
 
+## The runtime options.
+[runtime]
+## The number of threads to execute the runtime for global read operations.
+read_rt_size = 8
+## The number of threads to execute the runtime for global write operations.
+write_rt_size = 8
+## The number of threads to execute the runtime for global background operations.
+bg_rt_size = 8
+
 ## The heartbeat options.
 [heartbeat]
 ## Interval for sending heartbeat messages to the metasrv.

config/frontend.example.toml

@@ -5,6 +5,15 @@ mode = "standalone"
 ## +toml2docs:none-default
 default_timezone = "UTC"
 
+## The runtime options.
+[runtime]
+## The number of threads to execute the runtime for global read operations.
+read_rt_size = 8
+## The number of threads to execute the runtime for global write operations.
+write_rt_size = 8
+## The number of threads to execute the runtime for global background operations.
+bg_rt_size = 8
+
 ## The heartbeat options.
 [heartbeat]
 ## Interval for sending heartbeat messages to the metasrv.