### Benchmarks/Tsbs/README # How to run TSBS Benchmark This document contains the steps to run TSBS Benchmark. Our results are listed in other files in the same directory. ## Prerequires You need the following tools to run TSBS Benchmark: - Go - git - make - rust (optional, if you want to build the DB from source) ## Build TSBS suite Clone our fork of TSBS: ```shell git clone https://github.com/GreptimeTeam/tsbs.git ``` Then build it: ```shell cd tsbs make ``` You can check the `bin/` directory for compiled binaries. We will only use some of them. ```shell ls ./bin/ ``` Binaries we will use later: - `tsbs_generate_data` - `tsbs_generate_queries` - `tsbs_load_greptime` - `tsbs_run_queries_influx` ## Generate test data and queries The data is generated by `tsbs_generate_data` ```shell mkdir bench-data ./bin/tsbs_generate_data --use-case="cpu-only" --seed=123 --scale=4000 \ --timestamp-start="2023-06-11T00:00:00Z" \ --timestamp-end="2023-06-14T00:00:00Z" \ --log-interval="10s" --format="influx" \ > ./bench-data/influx-data.lp ``` Here we generates 4000 time-series in 3 days with 10s interval. We'll use influx line protocol to write so the target format is `influx`. Queries are generated by `tsbs_generate_queries`. You can change the parameters but need to make sure it matches with `tsbs_generate_data`. ``` /* Detailed source-code truncated for AI context efficiency. */ ``` ## Start GreptimeDB Reference to our [document](https://docs.greptime.com/getting-started/installation/overview) for how to install and start a GreptimeDB. Or you can also check this [document](https://docs.greptime.com/contributor-guide/getting-started#compile-and-run) for how to build a GreptimeDB from source. ## Write Data After the DB is started, we can use `tsbs_load_greptime` to test the write performance. ```shell ./bin/tsbs_load_greptime \ --urls=http://localhost:4000 \ --file=./bench-data/influx-data.lp \ --batch-size=3000 \ --gzip=false \ --workers=6 ``` Parameters here are only provided as an example. You can choose whatever you like or adjust them to match your target scenario. Notice that if you want to rerun `tsbs_load_greptime`, please destroy and restart the DB and clear its previous data first. Existing duplicated data will impact the write and query performance. ## Query Data After the data is imported, you can then run queries. The following script runs all queries. You can also choose a subset of queries to run. ``` /* Detailed source-code truncated for AI context efficiency. */ ``` Rerun queries need not to re-import data. Just execute the corresponding command again is fine. --- ### Benchmarks/Tsbs/V0.3.2 # TSBS benchmark - v0.3.2 ## Environment | | | | --- | --- | | CPU | AMD Ryzen 7 7735HS (8 core 3.2GHz) | | Memory | 32GB | | Disk | SOLIDIGM SSDPFKNU010TZ | | OS | Ubuntu 22.04.2 LTS | ## Write performance | Write buffer size | Ingest rate（rows/s） | | --- | --- | | 512M | 139583.04 | | 32M | 279250.52 | ## Query performance | Query type | v0.3.2 write buffer 32M (ms) | v0.3.2 write buffer 512M (ms) | v0.3.1 write buffer 32M (ms) | | --- | --- | --- | --- | | cpu-max-all-1 | 921.12 | 241.23 | 553.63 | | cpu-max-all-8 | 2657.66 | 502.78 | 3308.41 | | double-groupby-1 | 28238.85 | 27367.42 | 52148.22 | | double-groupby-5 | 33094.65 | 32421.89 | 56762.37 | | double-groupby-all | 38565.89 | 38635.52 | 59596.80 | | groupby-orderby-limit | 23321.60 | 22423.55 | 53983.23 | | high-cpu-1 | 1167.04 | 254.15 | 832.41 | | high-cpu-all | 32814.08 | 29906.94 | 62853.12 | | lastpoint | 192045.05 | 153575.42 | NA | | single-groupby-1-1-1 | 63.97 | 87.35 | 92.66 | | single-groupby-1-1-12 | 666.24 | 326.98 | 781.50 | | single-groupby-1-8-1 | 225.29 | 137.97 |281.95 | | single-groupby-5-1-1 | 70.40 | 81.64 | 86.15 | | single-groupby-5-1-12 | 722.75 | 356.01 | 805.18 | | single-groupby-5-8-1 | 285.60 | 115.88 | 326.29 | --- ### Benchmarks/Tsbs/V0.4.0 # TSBS benchmark - v0.4.0 ## Environment ### Local | | | | ------ | ---------------------------------- | | CPU | AMD Ryzen 7 7735HS (8 core 3.2GHz) | | Memory | 32GB | | Disk | SOLIDIGM SSDPFKNU010TZ | | OS | Ubuntu 22.04.2 LTS | ### Aliyun amd64 | | | | ------- | -------------- | | Machine | ecs.g7.4xlarge | | CPU | 16 core | | Memory | 64GB | | Disk | 100G | | OS | Ubuntu 22.04 | ### Aliyun arm64 | | | | ------- | ----------------- | | Machine | ecs.g8y.4xlarge | | CPU | 16 core | | Memory | 64GB | | Disk | 100G | | OS | Ubuntu 22.04 ARM | ## Write performance | Environment | Ingest rate（rows/s） | | ------------------ | --------------------- | | Local | 365280.60 | | Aliyun g7.4xlarge | 341368.72 | | Aliyun g8y.4xlarge | 320907.29 | ## Query performance | Query type | Local (ms) | Aliyun g7.4xlarge (ms) | Aliyun g8y.4xlarge (ms) | | --------------------- | ---------- | ---------------------- | ----------------------- | | cpu-max-all-1 | 50.70 | 31.46 | 47.61 | | cpu-max-all-8 | 262.16 | 129.26 | 152.43 | | double-groupby-1 | 2512.71 | 1408.19 | 1586.10 | | double-groupby-5 | 3896.15 | 2304.29 | 2585.29 | | double-groupby-all | 5404.67 | 3337.61 | 3773.91 | | groupby-orderby-limit | 3786.98 | 2065.72 | 2312.57 | | high-cpu-1 | 71.96 | 37.29 | 54.01 | | high-cpu-all | 9468.75 | 7595.69 | 8467.46 | | lastpoint | 13379.43 | 11253.76 | 12949.40 | | single-groupby-1-1-1 | 20.72 | 12.16 | 13.35 | | single-groupby-1-1-12 | 28.53 | 15.67 | 21.62 | | single-groupby-1-8-1 | 72.23 | 37.90 | 43.52 | | single-groupby-5-1-1 | 26.75 | 15.59 | 17.48 | | single-groupby-5-1-12 | 45.41 | 22.90 | 31.96 | | single-groupby-5-8-1 | 107.96 | 59.76 | 69.58 | --- ### Benchmarks/Tsbs/V0.7.0 # TSBS benchmark - v0.7.0 ## Environment ### Local | | | | ------ | ---------------------------------- | | CPU | AMD Ryzen 7 7735HS (8 core 3.2GHz) | | Memory | 32GB | | Disk | SOLIDIGM SSDPFKNU010TZ | | OS | Ubuntu 22.04.2 LTS | ### Amazon EC2 | | | | ------- | -------------- | | Machine | c5d.2xlarge | | CPU | 8 core | | Memory | 16GB | | Disk | 50GB (GP3) | | OS | Ubuntu 22.04.1 | ## Write performance | Environment | Ingest rate (rows/s) | | --------------- | -------------------- | | Local | 369581.464 | | EC2 c5d.2xlarge | 298716.664 | ## Query performance | Query type | Local (ms) | EC2 c5d.2xlarge (ms) | | --------------------- | ---------- | -------------------- | | cpu-max-all-1 | 30.56 | 54.74 | | cpu-max-all-8 | 52.69 | 70.50 | | double-groupby-1 | 664.30 | 1366.63 | | double-groupby-5 | 1391.26 | 2141.71 | | double-groupby-all | 2828.94 | 3389.59 | | groupby-orderby-limit | 718.92 | 1213.90 | | high-cpu-1 | 29.21 | 52.98 | | high-cpu-all | 5514.12 | 7194.91 | | lastpoint | 7571.40 | 9423.41 | | single-groupby-1-1-1 | 19.09 | 7.77 | | single-groupby-1-1-12 | 27.28 | 51.64 | | single-groupby-1-8-1 | 31.85 | 11.64 | | single-groupby-5-1-1 | 16.14 | 9.67 | | single-groupby-5-1-12 | 27.21 | 53.62 | | single-groupby-5-8-1 | 39.62 | 14.96 | --- ### Benchmarks/Tsbs/V0.8.0 # TSBS benchmark - v0.8.0 ## Environment ### Local | | | | ------ | ---------------------------------- | | CPU | AMD Ryzen 7 7735HS (8 core 3.2GHz) | | Memory | 32GB | | Disk | SOLIDIGM SSDPFKNU010TZ | | OS | Ubuntu 22.04.2 LTS | ### Amazon EC2 | | | | ------- | -------------- | | Machine | c5d.2xlarge | | CPU | 8 core | | Memory | 16GB | | Disk | 50GB (GP3) | | OS | Ubuntu 22.04.1 | ## Write performance | Environment | Ingest rate (rows/s) | | --------------- | -------------------- | | Local | 315369.66 | | EC2 c5d.2xlarge | 222148.56 | ## Query performance | Query type | Local (ms) | EC2 c5d.2xlarge (ms) | | --------------------- | ---------- | -------------------- | | cpu-max-all-1 | 24.63 | 15.29 | | cpu-max-all-8 | 51.69 | 33.53 | | double-groupby-1 | 673.51 | 1295.38 | | double-groupby-5 | 1244.93 | 1993.91 | | double-groupby-all | 2215.44 | 3056.77 | | groupby-orderby-limit | 754.50 | 1546.49 | | high-cpu-1 | 19.62 | 11.58 | | high-cpu-all | 5402.31 | 8011.43 | | lastpoint | 6756.12 | 9312.67 | | single-groupby-1-1-1 | 15.70 | 7.67 | | single-groupby-1-1-12 | 16.72 | 9.29 | | single-groupby-1-8-1 | 26.72 | 17.97 | | single-groupby-5-1-1 | 18.17 | 10.09 | | single-groupby-5-1-12 | 20.04 | 12.37 | | single-groupby-5-8-1 | 35.63 | 23.13 | `single-groupby-1-1-1` query throughput | Environment | Client concurrency | mean time (ms) | qps (queries/sec) | | --------------- | ------------------ | -------------- | ----------------- | | Local | 50 | 42.87 | 1165.73 | | Local | 100 | 89.29 | 1119.38 | | EC2 c5d.2xlarge | 50 | 69.25 | 721.73 | | EC2 c5d.2xlarge | 100 | 140.93 | 709.35 | --- ### Benchmarks/Tsbs/V0.9.1 # TSBS benchmark - v0.9.1 ## Environment ### Local | | | | ------ | ---------------------------------- | | CPU | AMD Ryzen 7 7735HS (8 core 3.2GHz) | | Memory | 32GB | | Disk | SOLIDIGM SSDPFKNU010TZ | | OS | Ubuntu 22.04.2 LTS | ### Amazon EC2 | | | | ------- | ----------------------- | | Machine | c5d.2xlarge | | CPU | 8 core | | Memory | 16GB | | Disk | 100GB (GP3) | | OS | Ubuntu Server 24.04 LTS | ## Write performance | Environment | Ingest rate (rows/s) | | --------------- | -------------------- | | Local | 387697.68 | | EC2 c5d.2xlarge | 234620.19 | ## Query performance | Query type | Local (ms) | EC2 c5d.2xlarge (ms) | | --------------------- | ---------- | -------------------- | | cpu-max-all-1 | 21.14 | 14.75 | | cpu-max-all-8 | 36.79 | 30.69 | | double-groupby-1 | 529.02 | 987.85 | | double-groupby-5 | 1064.53 | 1455.95 | | double-groupby-all | 1625.33 | 2143.96 | | groupby-orderby-limit | 529.19 | 1353.49 | | high-cpu-1 | 12.09 | 8.24 | | high-cpu-all | 3619.47 | 5312.82 | | lastpoint | 224.91 | 576.06 | | single-groupby-1-1-1 | 10.82 | 6.01 | | single-groupby-1-1-12 | 11.16 | 7.42 | | single-groupby-1-8-1 | 13.50 | 10.20 | | single-groupby-5-1-1 | 11.99 | 6.70 | | single-groupby-5-1-12 | 13.17 | 8.72 | | single-groupby-5-8-1 | 16.01 | 12.07 | `single-groupby-1-1-1` query throughput | Environment | Client concurrency | mean time (ms) | qps (queries/sec) | | --------------- | ------------------ | -------------- | ----------------- | | Local | 50 | 33.04 | 1511.74 | | Local | 100 | 67.70 | 1476.14 | | EC2 c5d.2xlarge | 50 | 61.93 | 806.97 | | EC2 c5d.2xlarge | 100 | 126.31 | 791.40 | --- ### Benchmarks/Tsbs/V0.12.0 # TSBS benchmark - v0.12.0 ## Environment ### Amazon EC2 | | | |---------|-------------------------| | Machine | c5d.2xlarge | | CPU | 8 core | | Memory | 16GB | | Disk | 100GB (GP3) | | OS | Ubuntu Server 24.04 LTS | ## Write performance | Environment | Ingest rate (rows/s) | |-----------------|----------------------| | EC2 c5d.2xlarge | 326839.28 | ## Query performance | Query type | EC2 c5d.2xlarge (ms) | |-----------------------|----------------------| | cpu-max-all-1 | 12.46 | | cpu-max-all-8 | 24.20 | | double-groupby-1 | 673.08 | | double-groupby-5 | 963.99 | | double-groupby-all | 1330.05 | | groupby-orderby-limit | 952.46 | | high-cpu-1 | 5.08 | | high-cpu-all | 4638.57 | | lastpoint | 591.02 | | single-groupby-1-1-1 | 4.06 | | single-groupby-1-1-12 | 4.73 | | single-groupby-1-8-1 | 8.23 | | single-groupby-5-1-1 | 4.61 | | single-groupby-5-1-12 | 5.61 | | single-groupby-5-8-1 | 9.74 | --- ### Benchmarks/Log/Query # Query URL and payload for Elastic Search ## Count URL: `http://127.0.0.1:9200/_count` ## Query by timerange URL: `http://127.0.0.1:9200/_search` You can use the following payload to get the full timerange first. ```JSON {"size":0,"aggs":{"max_timestamp":{"max":{"field":"timestamp"}},"min_timestamp":{"min":{"field":"timestamp"}}}} ``` And then use this payload to query by timerange. ```JSON { "from": 0, "size": 1000, "query": { "range": { "timestamp": { "gte": "2024-08-16T04:30:44.000Z", "lte": "2024-08-16T04:51:52.000Z" } } } } ``` ## Query by condition URL: `http://127.0.0.1:9200/_search` ### Structured payload ```JSON { "from": 0, "size": 10000, "query": { "bool": { "must": [ { "term": { "user.keyword": "CrucifiX" } }, { "term": { "method.keyword": "OPTION" } }, { "term": { "path.keyword": "/user/booperbot124" } }, { "term": { "http_version.keyword": "HTTP/1.1" } }, { "term": { "status": "401" } } ] } } } ``` ### Unstructured payload ```JSON { "from": 0, "size": 10000, "query": { "bool": { "must": [ { "match_phrase": { "message": "CrucifiX" } }, { "match_phrase": { "message": "OPTION" } }, { "match_phrase": { "message": "/user/booperbot124" } }, { "match_phrase": { "message": "HTTP/1.1" } }, { "match_phrase": { "message": "401" } } ] } } } ``` ## Query by condition and timerange URL: `http://127.0.0.1:9200/_search` ### Structured payload ```JSON { "size": 10000, "query": { "bool": { "must": [ { "term": { "user.keyword": "CrucifiX" } }, { "term": { "method.keyword": "OPTION" } }, { "term": { "path.keyword": "/user/booperbot124" } }, { "term": { "http_version.keyword": "HTTP/1.1" } }, { "term": { "status": "401" } }, { "range": { "timestamp": { "gte": "2024-08-19T07:03:37.383Z", "lte": "2024-08-19T07:24:58.883Z" } } } ] } } } ``` ### Unstructured payload ```JSON { "size": 10000, "query": { "bool": { "must": [ { "match_phrase": { "message": "CrucifiX" } }, { "match_phrase": { "message": "OPTION" } }, { "match_phrase": { "message": "/user/booperbot124" } }, { "match_phrase": { "message": "HTTP/1.1" } }, { "match_phrase": { "message": "401" } }, { "range": { "timestamp": { "gte": "2024-08-19T05:16:17.099Z", "lte": "2024-08-19T05:46:02.722Z" } } } ] } } } ``` --- ### Benchmarks/Log/README # Log benchmark configuration This repo holds the configuration we used to benchmark GreptimeDB, Clickhouse and Elastic Search. Here are the versions of databases we used in the benchmark | name | version | | :------------ | :--------- | | GreptimeDB | v0.9.2 | | Clickhouse | 24.9.1.219 | | Elasticsearch | 8.15.0 | ## Structured model vs Unstructured model We divide test into two parts, using structured model and unstructured model accordingly. You can also see the difference in create table clause. __Structured model__ The log data is pre-processed into columns by vector. For example an insert request looks like following ```SQL INSERT INTO test_table (bytes, http_version, ip, method, path, status, user, timestamp) VALUES () ``` The goal is to test string/text support for each database. In real scenarios it means the datasource(or log data producers) have separate fields defined, or have already processed the raw input. __Unstructured model__ The log data is inserted as a long string, and then we build fulltext index upon these strings. For example an insert request looks like following ```SQL INSERT INTO test_table (message, timestamp) VALUES () ``` The goal is to test fuzzy search performance for each database. In real scenarios it means the log is produced by some kind of middleware and inserted directly into the database. ## Creating tables See [here](./create_table.sql) for GreptimeDB and Clickhouse's create table clause. The mapping of Elastic search is created automatically. ## Vector Configuration We use vector to generate random log data and send inserts to databases. Please refer to [structured config](./structured_vector.toml) and [unstructured config](./unstructured_vector.toml) for detailed configuration. ## SQLs and payloads Please refer to [SQL query](./query.sql) for GreptimeDB and Clickhouse, and [query payload](./query.md) for Elastic search. ## Steps to reproduce 0. Decide whether to run structured model test or unstructured mode test. 1. Build vector binary(see vector's config file for specific branch) and databases binaries accordingly. 2. Create table in GreptimeDB and Clickhouse in advance. 3. Run vector to insert data. 4. When data insertion is finished, run queries against each database. Note: you'll need to update timerange value after data insertion. ## Addition - You can tune GreptimeDB's configuration to get better performance. - You can setup GreptimeDB to use S3 as storage, see [here](https://docs.greptime.com/user-guide/deployments-administration/configuration#storage-options). --- ### How To/Memory Profile Scripts/Scripts/README # Memory Analysis Process This section will guide you through the process of analyzing memory usage for greptimedb. 1. Get the `jeprof` tool script, see the next section("Getting the `jeprof` tool") for details. 2. After starting `greptimedb`(with env var `MALLOC_CONF=prof:true`), execute the `dump.sh` script with the PID of the `greptimedb` process as an argument. This continuously monitors memory usage and captures profiles when exceeding thresholds (e.g. +20MB within 10 minutes). Outputs `greptime-{timestamp}.gprof` files. 3. With 2-3 gprof files, run `gen_flamegraph.sh` in the same environment to generate flame graphs showing memory allocation call stacks. 4. **NOTE:** The `gen_flamegraph.sh` script requires `jeprof` and optionally `flamegraph.pl` to be in the current directory. If needed to gen flamegraph now, run the `get_flamegraph_tool.sh` script, which downloads the flame graph generation tool `flamegraph.pl` to the current directory. The usage of `gen_flamegraph.sh` is: `Usage: ./gen_flamegraph.sh ` where `` is the path to the greptimedb binary, `` is the directory containing the gprof files(the directory `dump.sh` is dumping profiles to). Example call: `./gen_flamegraph.sh ./greptime .` Generating the flame graph might take a few minutes. The generated flame graphs are located in the `/flamegraphs` directory. Or if no `flamegraph.pl` is found, it will only contain `.collapse` files which is also fine. 5. You can send the generated flame graphs(the entire folder of `/flamegraphs`) to developers for further analysis. ## Getting the `jeprof` tool there are three ways to get `jeprof`, list in here from simple to complex, using any one of those methods is ok, as long as it's the same environment as the `greptimedb` will be running on: 1. If you are compiling greptimedb from source, then `jeprof` is already produced during compilation. After running `cargo build`, execute `find_compiled_jeprof.sh`. This will copy `jeprof` to the current directory. 2. Or, if you have the Rust toolchain installed locally, simply follow these commands: ```bash cargo new get_jeprof cd get_jeprof ``` Then add this line to `Cargo.toml`: ```toml [dependencies] tikv-jemalloc-ctl = { version = "0.6", features = ["use_std", "stats"] } ``` then run: ```bash cargo build ``` after that the `jeprof` tool is produced. Now run `find_compiled_jeprof.sh` in current directory, it will copy the `jeprof` tool to the current directory. 3. compile jemalloc from source you can first clone this repo, and checkout to this commit: ```bash git clone https://github.com/tikv/jemalloc.git cd jemalloc git checkout e13ca993e8ccb9ba9847cc330696e02839f328f7 ``` then run: ```bash ./configure make ``` and `jeprof` is in `.bin/` directory. Copy it to the current directory. --- ### How To/How To Change Log Level On The Fly # Change Log Level on the Fly ## HTTP API example: ```bash curl --data "trace,flow=debug" 127.0.0.1:4000/debug/log_level ``` And database will reply with something like: ```bash Log Level changed from Some("info") to "trace,flow=debug"% ``` The data is a string in the format of `global_level,module1=level1,module2=level2,...` that follows the same rule of `RUST_LOG`. The module is the module name of the log, and the level is the log level. The log level can be one of the following: `trace`, `debug`, `info`, `warn`, `error`, `off`(case insensitive). # Enable/Disable Trace on the Fly ## HTTP API example: ```bash curl --data "true" 127.0.0.1:4000/debug/enable_trace ``` And database will reply with something like: ``` trace enabled% ``` Possible values are "true" or "false". --- ### How To/How To Implement Sql Statement This document introduces how to implement SQL statements in GreptimeDB. The execution entry point for SQL statements locates at Frontend Instance. You can see it has implemented `SqlQueryHandler`: ```rust impl SqlQueryHandler for Instance { async fn do_query( &self, query: &str, query_ctx: QueryContextRef, ) -> Vec> { // ... } } ``` Normally, when a SQL query arrives at GreptimeDB, the `do_query` method will be called. After some parsing work, the SQL will be fed into `StatementExecutor`: ```rust // in Frontend Instance: self.statement_executor.execute_sql(stmt, query_ctx).await ``` That's where we handle our SQL statements. You can just create a new match arm for your statement there, then the statement is implemented for both GreptimeDB Standalone and Cluster. You can see how `DESCRIBE TABLE` is implemented as an example. Now, what if the statements should be handled differently for GreptimeDB Standalone and Cluster? You can see there's a `SqlStatementExecutor` field in `StatementExecutor`. Each GreptimeDB Standalone and Cluster has its own implementation of `SqlStatementExecutor`. If you are going to implement the statements differently in the two modes ( like `CREATE TABLE`), you have to implement them in their own `SqlStatementExecutor`s. Summarize as the diagram below: ``` /* Detailed source-code truncated for AI context efficiency. */ ``` Note that some SQL statements can be executed in our QueryEngine, in the form of `LogicalPlan`. You can follow the invocation path down to the `QueryEngine` implementation from `StatementExecutor::plan_exec`. For now, there's only one `DatafusionQueryEngine` for both GreptimeDB Standalone and Cluster. That lone query engine works for both modes is because GreptimeDB read/write data through `Table` trait, and each mode has its own `Table` implementation. We don't have any bias towards whether statements should be handled in query engine or `StatementExecutor`. You can implement one kind of statement in both places. For example, `Insert` with selection is handled in query engine, because we can easily do the query part there. However, `Insert` without selection is not, for the cost of parsing statement to `LogicalPlan` is not neglectable. So generally if the SQL query is simple enough, you can handle it in `StatementExecutor`; otherwise if it is complex or has some part of selection, it should be parsed to `LogicalPlan` and handled in query engine. --- ### How To/How To Profile Cpu # Profiling CPU ## HTTP API Sample at 99 Hertz, for 5 seconds, output report in [protobuf format](https://github.com/google/pprof/blob/master/proto/profile.proto). ```bash curl -X POST -s '0:4000/debug/prof/cpu' > /tmp/pprof.out ``` Then you can use `pprof` command with the protobuf file. ```bash go tool pprof -top /tmp/pprof.out ``` Sample at 99 Hertz, for 60 seconds, output report in flamegraph format. ```bash curl -X POST -s '0:4000/debug/prof/cpu?seconds=60&output=flamegraph' > /tmp/pprof.svg ``` Sample at 49 Hertz, for 10 seconds, output report in text format. ```bash curl -X POST -s '0:4000/debug/prof/cpu?seconds=10&frequency=49&output=text' > /tmp/pprof.txt ``` --- ### How To/How To Profile Memory # Profile memory usage of GreptimeDB This crate provides an easy approach to dump memory profiling info. A set of ready to use scripts is provided in [docs/how-to/memory-profile-scripts](./memory-profile-scripts/scripts). ## Prerequisites ### jemalloc jeprof is already compiled in the target directory of GreptimeDB. You can find the binary and use it. ``` # find jeprof binary find . -name 'jeprof' # add executable permission chmod +x ``` The path is usually under `./target/${PROFILE}/build/tikv-jemalloc-sys-${HASH}/out/build/bin/jeprof`. The default version of jemalloc installed from the package manager may not have the `--collapsed` option. You may need to check the whether the `jeprof` version is >= `5.3.0` if you want to install it from the package manager. ```bash # for macOS brew install jemalloc # for Ubuntu sudo apt install libjemalloc-dev ``` ### [flamegraph](https://github.com/brendangregg/FlameGraph) ```bash curl https://raw.githubusercontent.com/brendangregg/FlameGraph/master/flamegraph.pl > ./flamegraph.pl ``` ## Profiling ### Enable memory profiling for greptimedb binary Start GreptimeDB instance with environment variables: ```bash # for Linux MALLOC_CONF=prof:true ./target/debug/greptime standalone start # for macOS _RJEM_MALLOC_CONF=prof:true ./target/debug/greptime standalone start ``` ### Memory profiling for greptimedb docker image We have memory profiling enabled and activated by default in our official docker image. This behavior is controlled by configuration `enable_heap_profiling`: ```toml [memory] # Whether to enable heap profiling activation during startup. # Default is true. enable_heap_profiling = true ``` To disable memory profiling, set `enable_heap_profiling` to `false`. ### Memory profiling control You can control heap profiling activation using the new HTTP APIs: ```bash # Check current profiling status curl -X GET localhost:4000/debug/prof/mem/status # Activate heap profiling (if not already active) curl -X POST localhost:4000/debug/prof/mem/activate # Deactivate heap profiling curl -X POST localhost:4000/debug/prof/mem/deactivate # Activate gdump feature that dumps memory profiling data every time virtual memory usage exceeds previous maximum value. curl -X POST localhost:4000/debug/prof/mem/gdump -d 'activate=true' # Deactivate gdump. curl -X POST localhost:4000/debug/prof/mem/gdump -d 'activate=false' # Retrieve current gdump status. curl -X GET localhost:4000/debug/prof/mem/gdump ``` ### Dump memory profiling data Dump memory profiling data through HTTP API: ```bash curl -X POST localhost:4000/debug/prof/mem > greptime.hprof # or output flamegraph directly curl -X POST "localhost:4000/debug/prof/mem?output=flamegraph" > greptime.svg # or output pprof format curl -X POST "localhost:4000/debug/prof/mem?output=proto" > greptime.pprof ``` You can periodically dump profiling data and compare them to find the delta memory usage. ### Symbolicate external dump files If you have jemalloc heap dump files generated externally (e.g., via `MALLOC_CONF="prof:true,prof_prefix:jeprof.out,lg_prof_interval:26"` or `prof_gdump`), you can upload them to a running GreptimeDB instance for symbolication and flamegraph generation: ```bash # Upload a jemalloc heap dump file and get a symbolicated flamegraph curl -X POST --data-binary @/path/to/jeprof.out.12345.0.i0.heap \ localhost:4000/debug/prof/mem/symbol > flamegraph.svg ``` This is useful when: - You collected heap dumps from a production environment - You want to symbolicate dumps on a machine with debug symbols - You need to analyze dumps generated by jemalloc's automatic dump mechanisms ## Analyze profiling data with flamegraph To create flamegraph according to dumped profiling data: ```bash sudo apt install -y libjemalloc-dev jeprof --collapse | ./flamegraph.pl > mem-prof.svg jeprof --base --collapse | ./flamegraph.pl > output.svg ``` --- ### How To/How To Write Fuzz Tests # How to write fuzz tests This document introduces how to write fuzz tests in GreptimeDB. ## What is a fuzz test Fuzz test is tool that leverages deterministic random generation to assist in finding bugs. The goal of fuzz tests is to identify inputs generated by the fuzzer that cause system panics, crashes, or unexpected behaviors to occur. And we are using the [cargo-fuzz](https://github.com/rust-fuzz/cargo-fuzz) to run our fuzz test targets. ## Why we need them - Find bugs by leveraging random generation - Integrate with other tests (e.g., e2e) ## Resources All fuzz test-related resources are located in the `/tests-fuzz` directory. There are two types of resources: (1) fundamental components and (2) test targets. ### Fundamental components They are located in the `/tests-fuzz/src` directory. The fundamental components define how to generate SQLs (including dialects for different protocols) and validate execution results (e.g., column attribute validation), etc. ### Test targets They are located in the `/tests-fuzz/targets` directory, with each file representing an independent fuzz test case. The target utilizes fundamental components to generate SQLs, sends the generated SQLs via specified protocol, and validates the results of SQL execution. Figure 1 illustrates the fundamental components of the fuzz test provide the ability to generate random SQLs. It utilizes a Random Number Generator (Rng) to generate the Intermediate Representation (IR), then employs a DialectTranslator to produce specified dialects for different protocols. Finally, the fuzz tests send the generated SQL via the specified protocol and verify that the execution results meet expectations. ``` Rng | | v ExprGenerator | | v Intermediate representation (IR) | | +----------------------+----------------------+ | | | v v v MySQLTranslator PostgreSQLTranslator OtherDialectTranslator | | | | | | v v v SQL(MySQL Dialect) ..... ..... | | v Fuzz Test ``` (Figure1: Overview of fuzz tests) For more details about fuzz targets and fundamental components, please refer to this [tracking issue](https://github.com/GreptimeTeam/greptimedb/issues/3174). ## How to add a fuzz test target 1. Create an empty rust source file under the `/tests-fuzz/targets/.rs` directory. 2. Register the fuzz test target in the `/tests-fuzz/Cargo.toml` file. ```toml [[bin]] name = "" path = "targets/.rs" test = false bench = false doc = false ``` 3. Define the `FuzzInput` in the `/tests-fuzz/targets/.rs`. ```rust #![no_main] use libfuzzer_sys::arbitrary::{Arbitrary, Unstructured}; #[derive(Clone, Debug)] struct FuzzInput { seed: u64, } impl Arbitrary<'_> for FuzzInput { fn arbitrary(u: &mut Unstructured<'_>) -> arbitrary::Result { let seed = u.int_in_range(u64::MIN..=u64::MAX)?; Ok(FuzzInput { seed }) } } ``` 4. Write your first fuzz test target in the `/tests-fuzz/targets/.rs`. ``` /* Detailed source-code truncated for AI context efficiency. */ ``` 5. Run your fuzz test target ```bash cargo fuzz run --fuzz-dir tests-fuzz ``` For more details, please refer to this [document](/tests-fuzz/README.md). --- ### How To/Migrate Local Sql File Access # Migrate Local SQL File Access SQL access to local files is sandboxed in standalone deployments and disabled in distributed deployments. ## Standalone The default sandbox is `/copy`. Relative paths in `COPY` and external-table locations are resolved below this directory. Absolute paths work only when they are inside the sandbox. Before upgrading, identify existing `COPY` workflows and external tables that use local paths outside the default sandbox. Choose one of these migrations: - Move the files below `/copy` and update the SQL locations. - Set `storage.copy_root` to a dedicated local directory containing the files. - Move the files to S3, OSS, GCS, or AzBlob and update the SQL locations. Do not set `storage.copy_root` to `storage.data_home` or to a directory that contains GreptimeDB data, WAL, manifests, or configuration files. GreptimeDB rejects copy roots that expose its internal data directory. When `storage.data_home` is an object-storage URL, local SQL file access is disabled unless `storage.copy_root` explicitly names a local directory. ## Distributed Distributed frontend and datanode processes reject local paths for `COPY TABLE`, `COPY QUERY`, `COPY DATABASE`, and external tables. Migrate these workflows and tables to S3, OSS, GCS, or AzBlob before upgrading. --- ### Rfcs/2022 12 20 Promql In Rust/Rfc --- Feature Name: "promql-in-rust" Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/596 Date: 2022-12-20 Author: "Ruihang Xia " --- Rewrite PromQL in Rust ---------------------- # Summary A Rust native implementation of PromQL, for GreptimeDB. # Motivation Prometheus and its query language PromQL prevails in the cloud-native observability area, which is an important scenario for time series database like GreptimeDB. We already have support for its remote read and write protocols. Users can now integrate GreptimeDB as the storage backend to existing Prometheus deployment, but cannot run PromQL query directly on GreptimeDB like SQL. This RFC proposes to add support for PromQL. Because it was created in Go, we can't use the existing code easily. For interoperability, performance and extendability, porting its logic to Rust is a good choice. # Details ## Overview One of the goals is to make use of our existing basic operators, execution model and runtime to reduce the work. So the entire proposal is built on top of Apache Arrow DataFusion. The rewrote PromQL logic is manifested as `Expr` or `Execution Plan` in DataFusion. And both the intermediate data structure and the result is in the format of `Arrow`'s `RecordBatch`. The following sections are organized in a top-down manner. Starts with evaluation procedure. Then introduces the building blocks of our new PromQL operation. Follows by an explanation of data model. And end with an example logic plan. *This RFC is heavily related to Prometheus and PromQL. It won't repeat some basic concepts of them.* ## Evaluation The original implementation is like an interpreter of parsed PromQL AST. It has two characteristics: (1) Operations are evaluated in place after they are parsed to AST. And some key parameters are separated from the AST because they do not present in the query, but come from other places like another field in the HTTP payload. (2) calculation is performed per timestamp. You can see this pattern many times: ```go for ts := ev.startTimestamp; ts <= ev.endTimestamp; ts += ev.interval {} ``` These bring out two differences in the proposed implementation. First, to make it more general and clear, the evaluation procedure is reorganized into serval phases (and is the same as DataFusion's). And second, data are evaluated by time series (corresponding to "columnar calculation", if think timestamp as row number). ``` Logic Query AST Plan ─────────► Parser ───────► Logical ────────► Physical ────┐ Planner Planner │ │ ◄───────────────────────────── Executor ◄────────────────┘ Evaluation Result Execution Plan ``` - Parser Provided by [`promql-parser`](https://github.com/GreptimeTeam/promql-parser) crate. Same as the original implementation. - Logical Planner Generates a logical plan with all the needed parameters. It should accept something like `EvalStmt` in Go's implementation, which contains query time range, evaluation interval and lookback range. Another important thing done here is assembling the logic plan, with all the operations baked into logically. Like what's the filter and time range to read, how the data then flows through a selector into a binary operation, etc. Or what's the output schema of every single step. The generated logic plan is deterministic without variables, and can be `EXPLAIN`ed clearly. - Physical Planner This step converts a logic plan into evaluatable execution plan. There are not many special things like the previous step. Except when a query is going to be executed distributedly. In this case, a logic plan will be divided into serval parts and sent to serval nodes. One physical planner only sees its own part. - Executor As its name shows, this step calculates data to result. And all new calculation logic, the implementation of PromQL in rust, is placed here. And the rewrote functions are using `RecordBatch` and `Array` from `Arrow` as the intermediate data structure. Each "batch" contains only data from single time series. This is from the underlying storage implementation. Though it's not a requirement of this RFC, having this property can simplify some functions. Another thing to mention is the rewrote functions don't aware of timestamp or value columns, they are defined only based on the input data types. For example, `increase()` function in PromQL calculates the unbiased delta of data, its implementation here only does this single thing. Let's compare the signature of two implementations: - Go ```go func funcIncrease(vals []parser.Value, args parser.Expressions) Vector {} ``` - Rust ```rust fn prom_increase(input: Array) -> Array {} ``` Some unimportant parameters are omitted. The original Go version only writes the logic for `Point`'s value, either float or histogram. But the proposed rewritten one accepts a generic `Array` as input, which can be any type that suits, from `i8` to `u64` to `TimestampNanosecond`. ## Plan and Expression They are structures to express logic from PromQL. The proposed implementation is built on top of DataFusion, thus our plan and expression are in form of `ExtensionPlan` and `ScalarUDF`. The only difference between them in this context is the return type: plan returns a record batch while expression returns a single column. This RFC proposes to add four new plans, they are fundamental building blocks that mainly handle data selection logic in PromQL, for the following calculation expressions. - `SeriesNormalize` Sort data inside one series on the timestamp column, and bias "offset" if has. This plan usually comes after `TableScan` (or `TableScan` and `Filter`) plan. - `VectorManipulator` and `MatrixManipulator` Corresponding to `InstantSelector` and `RangeSelector`. We don't calculate timestamp by timestamp, thus use "vector" instead of "instant", this image shows the difference. And "matrix" is another name for "range vector", for not confused with our "vector". The following section will detail how they are implemented using Arrow. Due to "interval" parameter in PromQL, data after "selector" (or "manipulator" here) are usually shorter than input. And we have to modify the entire record batch to shorten both timestamp, value and tag columns. So they are formed as plan. - `PromAggregator` The carrier of aggregator expressions. This should not be very different from the DataFusion built-in `Aggregate` plan, except PromQL can use "group without" to do reverse selection. PromQL has around 70 expressions and functions. But luckily we can reuse lots of them from DataFusion. Like unary expression, binary expression and aggregator. We only need to implement those PromQL-specific expressions, like `rate` or `percentile`. The following table lists some typical functions in PromQL, and their signature in the proposed implementation. Other function should be the same. | Name | In Param(s) | Out Param(s) | Explain | |-------------------- |------------------------------------------------------ |-------------- |-------------------- | | instant_delta | Matrix T | Array T | idelta in PromQL | | increase | Matrix T | Array T | increase in PromQL | | extrapolate_factor | - Matrix T
- Array Timestamp
- Array Timestamp | Array T | * | *: *`extrapolate_factor` is one of the "dark sides" in PromQL. In short it's a translation of this [paragraph](https://github.com/prometheus/prometheus/blob/0372e259baf014bbade3134fd79bcdfd8cbdef2c/promql/functions.go#L134-L159)* To reuse those common calculation logic, we can break them into serval expressions, and assemble in the logic planning phase. Like `rate()` in PromQL can be represented as `increase / extrapolate_factor`. ## Data Model This part explains how data is represented. Following the data model in GreptimeDB, all the data are stored as table, with tag columns, timestamp column and value column. Table to record batch is very straightforward. So an instant vector can be thought of as a row (though as said before, we don't use instant vectors) in the table. Given four basic types in PromQL: scalar, string, instant vector and range vector, only the last "range vector" need some tricks to adapt our columnar calculation. Range vector is some sort of matrix, it's consisted of small one-dimension vectors, with each being an input of range function. And, applying range function to a range vector can be thought of kind of convolution. (Left is an illustration of range vector. Notice the Y-axis has no meaning, it's just put different pieces separately. The right side is an imagined "matrix" as range function. Multiplying the left side to it can get a one-dimension "matrix" with four elements. That's the evaluation result of a range vector.) To adapt this range vector to record batch, it should be represented by a column. This RFC proposes to use `DictionaryArray` from Arrow to represent range vector, or `Matrix`. This is "misusing" `DictionaryArray` to ship some additional information about an array. Because the range vector is sliding over one series, we only need to know the `offset` and `length` of each slides to reconstruct the matrix from an array: The length is not fixed, it depends on the input's timestamp. An PoC implementation of `Matrix` and `increase()` can be found in [this repo](https://github.com/waynexia/corroding-prometheus). ## Example The logic plan of PromQL query ```promql # start: 2022-12-20T10:00:00 # end: 2022-12-21T10:00:00 # interval: 1m # lookback: 30s sum (rate(request_duration[5m])) by (idc) ``` looks like # Drawbacks Human-being is always error-prone. It's harder to endeavor to rewrite from the ground and requires more attention to ensure correctness, than translate line-by-line. And, since the evaluator's architecture are different, it might be painful to catch up with PromQL's breaking update (if any) in the future. Misusing Arrow's DictionaryVector as Matrix is another point. This hack needs some `unsafe` function call to bypass Arrow's check. And though Arrow's API is stable, this is still an undocumented behavior. # Alternatives There are a few alternatives we've considered: - Wrap the existing PromQL's implementation via FFI, and import it to GreptimeDB. - Translate its evaluator engine line-by-line, rather than rewrite one. - Integrate the Prometheus server into GreptimeDB via RPC, making it a detached execution engine for PromQL. The first and second options are making a separate execution engine in GreptimeDB, they may alleviate the pain during rewriting, but will have negative impacts to afterward evolve like resource management. And introduce another deploy component in the last option will bring a complex deploy architecture. And all of them are more or less redundant in data transportation that affects performance and resources. The proposed built-in executing procedure is also easy to integrate and expose to the existing SQL interface GreptimeDB currently provides. Some concepts in PromQL like sliding windows (range vector in PromQL) are very convenient and ergonomic in analyzing series data. This makes it not only a PromQL evaluator, but also an enhancement to our query system. --- ### Rfcs/2023 03 29 File External Table/Rfc --- Feature Name: "File external table" Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/1041 Date: 2023-03-08 Author: "Xu Wenkang " --- File external table --- # Summary Allows users to perform SQL queries on files # Motivation User data may already exist in other storages, i.g., file systems/s3, etc. in CSV, parquet, JSON format, etc. We can provide users the ability to perform SQL queries on these files. # Details ## Overview The file external table providers users ability to perform SQL queries on these files. For example, a user has a CSV file on the local file system `/var/data/city.csv`: ``` Rank , Name , State , 2023 Population , 2020 Census , Annual Change , Density (mi²) 1 , New York City , New York , 8,992,908 , 8,804,190 , 0.7% , 29,938 2 , Los Angeles , California , 3,930,586 , 3,898,747 , 0.27% , 8,382 3 , Chicago , Illinois , 2,761,625 , 2,746,388 , 0.18% , 12,146 ..... ``` Then user can create a file external table with: ```sql CREATE EXTERNAL TABLE city with(location='/var/data/city.csv', format="CSV", field_delimiter = ',', record_delimiter = '\n', skip_header = 1); ``` Then query the external table with: ```bash MySQL> select * from city; ``` | Rank | Name | State | 2023 Population | 2020 Census | Annual Change | Density (mi²) | | :--- | :------------ | :--------- | :-------------- | :---------- | :------------ | :------------ | | 1 | New York City | New York | 8,992,908 | 8,804,190 | 0.7% | 29,938 | | 2 | Los Angeles | California | 3,930,586 | 3,898,747 | 0.27% | 8,382 | | 3 | Chicago | Illinois | 2,761,625 | 2,746,388 | 0.18% | 12,146 | Drop the external table, if needs with: ```sql DROP EXTERNAL TABLE city ``` ### Syntax ``` CREATE EXTERNAL [.] [ ( [NULL | NOT NULL] [COMMENT ""] ) ] [ WITH ( LOCATION = 'url' [,FIELD_DELIMITER = 'delimiter' ] [,RECORD_DELIMITER = 'delimiter' ] [,SKIP_HEADER = '' ] [,FORMAT = { csv | json | parquet } ] [,PATTERN = '' ] [,ENDPOINT = '' ] [,ACCESS_KEY_ID = '' ] [,SECRET_ACCESS_KEY = '' ] [,SESSION_TOKEN = '' ] [,REGION = '' ] [,ENABLE_VIRTUAL_HOST_STYLE = ''] .. ) ] ``` ### Supported File Format The external file table supports multiple formats; We divide formats into row format and columnar format. Row formats: - CSV, JSON Columnar formats: - Parquet Some of these formats support filter pushdown, and others don't. If users use very large files, that format doesn't support pushdown, which might consume a lot of IO for scanning full files and cause a long running query. ### File Table Engine We implement a file table engine that creates an external table by accepting user-specified file paths and treating all records as immutable. 1. File Format Decoder: decode files to the `RecordBatch` stream. 2. File Table Engine: implement the `TableProvider` trait, store necessary metadata in memory, and provide scan ability. Our implementation is better for small files. For large files(i.g., a GB-level CSV file), suggests our users import data to the database. ## Drawbacks - Some formats don't support filter pushdown - Hard to support indexing ## Life cycle ### Register a table 1. Write metadata to manifest. 2. Create the table via file table engine. 3. Register table to `CatalogProvider` and register table to `SystemCatalog`(persist tables to disk). ### Deregister a table (Drop a table) 1. Fetch the target table info (figure out table engine type). 2. Deregister the target table in `CatalogProvider` and `SystemCatalog`. 3. Find the target table engine. 4. Drop the target table. ### Recover a table when restarting 1. Collect tables name and engine type info. 2. Find the target tables in different engines. 3. Open and register tables. # Alternatives ## Using DataFusion API We can use datafusion API to register a file table: ```rust let ctx = SessionContext::new(); ctx.register_csv("example", "tests/data/example.csv", CsvReadOptions::new()).await?; // create a plan let df = ctx.sql("SELECT a, MIN(b) FROM example WHERE a <= b GROUP BY a LIMIT 100").await?; ``` ### Drawbacks The DataFusion implements its own `Object Store` abstraction and supports parsing the partitioned directories, which can push down the filter and skips some directories. However, this makes it impossible to use our's `LruCacheLayer`(The parsing of the partitioned directories required paths as input). If we want to manage memory entirely, we should implement our own `TableProvider` or `Table`. - Impossible to use `CacheLayer` ## Introduce an intermediate representation layer We convert all files into `parquet` as an intermediate representation. Then we only need to implement a `parquet` file table engine, and we already have a similar one. Also, it supports limited filter pushdown via the `parquet` row group stats. ### Drawbacks - Computing overhead - Storage overhead --- ### Rfcs/2024 02 21 Multi Dimension Partition Rule/Rfc --- Feature Name: Multi-dimension Partition Rule Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/3351 Date: 2024-02-21 Author: "Ruihang Xia " --- # Summary A new region partition scheme that runs on multiple dimensions of the key space. The partition rule is defined by a set of simple expressions on the partition key columns. # Motivation The current partition rule is from MySQL's [`RANGE Partition`](https://dev.mysql.com/doc/refman/8.0/en/partitioning-range.html), which is based on a single dimension. It is sort of a [Hilbert Curve](https://en.wikipedia.org/wiki/Hilbert_curve) and pick several point on the curve to divide the space. It is neither easy to understand how the data get partitioned nor flexible enough to handle complex partitioning requirements. Considering the future requirements like region repartitioning or autonomous rebalancing, where both workload and partition may change frequently. Here proposes a new region partition scheme that uses a set of simple expressions on the partition key columns to divide the key space. # Details ## Partition rule First, we define a simple expression that can be used to define the partition rule. The simple expression is a binary expression expression on the partition key columns that can be evaluated to a boolean value. The binary operator is limited to comparison operators only, like `=`, `!=`, `>`, `>=`, `<`, `<=`. And the operands are limited either literal value or partition column. Example of valid simple expressions are $`col_A = 10`$, $`col_A \gt 10 \& col_B \gt 20`$ or $`col_A \ne 10`$. Those expressions can be used as predicates to divide the key space into different regions. The following example have two partition columns `Col A` and `Col B`, and four partitioned regions. ```math \left\{\begin{aligned} &col_A \le 10 &Region_1 \\ &10 \lt col_A \& col_A \le 20 &Region_2 \\ &20 \lt col_A \space \& \space col_B \lt 100 &Region_3 \\ &20 \lt col_A \space \& \space col_B \ge 100 &Region_4 \end{aligned}\right\} ``` An advantage of this scheme is that it is easy to understand how the data get partitioned. The above example can be visualized in a 2D space (two partition column is involved in the example). Here each expression draws a line in the 2D space. Managing data partitioning becomes a matter of drawing lines in the key space. To make it easy to use, there is a "default region" which catches all the data that doesn't match any of previous expressions. The default region exist by default and do not need to specify. It is also possible to remove this default region if the DB finds it is not necessary. ## SQL interface The SQL interface is in response to two parts: specifying the partition columns and the partition rule. Thouth we are targeting an autonomous system, it's still allowed to give some bootstrap rules or hints on creating table. Partition column is specified by `PARTITION ON COLUMNS` sub-clause in `CREATE TABLE`: ```sql CREATE TABLE t (...) PARTITION ON COLUMNS (...) (); ``` Two following brackets are for partition columns and partition rule respectively. Columns provided here are only used as an allow-list of how the partition rule can be defined. Which means (a) the sequence between columns doesn't matter, (b) the columns provided here are not necessarily being used in the partition rule. The partition rule part is a list of comma-separated simple expressions. Expressions here are not corresponding to region, as they might be changed by system to fit various workload. A full example of `CREATE TABLE` with partition rule is: ```sql CREATE TABLE IF NOT EXISTS demo ( a STRING, b STRING, c STRING, d STRING, ts TIMESTAMP, memory DOUBLE, TIME INDEX (ts), PRIMARY KEY (a, b, c, d) ) PARTITION ON COLUMNS (c, b, a) ( a < 10, 10 >= a AND a < 20, 20 >= a AND b < 100, 20 >= a AND b > 100 ) ``` ## Combine with storage Examining columns separately suits our columnar storage very well in two aspects. 1. The simple expression can be pushed down to storage and file format, and is likely to hit existing index. Makes pruning operation very efficient. 2. Columns in columnar storage are not tightly coupled like in the traditional row storages, which means we can easily add or remove columns from partition rule without much impact (like a global reshuffle) on data. The data file itself can be "projected" to the key space as a polyhedron, it is guaranteed that each plane is in parallel with some coordinate planes (in a 2D scenario, this is saying that all the files can be projected to a rectangle). Thus partition or repartition also only need to consider related columns. An additional limitation is that considering how the index works and how we organize the primary keys at present, the partition columns are limited to be a subset of primary keys for better performance. # Drawbacks This is a breaking change. --- ### Rfcs/2023 01 03 Procedure Framework --- Feature Name: "procedure-framework" Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/286 Date: 2023-01-03 Author: "Yingwen " --- Procedure Framework ---------------------- # Summary A framework for executing operations in a fault-tolerant manner. # Motivation Some operations in GreptimeDB require multiple steps to implement. For example, creating a table needs: 1. Check whether the table exists 2. Create the table in the table engine 1. Create a region for the table in the storage engine 2. Persist the metadata of the table to the table manifest 3. Add the table to the catalog manager If the node dies or restarts in the middle of creating a table, it could leave the system in an inconsistent state. The procedure framework, inspired by [Apache HBase's ProcedureV2 framework](https://github.com/apache/hbase/blob/bfc9fc9605de638785435e404430a9408b99a8d0/src/main/asciidoc/_chapters/pv2.adoc) and [Apache Accumulo’s FATE framework](https://accumulo.apache.org/docs/2.x/administration/fate), aims to provide a unified way to implement multi-step operations that is tolerant to failure. # Details ## Overview The procedure framework consists of the following primary components: - A `Procedure` represents an operation or a set of operations to be performed step-by-step - `ProcedureManager`, the runtime to run `Procedures`. It executes the submitted procedures, stores procedures' states to the `ProcedureStore` and restores procedures from `ProcedureStore` while the database restarts. - `ProcedureStore` is a storage layer for persisting the procedure state ## Procedures The `ProcedureManager` keeps calling `Procedure::execute()` until the Procedure is done, so the operation of the Procedure should be [idempotent](https://developer.mozilla.org/en-US/docs/Glossary/Idempotent): it needs to be able to undo or replay a partial execution of itself. ```rust trait Procedure { fn execute(&mut self, ctx: &Context) -> Result; fn dump(&self) -> Result; fn rollback(&self) -> Result<()>; // other methods... } ``` The `Status` is an enum that has the following variants: ```rust enum Status { Executing { persist: bool, }, Suspended { subprocedures: Vec, persist: bool, }, Done, } ``` A call to `execute()` can result in the following possibilities: - `Ok(Status::Done)`: we are done - `Ok(Status::Executing { .. })`: there are remaining steps to do - `Ok(Status::Suspend { sub_procedure, .. })`: execution is suspended and can be resumed later after the sub-procedure is done. - `Err(e)`: error occurs during execution and the procedure is unable to proceed anymore. Users need to assign a unique `ProcedureId` to the procedure and the procedure can get this id via the `Context`. The `ProcedureId` is typically a UUID. ```rust struct Context { id: ProcedureId, // other fields ... } ``` The `ProcedureManager` calls `Procedure::dump()` to serialize the internal state of the procedure and writes to the `ProcedureStore`. The `Status` has a field `persist` to tell the `ProcedureManager` whether it needs persistence. ## Sub-procedures A procedure may need to create some sub-procedures to process its subtasks. For example, creating a distributed table with multiple regions (partitions) needs to set up the regions in each node, thus the parent procedure should instantiate a sub-procedure for each region. The `ProcedureManager` makes sure that the parent procedure does not proceed till all sub-procedures are successfully finished. The procedure can submit sub-procedures to the `ProcedureManager` by returning `Status::Suspended`. It needs to assign a procedure id to each procedure manually so it can track the status of the sub-procedures. ```rust struct ProcedureWithId { id: ProcedureId, procedure: BoxedProcedure, } ``` ## ProcedureStore We might need to provide two different ProcedureStore implementations: - In standalone mode, it stores data on the local disk. - In distributed mode, it stores data on the meta server or the object store service. These implementations should share the same storage structure. They store each procedure's state in a unique path based on the procedure id: ``` Sample paths: /procedures/{PROCEDURE_ID}/000001.step /procedures/{PROCEDURE_ID}/000002.step /procedures/{PROCEDURE_ID}/000003.commit ``` `ProcedureStore` behaves like a WAL. Before performing each step, the `ProcedureManager` can write the procedure's current state to the ProcedureStore, which stores the state in the `.step` file. The `000001` in the path is a monotonic increasing sequence of the step. After the procedure is done, the `ProcedureManager` puts a `.commit` file to indicate the procedure is finished (committed). The `ProcedureManager` can remove the procedure's files once the procedure is done, but it needs to leave the `.commit` as the last file to remove in case of failure during removal. ## ProcedureManager `ProcedureManager` executes procedures submitted to it. ```rust trait ProcedureManager { fn register_loader(&self, name: &str, loader: BoxedProcedureLoader) -> Result<()>; async fn submit(&self, procedure: ProcedureWithId) -> Result<()>; } ``` It supports the following operations: - Register a `ProcedureLoader` by the type name of the `Procedure`. - Submit a `Procedure` to the manager and execute it. When `ProcedureManager` starts, it loads procedures from the `ProcedureStore` and restores the procedures by the `ProcedureLoader`. The manager stores the type name from `Procedure::type_name()` with the data from `Procedure::dump()` in the `.step` file and uses the type name to find a `ProcedureLoader` to recover the procedure from its data. ```rust type BoxedProcedureLoader = Box Result + Send>; ``` ## Rollback The rollback step is supposed to clean up the resources created during the execute() step. When a procedure has failed, the `ProcedureManager` puts a `rollback` file and calls the `Procedure::rollback()` method. ```text /procedures/{PROCEDURE_ID}/000001.step /procedures/{PROCEDURE_ID}/000002.rollback ``` Rollback is complicated to implement so some procedures might not support rollback or only provide a best-efforts approach. ## Locking The `ProcedureManager` can provide a locking mechanism that gives a procedure read/write access to a database object such as a table so other procedures are unable to modify the same table while the current one is executing. # Drawbacks The `Procedure` framework introduces additional complexity and overhead to our database. - To execute a `Procedure`, we need to write to the `ProcedureStore` multiple times, which may slow down the server - We need to rewrite the logic of creating/dropping/altering a table using the procedure framework # Alternatives Another approach is to tolerate failure during execution and allow users to retry the operation until it succeeds. But we still need to: - Make each step idempotent - Record the status in some place to check whether we are done --- ### Rfcs/2023 02 01 Table Compaction --- Feature Name: "table-compaction" Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/930 Date: 2023-02-01 Author: "Lei, HUANG " --- # Table Compaction --- ## Background GreptimeDB uses an LSM-tree based storage engine that flushes memtables to SSTs for persistence. But currently it only supports level 0. SST files in level 0 does not guarantee to contain only rows with disjoint time ranges. That is to say, different SST files in level 0 may contain overlapped timestamps. The consequence is, in order to retrieve rows in some time range, all files need to be scanned, which brings a lot of IO overhead. Also, just like other LSMT engines, delete/update to existing primary keys are converted to new rows with delete/update mark and appended to SSTs on flushing. We need to merge the operations to same primary keys so that we don't have to go through all SST files to find the final state of these primary keys. ## Goal Implement a compaction framework to: - maintain SSTs in timestamp order to accelerate queries with timestamp condition; - merge rows with same primary key; - purge expired SSTs; - accommodate other tasks like data rollup/indexing. ## Overview Table compaction involves following components: - Compaction scheduler: run compaction tasks, limit the consumed resources; - Compaction strategy: find the SSTs to compact and determine the output files of compaction. - Compaction task: read the rows from input SSTs and write to the output files. ## Implementation ### Compaction scheduler `CompactionScheduler` is an executor that continuously polls and executes compaction request from a task queue. ```rust #[async_trait] pub trait CompactionScheduler { /// Schedules a compaction task. async fn schedule(&self, task: CompactionRequest) -> Result<()>; /// Stops compaction scheduler. async fn stop(&self) -> Result<()>; } ``` ### Compaction triggering Currently, we can check whether to compact tables when memtable is flushed to SST. https://github.com/GreptimeTeam/greptimedb/blob/4015dd80752e1e6aaa3d7cacc3203cb67ed9be6d/src/storage/src/flush.rs#L245 ### Compaction strategy `CompactionStrategy` defines how to pick SSTs in all levels for compaction. ```rust pub trait CompactionStrategy { fn pick( &self, ctx: CompactionContext, levels: &LevelMetas, ) -> Result; } ``` The most suitable compaction strategy for time-series scenario would be a hybrid strategy that combines time window compaction with size-tired compaction, just like [Cassandra](https://cassandra.apache.org/doc/latest/cassandra/managing/operating/compaction/twcs.html) and [ScyllaDB](https://docs.scylladb.com/stable/architecture/compaction/compaction-strategies.html#time-window-compaction-strategy-twcs) does. We can first group SSTs in level n into buckets according to some predefined time window. Within that window, SSTs are compacted in a size-tired manner (find SSTs with similar size and compact them to level n+1). SSTs from different time windows are neven compacted together. That strategy guarantees SSTs in each level are mainly sorted in timestamp order which boosts queries with explicit timestamp condition, while size-tired compaction minimizes the impact to foreground writes. ### Alternatives Currently, GreptimeDB's storage engine [only support two levels](https://github.com/GreptimeTeam/greptimedb/blob/43aefc5d74dfa73b7819cae77b7eb546d8534a41/src/storage/src/sst.rs#L32). For level 0, we can start with a simple time-window based leveled compaction, which reads from all SSTs in level 0, align them to time windows with a fixed duration, merge them with SSTs in level 1 within the same time window to ensure there is only one sorted run in level 1. --- ### Rfcs/2023 03 08 Region Fault Tolerance --- Feature Name: "Fault Tolerance for Region" Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/1126 Date: 2023-03-08 Author: "Luo Fucong " --- Fault Tolerance for Region ---------------------- # Summary This RFC proposes a method to achieve fault tolerance for regions in GreptimeDB's distributed mode. Or, put it in another way, achieving region high availability("HA") for GreptimeDB cluster. In this RFC, we mainly describe two aspects of region HA: how region availability is detected, and what recovery process is need to be taken. We also discuss some alternatives and future work. When this feature is done, our users could expect a GreptimeDB cluster that can always handle their requests to regions, despite some requests may failed during the region failover. The optimization to reduce the MTTR(Mean Time To Recovery) is not a concern of this RPC, and is left for future work. # Motivation Fault tolerance for regions is a critical feature for our clients to use the GreptimeDB cluster confidently. High availability for users to interact with their stored data is a "must have" for any TSDB products, that include our GreptimeDB cluster. # Details ## Background Some backgrounds about region in distributed mode: - A table is logically split into multiple regions. Each region stores a part of non-overlapping table data. - Regions are distributed in Datanodes, the mappings are not static, are assigned and governed by Metasrv. - In distributed mode, client requests are scoped in regions. To be more specific, when a request that needs to scan multiple regions arrived in Frontend, Frontend splits the request into multiple sub-requests, each of which scans one region only, and submits them to Datanodes that hold corresponding regions. In conclusion, as long as regions remain available, and regions could regain availability when failures do occur, the overall region HA could be achieved. With this in mind, let's see how region failures are detected first. ## Failure Detection We detect region failures in Metasrv, and do it both passively and actively. Passively means that Metasrv do not fire some "are you healthy" requests to regions. Instead, we carry region healthy information in the heartbeat requests that are submit to Metasrv by Datanodes. Datanode already carries its regions stats in the heartbeat request (the non-relevant fields are omitted): ```protobuf message HeartbeatRequest { ... // Region stats on this node repeated RegionStat region_stats = 6; ... } message RegionStat { uint64 region_id = 1; TableName table_name = 2; ... } ``` For the sake of simplicity, we don't add another field `bool available = 3` to the `RegionStat` message; instead, if the region were unavailable in the view of the Datanode that contains it, the Datanode just not includes the `RegionStat` of it in the heartbeat request. Or, if the Datanode itself is not unavailable, the heartbeat request is not submitted, effectively the same with not carrying the `RegionStat`. > The heartbeat interval is now hardcoded to five seconds. Metasrv gathers the heartbeat requests, extracts the `RegionStat`s, and treat them as region heartbeat. In this way, Metasrv maintains all regions healthy information. If some region's heartbeats were not received in a period of time, Metasrv speculates the region might be unavailable. To make the decision whether a region is failed or not, Metasrv uses a failure detection algorithm called the "[Phi φ Accrual Failure Detection](https://medium.com/@arpitbhayani/phi-%CF%86-accrual-failure-detection-79c21ce53a7a)". Basically, the algorithm calculates a value called "phi" to represent the possibility of a region's unavailability, based on the historical heartbeats' arrived rate. Once the "phi" is above some pre-defined threshold, Metasrv knows the region is failed. > This algorithm has been widely adopted in some well known products, like Akka and Cassandra. When Metasrv decides some region is failed from heartbeats, it's not the final decision. Here comes the "actively" detection. Before Metasrv decides to do region failover, it actively invokes the healthy check interface of the Datanode that the failure region resides. Only this healthy check is failed does Metasrv actually start doing failover upon the region. To conclude, the failure detection pseudo-codes are like this: ```rust // in Metasrv: fn failure_detection() { loop { // passive detection let failed_regions = all_regions.iter().filter(|r| r.estimated_failure_possibility() > config.phi).collect(); // find the datanodes that contains the failed regions let datanodes_and_regions = find_region_resides_datanodes(failed_regions); // active detection for (datanode, regions) in datanodes_and_regions { if !datanode.is_healthy(regions) { do_failover(datanode, regions); } } sleep(config.detect_interval); } } ``` Some design considerations: - Why active detecting while we have passively detection? Because it could be happened that the network is singly connectable sometimes (especially in the complex Cloud environment), then the Datanode's heartbeats cannot reach Metasrv, while Metasrv could request Datanode. Active detecting avoid this false positive situation. - Why the detection works on region instead of Datanode? Because we might face the possibility that only part of the regions in the Datanode are not available, not ALL regions. Especially the situation that Datanodes are used by multiple tenants. If this is the case, it's better to do failover upon the designated regions instead of the whole regions that reside on the Datanode. All in all, we want a more subtle control over region failover. So we detect some regions are not available. How to regain the availability back? ## Region Failover Region Failover largely relies on remote WAL, aka "[Bunshin](https://github.com/GreptimeTeam/bunshin)". I'm not including any of the details of it in this RFC, let's just assume we already have it. In general, region failover is fairly simple. Once Metasrv decides to do failover upon some regions, it first chooses one or more Datanodes to hold the failed region. This can be done easily, as the Metasrv already has the whole picture of Datanodes: it knows which Datanode has the minimum regions, what Datanode historically had the lowest CPU usage and IO rate, and how the Datanodes are assigned to tenants, among other information that can all help the Metasrv choose the most suitable Datanodes. Let's call these chosen Datanodes as "candidates". > The strategy to choose the most suitable candidates required careful design, but it's another RFC. Then, Metasrv sets the states of these failed regions as "passive". We should add a field to `Region`: ```protobuf message Region { uint64 id = 1; string name = 2; Partition partition = 3; message State { Active, Passive, } State state = 4; map attrs = 100; } ``` Here `Region` is used in message `RegionRoute`, which indicates how the write request is split among regions. When a region is set as "passive", Frontend knows the write to it should be rejected at the moment (the region read is not blocked, however). > Making a region "passive" here is effectively blocking the write to it. It's ok in the failover situation, the region is failed anyway. However, when dealing with active maintenance operations, region state requires more refined design. But that's another story. Third, Metasrv fires the "close region" requests to the failed Datanodes, and fires the "open region" requests to those candidates. "Close region" requests might be failed due to the unavailability of Datanodes, but that's fine, it's just a best-effort attempt to reduce the chance of any in-flight writes got handled unintentionally after the region is set as "passive". The "open region" requests must have succeeded though. Datanodes open regions from remote WAL. > Currently the "close region" is undefined in Datanode. It could be a local cache clean up of region data or other resources tidy up. Finally, when a candidate successfully opens its region, it calls back to Metasrv, indicating it is ready to handle region. "call back" here is backed by its heartbeat to Metasrv. Metasrv updates the region's state to "active", so as to let Frontend lifts the restrictions of region writes (again, the read part of region is untouched). All the above steps should be managed by remote procedure framework. It's another implementation challenge in the region failover feature. (One is the remote WAL of course.) A picture is worth a 1000 words: ``` /* Detailed source-code truncated for AI context efficiency. */ ``` # Alternatives ## The "Neon" Way Remote WAL raises a problem that could harm the write throughput of GreptimeDB cluster: each write request has to do at least two remote call, one is from Frontend to Datanode, and one is from Datanode to remote WAL. What if we do it the "[Neon](https://github.com/neondatabase/neon)" way, making remote WAL sits in between the Frontend and Datanode, couldn't that improve our write throughput? It could, though there're some consistency issues like "read-your-writes" to solve. However, the main concerns we don't adopt this method are two-fold: 1. Remote WAL is planned to be quorum based, it can be efficiently written; 2. More importantly, we are planning to make the remote WAL an option that users could choose not to enable it (at the cost of some reliability reduction). ## No WAL, Replication instead This method replicates region across Datanodes directly, like the common way in shared-nothing database. Were the main region failed, a standby region in the replicate group is elected as new "main" and take the read/write requests. The main concern to this method is the incompatibility to our current architecture and code structure. It requires a major redesign, but gains no significant advantage over the remote WAL method. However, the replication does have its own advantage that we can learn from to optimize this failover procedure. # Future Work Some optimizations we could take: - To reduce the MTTR, we could make Metasrv chooses the candidate to each region at normal time. The candidate does some preparation works to reduce the open region time, effectively accelerate the failover procedure. - We can adopt the replication method, to the degree that region replicas are used as the fast catch-up candidates. The data difference among replicas is minor, region failover does not need to load or exchange too much data, greatly reduced the region failover time. --- ### Rfcs/2023 05 09 Distributed Planner --- Feature Name: distributed-planner Tracking Issue: TBD Date: 2023-05-09 Author: "Ruihang Xia " --- Distributed Planner ------------------- # Summary Enhance the logical planner with aware of distributed, multi-region table topology. To achieve "push computation down" execution rather than the current "pull data up" manner. # Motivation Query distributively can leverage the advantage of GreptimeDB's architecture to process large dataset that exceeds the capacity of a single node, or accelerate the query execution by executing it in parallel. This task includes two sub-tasks - Be able to transform the plan that can push as much as possible computation down to data source. - Be able to handle pipeline breaker (like `Join` or `Sort`) on multiple computation nodes. This is a relatively complex topic. To keep this RFC concentrated I'll focus on the first one. # Details ## Background: Partition and Region GreptimeDB supports table partitioning, where the partition rule is set during table creation. Each partition can be further divided into one or more physical storage units known as "regions". Both partitions and regions are divided based on rows: ``` text ┌────────────────────────────────────┐ │ │ │ Table │ │ │ └─────┬────────────┬────────────┬────┘ │ │ │ │ │ │ ┌─────▼────┐ ┌─────▼────┐ ┌─────▼────┐ │ Region 1 │ │ Region 2 │ │ Region 3 │ └──────────┘ └──────────┘ └──────────┘ Row 1~10 Row 11~20 Row 21~30 ``` General speaking, region is the minimum element of data distribution, and we can also use it as the unit to distribute computation. This can greatly simplify the routing logic of this distributed planner, by always schedule the computation to the node that currently opening the corresponding region. And is also easy to scale more node for computing since GreptimeDB's data is persisted on shared storage backend like S3. But this is a bit beyond the scope of this specific topic. ## Background: Commutativity Commutativity is an attribute that describes whether two operation can exchange their apply order: $P1(P2(R)) \Leftrightarrow P2(P1(R))$. If the equation keeps, we can transform one expression into another form without changing its result. This is useful on rewriting SQL expression, and is the theoretical basis of this RFC. Take this SQL as an example ``` sql SELECT a FROM t WHERE a > 10; ``` As we know projection and filter are commutative (todo: latex), it can be translated to the following two identical plan trees: ```text ┌─────────────┐ ┌─────────────┐ │Projection(a)│ │Filter(a>10) │ └──────▲──────┘ └──────▲──────┘ │ │ ┌──────┴──────┐ ┌──────┴──────┐ │Filter(a>10) │ │Projection(a)│ └──────▲──────┘ └──────▲──────┘ │ │ ┌──────┴──────┐ ┌──────┴──────┐ │ TableScan │ │ TableScan │ └─────────────┘ └─────────────┘ ``` ## Merge Operation This RFC proposes to add a new expression node `MergeScan` to merge result from several regions in the frontend. It wrap the abstraction of remote data and execution, and expose a `TableScan` interface to upper level. ``` text ▲ │ ┌───────┼───────┐ │ │ │ │ ┌──┴──┐ │ │ └──▲──┘ │ │ │ │ │ ┌──┴──┐ │ │ └──▲──┘ │ ┌─────────────────────────────┐ │ │ │ │ │ │ ┌────┴────┐ │ │ ┌──────────┐ ┌───┐ ┌───┐ │ │ │MergeScan◄──┼────┤ │ Region 1 │ │ │ .. │ │ │ │ └─────────┘ │ │ └──────────┘ └───┘ └───┘ │ │ │ │ │ └─Frontend──────┘ └─Remote-Sources──────────────┘ ``` This merge operation simply chains all the underlying remote data sources and return `RecordBatch`, just like a coalesce op. And each remote sources is a gRPC query to datanode via the substrait logical plan interface. The plan is transformed and divided from the original query that comes to frontend. ## Commutativity of MergeScan Obviously, The position of `MergeScan` is the key of the distributed plan. The more closer to the underlying `TableScan`, the less computation is taken by datanodes. Thus the goal is to pull the `MergeScan` up as more as possible. The word "pull up" means exchange `MergeScan` with its parent node in the plan tree, which means we should check the commutativity between the existing expression nodes and the `MergeScan`. Here I classify all the possibility into five categories: - Commutative: $P1(P2(R)) \Leftrightarrow P2(P1(R))$ - filter - projection - operations that match the partition key - Partial Commutative: $P1(P2(R)) \Leftrightarrow P1(P2(P1(R)))$ - $min(R) \rightarrow min(MERGE(min(R)))$ - $max(R) \rightarrow max(MERGE(max(R)))$ - Conditional Commutative: $P1(P2(R)) \Leftrightarrow P3(P2(P1(R)))$ - $count(R) \rightarrow sum(count(R))$ - Transformed Commutative: $P1(P2(R)) \Leftrightarrow P1(P3(R)) \Leftrightarrow P3(P1(R))$ - $avg(R) \rightarrow sum(R)/count(R)$ - Non-commutative - sort - join - percentile ## Steps to plan After establishing the set of commutative relations for all expressions, we can begin transforming the logical plan. There are four steps: - Add a merge node before table scan - Evaluate commutativity in a bottom-up way, stop at the first non-commutative node - Divide the TableScan to scan over partitions - Execute First insert the `MergeScan` on top of the bottom `TableScan` node. Then examine the commutativity start from the `MergeScan` node transform the plan tree based on the result. Stop this process on the first non-commutative node. ``` text ┌─────────────┐ ┌─────────────┐ │ Sort │ │ Sort │ └──────▲──────┘ └──────▲──────┘ │ │ ┌─────────────┐ ┌──────┴──────┐ ┌──────┴──────┐ │ Sort │ │Projection(a)│ │ MergeScan │ └──────▲──────┘ └──────▲──────┘ └──────▲──────┘ │ │ │ ┌──────┴──────┐ ┌──────┴──────┐ ┌──────┴──────┐ │Projection(a)│ │ MergeScan │ │Projection(a)│ └──────▲──────┘ └──────▲──────┘ └──────▲──────┘ │ │ │ ┌──────┴──────┐ ┌──────┴──────┐ ┌──────┴──────┐ │ TableScan │ │ TableScan │ │ TableScan │ └─────────────┘ └─────────────┘ └─────────────┘ (a) (b) (c) ``` Then in the physical planning phase, convert the sub-tree below `MergeScan` into a remote query request and dispatch to all the regions. And let the `MergeScan` to receive the results and feed to it parent node. To simplify the overall complexity, any error in the procedure will lead to a failure to the entire query, and cancel all other parts. # Alternatives ## Spill If only consider the ability of processing large dataset, we can enable DataFusion's spill ability to temporary persist intermediate data into disk, like the "swap" memory. But this will lead to a super slow performance and very large write amplification. # Future Work As described in the `Motivation` section we can further explore the distributed planner on the physical execution level, by introducing mechanism like Spark's shuffle to improve parallelism and reduce intermediate pipeline breaker's stage. --- ### Rfcs/2023 07 06 Table Engine Refactor --- Feature Name: table-engine-refactor Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/1869 Date: 2023-07-06 Author: "Yingwen " --- Refactor Table Engine ---------------------- # Summary Refactor table engines to address several historical tech debts. # Motivation Both `Frontend` and `Datanode` have to deal with multiple regions in a table. This results in code duplication and additional burden to the `Datanode`. Before: ```mermaid graph TB subgraph Frontend["Frontend"] subgraph MyTable A("region 0, 2 -> Datanode0") B("region 1, 3 -> Datanode1") end end MyTable --> Metasrv Metasrv --> ETCD MyTable-->TableEngine0 MyTable-->TableEngine1 subgraph Datanode0 Procedure0("procedure") TableEngine0("table engine") region0 region2 mytable0("my_table") Procedure0-->mytable0 TableEngine0-->mytable0 mytable0-->region0 mytable0-->region2 end subgraph Datanode1 Procedure1("procedure") TableEngine1("table engine") region1 region3 mytable1("my_table") Procedure1-->mytable1 TableEngine1-->mytable1 mytable1-->region1 mytable1-->region3 end subgraph manifest["table manifest"] M0("my_table") M1("regions: [0, 1, 2, 3]") end mytable1-->manifest mytable0-->manifest RegionManifest0("region manifest 0") RegionManifest1("region manifest 1") RegionManifest2("region manifest 2") RegionManifest3("region manifest 3") region0-->RegionManifest0 region1-->RegionManifest1 region2-->RegionManifest2 region3-->RegionManifest3 ``` `Datanodes` can update the same manifest file for a table as regions are assigned to different nodes in the cluster. We also have to run procedures on `Datanode` to ensure the table manifest is consistent with region manifests. "Table" in a `Datanode` is a subset of the table's regions. The `Datanode` is much closer to `RegionServer` in `HBase` which only deals with regions. In cluster mode, we store table metadata in etcd and table manifest. The table manifest becomes redundant. We can remove the table manifest if we refactor the table engines to region engines that only care about regions. What's more, we don't need to run those procedures on `Datanode`. After: ```mermaid graph TB subgraph Frontend["Frontend"] direction LR subgraph MyTable A("region 0, 2 -> Datanode0") B("region 1, 3 -> Datanode1") end end MyTable --> Metasrv Metasrv --> ETCD MyTable-->RegionEngine MyTable-->RegionEngine1 subgraph Datanode0 RegionEngine("region engine") region0 region2 RegionEngine-->region0 RegionEngine-->region2 end subgraph Datanode1 RegionEngine1("region engine") region1 region3 RegionEngine1-->region1 RegionEngine1-->region3 end RegionManifest0("region manifest 0") RegionManifest1("region manifest 1") RegionManifest2("region manifest 2") RegionManifest3("region manifest 3") region0-->RegionManifest0 region1-->RegionManifest1 region2-->RegionManifest2 region3-->RegionManifest3 ``` This RFC proposes to refactor table engines into region engines as a first step to make the `Datanode` acts like a `RegionServer`. # Details ## Overview We plan to refactor the `TableEngine` trait into `RegionEngine` gradually. This RFC focuses on the `mito` engine as it is the default table engine and the most complicated engine. Currently, we built `MitoEngine` upon `StorageEngine` that manages regions of the `mito` engine. Since `MitoEngine` becomes a region engine, we could combine `StorageEngine` with `MitoEngine` to simplify our code structure. The chart below shows the overall architecture of the `MitoEngine`. ```mermaid classDiagram class MitoEngine~LogStore~ { -WorkerGroup workers } class MitoRegion { +VersionControlRef version_control -RegionId region_id -String manifest_dir -AtomicI64 last_flush_millis +region_id() RegionId +scan() ChunkReaderImpl } class RegionMap { -HashMap<RegionId, MitoRegionRef> regions } class ChunkReaderImpl class WorkerGroup { -Vec~RegionWorker~ workers } class RegionWorker { -RegionMap regions -Sender sender -JoinHandle handle } class RegionWorkerThread~LogStore~ { -RegionMap regions -Receiver receiver -Wal~LogStore~ wal -ObjectStore object_store -MemtableBuilderRef memtable_builder -FlushSchedulerRef~LogStore~ flush_scheduler -FlushStrategy flush_strategy -CompactionSchedulerRef~LogStore~ compaction_scheduler -FilePurgerRef file_purger } class Wal~LogStore~ { -LogStore log_store } class MitoConfig MitoEngine~LogStore~ o-- MitoConfig MitoEngine~LogStore~ o-- MitoRegion MitoEngine~LogStore~ o-- WorkerGroup MitoRegion o-- VersionControl MitoRegion -- ChunkReaderImpl WorkerGroup o-- RegionWorker RegionWorker o-- RegionMap RegionWorker -- RegionWorkerThread~LogStore~ RegionWorkerThread~LogStore~ o-- RegionMap RegionWorkerThread~LogStore~ o-- Wal~LogStore~ ``` We replace the `RegionWriter` with `RegionWorker` to process write requests and DDL requests. ## Metadata We also merge region's metadata with table's metadata. It should make metadata much easier to maintain. ``` /* Detailed source-code truncated for AI context efficiency. */ ``` # Drawback This is a breaking change. # Future Work - Rename `TableEngine` to `RegionEngine` - Simplify schema relationship in the `mito` engine - Refactor the `Datanode` into a `RegionServer`. --- ### Rfcs/2023 07 10 Metric Engine --- Feature Name: metric-engine Tracking Issue: TBD Date: 2023-07-10 Author: "Ruihang Xia " --- # Summary A new metric engine that can significantly enhance our ability to handle the tremendous number of small tables in scenarios like Prometheus metrics, by leveraging a synthetic wide table that offers storage and metadata multiplexing capabilities over the existing engine. # Motivation The concept "Table" in GreptimeDB is a bit "heavy" compared to other time-series storage like Prometheus or VictoriaMetrics. This has lots of disadvantages in aspects from performance, footprint, and storage to cost. # Details ## Top level description - User Interface This feature will add a new type of storage engine. It might be available to be an option like `with ENGINE=mito` or an internal interface like auto create table on Prometheus remote write. From the user side, there is no difference from tables in mito engine. All the DDL like `CREATE`, `ALTER` and DML like `SELECT` should be supported. - Implementation Overlook This new engine doesn't re-implement low level components like file R/W etc. It's a wrapper layer over the existing mito engine, with extra storage and metadata multiplexing capabilities. I.e., it expose multiple table based on one mito engine table like this: ``` plaintext ┌───────────────┐ ┌───────────────┐ ┌───────────────┐ │ Metric Engine │ │ Metric Engine │ │ Metric Engine │ │ Table 1 │ │ Table 2 │ │ Table 3 │ └───────────────┘ └───────────────┘ └───────────────┘ ▲ ▲ ▲ │ │ │ └───────────────┼───────────────────┘ │ ┌─────────┴────────┐ │ Metric Region │ │ Engine │ │ ┌─────────────┤ │ │ Mito Region │ │ │ Engine │ └────▲─────────────┘ │ │ ┌─────┴───────────────┐ │ │ │ Mito Engine Table │ │ │ └─────────────────────┘ ``` The following parts will describe these implementation details: - How to route these metric region tables and how those table are distributed - How to maintain the schema and other metadata of the underlying mito engine table - How to maintain the schema of metric engine table - How the query goes ## Routing Before this change, the region route rule was based on a group of partition keys. Relation of physical table to region is one-to-many. ``` rust pub struct PartitionDef { partition_columns: Vec, partition_bounds: Vec, } ``` And for metric engine tables, the key difference is we split the concept of "physical table" and "logical table". Like the previous ASCII chart, multiple logical tables are based on one physical table. The relationship of logical table to region becomes many-to-many. Thus, we must include the table name (of logical table) into partition rules. Consider the partition/route interface is a generic map of string array to region id, all we need to do is to insert logical table name into the request: ``` rust fn route(request: Vec) -> RegionId; ``` The next question is, where to do this conversion? The basic idea is to dispatch different routing behavior based on the engine type. Since we have all the necessary information in frontend, it's a good place to do that. And can leave meta server untouched. The essential change is to associate engine type with route rule. ## Physical Region Schema The idea "physical wide table" is to perform column-level multiplexing. I.e., map all logical columns to physical columns by their names. ``` ┌────────────┐ ┌────────────┐ ┌────────────┐ │ Table 1 │ │ Table 2 │ │ Table 3 │ ├───┬────┬───┤ ├───┬────┬───┤ ├───┬────┬───┤ │C1 │ C2 │ C3│ │C1 │ C3 │ C5├──────┐ │C2 │ C4 │ C6│ └─┬─┴──┬─┴─┬─┘ ┌────┴───┴──┬─┴───┘ │ └─┬─┴──┬─┴─┬─┘ │ │ │ │ │ │ │ │ │ │ │ │ │ └──────────┐ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ┌─────────────────┐ │ │ │ │ │ │ │ │ │ │ Physical Table │ │ │ │ │ │ │ │ │ │ ├──┬──┬──┬──┬──┬──┘ │ │ │ │ │ └────x───x───┴─►│C1│C2│C3│C4│C5│C6◄─┼─x────x────x───┘ │ │ └──┘▲─┘▲─┴─▲└─▲└──┘ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ├───x──────────┘ ├───x──x─────┘ │ │ │ │ │ │ │ │ │ │ │ │ └─────────────┘ │ └───────┘ │ │ │ │ │ │ └─────────────────────x───────────────┘ │ │ │ └────────────────────┘ ``` This approach is very straightforward but has one problem. It only works when two columns have different semantic type (time index, tag or field) or data types but with the same name. E.g., `CREATE TABLE t1 (c1 timestamp(3) TIME INDEX)` and `CREATE TABLE t2 (c1 STRING PRIMARY KEY)`. One possible workaround is to prefix each column with its data type and semantic type, like `_STRING_PK_c1`. However, considering the primary goal at present is to support data from monitoring metrics like Prometheus remote write, it's acceptable not to support this at first because data types are often simple and limited here. The next point is changing the physical table's schema. This is only needed when creating a new logical table or altering the existing table. Typically speaking, table creating and altering are explicit. We only need to emit an add column request to underlying physical table on processing logical table's DDL. GreptimeDB can create or alter table automatically on some protocols, but the internal logic is the same. Also for simplicity, we don't support shrinking the underlying table at first. This can be achieved by introducing mechanism on the physical column. Frontend needs not to keep physical table's schema. ## Metadata of physical regions Those metric engine regions need to store extra metadata like the schema of logical table or all logical table's name. That information is relatively simple and can be stored in a format like key-value pair. For now, we have to use another physical mito region for metadata. This involves an issue with region scheduling. Since we don't have the ability to perform affinity scheduling, the initial version will just assume the data region and metadata region are in the same instance. See alternatives - other storage for physical region's metadata for possible future improvement. Here is the schema of metadata region and how we would use it. The `CREATE TABLE` clause of metadata region looks like the following. Notice that it wouldn't be actually created by SQL. ``` sql CREATE TABLE metadata( ts timestamp time index, key string primary key, value string ); ``` The `ts` field is just a placeholder -- for the constraints that a mito region must contain a time index field. It will always be `0`. The other two fields `key` and `value` will be used as a k-v storage. It contains two group of key - `__table_` is used for marking table existence. It doesn't have value. - `__column__` is used for marking table existence, the value is column's semantic type. ## Physical region implementation This RFC proposes to add a new region implementation named "MetricRegion". As showed in the first chart, it's wrapped over the existing mito region. This section will describe the implementation details. Firstly, here is a chart shows how the region hierarchy looks like: ```plaintext ┌───────────────────────┐ │ Metric Region │ │ │ │ ┌────────┬──────────┤ │ │ Mito │ Mito │ │ │ Region │ Region │ │ │ for │ for │ │ │ Data │ Metadata │ └───┴────────┴──────────┘ ``` All upper levels only see the Metric Region. E.g., Meta Server schedules on this region, or Frontend routes requests to this Metrics Region's id. To be scheduled (open or close etc.), Metric Region needs to implement its own procedures. Most of those procedures can be simply assembled from underlying Mito Regions', but those related to data like alter or drop will have its own new logic. Another point is region id. Since the region id is used widely from meta server to persisted state, it's better to keep it unchanged. This means we can't use the same id for two regions, but one for each. To achieve this, this RFC proposes a concept named "region id group". A region id group is a group of region ids that are bound for different purposes. Like the two underlying regions here. This preserves the first 8 bits of the `u32` region number for grouping. Each group has one main id (the first one) and other sub ids (the rest non-zero ids). All components other than the region implementation itself doesn't aware of the existence of region id group. They only see the main id. The region implementation is in response of managing and using the region id group. ```plaintext 63 31 23 0 ┌────────────────────────────────────┬──────────┬──────────────────┐ │ Table Id(32) │ Group(8) │ Region Number(24)│ └────────────────────────────────────┴──────────┴──────────────────┘ Region Id(32) ``` ## Routing in meta server From previous sections, we can conclude the following points about routing: - Each "logical table" has its own, universe unique table id. - Logical table doesn't have physical region, they share the same physical region with other logical tables. - Route rule of logical table's is a strict subset of physical table's. To associate the logical table with physical region, we need to specify necessary information in the create table request. Specifically, the table type and its parent table. This require to change our gRPC proto's definition. And once meta recognize the table to create is a logical table, it will use the parent table's region to create route entry. And to reduce the consumption of region failover (which need to update the physical table route info), we'd better to split the current route table structure into two parts: ```rust region_route: Map, node_route: Map, ``` By doing this on each failover the meta server only needs to update the second `node_route` map and leave the first one untouched. ## Query Like other existing components, a user query always starts in the frontend. In the planning phase, frontend needs to fetch related schemas of the queried table. This part is the same. I.e., changes in this RFC don't affect components above the `Table` abstraction. # Alternatives ## Other routing method We can also do this "special" route rule in the meta server. But there is no difference with the proposed method. ## Other storage for physical region's metadata Once we have implemented the "region family" that allows multiple physical schemas exist in one region, we can store the metadata and table data into one region. Before that, we can also let the `MetricRegion` holds a `KvBackend` to access the storage layer directly. But this breaks the abstraction in some way. # Drawbacks Since the physical storage is mixed together. It's hard to do fine-grained operations in table level. Like configuring TTL, memtable size or compaction strategy in table level. Or define different partition rules for different tables. For scenarios like this, it's better to move the table out of metrics engine and "upgrade" it to a normal mito engine table. This requires a migration process in a low cost. And we have to ensure data consistency during the migration, which may require a out-of-service period. --- ### Rfcs/2023 08 04 Table Trait Refactor --- Feature Name: table-trait-refactor Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/2065 Date: 2023-08-04 Author: "Ruihang Xia " --- Refactor Table Trait -------------------- # Summary Refactor `Table` trait to adapt the new region server architecture and make code more straightforward. # Motivation The `Table` is designed in the background of both frontend and datanode keeping the same concepts. And all the operations are served by a `Table`. However, in our practice, we found that not all the operations are suitable to be served by a `Table`. For example, the `Table` doesn't hold actual physical data itself, thus operations like write or alter are simply a proxy over underlying regions. And in the recent refactor to datanode ([rfc table-engine-refactor](./2023-07-06-table-engine-refactor.md)), we are changing datanode to region server that is only aware of `Region` things. This also calls for a refactor to the `Table` trait. # Details ## Definitions The current `Table` trait contains the following methods: ``` /* Detailed source-code truncated for AI context efficiency. */ ``` We can divide those methods into three categories from the perspective of functionality: | Retrieve Metadata | Manipulate Data | Read Data | | :------------------------: | :-------------: | :--------------: | | `schema` | `insert` | `scan_to_stream` | | `table_info` | `alter` | | | `table_type` | `delete` | | | `supports_filter_pushdown` | `flush` | | | `region_stats` | `close` | | | `contains_region` | `compact` | | | `statistics` | | | And considering most of the access to metadata happens in frontend, like route or query; and all the persisted data are stored in regions; while only the query engine needs to read data. We can divide the `Table` trait into three concepts: - struct `Table` provides metadata: ```rust impl Table { /// Get a reference to the schema for this table fn schema(&self) -> SchemaRef; /// Get a reference to the table info. fn table_info(&self) -> TableInfoRef; /// Get the type of this table for metadata/catalog purposes. fn table_type(&self) -> TableType; /// Get statistics for this table, if available fn statistics(&self) -> Option; fn to_data_source(&self) -> DataSourceRef; } ``` - Requests to region server - `InsertRequest` - `AlterRequest` - `DeleteRequest` - `FlushRequest` - `CompactRequest` - `CloseRequest` - trait `DataSource` provides data (`RecordBatch`) ```rust trait DataSource { fn get_stream(&self, request: ScanRequest) -> Result; } ``` ## Use `Table` `Table` will only be used in frontend. It's constructed from the `OpenTableRequest` or `CreateTableRequest`. `Table` also provides a method `to_data_source` to generate a `DataSource` from itself. But this method is only for non-`TableType::Base` tables (i.e., `TableType::View` and `TableType::Temporary`) because `TableType::Base` table doesn't hold actual data itself. Its `DataSource` should be constructed from the `Region` directly (in other words, it's a remote query). And it requires some extra information to construct a `DataSource`, named `TableSourceProvider`: ```rust type TableFactory = Arc DataSourceRef>; pub enum TableSourceProvider { Base, View(LogicalPlan), Temporary(TableFactory), } ``` ## Use `DataSource` `DataSource` will be adapted to the `TableProvider` from DataFusion that can be `scan()`ed in a `TableScan` plan. In frontend this is done in the planning phase. And datanode will have one implementation for `Region` to generate record batch stream. ## Interact with RegionServer Previously, persisted state change operations were through the old `Table` trait, like said before. Now they will come from the action source, like the procedure or protocol handler directly to the region server. E.g., on alter table, the corresponding procedure will generate its `AlterRequest` and send it to regions. Or write request will be split in frontend handler, and sent to regions. `Table` only provides necessary metadata like route information if needed, but not the necessary part anymore. ## Implement temporary table Temporary table is a special table that doesn't revolves to any persistent physical region. Examples are: - the `Numbers` table for testing, which produces a record batch that contains 0-100 integers. - tables in information schema. It is an interface for querying catalog's metadata. The contents are generated on the fly with information from `CatalogManager`. The `CatalogManager` can be held in `TableFactory`. - Function table that produces data generated by a formula or a function. Like something that always `sin(current_timestamp())`. ## Relationship among those components Here is a diagram to show the relationship among those components, and how they interact with each other. ```mermaid erDiagram CatalogManager ||--|{ Table : manages Table ||--|{ DataStream : generates Table ||--|{ Region : routes Region ||--|{ DataStream : implements DataStream }|..|| QueryEngine : adapts-to Procedure ||--|{ Region : requests Protocol ||--|{ Region : writes Protocol ||--|{ QueryEngine : queries ``` # Drawback This is a breaking change. --- ### Rfcs/2023 08 13 Metadata Txn --- Feature Name: Update Metadata in single transaction Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/1715 Date: 2023-08-13 Author: "Feng Yangsen , Xu Wenkang " --- # Summary Update Metadata in single transaction. # Motivation Currently, multiple transactions are involved during the procedure. This implementation is inefficient, and it's hard to make data consistent. Therefore, We can update multiple metadata in a single transaction. # Details Now we have the following table metadata keys: **TableInfo** ```rust // __table_info/{table_id} pub struct TableInfoKey { table_id: TableId, } pub struct TableInfoValue { pub table_info: RawTableInfo, version: u64, } ``` **TableRoute** ```rust // __table_route/{table_id} pub struct NextTableRouteKey { table_id: TableId, } pub struct TableRoute { pub region_routes: Vec, } ``` **DatanodeTable** ```rust // __table_route/{datanode_id}/{table_id} pub struct DatanodeTableKey { datanode_id: DatanodeId, table_id: TableId, } pub struct DatanodeTableValue { pub table_id: TableId, pub regions: Vec, version: u64, } ``` **TableNameKey** ```rust // __table_name/{CatalogName}/{SchemaName}/{TableName} pub struct TableNameKey<'a> { pub catalog: &'a str, pub schema: &'a str, pub table: &'a str, } pub struct TableNameValue { table_id: TableId, } ``` These table metadata only updates in the following operations. ## Region Failover It needs to update `TableRoute` key and `DatanodeTable` keys. If the `TableRoute` equals the Snapshot of `TableRoute` submitting the Failover task, then we can safely update these keys. After submitting Failover tasks to acquire locks for execution, the `TableRoute` may be updated by another task. After acquiring the lock, we can get the latest `TableRoute` again and then execute it if needed. ## Create Table DDL Creates all of the above keys. `TableRoute`, `TableInfo`, should be empty. The **TableNameKey**'s lock will be held by the procedure framework. ## Drop Table DDL `TableInfoKey` and `NextTableRouteKey` will be added with `__removed-` prefix, and the other above keys will be deleted. The transaction will not compare any keys. ## Alter Table DDL 1. Rename table, updates `TableInfo` and `TableName`. Compares `TableInfo`, and the new `TableNameKey` should be empty, and TableInfo should equal the Snapshot when submitting DDL. The old and new **TableNameKey**'s lock will be held by the procedure framework. 2. Alter table, updates `TableInfo`. `TableInfo` should equal the Snapshot when submitting DDL. --- ### Rfcs/2023 11 03 Inverted Index --- Feature Name: Inverted Index for SST File Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/2705 Date: 2023-11-03 Author: "Zhong Zhenchi " --- # Summary This RFC proposes an optimization towards the storage engine by introducing an inverted indexing methodology aimed at optimizing label selection queries specifically pertaining to Metrics with tag columns as the target for optimization. # Introduction In the current system setup, in the Mito Engine, the first column of Primary Keys has a Min-Max index, which significantly optimizes the outcome. However, there are limitations when it comes to other columns, primarily tags. This RFC suggests the implementation of an inverted index to provide enhanced filtering benefits to bridge these limitations and improve overall system performance. # Design Detail ## Inverted Index The primary aim of the proposed inverted index is to optimize tag columns in the SST Parquet Files within the Mito Engine. The mapping and construction of an inverted index, from Tag Values to Row Groups, enables efficient logical structures that provide faster and more flexible queries. When scanning SST Files, pushed-down filters applied to a respective Tag's inverted index, determine the final Row Groups to be indexed and scanned, further bolstering the speed and efficiency of data retrieval processes. ## Index Format The Inverted Index for each SST file follows the format shown below: ``` inverted_index₀ inverted_index₁ ... inverted_indexₙ footer ``` The structure inside each Inverted Index is as followed: ``` bitmap₀ bitmap₁ bitmap₂ ... bitmapₙ null_bitmap fst ``` The format is encapsulated by a footer: ``` footer_payload footer_payload_size ``` The `footer_payload` is presented in protobuf encoding of `InvertedIndexFooter`. The complete format is containerized in [Puffin](https://iceberg.apache.org/puffin-spec/) with the type defined as `greptime-inverted-index-v1`. ## Protobuf Details The `InvertedIndexFooter` is defined in the following protobuf structure: ```protobuf message InvertedIndexFooter { repeated InvertedIndexMeta metas; } message InvertedIndexMeta { string name; uint64 row_count_in_group; uint64 fst_offset; uint64 fst_size; uint64 null_bitmap_offset; uint64 null_bitmap_size; InvertedIndexStats stats; } message InvertedIndexStats { uint64 null_count; uint64 distinct_count; bytes min_value; bytes max_value; } ``` ## Bitmap Bitmaps are used to represent indices of fixed-size groups. Rows are divided into groups of a fixed size, defined in the `InvertedIndexMeta` as `row_count_in_group`. For example, when `row_count_in_group` is `4096`, it means each group has `4096` rows. If there are a total of `10000` rows, there will be `3` groups in total. The first two groups will have `4096` rows each, and the last group will have `1808` rows. If the indexed values are found in row `200` and `9000`, they will correspond to groups `0` and `2`, respectively. Therefore, the bitmap should show `0` and `2`. Bitmap is implemented using [BitVec](https://docs.rs/bitvec/latest/bitvec/), selected due to its efficient representation of dense data arrays typical of indices of groups. ## Finite State Transducer (FST) [FST](https://docs.rs/fst/latest/fst/) is a highly efficient data structure ideal for in-memory indexing. It represents ordered sets or maps where the keys are bytes. The choice of the FST effectively balances the need for performance, space efficiency, and the ability to perform complex analyses such as regular expression matching. The conventional usage of FST and `u64` values has been adapted to facilitate indirect indexing to row groups. As the row groups are represented as Bitmaps, we utilize the `u64` values split into bitmap's offset (higher 32 bits) and size (lower 32 bits) to represent the location of these Bitmaps. ## API Design Two APIs `InvertedIndexBuilder` for building indexes and `InvertedIndexSearcher` for querying indexes are designed: ```rust type Bytes = Vec; type GroupId = u64; trait InvertedIndexBuilder { fn add(&mut self, name: &str, value: Option<&Bytes>, group_id: GroupId) -> Result<()>; fn finish(&mut self) -> Result<()>; } enum Predicate { Gt(Bytes), GtEq(Bytes), Lt(Bytes), LtEq(Bytes), InList(Vec), RegexMatch(String), } trait InvertedIndexSearcher { fn search(&mut self, name: &str, predicates: &[Predicate]) -> Result>; } ``` --- ### Rfcs/2023 11 07 Region Migration --- Feature Name: Region Migration Procedure Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/2700 Date: 2023-11-03 Author: "Xu Wenkang " --- # Summary This RFC proposes a way that brings the ability of Meta Server to move regions between the Datanodes. # Motivation Typically, We need this ability in the following scenarios: - Migrate hot-spot Regions to idle Datanode - Move the failure Regions to an available Datanode # Details ``` /* Detailed source-code truncated for AI context efficiency. */ ``` **Only the red nodes will persist state after it has succeeded**, and other nodes won't persist state. (excluding the Start and End nodes). ## Steps **The persistent context:** It's shared in each step and available after recovering. It will only be updated/stored after the Red node has succeeded. Values: - `region_id`: The target leader region. - `peer`: The target datanode. - `close_old_leader`: Indicates whether close the region. - `leader_may_unreachable`: It's used to support the failover procedure. **The Volatile context:** It's shared in each step and available in executing (including retrying). It will be dropped if the procedure runner crashes. ### Select Candidate The Persistent state: Selected Candidate Region. ### Update Metadata(Down) **The Persistent context:** - The (latest/updated) `version` of `TableRouteValue`, It will be used in the step of `Update Metadata(Up)`. ### Downgrade Leader This step sends an instruction via heartbeat and performs: 1. Downgrades leader region. 2. Retrieves the `last_entry_id` (if available). If the target leader region is not found: - Sets `close_old_leader` to true. - Sets `leader_may_unreachable` to true. If the target Datanode is unreachable: - Waits for region lease expired. - Sets `close_old_leader` to true. - Sets `leader_may_unreachable` to true. **The Persistent context:** None **The Persistent state:** - `last_entry_id` *Passes to next step. ### Upgrade Candidate This step sends an instruction via heartbeat and performs: 1. Replays the WAL to latest(`last_entry_id`). 2. Upgrades the candidate region. If the target region is not found: - Rollbacks. - Notifies the failover detector if `leader_may_unreachable` == true. - Exits procedure. If the target Datanode is unreachable: - Rollbacks. - Notifies the failover detector if `leader_may_unreachable` == true. - Exits procedure. **The Persistent context:** None ### Update Metadata(Up) This step performs 1. Switches Leader. 2. Removes Old Leader(Opt.). 3. Moves Old Leader to follower(Opt.). The `TableRouteValue` version should equal the `TableRouteValue`'s `version` in Persistent context. Otherwise, verifies whether `TableRouteValue` already updated. **The Persistent context:** None ### Close Old Leader(Opt.) This step sends a close region instruction via heartbeat. If the target leader region is not found: - Ignore. If the target Datanode is unreachable: - Ignore. ### Open Candidate(Opt.) This step sends an open region instruction via heartbeat and waits for conditions to be met (typically, the condition is that the `last_entry_id` of the Candidate Region is very close to that of the Leader Region or the latest). If the target Datanode is unreachable: - Exits procedure. --- ### Rfcs/2023 12 22 Enclose Column Id --- Feature Name: Enclose Column Id Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/2982 Date: 2023-12-22 Author: "Ruihang Xia " --- # Summary This RFC proposes to enclose the usage of `ColumnId` into the region engine only. # Motivation `ColumnId` is an identifier for columns. It's assigned by meta server, stored in `TableInfo` and `RegionMetadata` and used in region engine to distinguish columns. At present, Both Frontend, Datanode and Metasrv are aware of `ColumnId` but it's only used in region engine. Thus this RFC proposes to remove it from Frontend (mainly used in `TableInfo`) and Metasrv. # Details `ColumnId` is used widely on both read and write paths. Removing it from Frontend and Metasrv implies several things: - A column may have different column id in different regions. - A column is identified by its name in all components. - Column order in the region engine is not restricted, i.e., no need to be in the same order with table info. The first thing doesn't matter IMO. This concept doesn't exist anymore outside of region server, and each region is autonomous and independent -- the only guarantee it should hold is those columns exist. But if we consider region repartition, where the SST file would be re-assign to different regions, things would become a bit more complicated. A possible solution is store the relation between name and ColumnId in the manifest, but it's out of the scope of this RFC. We can likely give a workaround by introducing a indirection mapping layer of different version of partitions. And more importantly, we can still assume columns have the same column ids across regions. We have procedure to maintain consistency between regions and the region engine should ensure alterations are idempotent. So it is possible that region repartition doesn't need to consider column ids or other region metadata in the future. Users write and query column by their names, not by ColumnId or something else. The second point also means to change the column reference in ScanRequest from index to name. This change can hugely alleviate the misuse of the column index, which has given us many surprises. And for the last one, column order only matters in table info. This order is used in user-faced table structure operation, like add column, describe column or as the default order of INSERT clause. None of them is connected with the order in storage. # Drawback Firstly, this is a breaking change. Delivering this change requires a full upgrade of the cluster. Secondly, this change may introduce some performance regression. For example, we have to pass the full table name in the `ScanRequest` instead of the `ColumnId`. But this influence is very limited, since the column index is only used in the region engine. # Alternatives There are two alternatives from the perspective of "what can be used as the column identifier": - Index of column to the table schema - `ColumnId` of that column The first one is what we are using now. By choosing this way, it's required to keep the column order in the region engine the same as the table info. This is not hard to achieve, but it's a bit annoying. And things become tricky when there is internal column or different schemas like those stored in file format. And this is the initial purpose of this RFC, which is trying to decouple the table schema and region schema. The second one, in other hand, requires the `ColumnId` should be identical in all regions and `TableInfo`. It has the same drawback with the previous alternative, that the `TableInfo` and `RegionMetadata` are tighted together. Another point is that the `ColumnId` is assigned by the Metasrv, who doesn't need it but have to maintain it. And this also limits the functionality of `ColumnId`, by taking the ability of assigning it from concrete region engine. --- ### Rfcs/2024 01 17 Dataflow Framework --- Feature Name: Dataflow Framework Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/3187 Date: 2024-01-17 Author: "Discord9 " --- # Summary This RFC proposes a Lightweight Module for executing continuous aggregation queries on a stream of data. # Motivation Being able to do continuous aggregation is a very powerful tool. It allows you to do things like: 1. downsample data from i.e. 1 milliseconds to 1 second 2. calculate the average of a stream of data 3. Keeping a sliding window of data in memory In order to do those things while maintaining a low memory footprint, you need to be able to manage the data in a smart way. Hence, we only store necessary data in memory, and send/recv data deltas to/from the client. # Details ## System boundary / What it's and isn't - GreptimeFlow provides a way to perform continuous aggregation over time-series data. - It's not a complete streaming-processing system. Only a must subset functionalities are provided. - Flow can process a configured range of fresh data. Data exceeding this range will be dropped directly. Thus it cannot handle random datasets (random on timestamp). - Both sliding windows (e.g., latest 5m from present) and fixed windows (every 5m from some time) are supported. And these two are the major targeting scenarios. - Flow can handle most aggregate operators within one table(i.e. Sum, avg, min, max and comparison operators). But others (join, trigger, txn etc.) are not the target feature. ## Framework - Greptime Flow's is built on top of [Hydroflow](https://github.com/hydro-project/hydroflow). - We have three choices for the Dataflow/Streaming process framework for our simple continuous aggregation feature: 1. Based on the timely/differential dataflow crate that [materialize](https://github.com/MaterializeInc/materialize) based on. Later, it's proved too obscure for a simple usage, and is hard to customize memory usage control. 2. Based on a simple dataflow framework that we write from ground up, like what [arroyo](https://www.arroyo.dev/) or [risingwave](https://www.risingwave.dev/) did, for example the core streaming logic of [arroyo](https://github.com/ArroyoSystems/arroyo/blob/master/crates/arroyo-datastream/src/lib.rs) only takes up to 2000 line of codes. However, it means maintaining another layer of dataflow framework, which might seem easy in the beginning, but I fear it might be too burdensome to maintain once we need more features. 3. Based on a simple and lower level dataflow framework that someone else write, like [hydroflow](https://github.com/hydro-project/hydroflow), this approach combines the best of both worlds. Firstly, it boasts ease of comprehension and customization. Secondly, the dataflow framework offers precisely the necessary features for crafting uncomplicated single-node dataflow programs while delivering decent performance. Hence, we choose the third option, and use a simple logical plan that's anagonistic to the underlying dataflow framework, as it only describe how the dataflow graph should be doing, not how it do that. And we built operator in hydroflow to execute the plan. And the result hydroflow graph is wrapped in a engine that only support data in/out and tick event to flush and compute the result. This provide a thin middle layer that's easy to maintain and allow switching to other dataflow framework if necessary. ## Deploy mode and protocol - Greptime Flow is an independent streaming compute component. It can be used either within a standalone node or as a dedicated node at the same level as frontend in distributed mode. - It accepts insert request Rows, which is used between frontend and datanode. - New flow job is submitted in the format of modified SQL query like snowflake do, like: `CREATE TASK avg_over_5m WINDOW_SIZE = "5m" AS SELECT avg(value) FROM table WHERE time > now() - 5m GROUP BY time(1m)`. Flow job then got stored in Metasrv. - It also persists results in the format of Rows to frontend. - The query plan uses Substrait as codec format. It's the same with GreptimeDB's query engine. - Greptime Flow needs a WAL for recovering. It's possible to reuse datanode's. The workflow is shown in the following diagram ```mermaid graph TB subgraph Flownode["Flownode"] subgraph Dataflows df1("Dataflow_1") df2("Dataflow_2") end end subgraph Frontend["Frontend"] newLines["Mirror Insert Create Task From Query Write result from flow node"] end subgraph Datanode["Datanode"] end User --> Frontend Frontend -->|Register Task| Metasrv Metasrv -->|Read Task Metadata| Frontend Frontend -->|Create Task| Flownode Frontend -->|Mirror Insert| Flownode Flownode -->|Write back| Frontend Frontend --> Datanode Datanode --> Frontend ``` ## Lifecycle of data - New data is inserted into frontend like before. Frontend will mirror insert request to Flow node if there is configured flow job. - Depending on the timestamp of incoming data, flow will either drop it (outdated data) or process it (fresh data). - Greptime Flow will periodically write results back to the result table through frontend. - Those result will then be written into a result table stored in datanode. - A small table of intermediate state is kept in memory, which is used to calculate the result. ## Supported operations - Greptime Flow accepts a configurable "materialize window", data point exceeds that time window is discarded. - Data within that "materialize window" is queryable and updateable. - Greptime Flow can handle partitioning, if and only if the input query can be transformed to a fully partitioned plan according to the existing commutative rules. Otherwise the corresponding flow job has to be calculated in a single node. - Notice that Greptime Flow has to see all the data belongs to one partition. - Deletion and duplicate insertion are not supported at early stage. ## Miscellaneous - Greptime Flow can translate SQL to it's own plan, however only a selected few aggregate function is supported for now, like min/max/sum/count/avg - Greptime Flow's operator is configurable in terms of the size of the materialize window, whether to allow delay of incoming data etc., so simplest operator can choose to not tolerate any delay to save memory. # Future Work - Support UDF that can do one-to-one mapping. Preferably, we can reuse the UDF mechanism in GreptimeDB. - Support join operator. - Design syntax for config operator for different materialize window and delay tolerance. - Support cross partition merge operator that allows complex query plan that not necessary accord with partitioning rule to communicate between nodes and create final materialize result. - Duplicate insertion, which can be reverted easily within the current framework, so supporting it could be easy - Deletion within "materialize window", this requires operators like min/max to store all inputs within materialize window, which might require further optimization. --- ### Rfcs/2024 08 06 Json Datatype --- Feature Name: Json Datatype Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/4230 Date: 2024-8-6 Author: "Yuhan Wang " --- # Summary This RFC proposes a method for storing and querying JSON data in the database. # Motivation JSON is widely used across various scenarios. Direct support for writing and querying JSON can significantly enhance the database's flexibility. # Details ## Storage and Query GreptimeDB's type system is built on Arrow/DataFusion, where each data type in GreptimeDB corresponds to a data type in Arrow/DataFusion. The proposed JSON type will be implemented on top of the existing `Binary` type, leveraging the current `datatype::value::Value` and `datatype::vectors::BinaryVector` implementations, utilizing the JSONB format as the encoding of JSON data. JSON data is stored and processed similarly to binary data within the storage layer and query engine. This approach brings problems when dealing with insertions and queries of JSON columns. ## Insertion Users commonly write JSON data as strings. Thus we need to make conversions between string and JSONB. There are 2 ways to do this: 1. MySQL and PostgreSQL servers provide auto-conversions between strings and JSONB. When a string is inserted into a JSON column, the server will try to parse the string as JSON and convert it to JSONB. The non-JSON strings will be rejected. 2. A function `parse_json` is provided to convert string to JSONB. If the string is not a valid JSON string, the function will return an error. For example, in MySQL client: ```SQL CREATE TABLE IF NOT EXISTS test ( ts TIMESTAMP TIME INDEX, a INT, b JSON ); INSERT INTO test VALUES( 0, 0, '{ "name": "jHl2oDDnPc1i2OzlP5Y", "timestamp": "2024-07-25T04:33:11.369386Z", "attributes": { "event_attributes": 48.28667 } }' ); INSERT INTO test VALUES( 0, 0, parse_json('{ "name": "jHl2oDDnPc1i2OzlP5Y", "timestamp": "2024-07-25T04:33:11.369386Z", "attributes": { "event_attributes": 48.28667 } }') ); ``` Are both valid. The dataflow of the insertion process is as follows: ``` Insert JSON strings directly through client: Parse Insert String(Serialized JSON)┌──────────┐Arrow Binary(JSONB)┌──────┐Arrow Binary(JSONB) Client ---------------------->│ Server │------------------>│ Mito │------------------> Storage └──────────┘ └──────┘ (Server identifies JSON type and performs auto-conversion) Insert JSON strings through parse_json function: Parse Insert String(Serialized JSON)┌──────────┐String(Serialized JSON)┌─────┐Arrow Binary(JSONB)┌──────┐Arrow Binary(JSONB) Client ---------------------->│ Server │---------------------->│ UDF │------------------>│ Mito │------------------> Storage └──────────┘ └─────┘ └──────┘ (Conversion is performed by UDF inside Query Engine) ``` Servers identify JSON column through column schema and perform auto-conversions. But when using prepared statements and binding parameters, the corresponding cached plans in datafusion generated by prepared statements cannot identify JSON columns. Under this circumstance, the servers identify JSON columns through the given parameters and perform auto-conversions. The following is an example of inserting JSON data through prepared statements: ```Rust sqlx::query( "create table test(ts timestamp time index, j json)", ) .execute(&pool) .await .unwrap(); let json = serde_json::json!({ "code": 200, "success": true, "payload": { "features": [ "serde", "json" ], "homepage": null } }); // Valid, can identify serde_json::Value as JSON type sqlx::query("insert into test values($1, $2)") .bind(i) .bind(json) .execute(&pool) .await .unwrap(); // Invalid, cannot identify String as JSON type sqlx::query("insert into test values($1, $2)") .bind(i) .bind(json.to_string()) .execute(&pool) .await .unwrap(); ``` ## Query Correspondingly, users prefer to display JSON data as strings. Thus we need to make conversions between JSON data and strings before presenting JSON data. There are also 2 ways to do this: auto-conversions on MySQL and PostgreSQL servers, and function `json_to_string`. For example, in MySQL client: ```SQL SELECT b FROM test; SELECT json_to_string(b) FROM test; ``` Will both return the JSON as human-readable strings. Specifically, to perform auto-conversions, we attach a message to JSON data in the `metadata` of `Field` in Arrow/Datafusion schema when scanning a JSON column. Frontend servers could identify JSON data and convert it to strings. The dataflow of the query process is as follows: ``` Query directly through client: Decode Scan String(Serialized JSON)┌──────────┐Arrow Binary(JSONB)┌──────────────┐Arrow Binary(JSONB) Client <----------------------│ Server │<------------------│ Query Engine │<----------------- Storage └──────────┘ └──────────────┘ (Server identifies JSON type and performs auto-conversion based on column metadata) Query through json_to_string function: Scan & Decode String(Serialized JSON)┌──────────┐String(Serialized JSON)┌──────────────┐Arrow Binary(JSONB) Client <----------------------│ Server │<----------------------│ Query Engine │<----------------- Storage └──────────┘ └──────────────┘ (Conversion is performed by UDF inside Query Engine) ``` However, if a function uses JSON type as its return type, the metadata method mentioned above is not applicable. Thus the functions of JSON type should specify the return type explicitly instead of returning a JSON type, such as `json_get_int` and `json_get_float` which return corresponding data of `INT` and `FLOAT` type respectively. ## Functions Similar to the common JSON type, JSON data can be queried with functions. For example: ```SQL CREATE TABLE IF NOT EXISTS test ( ts TIMESTAMP TIME INDEX, a INT, b JSON ); INSERT INTO test VALUES( 0, 0, '{ "name": "jHl2oDDnPc1i2OzlP5Y", "timestamp": "2024-07-25T04:33:11.369386Z", "attributes": { "event_attributes": 48.28667 } }' ); SELECT json_get_string(b, 'name') FROM test; +---------------------+ | b.name | +---------------------+ | jHl2oDDnPc1i2OzlP5Y | +---------------------+ SELECT json_get_float(b, 'attributes.event_attributes') FROM test; +--------------------------------+ | b.attributes.event_attributes | +--------------------------------+ | 48.28667 | +--------------------------------+ ``` And more functions can be added in the future. # Drawbacks As a general purpose JSON data type, JSONB may not be as efficient as specialized data types for specific scenarios. The auto-conversion mechanism is not supported in all scenarios. We need to find workarounds for these scenarios. # Alternatives Extract and flatten JSON schema to store in a structured format through pipeline. For nested data, we can provide nested types like `STRUCT` or `ARRAY`. --- ### Rfcs/2025 02 06 Remote Wal Purge --- Feature Name: Remote WAL Purge Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/5474 Date: 2025-02-06 Author: "Yuhan Wang " --- # Summary This RFC proposes a method for purging remote WAL in the database. # Motivation Currently only local wal entries are purged when flushing, while remote wal does nothing. # Details ```mermaid sequenceDiagram Region0->>Kafka: Last entry id of the topic in use Region0->>WALPruner: Heartbeat with last entry id WALPruner->>+WALPruner: Time Loop WALPruner->>+ProcedureManager: Submit purge procedure ProcedureManager->>Region0: Flush request ProcedureManager->>Kafka: Prune WAL entries Region0->>Region0: Flush ``` ## Steps ### Before purge Before purging remote WAL, metasrv needs to know: 1. `last_entry_id` of each region. 2. `kafka_topic_last_entry_id` which is the last entry id of the topic in use. Can be lazily updated and needed when region has empty memtable. 3. Kafka topics that each region uses. The states are maintained through: 1. Heartbeat: Datanode sends `last_entry_id` to metasrv in heartbeat. As for regions with empty memtable, `last_entry_id` should equals to `kafka_topic_last_entry_id`. 2. Metasrv maintains a topic-region map to know which region uses which topic. `kafka_topic_last_entry_id` will be maintained by the region itself. Region will update the value after `k` heartbeats if the memtable is empty. ### Purge procedure We can better handle locks utilizing current procedure. It's quite similar to the region migration procedure. After a period of time, metasrv will submit a purge procedure to ProcedureManager. The purge will apply to all topics. The procedure is divided into following stages: 1. Preparation: - Retrieve `last_entry_id` of each region kvbackend. - Choose regions that have a relatively small `last_entry_id` as candidate regions, which means we need to send a flush request to these regions. 2. Communication: - Send flush requests to candidate regions. 3. Purge: - Choose proper entry id to delete for each topic. The entry should be the smallest `last_entry_id - 1` among all regions. - Delete legacy entries in Kafka. - Store the `last_purged_entry_id` in kvbackend. It should be locked to prevent other regions from replaying the purged entries. ### After purge After purge, there may be some regions that have `last_entry_id` smaller than the entry we just deleted. It's legal since we only delete the entries that are not needed anymore. When restarting a region, it should query the `last_purged_entry_id` from metasrv and replay from `min(last_entry_id, last_purged_entry_id)`. ### Error handling No persisted states are needed since all states are maintained in kvbackend. Retry when failed to retrieving metadata from kvbackend. # Alternatives Purge time can depend on the size of the WAL entries instead of a fixed period of time, which may be more efficient. --- ### Rfcs/2025 06 20 Repartition --- Feature Name: Repartition Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/6558 Date: 2025-06-20 Author: "Ruihang Xia " --- # Summary This RFC proposes a method for repartitioning a table, to adjust the partition rule and data distribution. # Motivation With time passing, the data distribution and skew pattern of a table might change. We need a way to repartition the table to suit the new pattern. # Details Here is a rough workflow diagram of the entire repartition process, each step is described in detail below. ```mermaid sequenceDiagram participant Frontend participant Metasrv participant Datanodes participant Region0 as Region 0 Frontend->>Frontend: Process request, validation etc. Frontend->>Metasrv: Submit procedure Metasrv->>Metasrv: Compute diff and generate migration plan Metasrv->>Metasrv: Allocate necessary region resources (with Paas) Metasrv->>Datanodes: Stop compaction and snapshot rect rgb(255, 225, 225) note over Frontend, Region0: No Ingestion Period Metasrv->>Frontend: Stop processing write requests Metasrv->>Metasrv: Update metadata Metasrv->>Frontend: Start processing read requests end Metasrv->>Datanodes: Update region rule, stage version changes from now on Region0->>Region0: Compute new manifests for all regions Region0->>Datanodes: Submit manifest changes Metasrv->>Datanodes: Recover compaction and snapshot, make staged changes visible note over Frontend, Datanodes: Reload Cache Metasrv->>Metasrv: Release resources (with Paas) Metasrv->>Metasrv: Schedule optional compaction (to remote compactor) ``` ## Preprocessing This phase is for static analysis of the new partition rule. The server can know whether the repartitioning is possible, how to do the repartitioning, and how much resources are needed. In theory, the input and output partition rules for repartitioning can be completely unrelated. But in practice, to avoid a very large change set, we'll only allow two simple kinds of change. One splits one region into two regions (region split) and another merges two regions into one (region merge). After validating the new partition rule using the same validation logic as table creation, we compute the difference between the old and new partition rules. The resulting diff may contain several independent groups of changes. During subsequent processing, each group of changes can be handled independently and can succeed or fail without affecting other groups or creating non-idempotently retryable scenarios. Next, we generate a repartition plan for each group of changes. Each plan contains this information for all regions involved in that particular plan. And one target region will only be referenced by a single plan. With those plans, we can determine the resource requirements for the repartition operation, where resources here primarily refer to Regions. Metasrv will coordinate with PaaS layer to pre-allocate the necessary regions at this stage. These new regions start completely empty, and their metadata and manifests will be populated during subsequent modification steps. ## Data Processing This phase is primarily for region's change, including region's metadata (route table and the corresponding rule) and manifest. Once we start processing one plan through a procedure, we'll first stop the region's compaction and snapshot. This is to avoid any states being removed due to compaction (which may removes old SST files) and snapshot (which may removes old manifest files). Metasrv will trying to update the metadata of partition, or the region route table (related to `PartitionRuleManager`). This step is in the "no ingestion" scope, so no new data will be ingested. Since this won't take much time, the affection to the cluster is minimized. Metasrv will also update the region rule to corresponding regions on Datanodes. Every regions and all the ingestion requests to the region server will have a version of region rule, to identify under which rule the request is processed. The version can be something like `hash(region_rule)`. Once the region rule on region server is updated, all ingestion request with old rule will be rejected, and all requests with new rule will be accepted but not visible. They can still be flushed to persisted storage, but their version change (new manifest) will be staged. Then region 0 (or let metasrv to pick any operational region) will compute the new manifests for all target regions. This step is done by first reading all old manifests, and remapping the files with new partition rule, to get the content of new manifests. Notice this step only handles the manifests before region rule change on region server, and won't touch those staged manifests, as they are already with the new rule. Those new manifest will be submitted to the corresponding target regions by region 0 via a `RegionEdit` request. If this request falls after a few retries, region 0 will try to rollback this change by directly overwriting the manifest on object storage. and report this failure to metasrv and let the entire repartition procedure to fail. And we can also optionally compute the new manifest for those staged version changes (like another repartition) and submit them to the target regions to make the also visible even if the repartition fails. In the other hand, a successful `RegionEdit` request also acknowledges those staged version changes and make them visible. After this step, the repartition is done in the data plane. We can start to process compaction and snapshot again. ## Postprocessing After the main processing is done, we can do some extra postprocessing to reduce the performance impact of repartition. Including reloading caches in frontend's route table, metasrv's kv cache and datanode's read/write/page cache etc. We can also schedule an optional compaction to reorganize all the data file under the new partition rule to reduce potential fragmentation or read amplification. ## Procedure Here describe the repartition procedure step by step: - Validating repartition request - Initialize the repartition procedure - Calculate rule diff and repartition plan group - Allocate necessary new regions - Lock the table key - For each repartition subprocedure - Stop compaction and snapshot - Forbid new ingestion requests, update metadata, allow ingestion requests. - Update region rule to regions - Pick one region to calculate new manifest for all regions in this repartition group - Let that region to apply new manifest to each region via `RegionEdit` - If failed after some retries, revert this manifest change to other succeeded regions and mark this failure. - If all succeeded, acknowledge those staged version changes and make them visible. - Return result - Collect results from subprocedure. - For those who failed, we need to restart those regions to force reconstruct their status from manifests - For those who succeeded, collect and merge their rule diff - Unlock the table key - Report the result to user. - Reload cache - Maybe trigger a special compaction In addition of sequential step, rollback is also an important part of this procedure. There are three steps can be rolled back when unrecoverable failure occurs. If the metadata update is not committed, we can overwrite the metadata to previous version. This step is scoped in the "no ingestion" period, so no new data will be ingested and the status of both datanode and metasrv will be consistent. If the `RegionEdit` to other regions is not acknowledged, or partial acknowledged, we can directly overwrite the manifest on object storage from the central region (who computes the new manifest), and force region server to reload corresponding region to load its state from object storage to recover. If the staged version changes are not acknowledged, we can re-compute manifest based on old rule for staged data, and apply them directly like above. This is like another smaller repartition for those staged data. ## Region rule validation and diff calculation In the current codebase, the rule checker is not complete. It can't check uniqueness and completeness of the rule. This RFC also propose a new way to validate the rule. The proposed validation way is based on a check-point system, which first generates a group of check-points from the rule, and then check if all the point is covered and only covered by one rule. All the partition rule expressionis limited to be the form of ` `, and the operator is limited to be comparison operators. Those expressions are allowed to be nested with `AND` and `OR` operators. Based on this, we can first extract all the unique values on each column, adding and subtracting a little epsilon to cover its left and right boundary. Since we accept integer, float and string as the value type, compute on them directly is not convenient. So we'll first normalize them to a common type and only need to preserve the relative partial ordering. This also avoids the problem of "what is next/previous value" of string and "what's a good precision" for float. After normalization, we get a set of scatter points for each column. Then we can generate a set of check-points by combining all the scatter points like building a cartesian product. This might bring a large number of check-points, so we can do an prune optimization to remove some of them by merging some of the expression zones. Those expressions who have identical N-1 edge sub-expressions with one adjacent edge can be merged together. This prune check is with a time complexity of O(N * M * log(M)), where N is the number of active dimensions and M is the number of expression zones. Diff calculation is also done by finding different expression zones between the old and new rule set, and check if we can transform one to another by merging some of the expression zones. The step to validate the check-points set against expressions can be treated as a tiny expression of `PhysicalExpr`. This evaluation will give a boolean matrix of K*M shape, where K is the number of check-points. We then check in each row of the matrix, if there is one and only one true value. ## Compute and use new manifest We can generate a new set of manifest file based on old manifest and two versions of rule. From abvoe rule processing part, we can tell how a new rule & region is from previous one. So a simple way to get the new manifest is also apply the step of change to manifest files. E.g., if region A is from region B and C, we simply combine all file IDs from B and C to generate the content of A. If necessary, we can do this better by involving some metadata related to data, like min-max statistics of each file, and pre-evaluate over min-max to filter out unneeded files when generating new manifest. The way to use new manifest needs one more extra step based on the current implementation. We'll need to record either in manifest or in file metadata, of what rule is used when generating (flush or compaction) a SST file. Then in every single read request, we need to append the current region rule as predicate to the read request, to ensure no data belong to other regions will be read. We can use the stored region rule to reduce the number of new predicates to apply, by removing the identical predicate between the current region rule and the stored region rule. So ideally in a table that has not been repartitioned recently, the overhead of checking region rule is minimal. ## Pre-required tasks In above steps, we assume some functionalities are implemented. Here list them with where they are used and how to implement them. ### Cross-region read The current data directory structure is `{table_id}/{region_id}/[data/metadata]/{file_id}`, every region can only access files under their own directory. After repartition, data file may be placed in other previous old regions. So we need to support cross-region read. This new access method allows region to access any file under the same table. Related tracking issue is . ### Global GC worker This is to simplify state management of data files. As one file may be referenced in multiple manifests, or no manifest at all. After this, every region and the repartition process only need to care about generateing and using new files, without tracking whether a file should be deleted or not. Leaving the deletion to the global GC worker. This worker basically works by counting reference from manifest file, and remove unused one. Related tracking issue is **TBD**. # Alternatives In the "Data Processing" section, we can enlarge the "no ingestion" period to include almost all the steps. This can simplify the entire procedure by a lot, but will bring a longer time of ingestion pause which may not be acceptable. --- ### Rfcs/2025 07 04 Compatibility Test Framework --- Feature Name: Compatibility Test Framework Tracking Issue: TBD Date: 2025-07-04 Author: "Ruihang Xia " --- # Summary This RFC proposes a compatibility test framework for GreptimeDB to ensure backward/forward compatibility for different versions of GreptimeDB. # Motivation In current practice, we don't have a systematic way to test and ensure the compatibility of different versions of GreptimeDB. Each time we release a new version, we need to manually test the compatibility with ad-hoc cases. This is not only time-consuming, but also prone to errors and unmaintainable. Highly rely on the release manager to ensure the compatibility of different versions of GreptimeDB. We don't have a detailed guide on the release SoP of how to test and ensure the compatibility of the new version. And has broken the compatibility of the new version many times (`v0.14.1` and `v0.15.1` are two examples, which are both released right after the major release). # Details This RFC proposes a compatibility test framework that is easy to maintain, extend and run. It can tell the compatibility between any given two versions of GreptimeDB, both backward and forward. It's based on the Sqlness library but used in a different way. Generally speaking, the framework is composed of two parts: 1. Test cases: A set of test cases that are maintained dedicatedly for the compatibility test. Still in the `.sql` and `.result` format. 2. Test framework: A new sqlness runner that is used to run the test cases. With some new features that is not required by the integration sqlness test. ## Test Cases ### Structure The case set is organized in three parts: - `1.feature`: Use a new feature - `2.verify`: Verify database behavior - `3.cleanup`: Paired with `1.feature`, cleanup the test environment. These three parts are organized in a tree structure, and should be run in sequence: ``` compatibility_test/ ├── 1.feature/ │ ├── feature-a/ │ ├── feature-b/ │ └── feature-c/ ├── 2.verify/ │ ├── verify-metadata/ │ ├── verify-data/ │ └── verify-schema/ └── 3.cleanup/ ├── cleanup-a/ ├── cleanup-b/ └── cleanup-c/ ``` ### Example For example, for a new feature like adding new index option ([#6416](https://github.com/GreptimeTeam/greptimedb/pull/6416)), we (who implement the feature) create a new test case like this: ```sql -- path: compatibility_test/1.feature/index-option/granularity_and_false_positive_rate.sql -- SQLNESS ARG since=0.15.0 -- SQLNESS IGNORE_RESULT CREATE TABLE granularity_and_false_positive_rate (ts timestamp time index, val double) with ("index.granularity" = "8192", "index.false_positive_rate" = "0.01"); ``` And ```sql -- path: compatibility_test/3.cleanup/index-option/granularity_and_false_positive_rate.sql drop table granularity_and_false_positive_rate; ``` Since this new feature don't require some special way to verify the database behavior, we can reuse existing test cases in `2.verify/` to verify the database behavior. For example, we can reuse the `verify-metadata` test case to verify the metadata of the table. ```sql -- path: compatibility_test/2.verify/verify-metadata/show-create-table.sql -- SQLNESS TEMPLATE TABLE="SHOW TABLES"; SHOW CREATE TABLE $TABLE; ``` In this example, we use some new sqlness features that will be introduced in the next section (`since`, `IGNORE_RESULT`, `TEMPLATE`). ### Maintenance Each time implement a new feature that should be covered by the compatibility test, we should create a new test case in `1.feature/` and `3.cleanup/` for them. And check if existing cases in `2.verify/` can be reused to verify the database behavior. This simulates an enthusiastic user who uses all the new features at the first time. All the new Maintenance burden is on the feature implementer to write one more test case for the new feature, to "fixation" the behavior. And once there is a breaking change in the future, it can be detected by the compatibility test framework automatically. Another topic is about deprecation. If a feature is deprecated, we should also mark it in the test case. Still use above example, assume we deprecate the `index.granularity` and `index.false_positive_rate` index options in `v0.99.0`, we can mark them as: ```sql -- SQLNESS ARG since=0.15.0 till=0.99.0 ... ``` This tells the framework to ignore this feature in version `v0.99.0` and later. Currently, we have so many experimental features that are scheduled to be broken in the future, this is a good way to mark them. ## Test Framework This section is about new sqlness features required by this framework. ### Since and Till Follows the `ARG` interceptor in sqlness, we can mark a feature is available between two given versions. Only the `since` is required: ```sql -- SQLNESS ARG since=VERSION_STRING [till=VERSION_STRING] ``` ### IGNORE_RESULT `IGNORE_RESULT` is a new interceptor, it tells the runner to ignore the result of the query, only check whether the query is executed successfully. This is useful to reduce the Maintenance burden of the test cases, unlike the integration sqlness test, in most cases we don't care about the result of the query, only need to make sure the query is executed successfully. ### TEMPLATE `TEMPLATE` is another new interceptor, it can generate queries from a template based on a runtime data. In above example, we need to run the `SHOW CREATE TABLE` query for all existing tables, so we can use the `TEMPLATE` interceptor to generate the query with a dynamic table list. ### RUNNER There are also some extra requirement for the runner itself: - It should run the test cases in sequence, first `1.feature/`, then `2.verify/`, and finally `3.cleanup/`. - It should be able to fetch required version automatically to finish the test. - It should handle the `since` and `till` properly. On the `1.feature` phase, the runner needs to identify all features need to be tested by version number. And then restart with a new version (the `to` version) to run `2.verify/` and `3.cleanup/` phase. ## Test Report Finally, we can run the compatibility test to verify the compatibility between any given two versions of GreptimeDB, for example: ```bash # check backward compatibility between v0.15.0 and v0.16.0 when releasing v0.16.0 ./sqlness run --from=0.15.0 --to=0.16.0 # check forward compatibility when downgrading from v0.15.0 to v0.13.0 ./sqlness run --from=0.15.0 --to=0.13.0 ``` We can also use a script to run the compatibility test for all the versions in a given range to give a quick report with all versions we need. And we always bump the version in `Cargo.toml` to the next major release version, so the next major release version can be used as "latest" unpublished version for scenarios like local testing. # Alternatives There was a previous attempt to implement a compatibility test framework that was disabled due to some reasons [#3728](https://github.com/GreptimeTeam/greptimedb/issues/3728). --- ### Rfcs/2025 07 23 Global Gc Worker --- Feature Name: "global-gc-worker" Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/6571 Date: 2025-07-23 Author: "discord9 " --- # Global GC Worker ## Summary This RFC proposes the integration of a garbage collection (GC) mechanism within the Compaction process. This mechanism aims to manage and remove stale files that are no longer actively used by any system component, thereby reclaiming storage space. ## Motivation With the introduction of features such as table repartitioning, a substantial number of Parquet files can become obsolete. Furthermore, failures during manifest updates may result in orphaned files that are never referenced by the system. Therefore, a periodic garbage collection mechanism is essential to reclaim storage space by systematically removing these unused files. ## Details ### Overview The garbage collection process will be integrated directly into the Compaction process. Upon the completion of a Compaction for a given region, the GC worker will be automatically triggered. Its primary function will be to identify and subsequently delete obsolete files that have persisted beyond their designated retention period. This integration ensures that garbage collection is performed in close conjunction with data lifecycle management, effectively leveraging the compaction process's inherent knowledge of file states. This design prioritizes correctness and safety by explicitly linking GC execution to a well-defined operational boundary: the successful completion of a compaction cycle. ### Terminology - **Unused File**: Refers to a file present in the storage directory that has never been formally recorded in any manifest. A common scenario for this includes cases where a new SST file is successfully written to storage, but the subsequent update to the manifest fails, leaving the file unreferenced. - **Obsolete File**: Denotes a file that was previously recorded in a manifest but has since been explicitly marked for removal. This typically occurs following operations such as data repartitioning or compaction. ### GC Worker Process The GC worker operates as an integral part of the Compaction process. Once a Compaction for a specific region is completed, the GC worker is automatically triggered. Executing this process on a `datanode` is preferred to eliminate the overhead associated with having to set object storage configurations in the `metasrv`. The detailed process is as follows: 1. **Invocation**: Upon the successful completion of a Compaction for a region, the GC worker is invoked. 2. **Manifest Reading**: The worker reads the region's primary manifest to obtain a comprehensive list of all files marked as obsolete. Concurrently, it reads any temporary manifests generated by long-running queries to identify files that are currently in active use, thereby preventing their premature deletion. 3. **Lingering Time Check (Obsolete Files)**: For each identified obsolete file, the GC worker evaluates its "lingering time." Which is the time passed after it had been removed from manifest. 4. **Deletion Marking (Obsolete Files)**: Files that have exceeded their maximum configurable lingering time and are not referenced by any active temporary manifests are marked for deletion. 5. **Lingering Time (Unused Files)**: Unused files (those never recorded in any manifest) are also subject to a configurable maximum lingering time before they are eligible for deletion. Following flowchart illustrates the GC worker's process: ```mermaid flowchart TD A[Compaction Completed] --> B[Trigger GC Worker] B --> C[Scan Region Manifest] C --> D[Identify File Types] D --> E[Unused Files
Never recorded in manifest] D --> F[Obsolete Files
Previously in manifest
but marked for removal] E --> G[Check Lingering Time] F --> G G --> H{File exceeds
configured lingering time?} H -->|No| I[Skip deletion] H -->|Yes| J[Check Temporary Manifest] J --> K{File in use by
active queries?} K -->|Yes| L[Retain file
Wait for next GC cycle] K -->|No| M[Safely delete file] I --> N[End GC cycle] L --> N M --> O[Update Manifest] O --> N N --> P[Wait for next Compaction] P --> A style A fill:#e1f5fe style B fill:#f3e5f5 style M fill:#e8f5e8 style L fill:#fff3e0 ``` #### Handling Obsolete Files An obsolete file is permanently deleted only if two conditions are met: 1. The time elapsed since its removal from the manifest (its obsolescence timestamp) exceeds a configurable threshold. 2. It is not currently referenced by any active temporary manifests. #### Handling Unused Files With the integration of the GC worker into the Compaction process, the risk of accidentally deleting newly created SST files that have not yet been recorded in the manifest is significantly mitigated. Consequently, the concept of "Unused Files" as a distinct category primarily susceptible to accidental deletion is largely resolved. Any files that are genuinely "unused" (i.e., never referenced by any manifest, including temporary ones) can be safely deleted after a configurable maximum lingering time. For debugging and auditing purposes, a comprehensive list of recently deleted files can be maintained. ### Ensuring Read Consistency To prevent the GC worker from inadvertently deleting files that are actively being utilized by long-running analytical queries, a robust protection mechanism is introduced. This mechanism relies on temporary manifests that are actively kept "alive" by the queries using them. When a long-running query is detected (e.g., by a slow query recorder), it will write a temporary manifest to the region's manifest directory. This manifest lists all files required for the query. However, simply creating this file is not enough, as a query runner might crash, leaving the temporary manifest orphaned and preventing garbage collection indefinitely. To address this, the following "heartbeat" mechanism is implemented: 1. **Periodic Updates**: The process executing the long-running query is responsible for periodically updating the modification timestamp of its temporary manifest file (i.e., "touching" the file). This serves as a heartbeat, signaling that the query is still active. 2. **GC Worker Verification**: When the GC worker runs, it scans for temporary manifests. For each one it finds, it checks the file's last modification time. 3. **Stale File Handling**: If a temporary manifest's last modification time is older than a configurable threshold, the GC worker considers it stale (left over from a crashed or terminated query). The GC worker will then delete this stale temporary manifest. Files that were protected only by this stale manifest are no longer shielded from garbage collection. This approach ensures that only files for genuinely active queries are protected. The lifecycle of the temporary manifest is managed dynamically: it is created when a long query starts, kept alive through periodic updates, and is either deleted by the query upon normal completion or automatically cleaned up by the GC worker if the query terminates unexpectedly. This mechanism may be too complex to implement at once. We can consider a two-phased approach: 1. **Phase 1 (Simple Time-Based Deletion)**: Initially, implement a simpler GC strategy that deletes obsolete files based solely on a configurable lingering time. This provides a baseline for space reclamation without the complexity of temporary manifests. 2. **Phase 2 (Consistency-Aware GC)**: Based on the practical effectiveness and observed issues from Phase 1, we can then decide whether to implement the full temporary manifest and heartbeat mechanism to handle long-running queries. This iterative approach allows for a quicker initial implementation while gathering real-world data to justify the need for a more complex solution. ## Drawbacks - **Dependency on Compaction Frequency**: The integration of the GC worker with Compaction means that GC cycles are directly tied to the frequency of compactions. In environments with infrequent compaction operations, obsolete files may accumulate for extended periods before being reclaimed, potentially leading to increased storage consumption. - **Race Condition with Long-Running Queries**: A potential race condition exists if a long-running query initiates but haven't write its temporary manifest in time, while a compaction process simultaneously begins and marks files used by that query as obsolete. This scenario could lead to the premature deletion of files still required by the active query. To mitigate this, the threshold time for writing a temporary manifest should be significantly shorter than the lingering time configured for obsolete files, ensuring that next GC worker runs do not delete files that are now referenced by a temporary manifest if the query is still running. Also the read replica shouldn't be later in manifest version for more than the lingering time of obsolete files, otherwise it might ref to files that are already deleted by the GC worker. - need to upload tmp manifest to object storage, which may introduce additional complexity and potential performance overhead. But since long-running queries are typically not frequent, the performance impact is expected to be minimal. one potential race condition with region-migration is illustrated below: ```mermaid sequenceDiagram participant gc_worker as GC Worker(same dn as region 1) participant region1 as Region 1 (Leader → Follower) participant region2 as Region 2 (Follower → Leader) participant region_dir as Region Directory gc_worker->>region1: Start GC, get region manifest activate region1 region1-->>gc_worker: Region 1 manifest deactivate region1 gc_worker->>region_dir: Scan region directory Note over region1,region2: Region Migration Occurs region1-->>region2: Downgrade to Follower region2-->>region1: Becomes Leader region2->>region_dir: Add new file gc_worker->>region_dir: Continue scanning gc_worker-->>region_dir: Discovers new file Note over gc_worker: New file not in Region 1's manifest gc_worker->>gc_worker: Mark file as orphan(incorrectly) ``` which could cause gc worker to incorrectly mark the new file as orphan and delete it, if config the lingering time for orphan files(files not mentioned anywhere(in used or unused)) is not long enough. A good enough solution could be to use lock to prevent gc worker to happen on the region if region migration is happening on the region, and vise versa. The race condition between gc worker and repartition also needs to be considered carefully. For now, acquiring lock for both region-migration and repartition during gc worker process could be a simple solution. ## Conclusion and Rationale This section summarizes the key aspects and trade-offs of the proposed integrated GC worker, highlighting its advantages and potential challenges. | Aspect | Current Proposal (Integrated GC) | | :--- | :--- | | **Implementation Complexity** | **Medium**. Requires careful integration with the compaction process and the slow query recorder for temporary manifest management. | | **Reliability** | **High**. Integration with compaction and leveraging temporary manifests from long-running queries significantly mitigates the risk of incorrect deletion. Accurate management of lingering times for obsolete files and prevention of accidental deletion of newly created SSTs enhance data safety. | | **Performance Overhead** | **Low to Medium**. The GC worker runs post-compaction, minimizing direct impact on write paths. Overhead from temporary manifest management by the slow query recorder is expected to be acceptable for long-running queries. | | **Impact on Other Components** | **Moderate**. Requires modifications to the compaction process to trigger GC and the slow query recorder to manage temporary manifests. This introduces some coupling but enhances overall data safety. | | **Deletion Strategy** | **State- and Time-Based**. Obsolete files are deleted based on a configurable lingering time, which is paused if the file is referenced by a temporary manifest. Unused files (never in a manifest) are also subject to a lingering time. | ## Unresolved Questions and Future Work This section outlines key areas requiring further discussion and defines potential avenues for future development. * **Slow Query Recorder Implementation**: Detailed specifications for modify slow query recorder's implementation and its precise interaction mechanisms with temporary manifests are needed. * **Configurable Lingering Times**: Establish and make configurable the specific lingering times for both obsolete and unused files to optimize storage reclamation and data availability. ## Alternatives ### 1. Standalone GC Service Instead of integrating the GC worker directly into the Compaction process, a standalone GC service could be implemented. This service would operate independently, periodically scanning the storage for obsolete and unused files based on manifest information and predefined retention policies. **Pros:** * **Decoupling**: Separates GC logic from compaction, allowing independent scaling and deployment. * **Flexibility**: Can be configured to run at different frequencies and with different strategies than compaction. **Cons:** * **Increased Complexity**: Requires a separate service to manage, monitor, and coordinate with other components. * **Potential for Redundancy**: May duplicate some file scanning logic already present in compaction. * **Consistency Challenges**: Ensuring read consistency would require more complex coordination mechanisms between the standalone GC service and active queries, potentially involving a distributed lock manager or a more sophisticated temporary manifest system. This alternative could be implemented in the future if the integrated GC worker proves insufficient or if there is a need for more advanced GC strategies. ### 2. Manifest-Driven Deletion (No Lingering Time) This alternative would involve immediate deletion of files once they are removed from the manifest, without a lingering time. **Pros:** * **Simplicity**: Simplifies the GC logic by removing the need for lingering time management. * **Immediate Space Reclamation**: Storage space is reclaimed as soon as files are marked for deletion. **Cons:** * **Increased Risk of Data Loss**: Higher risk of deleting files still in use by long-running queries or other processes if not perfectly synchronized. * **Complex Read Consistency**: Requires extremely robust and immediate mechanisms to ensure that no active queries are referencing files marked for deletion, potentially leading to performance bottlenecks or complex error handling. * **Debugging Challenges**: Difficult to debug issues related to premature file deletion due to the immediate nature of the operation. --- ### Rfcs/2025 08 16 Async Index Build --- Feature Name: Async Index Build Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/6756 Date: 2025-08-16 Author: "SNC123 " --- # Summary This RFC proposes an asynchronous index build mechanism in the database, with a configuration option to choose between synchronous and asynchronous modes, aiming to improve flexibility and adapt to different workload requirements. # Motivation Currently, index creation is performed synchronously, which may lead to prolonged write suspension and impact business continuity. As data volume grows, the time required for index building increases significantly. An asynchronous solution is urgently needed to enhance user experience and system throughput. # Details ## Overview The following table highlights the difference between async and sync index approach: | Approach | Trigger | Data Source | Additional Index Metadata Installation | Fine-grained `FileMeta` Index | | :--- | :--- | :--- | :--- | :--- | | Sync Index | On `write_sst` | Memory (on flush) / Disk (on compact) | Not required(already installed synchronously) | Not required | | Async Index | 4 trigger types | Disk | Required | Required | The index build mode (synchronous or asynchronous) can be selected via configuration file. ### Four Trigger Types This RFC introduces four `IndexBuildType`s to trigger index building: - **Manual Rebuild**: Triggered by the user via `ADMIN build_index("table_name")`, for scenarios like recovering from failed builds or migrating data. SST files whose `ColumnIndexMetadata` (see below) is already consistent with the `RegionMetadata` will be skipped. - **Schema Change**: Automatically triggered when the schema of an indexed column is altered. - **Flush**: Automatically builds indexes for new SST files created by a flush. - **Compact**: Automatically builds indexes for new SST files created by a compaction. ### Additional Index Metadata Installation Previously, index information in the in-memory `FileMeta` was updated synchronously. The async approach requires an explicit installation step. A race condition can occur when compaction and index building run concurrently, leading to: 1. Building an index for a file that is about to be deleted by compaction. 2. Creating an unnecessary index file and an incorrect manifest record. 3. On restart, replaying the manifest could load metadata for a non-existent file. To prevent this, the system checks if a file's `FileMeta` is in a `compacting` state before updating the manifest. If it is, the installation is aborted. ### Fine-grained `FileMeta` Index The original `FileMeta` only stored file-level index information. However, manual rebuilds require column-level details to identify files inconsistent with the current DDL. Therefore, the `indexes` field in `FileMeta` is updated as follows: ```rust struct FileMeta { ... // From file-level: // available_indexes: SmallVec<[IndexType; 4]> // To column-level: indexes: Vec, ... } pub struct ColumnIndexMetadata { pub column_id: ColumnId, pub created_indexes: IndexTypes, } ``` ## Process The index building process is similar to a flush and is illustrated below: ```mermaid sequenceDiagram Region0->>Region0: Triggered by one of 4 conditions, targets specific files loop For each target file Region0->>IndexBuildScheduler: Submits an index build task end IndexBuildScheduler->>IndexBuildTask: Executes the task IndexBuildTask->>Storage Interfaces: Reads SST data from disk IndexBuildTask->>IndexBuildTask: Builds the index file alt Index file size > 0 IndexBuildTask->>Region0: Sends IndexBuildFinished notification end alt File exists in Version and is not compacting Region0->>Storage Interfaces: Updates manifest and Version end ``` ### Task Triggering and Scheduling The process starts with one of the four `IndexBuildType` triggers. In `handle_rebuild_index`, the `RegionWorkerLoop` identifies target SSTs from the request or the current region version. It then creates an `IndexBuildTask` for each file and submits it to the `index_build_scheduler`. Similar to Flush and Compact operations, index build tasks are ultimately dispatched to the LocalScheduler. Resource usage can be adjusted via configuration files. Since asynchronous index tasks are both memory-intensive and IO-intensive but have lower priority, it is recommended to allocate fewer resources to them compared to compaction and flush tasks—for example, limiting them to 1/8 of the CPU cores. ### Index Building and Notification The scheduled `IndexBuildTask` executes its `index_build` method. It uses an `indexer_builder` to create an `Indexer` that reads SST data and builds the index. If a new index file is created (`IndexOutput.file_size > 0`), the task sends an `IndexBuildFinished` notification back to the `RegionWorkerLoop`. ### Index Metadata Installation Upon receiving the `IndexBuildFinished` notification in `handle_index_build_finished`, the `RegionWorkerLoop` verifies that the file still exists in the current `version` and is not being compacted. If the check passes, it calls `manifest_ctx.update_manifest` to apply a `RegionEdit` with the new index information, completing the installation. # Drawbacks Asynchronous index building may consume extra system resources, potentially affecting overall performance during peak periods. There may be a delay before the new index becomes available for queries, which could impact certain use cases. # Unresolved Questions and Future Work **Resource Management and Throttling**: The resource consumption (CPU, I/O) of background index building can be managed and limited to some extent by configuring a dedicated background thread pool. However, this approach cannot fully eliminate resource contention, especially under heavy workloads or when I/O is highly competitive. Additional throttling mechanisms or dynamic prioritization may still be necessary to avoid impacting foreground operations. # Alternatives Instead of being triggered by events like Flush or Compact, index building could be performed in batches during scheduled maintenance windows. This offers predictable resource usage but delays index availability. ---