feat: Support remote engine server with flatbuffer write

.github/workflows/ci.yml

@@ -66,6 +66,7 @@ jobs:
         run: |
           sudo apt update
           sudo apt install --yes protobuf-compiler
+          sudo sh scripts/compile-flatc.sh
       - name: Install check binaries
         run: |
           rustup component add clippy
@@ -95,6 +96,7 @@ jobs:
         run: |
           sudo apt update
           sudo apt install --yes protobuf-compiler
+          sudo sh scripts/compile-flatc.sh
       - name: Backup Lock File
         run: |
           cp ${LOCK_FILE} ${LOCK_FILE}.bak
@@ -127,6 +129,7 @@ jobs:
         run: |
           sudo apt update
           sudo apt install --yes protobuf-compiler
+          sudo sh scripts/compile-flatc.sh
       - name: Run integration tests
         run: |
           make integration-test
@@ -164,6 +167,7 @@ jobs:
         run: |
           sudo apt update
           sudo apt install --yes protobuf-compiler
+          sudo sh scripts/compile-flatc.sh
       - name: Build and Run HoraeDB Cluster
         working-directory: integration_tests
         run: |
@@ -224,6 +228,7 @@ jobs:
         run: |
           sudo apt update
           sudo apt install --yes protobuf-compiler
+          sudo sh scripts/compile-flatc.sh
       - name: Run recovery tests
         working-directory: integration_tests
         run: |
@@ -257,6 +262,7 @@ jobs:
         run: |
           sudo apt update
           sudo apt install --yes protobuf-compiler
+          sudo sh scripts/compile-flatc.sh
       - name: Run dist query tests
         working-directory: integration_tests
         run: |

.github/workflows/tsbs.yml

@@ -47,6 +47,7 @@ jobs:
         run: |
           sudo apt update
           sudo apt install --yes protobuf-compiler
+          sudo sh scripts/compile-flatc.sh
       - name: Build server
         run: |
           make build

Cargo.toml

@@ -39,6 +39,7 @@ members = [
     "src/components/arrow_ext",
     "src/components/bytes_ext",
     "src/components/codec",
+    "src/components/fb_util",
     "src/components/future_ext",
     "src/components/hash_ext",
     "src/components/id_allocator",
@@ -182,6 +183,7 @@ table_engine = { path = "src/table_engine" }
 table_kv = { path = "src/components/table_kv" }
 tempfile = "3.1.0"
 test_util = { path = "src/components/test_util" }
+fb_util = { path = "src/components/fb_util" }
 time_ext = { path = "src/components/time_ext" }
 toml = "0.7"
 toml_ext = { path = "src/components/toml_ext" }
@@ -196,6 +198,7 @@ uuid = "1.6.1"
 wal = { path = "src/wal" }
 xorfilter-rs = { git = "https://github.com/CeresDB/xorfilter", rev = "ac8ef01" }
 zstd = { version = "0.12", default-features = false }
+flatbuffers = { version = "24.3.25" }
 
 # This profile optimizes for good runtime performance.
 [profile.release]

Dockerfile

@@ -22,11 +22,13 @@ FROM rust:${RUST_VERSION}-slim-bullseye as build
 # cache mounts below may already exist and owned by root
 USER root
 
-RUN apt update && apt install --yes git gcc g++ libssl-dev pkg-config cmake protobuf-compiler && rm -rf /var/lib/apt/lists/*
+RUN apt update && apt install --yes git gcc g++ libssl-dev pkg-config cmake protobuf-compiler wget && rm -rf /var/lib/apt/lists/*
 
 COPY . /horaedb
 WORKDIR /horaedb
 
+RUN sh scripts/compile-flatc.sh
+
 RUN make build
 
 ## HoraeDB

scripts/compile-flatc.sh

@@ -0,0 +1,30 @@
+#!/usr/bin/env bash
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+
+set -e
+wget 'https://github.com/google/flatbuffers/archive/refs/tags/v24.3.25.tar.gz'
+
+tar xzf v24.3.25.tar.gz
+cd flatbuffers-24.3.25
+cmake -G "Unix Makefiles"
+make
+make install
+ldconfig
+flatc --version
+cd -
+rm -rf flatbuffers-24.3.25

src/analytic_engine/src/memtable/skiplist/mod.rs

@@ -27,7 +27,10 @@ use arena::{Arena, BasicStats};
 use bytes_ext::Bytes;
 use codec::Encoder;
 use common_types::{
-    row::{contiguous::ContiguousRowWriter, Row},
+    row::{
+        contiguous::{ContiguousRowWriter, InnerType},
+        Row,
+    },
     schema::Schema,
     time::TimeRange,
     SequenceNumber,
@@ -158,7 +161,8 @@ impl<A: Arena<Stats = BasicStats> + Clone + Sync + Send + 'static> MemTable
 
         // Encode row value. The ContiguousRowWriter will clear the buf.
         let row_value = &mut ctx.value_buf;
-        let mut row_writer = ContiguousRowWriter::new(row_value, schema, &ctx.index_in_writer);
+        let mut row_writer =
+            ContiguousRowWriter::new(InnerType::Buffer(row_value), schema, &ctx.index_in_writer);
         row_writer.write_row(row).context("invalid row")?;
         let encoded_size = internal_key.len() + row_value.len();
         self.skiplist.put(internal_key, row_value);

src/analytic_engine/src/row_iter/tests.rs

@@ -20,7 +20,7 @@ use common_types::{
     projected_schema::{ProjectedSchema, RowProjector},
     record_batch::{FetchedRecordBatch, FetchedRecordBatchBuilder},
     row::{
-        contiguous::{ContiguousRowReader, ContiguousRowWriter, ProjectedContiguousRow},
+        contiguous::{ContiguousRowReader, ContiguousRowWriter, InnerType, ProjectedContiguousRow},
         Row,
     },
     schema::{IndexInWriterSchema, RecordSchemaWithKey, Schema},
@@ -95,7 +95,8 @@ pub fn build_fetched_record_batch_with_key(schema: Schema, rows: Vec<Row>) -> Fe
 
     let mut buf = Vec::new();
     for row in rows {
-        let mut writer = ContiguousRowWriter::new(&mut buf, &schema, &index_in_writer);
+        let mut writer =
+            ContiguousRowWriter::new(InnerType::Buffer(&mut buf), &schema, &index_in_writer);
 
         writer.write_row(&row).unwrap();
 

src/benchmarks/Cargo.toml

@@ -39,14 +39,17 @@ bytes_ext = { workspace = true }
 clap = { workspace = true }
 common_types = { workspace = true }
 env_logger = { workspace = true }
+fb_util ={ workspace = true }
 futures = { workspace = true }
 generic_error = { workspace = true }
+horaedbproto = { workspace = true }
 logger = { workspace = true }
 macros = { workspace = true }
 object_store = { workspace = true }
 parquet = { workspace = true }
 parquet_ext = { workspace = true }
 pprof = { workspace = true, features = ["flamegraph", "criterion"] }
+prost = { workspace = true }
 rand = { workspace = true }
 runtime = { workspace = true }
 serde = { workspace = true }

src/benchmarks/bench.toml

@@ -24,6 +24,10 @@ max_projections = 5
 read_batch_row_num = 500
 sst_meta_cache_cap = 1000
 sst_data_cache_cap = 10000
+is_async = false
+bench_sample_size = 10
+num_rows_per_row_group = 1
+reverse = false
 
 [sst_bench.predicate]
 # start_time_ms = 0
@@ -41,6 +45,17 @@ bench_measurement_time = "120s"
 max_projections = 5
 read_batch_row_num = 500
 sst_level = 0
+bench_sample_size = 10
+num_rows_per_row_group = 1
+
+[merge_memtable_bench]
+store_path = "/path/to/data"
+space_id = 1
+table_id = 1
+sst_file_ids = [ 37 ]
+runtime_thread_num = 1
+max_projections = 5
+arena_block_size = "64M"
 
 [merge_sst_bench.predicate]
 start_time_ms = 0
@@ -67,4 +82,9 @@ value_size = 1024
 
 [replay_bench]
 bench_measurement_time = "3s"
+bench_sample_size = 10
+batch_size = 10
+
+[request_serialization_bench]
+bench_measurement_time = "3s"
 bench_sample_size = 10