couchdb Documentation

Seamless multi-primary syncing database with an intuitive HTTP/JSON API, designed for reliability

6,932 stars Erlang Erlang 1 file · ~28,803 tokens #big-data#cloud#content#couchdb#database#erlang#http#javascript
RAW Doc

1. Project Overview & Quickstart (apache/couchdb)

File: build-aux/README.md

CouchDB and Jenkins CI

In 2019, ASF and CloudBees reached an agreement to allow
the Foundation to use CloudBees Core to have a farm of managed Jenkins
masters. This allows the ASF to give larger projects their own dedicated
Jenkins master, which can be custom configured for the project. They can
readily manage this farm of Jenkins masters centrally, including push
updates to all masters and their plugins. Naturally, this also reduces
contention, as well as providing increased security for project-level
credentials. CouchDB is the first project to use this setup, via
https://ci-couchdb.apache.org/ (aka https://jenkins-cm1.apache.org/)

Only members of the ASF LDAP group couchdb-pmc have write access to
the Jenkins CI server, and all jobs are set to not trust changes to the
Jenkinsfile from forked repos (see below).

Further, IBM is sponsoring CouchDB with cloud-based worker nodes, for
the project's exclusive use. Combined with the FreeBSD and OSX nodes the
project already had, this will provide the necessary compute resources
to meet the needs of the project for some time to come.

Jenkins Configuration

All jobs on the new Jenkins master are contained in a CouchDB folder.
Credentials for the project are placed within this folder; this is a
unique capability of CloudBees Core at this time.

Pull Requests

Blue Ocean link

To implement build-a-PR functionality in the same way that Travis
performs builds, Jenkins offers the Multibranch Pipeline job. Here's how
it's configured for CouchDB through the GUI:

  • Job name: PullRequests (jobs shouldn't have spaces in them, because
    the job name is used for workspace path naming.)
  • Display Name: "Pull Requests"
  • Description: "This job builds all GitHub pull requests against
    apache/couchdb."
  • Branch sources: Github
    • Credentials: a GitHub API key from wohali. These credentials are
      stored in the top-level Jenkins CouchDB folder on the server.
      The API token credentials are user:email and repo:status.
    • URL https://github.com/apache/couchdb
    • Behaviors
      • Discover branches: Exclude branches that are also filed as PRs
      • Discover PRs from origin: Merging the PR with the current target
        branch revision
      • Discover PRs from works: Merging the PR with the current target
        branch revision, trust Nobody 2
      • Advanced clone behaviours:
        • Fetch tags
        • Clear "Shallow clone" 1
      • Clean before checkout 1
      • Prune stale remote-tracking branches
    • Property strategy: All branches get the same properties
  • Build Configuration
    • Mode: by Jenkinsfile
    • Script path: build-aux/Jenkinsfile.pr
  • Scan Repository Triggers
    • Periodically if not other wise run
    • Interval: 1 day (do not set this too high, GitHub has an API token
      throttle that can cause issues!)
  • Orphaned Item Strategy
    • Discard old items
    • Days to keep old items:
    • Max # of old items to keep: 10
  • Everything else set as defaults.

1: https://issues.jenkins-ci.org/browse/JENKINS-44598 explains why we have the build set to Clean Before Checkout every time, and why clones are not shallow.

2: https://issues.apache.org/jira/browse/INFRA-17449 explains why "Discover pull requests from forks/Trust" has been set to "Nobody."

Full Platform Builds on `main` and release branches


Our original Jenkins job (formerly /Jenkinsfile) is now
build-aux/Jenkinsfile.full. This builds CouchDB on main, all of
our release branches (2.*, 3.*, etc.) as well as any branch prefixed with
jenkins- for testing on a wide variety of supported operating systems.

Settings are as follows:

  • Job name: FullPlatformMatrix (jobs shouldn't have spaces in them, because
    the job name is used for workspace path naming.)
  • Display Name: "Full Platform Builds"
  • Description: "This job builds on our master and release branches,
    and builds packages on all."
  • Branch sources: Github
    • Credentials: a GitHub API key from wohali. These credentials are
      stored in the top-level Jenkins CouchDB folder on the server.
      The API token credentials are user:email and repo:status.
    • URL https://github.com/apache/couchdb
    • Behaviors
      • Discover branches: All branches
      • Filter by name (with wildcards): Include: master 2..x 3..x 4..x jenkins-
      • Advanced clone behaviours:
        • Fetch tags
        • Clear "Shallow clone" 1
      • Clean before checkout 1
      • Prune stale remote-tracking branches
    • Property strategy: All branches get the same properties
  • Build Configuration
    • Mode: by Jenkinsfile
    • Script path: build-aux/Jenkinsfile.pull
  • Scan Repository Triggers
    • none
  • Orphaned Item Strategy
    • Discard old items
    • Days to keep old items:
    • Max # of old items to keep: 10
  • Everything else set as defaults.

Other Resources

The apache/couchdb-ci repo contains the
dockerfiles that we use to generate the container images used for our
container-based builds. These images are hosted on Docker Hub in the following
repos:

The apache/couchdb-pkg repo contains
a set of helper scripts to build binary packages for Debian / CentOS / Ubuntu
from the contents of a release tarball. The packaging stage of our "Full
Platform Builds" pipeline clones this repo to produces the package artifacts.

File: src/weatherreport/README.md

Weather Report

weatherreport is an escript and set of tools that diagnoses common problems which could affect a CouchDB node or cluster.

Overview

Here is a basic example of using weatherreport followed immediately by the command's output:

bash
$ ./weatherreport --etc /path/to/etc
[warning] Cluster member [email protected] is not connected to this node. Please check whether it is down.

Usage

For most cases, you can just run the weatherreport command as given at the top of this README. However, sometimes you might want to know some extra detail or run only specific checks. For that, there are command-line options. Execute weatherreport --help to learn more about these options:

bash
weatherreport --help
Usage: weatherreport [-c ] [-d <level>] [-e] [-h] [-l] [check_name ...]

  -c, --etc		Path to the CouchDB configuration directory
  -d, --level		Minimum message severity level (default: notice)
  -l, --list		Describe available diagnostic tasks
  -e, --expert		Perform more detailed diagnostics
  -h, --help		Display help/usage
  check_name		A specific check to run

To get an idea of what checks will be run, use the --list option:

bash
weatherreport --list
Available diagnostic checks:

  custodian            Shard safety/liveness checks
  disk                 Data directory permissions and atime
  internal_replication Check the number of pending internal replication jobs
  ioq                  Check the total number of active IOQ requests
  mem3_sync            Check there is a registered mem3_sync process
  membership           Cluster membership validity
  memory_use           Measure memory usage
  message_queues       Check for processes with large mailboxes
  node_stats           Check useful erlang statistics for diagnostics
  nodes_connected      Cluster node liveness
  process_calls        Check for large numbers of processes with the same current/initial call
  process_memory       Check for processes with high memory usage
  safe_to_rebuild      Check whether the node can safely be taken out of service
  search               Check the local search node is responsive
  tcp_queues           Measure the length of tcp queues in the kernel

If you want all the gory details about what WeatherReport is doing, you can run the checks at a more verbose logging level with the --level option:

bash
$ ./weatherreport --etc /path/to/etc -d debug
[debug] Not connected to the local cluster node, trying to connect. alive:false connect_failed:undefined
[debug] Starting distributed Erlang.
[debug] Connected to local cluster node '[email protected]'.
[debug] Local RPC: mem3:nodes([]) [5000]
[debug] Local RPC: os:getpid([]) [5000]
[debug] Running shell command: ps -o pmem,rss -p 73905
[debug] Shell command output:
%MEM    RSS
 0.3  25116


[debug] Local RPC: erlang:nodes([]) [5000]
[debug] Local RPC: mem3:nodes([]) [5000]
[warning] Cluster member [email protected] is not connected to this node. Please check whether it is down.
[info] Process is using 0.3% of available RAM, totalling 25116 KB of real memory.

Most times you'll want to use the defaults, but any Syslog severity name will do (from most to least verbose): debug, info, notice, warning, error, critical, alert, emergency.

Finally, if you want to run just a single diagnostic or a list of specific ones, you can pass their name(s):

bash
$ ./weatherreport --etc /path/to/etc nodes_connected
[warning] Cluster member [email protected] is not connected to this node. Please check whether it is down.

File: src/setup/README.md

This module implements /_cluster_setup and manages the setting up, duh, of a CouchDB cluster.

Testing

bash
git clone https://git-wip-us.apache.org/repos/asf/couchdb.git
cd couchdb
git checkout setup
./configure
make
dev/run --no-join -n 2 --admin a:b

Then, in a new terminal:

text
$ src/setup/test/t.sh

Before running each test, kill the dev/run script, then reset the
CouchDB instances with:

text
$ rm -rf dev/lib/ dev/logs/
$ dev/run --no-join -n 2 --admin a:b

before running the next shell script.

The Plan:

N. End User Action

  • What happens behind the scenes.
  1. Launch CouchDB with $ couchdb, or init.d, or any other way, exactly
    like it is done in 1.x.x.
  • CouchDB launches and listens on 127.0.0.1:5984

From here on, there are two paths, one is via Fauxton (a) the other is
using a HTTP endpoint (b). Fauxton just uses the HTTP endpoint in (b).
(b) can be used to set up a cluster programmatically.

When using (b) you POST HTTP requests with a JSON request body (the request content type has to be set to application/json).

If you have already setup a server admin account, you might need to pass the credentials to the HTTP calls using HTTP basic authentication.
Alternativaly, if you use the cURL command you can can add username and password inline, like so:

text
curl -X PUT "http://admin:[email protected]:5984/mydb"

2.a. Go to Fauxton. There is a “Cluster Setup” tab in the sidebar. Go
to the tab and get presented with a form that asks you to enter an admin
username, admin password and optionally a bind_address and port to bind
to publicly. Submit the form with the [Enable Cluster] button.

If this is a single node install that already has an admin set up, there
is no need to ask for admin credentials here. If the bind_address is !=
127.0.0.1, we can skip this entirely and Fauxton can show the add_node
UI right away.

  • POST a JSON entity to /_cluster_setup, the entity looks like:
text
{
  "action":"enable_cluster",
  "username":"username",
  "password":"password",
  "bind_address":"0.0.0.0",
  "port": 5984
}

This sets up the admin user on the current node and binds to 0.0.0.0:5984
or the specified ip:port. Logs admin user into Fauxton automatically.

2.b. POST to /_cluster_setup as shown above.

Repeat on all nodes.

  • keep the same username/password everywhere.
  1. Pick any one node, for simplicity use the first one, to be the
    “setup coordination node”.
  • this is a “master” node that manages the setup and requires all
    other nodes to be able to see it and vice versa. Setup won’t work
    with unavailable nodes (duh). The notion of “master” will be gone
    once the setup is finished. At that point, the system has no
    master node. Ignore I ever said “master”.

a. Go to Fauxton / Cluster Setup, once we have enabled the cluster, the
UI shows an “Add Node” interface with the fields admin, and node:

  • POST a JSON entity to /_cluster_setup, the entity looks like:
text
{
  "action":"add_node",
  "username":"username",
  "password":"password",
  "host":"192.168.1.100",
  ["port": 5984],
  "name": "node1"  // as in “node1@hostname”, same as in vm.args
}

In the example above, this adds the node with IP address 192.168.1.100 to the cluster.

b. as in a, but without the Fauxton bits, just POST to /_cluster_setup

  • this request will do this:
  • on the “setup coordination node”:
  • check if we have an Erlang Cookie Secret. If not, generate
    a UUID and set the erlang cookie to that UUID.
    • store the cookie in config.ini, re-set_cookie() on startup.
  • make a POST request to the node specified in the body above
    using the admin credentials in the body above:
    POST to http://username:password@node_b:5984/_cluster_setup with:
text
{
      "action": "receive_cookie",
      "cookie": "<secretcookie>",
    }

  • when the request to node B returns, we know the Erlang-level
    inter-cluster communication is enabled and we can start adding
    the node on the CouchDB level. To do that, the “setup
    coordination node” does this to it’s own HTTP endpoint:
    PUT /nodes/node_b:5984 or the same thing with internal APIs.

  • Repeat for all nodes.

  • Fauxton keeps a list of all set up nodes for users to see.

4.a. When all nodes are added, click the [Finish Cluster Setup] button
in Fauxton.

  • this does POST /_cluster_setup
text
{
    "action": "finish_cluster"
  }

b. Same as in a.

  • this manages the final setup bits, like creating the _users,
    _replicator and _metadata, _db_updates endpoints and
    whatever else is needed. // TBD: collect what else is needed.

Single node auto setup

Option single_node set to true in [couchdb] configuration executes single node configuration on startup so the node is ready for use immediately.

Testing single_node auto setup

Pass --config-overrides single_node=true and -n 1 to dev/run

text
$ dev/run --no-join -n 1 --admin a:b --config-overrides single_node=true

Then, in a new terminal:

text
$ src/setup/test/t-single_node.sh

The script should show that single node is enabled.

The Setup Endpoint

This is not a REST-y endpoint, it is a simple state machine operated
by HTTP POST with JSON bodies that have an action field.

State 1: No Cluster Enabled

This is right after starting a node for the first time, and any time
before the cluster is enabled as outlined above.

text
GET /_cluster_setup
{"state": "cluster_disabled"}

POST /_cluster_setup {"action":"enable_cluster"...} -> Transition to State 2
POST /_cluster_setup {"action":"enable_cluster"...} with empty admin user/pass or invalid host/post or host/port not available -> Error
POST /_cluster_setup {"action":"anything_but_enable_cluster"...} -> Error

State 2: Cluster enabled, admin user set, waiting for nodes to be added.

text
GET /_cluster_setup
{"state":"cluster_enabled","nodes":[]}

POST /_cluster_setup {"action":"enable_cluster"...} -> Error
POST /_cluster_setup {"action":"add_node"...} -> Stay in State 2, but return "nodes":["node B"}] on GET
POST /_cluster_setup {"action":"add_node"...} -> if target node not available, Error
POST /_cluster_setup {"action":"finish_cluster"} with no nodes set up -> Error
POST /_cluster_setup {"action":"finish_cluster"} -> Transition to State 3
POST /_cluster_setup {"action":"delete_node"...} -> Stay in State 2, but delete node from /nodes, reflect the change in GET /_cluster_setup
POST /_cluster_setup {"action":"delete_node","node":"unknown"} -> Error Unknown Node

State 3: Cluster set up, all nodes operational

text
GET /_cluster_setup
{"state":"cluster_finished","nodes":["node a", "node b", ...]}

POST /_cluster_setup {"action":"enable_cluster"...} -> Error
POST /_cluster_setup {"action":"finish_cluster"...} -> Stay in State 3, do nothing
POST /_cluster_setup {"action":"add_node"...} -> Error
POST /_cluster_setup?i_know_what_i_am_doing=true {"action":"add_node"...} -> Add node, stay in State 3.
POST /_cluster_setup {"action":"delete_node"...} -> Stay in State 3, but delete node from /nodes, reflect the change in GET /_cluster_setup
POST /_cluster_setup {"action":"delete_node","node":"unknown"} -> Error Unknown Node

// TBD: we need to persist the setup state somewhere.

File: src/rexi/README.md

Rexi is a tailor-made RPC server application for sending CouchDB operations to nodes in a cluster. It is used in BigCouch as the remote procedure vehicle to get fabric functions to execute on remote cluster nodes.

Rexi better fits the needs of the BigCouch distributed data store by dropping some unneeded overhead in rex, the RPC server that ships with Erlang/OTP. Rexi is optimized for the case when you need to spawn a bunch of remote processes. Cast messages are sent from the origin to the remote rexi server, and local processes are spawned from there, which is vastly more efficient than spawning remote processes from the origin. You still get monitoring of the remote processes, but the request-handling process doesn't get stuck trying to connect to an overloaded/dead node. 'rexi_DOWN' messages will arrive at the client eventually. This has been an extremely advantageous mix of latency and failure detection, vastly improving the performance of BigCouch.

Rexi is used in conjunction with 'Fabric' which is also an application within BigCouch, but can be used on a stand-alone basis.

Getting Started

Rexi requires R13B03 or higher and can be built with rebar, which comes bundled in the repository.

License

Apache 2.0

Contact

File: src/mem3/README.md

mem3

Mem3 is the node membership application for clustered CouchDB. It is used
in CouchDB since version 2.0 and tracks two very important things for the
cluster:

  1. member nodes
  2. node/shards mappings for each database

Both the nodes and shards are tracked in node-local couch databases. Shards
are heavily used, so an ETS cache is also maintained for low-latency lookups.
The nodes and shards are synchronized via continuous CouchDB replication,
which serves as 'gossip' in Dynamo parlance. The shards ETS cache is kept in
sync based on membership and database event listeners.

A very important point to make here is that CouchDB does not necessarily
divide up each database into equal shards across the nodes of a cluster. For
instance, in a 20-node cluster, you may have the need to create a small
database with very few documents. For efficiency reasons, you may create your
database with Q=4 and keep the default of N=3. This means you only have 12
shards total, so 8 nodes will hold none of the data for this database. Given
this feature, we even shard use out across the cluster by altering the 'start'
node for the database's shards.

Shards can be split using the /_reshard API endpoint. Refer to a separate
README regarding the technical detail on how shard
splitting works.

Getting Started

Mem3 requires R13B03 or higher and can be built with rebar, which comes
bundled in the repository. Rebar needs to be able to find the couch_db.hrl
header file; one way to accomplish this is to set ERL_LIBS to point to the
apps subdirectory of a CouchDB checkout, e.g.

text
ERL_LIBS="/usr/local/src/couchdb/apps" ./rebar compile

License

Apache 2.0

File: src/mem3/README_reshard.md

Developer Oriented Resharding Description

This is a technical description of the resharding logic. The discussion will focus on: job creation and life-cycle, data definitions, and on the state transition mechanisms.

Job Life-Cycle

Job creation happens in the mem3_reshard_httpd API handler module. That module uses mem3_reshard_http_util to do some validation, and eventually calls mem3_reshard:start_split_job/2 on one or more nodes in the cluster depending where the new jobs should run.

mem3_reshard:start_split_job/2 is the main Erlang API entry point. After some more validation it creates a #job{} record and calls the mem3_reshard job manager. The manager then will add the job to its jobs ets table, save it to a _local document in the shards db, and most importantly, start a new resharding process. That process will be supervised by the mem3_reshard_job_sup supervisor.

Each job will be running in a gen_server implemented in mem3_reshard_job module. When splitting a shard, a job will go through a series of steps such as initial_copy, build_indices, update_shard_map, etc. Between each step it will report progress and checkpoint with mem3_reshard manager. A checkpoint involved the mem3_reshard manager persisting that job's state to disk in _local document in _dbs db. Then job continues until completed state or until it failed in the failed state.

If a user stops shard splitting on the whole cluster, then all running jobs will stop. When shard splitting is resumed, they will try to recover from their last checkpoint.

A job can also be individually stopped or resumed. If a job is individually stopped it will stay so even if the global shard splitting state is running. A user has to explicitly set that job to a running state for it to resume. If a node with running jobs is turned off, when it is rebooted running jobs will resume from their last checkpoint.

Data Definitions

This section focuses on record definition and how data is transformed to and from various formats.

Right below the mem3_reshard:start_split_job/1 API level a job is converted to a #job{} record defined in the mem3_reshard.hrl header file. That record is then used throughout most of the resharding code. The job manager mem3_reshard stores it in its jobs ets table, then when a job process is spawn it single argument also just a #job{} record. As a job process is executing it will periodically report state back to the mem3_reshard manager as an updated #job{} record.

Some interesting fields from the #job{} record:

  • id Uniquely identifies a job in a cluster. It is derived from the source shard name, node and a version (currently = 1).
  • type Currently the only type supported is split but merge or rebalance might be added in the future.
  • job_state The running state of the job. Indicates if the job is running, stopped, completed or failed.
  • split_state Once the job is running this indicates how far along it got in the splitting process.
  • source Source shard file. If/when merge is implemented this will be a list.
  • target List of target shard files. This is expected to be a list of 2 items currently.
  • history A time-line of state transitions represented as a list of tuples.
  • pid When job is running this will be set to the pid of the process.

In the mem3_reshard_job_store module the #job{} record is translated to an json document so it can be persisted to disk. Translation functions to and from a json in that module are also used by the HTTP API layer to return a job's state and other information to the user.

Another important piece of data is the global resharding state. When a user disables resharding on a cluster, a call is made to mem3_reshard manager on each node and they store that in a #state{} record. This record is defined in the mem3_reshard.hrl module, and just like the #job{} record can be translated to/from ejson in the mem3_reshard_store and stored and loaded from disk.

State Transitions

Resharding logic has 3 separate states to keep track of:

  1. Per-node resharding state. This state can be toggled between running and stopped. That toggle happens via the mem3_reshard:start/0 and mem3_reshard:stop/1 function. This state is kept in the #state{} record of the mem3_reshard manager gen_server. This state is also persisted to the local shard map database as a _local document so that it is maintained through a node restart. When the state is running then all jobs that are not individually stopped, and have not failed or completed, will be running. When the state is stopped all the running jobs will be stopped.

  2. Job's running state held in the #job{} job_state field. This is the general running state of a resharding job. It can be new, running, stopped, completed or failed. This state is most relevant for the mem3_reshard manager. In other words, it is the mem3_reshard gen_server that starts the job, monitors it to see if it exits successfully on completion or with an error.

  3. Job's internal splitting state. This state tracks the steps taken during shard splitting by each job. This state is mostly relevant for the mem3_reshard_job module. This state is kept in #job{}'s split_state field. The progression of these states is linear going from one state to the next. That's reflected in the code as well, when one state is done, mem3_reshard_job:get_next_state/1 is called which returns the next state in the list. The list is defined in the SPLIT_STATES macro. This simplistic transition is also one of the reasons why a gen_fsm wasn't considered for mem3_reshard_job logic.

Another interesting aspect is how the split_state transitions happen in the mem3_reshard_job module. What follows is an examination of that.

A job starts running in the new state or from a previously checkpointed state. In the later case, the job goes through some recovery logic (see ?STATE_RESTART macro in mem3_reshard_job) where it tries to resume its work from where it left of. It means, for example, if it was in the initial_copy state and was interrupted it might have to reset the target files and copy everything again. After recovery, the state execution logic is driven by run(#job{}) which ends up calling ?MODULE:State(#job{}) state specific functions for each state.

In mem3_reshard_job:switch_to_next_state/2 job's history is updated, any current state_info is cleared, job state is switched in the #job{} record. Then, the new state is checkpointed in the checkpoint/1 function. Checkpoint will cast a message to the mem3_reshard manager. After that message is sent the job process sits and waits.

In the meantime mem3_reshard manager checkpoints the state, which means it updates both its ETS table with the new #job{} record, persists the state with the mem3_reshard_store module, then, finally, it notifies the job process that checkpointing is done by calling mem3_reshard_job:checkpoint_done/1.

mem3_reshard_job:checkpoint_done/1 function call sends a checkpoint_done message to the job's process, at which point it starts executing that state.

Most states in mem3_reshard_job try not to block the main job process and instead launch worker processes to perform long running operations. It is usually just one worker process but it could be multiple as well. After that it waits for the workers to finish and inspects their exit signal (see wait_for_workers/1 function). When all the workers exit for a particular split_state, the job is switched to the next state with switch_to_next_state/1 and the whole thing repeats until the completed state is reached when the whole job exits normally.

If the source is updated at high rate and the cluster is under load, there is a possibility for the resharding jobs to take longer to finish. The cluster would have to be running at the limit where both compaction and internal replication will have difficulty catching up as fundamentally the logic used for the initial bulk copy is similar the compaction code, and topoff states are just reusing the internal replicator code. Eventually when the load subsides the jobs should catch up and finish.

Individual Modules Description

These are mostly random notes about various modules involved in resharding. Most, but not all, are in the mem3 application.

  • mem3_reshard: Main API entry point and the job manager.

  • mem3_reshard_job : Individual job logic.

  • mem3_reshard_dbdoc : Responsible for updating shard doc in the _db's database. Besides just having a bunch of utility function there is a gen_server spawned which is used to update shard documents in a cluster in such a way as to minimize the risk of conflicts. That is accomplished by having each shard updater calling only one such updater for the whole cluster. This coordinator is picked by sorting the list of all the live mem3 nodes and picking the first one in the list.

  • mem3_reshard_httpd : API endpoint definitions.

  • mem3_reshard_api : Cluster API endpoint. This module is responsible for sending requests to all the nodes in a cluster and gathering results.

  • mem3_reshard_index : This is a helper module used by workers in the build_indices state.

  • mem3_reshard_job_sup : Simple one for one supervisor which keeps track of running jobs.

  • mem3_reshard_store : State persistence module. It knows how to save/restore #job{} and #state{} records to/from _local docs. It is also re-used for serializing #job{} into ejson by the HTTP API module.

  • mem3_reshard_validate : Validate that source exists, target ranges don't have gaps in them, etc.

  • couch_db_split : This module is not in mem3 app but it does all the heavy lifting during the initial data copy. Given a source db and some targets, and a function to decide which doc go to which target, it will copy all data from the source to the targets. It's best to think of this module as a form of compactor. Unlike couch_bt_engine_compactor this one lives above the couch_db_engine API, and instead of copying data to one new file it copies it to 2 or more. Unsurprisingly because of that it uses some lower level couch_db_engine API directly, including linking to a couch_db_updater, force setting db update sequences and others.

File: src/mango/README.md

Mango

A MongoDB inspired query language interface for Apache CouchDB.

Motivation

Mango provides a single HTTP API endpoint that accepts JSON bodies via HTTP POST. These bodies provide a set of instructions that will be handled with the results being returned to the client in the same order as they were specified. The general principle of this API is to be simple to implement on the client side while providing users a more natural conversion to Apache CouchDB than would otherwise exist using the standard RESTful HTTP interface that already exists.

Actions

The general API exposes a set of actions that are similar to what MongoDB exposes (although not all of MongoDB's API is supported). These are meant to be loosely and obviously inspired by MongoDB but without too much attention to maintaining the exact behavior.

Each action is specified as a JSON object with a number of keys that affect the behavior. Each action object has at least one field named "action" which must
have a string value indicating the action to be performed. For each action there are zero or more fields that will affect behavior. Some of these fields are required and some are optional.

For convenience, the HTTP API will accept a JSON body that is either a single JSON object which specifies a single action or a JSON array that specifies a list of actions that will then be invoked serially. While multiple commands can be batched into a single HTTP request, there are no guarantees about atomicity or isolation for a batch of commands.

Activating Query on a cluster

Query can be enabled by setting the following config:

text
rpc:multicall(config, set, ["native_query_servers", "query", "{mango_native_proc, start_link, []}"]).

HTTP API

This API adds a single URI endpoint to the existing CouchDB HTTP API. Creating databases, authentication, Map/Reduce views, etc are all still supported exactly as currently document. No existing behavior is changed.

The endpoint added is for the URL pattern /dbname/_query and has the following characteristics:

  • The only HTTP method supported is POST.
  • The request Content-Type must be application/json.
  • The response status code will either be 200, 4XX, or 5XX
  • The response Content-Type will be application/json
  • The response Transfer-Encoding will be chunked.
  • The response is a single JSON object or array that matches to the single command or list of commands that exist in the request.

This is intended to be a significantly simpler use of HTTP than the current APIs. This is motivated by the fact that this entire API is aimed at customers who are not as savvy at HTTP or non-relational document stores. Once a customer is comfortable using this API we hope to expose any other "power features" through the existing HTTP API and its adherence to HTTP semantics.

Supported Actions

This is a list of supported actions that Mango understands. For the time being it is limited to the four normal CRUD actions plus one meta action to create indices on the database.

insert

Insert a document or documents into the database.

Keys:

  • action - "insert"
  • docs - The JSON document to insert
  • w (optional) (default: 2) - An integer > 0 for the write quorum size

If the provided document or documents do not contain an "_id" field one will be added using an automatically generated UUID.

It is more performant to specify multiple documents in the "docs" field than it is to specify multiple independent insert actions. Each insert action is submitted as a single bulk update (ie, _bulk_docs in CouchDB terminology). This, however, does not make any guarantees on the isolation or atomicity of the bulk operation. It is merely a performance benefit.

find

Retrieve documents from the database.

Keys:

  • action - "find"
  • selector - JSON object following selector syntax, described below
  • limit (optional) (default: 25) - integer >= 0, Limit the number of rows returned
  • skip (optional) (default: 0) - integer >= 0, Skip the specified number of rows
  • sort (optional) (default: []) - JSON array following sort syntax, described below
  • fields (optional) (default: null) - JSON array following the field syntax, described below
  • r (optional) (default: 1) - By default a find will return the document that was found when traversing the index. Optionally there can be a quorum read for each document using r as the read quorum. This is obviously less performant than using the document local to the index.
  • conflicts (optional) (default: false) - boolean, whether or not to include information about any existing conflicts for the document.

The important thing to note about the find command is that it must execute over a generated index. If a selector is provided that cannot be satisfied using an existing index the list of basic indices that could be used will be returned.

For the most part, indices are generated in response to the "create_index" action (described below) although there are two special indices that can be used as well. The "_id" is automatically indexed and is similar to every other index. There is also a special "_seq" index to retrieve documents in the order of their update sequence.

Its also quite possible to generate a query that can't be satisfied by any index. In this case an error will be returned stating that fact. Generally speaking the easiest way to stumble onto this is to attempt to OR two separate fields which would require a complete table scan. In the future I expect to support these more complicated queries using an extended indexing API (which deviates from the current MongoDB model a bit).

update

Update an existing document in the database

Keys:

  • action - "update"
  • selector - JSON object following selector syntax, described below
  • update - JSON object following update syntax, described below
  • upsert - (optional) (default: false) - boolean, Whether or not to create a new document if the selector does not match any documents in the database
  • limit (optional) (default: 1) - integer > 0, How many documents returned from the selector should be modified. Currently has a maximum value of 100
  • sort - (optional) (default: []) - JSON array following sort syntax, described below
  • r (optional) (default: 1) - integer > 0, read quorum constant
  • w (optional) (default: 2) - integer > 0, write quorum constant

Updates are fairly straightforward other than to mention that the selector (like find) must be satisifiable using an existing index.

On the update field, if the provided JSON object has one or more update operator (described below) then the operation is applied onto the existing document (if one exists) else the entire contents are replaced with exactly the value of the update field.

delete

Remove a document from the database.

Keys:

  • action - "delete"
  • selector - JSON object following selector syntax, described below
  • force (optional) (default: false) - Delete all conflicted versions of the document as well
  • limit - (optional) (default: 1) - integer > 0, How many documents to delete from the database. Currently has a maximum value of 100
  • sort - (optional) (default: []) - JSON array following sort syntax, described below
  • r (optional) (default: 1) - integer > 1, read quorum constant
  • w (optional) (default: 2) - integer > 0, write quorum constant

Deletes behave quite similarly to update except they attempt to remove documents from the database. Its important to note that if a document has conflicts it may "appear" that delete's aren't having an effect. This is because the delete operation by default only removes a single revision. Specify "force":true if you would like to attempt to delete all live revisions.

If you wish to delete a specific revision of the document, you can specify it in the selector using the special "_rev" field.

create\_index

Create an index on the database

Keys:

  • action - "create_index"
  • index - JSON array following sort syntax, described below
  • type (optional) (default: "json") - string, specifying the index type to create. Currently only "json" indexes are supported but in the future we will provide full-text indexes as well as Geo spatial indexes
  • name (optional) - string, optionally specify a name for the index. If a name is not provided one will be automatically generated
  • ddoc (optional) - Indexes can be grouped into design documents underneath the hood for efficiency. This is an advanced feature. Don't specify a design document here unless you know the consequences of index invalidation. By default each index is placed in its own separate design document for isolation.

Anytime an operation is required to locate a document in the database it is required that an index must exist that can be used to locate it. By default the only two indices that exist are for the document "_id" and the special "_seq" index.

Indices are created in the background. If you attempt to create an index on a large database and then immediately utilize it, the request may block for a considerable amount of time before the request completes.

Indices can specify multiple fields to index simultaneously. This is roughly analogous to a compound index in SQL with the corresponding tradeoffs. For instance, an index may contain the (ordered set of) fields "foo", "bar", and "baz". If a selector specifying "bar" is received, it can not be answered. Although if a selector specifying "foo" and "bar" is received, it can be answered more efficiently than if there were only an index on "foo" and "bar" independently.

NB: while the index allows the ability to specify sort directions these are currently not supported. The sort direction must currently be specified as "asc" in the JSON. [INTERNAL]: This will require that we patch the view engine as well as the cluster coordinators in Fabric to follow the specified sort orders. The concepts are straightforward but the implementation may need some thought to fit into the current shape of things.

list\_indexes

List the indexes that exist in a given database.

Keys:

  • action - "list_indexes"

delete\_index

Delete the specified index from the database.

Keys:

  • action - "delete_index"
  • name - string, the index to delete
  • design_doc - string, the design doc id from which to delete the index. For auto-generated index names and design docs, you can retrieve this information from the list\_indexes action

Indexes require resources to maintain. If you find that an index is no longer necessary then it can be beneficial to remove it from the database.

describe\_selector

Shows debugging information for a given selector

Keys:

  • action - "describe_selector"
  • selector - JSON object in selector syntax, described below
  • extended (optional) (default: false) - Show information on what existing indexes could be used with this selector

This is a useful debugging utility that will show how a given selector is normalized before execution as well as information on what indexes could be used to satisfy it.

If "extended": true is included then the list of existing indices that could be used for this selector are also returned.

JSON Syntax Descriptions

This API uses a few defined JSON structures for various operations. Here we'll describe each in detail.

Selector Syntax

The Mango query language is expressed as a JSON object describing documents of interest. Within this structure it is also possible to express conditional logic using specially named fields. This is inspired by and intended to maintain a fairly close parity to the existing MongoDB behavior.

As an example, the simplest selector for Mango might look something like such:

text
{"_id": "Paul"}

Which would match the document named "Paul" (if one exists). Extending this example using other fields might look like such:

text
{"_id": "Paul", "location": "Boston"}

This would match a document named "Paul" AND having a "location" value of "Boston". Seeing as though I'm sitting in my basement in Omaha, this is unlikely.

There are two special syntax elements for the object keys in a selector. The first is that the period (full stop, or simply .) character denotes subfields in a document. For instance, here are two equivalent examples:

text
{"location": {"city": "Omaha"}}
{"location.city": "Omaha"}

If the object's key contains the period it could be escaped with backslash, i.e.

text
{"location\\.city": "Omaha"}

Note that the double backslash here is necessary to encode an actual single backslash.

The second important syntax element is the use of a dollar sign ($) prefix to denote operators. For example:

text
{"age": {"$gt": 21}}

In this example, we have created the boolean expression age > 21.

There are two core types of operators in the selector syntax: combination operators and condition operators. In general, combination operators contain groups of condition operators. We'll describe the list of each below.

Implicit Operators

For the most part every operator must be of the form {"$operator": argument}. Though there are two implicit operators for selectors.

First, any JSON object that is not the argument to a condition operator is an implicit $and operator on each field. For instance, these two examples are identical:

text
{"foo": "bar", "baz": true}
{"$and": [{"foo": {"$eq": "bar"}}, {"baz": {"$eq": true}}]}

And as shown, any field that contains a JSON value that has no operators in it is an equality condition. For instance, these are equivalent:

text
{"foo": "bar"}
{"foo": {"$eq": "bar"}}

And to be clear, these are also equivalent:

text
{"foo": {"bar": "baz"}}
{"foo": {"$eq": {"bar": "baz"}}}

Although, the previous example would actually be normalized internally to this:

text
{"foo.bar": {"$eq": "baz"}}

Combination Operators

These operators are responsible for combining groups of condition operators. Most familiar are the standard boolean operators plus a few extra for working with JSON arrays.

Each of the combining operators take a single argument that is either a condition operator or an array of condition operators.

The list of combining characters:

  • "$and" - array argument
  • "$or" - array argument
  • "$not" - single argument
  • "$nor" - array argument
  • "$all" - array argument (special operator for array values)
  • "$elemMatch" - single argument (special operator for array values)
  • "$allMatch" - single argument (special operator for array values)

Condition Operators

Condition operators are specified on a per field basis and apply to the value indexed for that field. For instance, the basic "$eq" operator matches when the indexed field is equal to its argument. There is currently support for the basic equality and inequality operators as well as a number of meta operators. Some of these operators will accept any JSON argument while some require a specific JSON formatted argument. Each is noted below.

The list of conditional arguments:

(In)equality operators

  • "$lt" - any JSON
  • "$lte" - any JSON
  • "$eq" - any JSON
  • "$ne" - any JSON
  • "$gte" - any JSON
  • "$gt" - any JSON

Object related operators

  • "$exists" - boolean, check whether the field exists or not regardless of its value
  • "$type" - string, check the document field's type

Array related operators

  • "$in" - array of JSON values, the document field must exist in the list provided
  • "$nin" - array of JSON values, the document field must not exist in the list provided
  • "$size" - integer, special condition to match the length of an array field in a document. Non-array fields cannot match this condition.

Misc related operators

  • "$mod" - [Divisor, Remainder], where Divisor is a non-zero integer and Remainder is any integer. Matches documents where (field % Divisor == Remainder) is true. This is false for any non-integer field
  • "$regex" - string, a regular expression pattern to match against the document field. Only matches when the field is a string value and matches the supplied matches

Update Syntax

Need to describe the syntax for update operators.

Sort Syntax

The sort syntax is a basic array of field name and direction pairs. It looks like such:

text
[{field1: dir1} | ...]

Where field1 can be any field (dotted notation is available for sub-document fields) and dir1 can be "asc" or "desc".

Note that it is highly recommended that you specify a single key per object in your sort ordering so that the order is not dependent on the combination of JSON libraries between your application and the internals of Mango's indexing engine.

Fields Syntax

When retrieving documents from the database you can specify that only a subset of the fields are returned. This allows you to limit your results strictly to the parts of the document that are interesting for the local application logic. The fields returned are specified as an array. Unlike MongoDB only the fields specified are included, there is no automatic inclusion of the "_id" or other metadata fields when a field list is included.

A trivial example:

text
["foo", "bar", "baz"]

HTTP API

Short summary until the full documentation can be brought over.

POST /dbname/\_find

Issue a query.

Request body is a JSON object that has the selector and the various options like limit/skip etc. Or we could post the selector and put the other options into the query string. Though I'd probably prefer to have it all in the body for consistency.

Response is streamed out like a view.

POST /dbname/\_index

Request body contains the index definition.

Response body is empty and the result is returned as the status code (200 OK -> created, 3something for exists).

GET /dbname/\_index

Request body is empty.

Response body is all of the indexes that are available for use by find.

DELETE /dbname/\_index/ddocid/viewname

Remove the specified index.

Request body is empty.

Response body is empty. The status code gives enough information.

2. Official Technical Reference & Guides (apache/docs)

File: README.md

Apache HoraeDB (incubating) Website

[!IMPORTANT]
Apache HoraeDB (incubating) is an effort undergoing incubation at the Apache
Software Foundation (ASF), sponsored by the Apache Incubator PMC.

Please read the DISCLAIMER and a full explanation of "incubating".

This repository stores all the source files of HoraeDB website, it's built with hugo and docsy.

How to run

bash
# Start a local server for preview
hugo serve

By default hugo will listen on http://localhost:1313

Contributing

Welcome any contributions from the community. You can

  • Open an issue with any suggestions, or
  • Submit a pull request to improve the website (trivial fixes are welcome).

File: content/en/blog/2024/release-2.0.0.md

title: Release 2.0.0 date: 2024-04-23

Upgrade from 1.x.x to 2.0.0

The transition from CeresDB to Apache HoraeDB introduces several breaking changes. To facilitate upgrading from older versions to v2.0.0, specific alterations are necessary.

Upgrade Steps

  1. Setup required envs
text
export HORAEDB_DEFAULT_CATALOG=ceresdb
  1. Update config

Etcd's root should be configured both in horaedb and horaemeta

For horaedb

text
[cluster_deployment.etcd_client]
server_addrs = ['127.0.0.1:2379']
root_path = "/rootPath"

For horaemeta

text
storage-root-path = "/rootPath"
  1. Upgrade horaemeta

Horaedb will throw following errors, which is expected

text
2024-01-23 14:37:57.726 ERRO [src/cluster/src/cluster_impl.rs:136] Send heartbeat to meta failed, err:Failed to send heartbeat, cluster:defaultCluster, err:status: Unimplemented, message: "unknown service meta_service.MetaRpcService", details: [], metadata: MetadataMap { headers: {"content-type": "application/grpc"} }
  1. Upgrade horaedb

After all server upgraded, the cluster should be ready for read/write, and old data could be queried like before.

What's Changed

Breaking Changes

Features

Refactor

Fixed

Docs

Chore

New Contributors

Full Changelog: https://github.com/apache/horaedb/compare/v1.2.7...v2.0.0

File: content/en/blog/2024/release-2.1.0.md

title: Release 2.1.0
date: 2024-11-18
tags:

  • release

The Apache HoraeDB(incubating) team are pleased to announce that v2.1.0 is released, which has closed over 60 issues, including two major features:

1. New WAL implementation based on local disk.

In previous version, there is a RocksDB-based WAL. Although it works well in most cases, it has following issues:

  • Compiling from source can be a challenging task, especially since RocksDB is primarily written in C++.
  • For WAL, RocksDB can be somewhat overkill. If you are not familiar with RocksDB, tuning it can be very challenging.

With this new WAL, the above two problems are solved very well, and in performance test, the new WAL slightly outperforms the previous implementation, giving a solid foundation for future optimizations.

Interested readers can refer to the design documentation [here]({{< ref "wal_on_disk" >}}) for more details on this feature.

How to enable

text
[analytic.wal]
type = "Local"
data_dir = "/path/to/local/wal"

2. Access object store with [Apache OpenDAL](https://github.com/apache/opendal)

OpenDAL (Open Data Access Layer) is a project that provides a unified API for accessing various data storage backends.
It offers several advantages for developers and organizations. Here are some key benefits:

  • Unified API. OpenDAL provides a consistent and unified API for accessing different storage backends, such as AWS S3, Azure Blob Storage, and local file systems.
  • Optimized for Efficiency: OpenDAL is built with performance in mind. It includes optimizations to ensure efficient data access and manipulation, making it suitable for high-performance applications.
  • Comprehensive Documentation: The project provides detailed documentation, making it easier for developers to get started and understand how to use the library effectively.

In newer versions of OpenDAL, object_store integration is provided, which is very beneficial for HoraeDB code migration, as the APIs used by the upper layers remain virtually unchanged, and only the object store part needs to be abstracted to a unified OpenDAL operator:

rust
// Create a new operator
let operator = Operator::new(S3::default())?.finish();

// Create a new object store
let object_store = Arc::new(OpendalStore::new(operator));

Additionally, since the Apache OpenDAL implementation of object_store is based on the latest version of the object_store, which has breaking changes from the previous version that HoraeDB is using, we've chosen to make it compatible in order to keep the scope of this upgrade as manageable as possible.

In the process of adapting to the new API, the put_multipart interface has changed the most, so the main adaptation logic is also here, HoraeDB's approach is to encapsulate the underlying put_multipart interface to ensure that the upper layer code is not modified, the details can be found in the reference:

https://github.com/apache/horaedb/blob/v2.1.0/src/components/object_store/src/multi_part.rs

Note: The adaptation logic is only practical when parquet version < 52.0.0.

Download

Go to download pages.

Conclusion

Other bug fixes and improvements can be seen here:

As always, we warmly welcome you to join our community and share your insights.

File: content/en/blog/_index.md

title: "Blog"
weight: 1
simple_list: true
menu:
main:
weight: 30
pre:

File: content/en/community/_index.md

title: "Community"
params:
contributingUrl: https://github.com/apache/horaedb/blob/main/CONTRIBUTING.md
menu:
main:
weight: 40

{{% blocks/section color="white" %}}

How to Subscribe Mail List

Mailing lists are a form of communication used by the Apache community. Generally speaking, many things in the Apache community are hosted by mailing lists, For example: project q&A, technical discussion, transaction decision, release vote, etc，By subscribing, you can get the first update of the HoraeDB community and keep up with the community. The following list is the main list we are using:

In order to subscribe to the [email protected] mailing list, the steps are as follows：

  1. Send an email without any content or subject: [email protected]
  2. Wait until you receive an email with the subject line confirm subscribe to [email protected] (if you have not received it for a long time, please confirm whether the email is blocked by your email, if you have not been blocked and will receive a reply for a long time, return to step 1)
  3. Reply directly to the email without changing the subject line or adding the email content.
  4. Wait until you receive an email with the subject line WELCOME to [email protected] .
  5. If you receive an email from dev, you have successfully subscribed to the email. To initiate a discussion, you can send an email directly to [email protected], which will be sent to everyone who subscribed to the mailing list.

Unsubscribe from the mail list

The steps for unsubscribing to a mailing list are similar to those for subscribing to a mailing list:

  1. Send an email without any content or subject to: [email protected]
  2. Wait until you receive an email with the subject line confirm unsubscribe from [email protected]
  3. Reply directly to the email without changing the subject line or adding the email content
  4. Wait until you receive an email with the subject line GOODBYE from [email protected]
  5. Unsubscribe success

{{% /blocks/section %}}

File: content/en/docs/design/architecture.md

title: "Introduction to HoraeDB's Architecture"

Target

  • Provide the overview of HoraeDB to the developers who want to know more about HoraeDB but have no idea where to start.
  • Make a brief introduction to the important modules of HoraeDB and the connections between these modules but details about their implementations are not be involved.

Motivation

HoraeDB is a timeseries database (TSDB). However, HoraeDB's goal is to handle both timeseries and analytic workloads compared with the classic TSDB, which usually have a poor performance in handling analytic workloads.

In the classic timeseries database, the Tag columns (InfluxDB calls them Tag and Prometheus calls them Label) are normally indexed by generating an inverted index. However, it is found that the cardinality of Tag varies in different scenarios. And in some scenarios the cardinality of Tag is very high (we name this case after analytic workload), and it takes a very high cost to store and retrieve the inverted index. On the other hand, it is observed that scanning+pruning often used by the analytical databases can do a good job to handle such analytic workload.

The basic design idea of HoraeDB is to adopt a hybrid storage format and the corresponding query method for a better performance in processing both timeseries and analytic workloads.

Architecture

plaintext
┌──────────────────────────────────────────┐
│       RPC Layer (HTTP/gRPC/MySQL)        │
└──────────────────────────────────────────┘
┌──────────────────────────────────────────┐
│                 SQL Layer                │
│ ┌─────────────────┐  ┌─────────────────┐ │
│ │     Parser      │  │     Planner     │ │
│ └─────────────────┘  └─────────────────┘ │
└──────────────────────────────────────────┘
┌───────────────────┐  ┌───────────────────┐
│    Interpreter    │  │      Catalog      │
└───────────────────┘  └───────────────────┘
┌──────────────────────────────────────────┐
│               Query Engine               │
│ ┌─────────────────┐  ┌─────────────────┐ │
│ │    Optimizer    │  │    Executor     │ │
│ └─────────────────┘  └─────────────────┘ │
└──────────────────────────────────────────┘
┌──────────────────────────────────────────┐
│         Pluggable Table Engine           │
│  ┌────────────────────────────────────┐  │
│  │              Analytic              │  │
│  │┌────────────────┐┌────────────────┐│  │
│  ││      Wal       ││    Memtable    ││  │
│  │└────────────────┘└────────────────┘│  │
│  │┌────────────────┐┌────────────────┐│  │
│  ││     Flush      ││   Compaction   ││  │
│  │└────────────────┘└────────────────┘│  │
│  │┌────────────────┐┌────────────────┐│  │
│  ││    Manifest    ││  Object Store  ││  │
│  │└────────────────┘└────────────────┘│  │
│  └────────────────────────────────────┘  │
│  ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─   │
│           Another Table Engine        │  │
│  └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─   │
└──────────────────────────────────────────┘

The figure above shows the architecture of HoraeDB stand-alone service and the details of some important modules will be described in the following part.

RPC Layer

module path: https://github.com/apache/incubator-horaedb/tree/main/server

The current RPC supports multiple protocols including HTTP, gRPC, MySQL.

Basically, HTTP and MySQL are used to debug HoraeDB, query manually and perform DDL operations (such as creating, deleting tables, etc.). And gRPC protocol can be regarded as a customized protocol for high-performance, which is suitable for massive reading and writing operations.

SQL Layer

module path: https://github.com/apache/incubator-horaedb/tree/main/query_frontend

SQL layer takes responsibilities for parsing sql and generating the query plan.

Based on sqlparser a sql dialect, which introduces some key concepts including Tag and Timestamp, is provided for processing timeseries data. And by utilizing DataFusion the planner is able to generate both regular logical plans and tailored ones which is used to implement the special operators defined by timeseries queries, e.g PromQL.

Interpreter

module path: https://github.com/apache/incubator-horaedb/tree/main/interpreters

The Interpreter module encapsulates the SQL CRUD operations. In the query procedure, a sql received by HoraeDB is parsed, converted into the query plan and then executed in some specific interpreter, such as SelectInterpreter, InsertInterpreter and etc.

Catalog

module path: https://github.com/apache/incubator-horaedb/tree/main/catalog_impls

Catalog is actually the module managing metadata and the levels of metadata adopted by HoraeDB is similar to PostgreSQL: Catalog > Schema > Table, but they are only used as namespace.

At present, Catalog and Schema have two different kinds of implementation for standalone and distributed mode because some strategies to generate ids and ways to persist metadata differ in different mode.

Query Engine

module path: https://github.com/apache/incubator-horaedb/tree/main/query_engine

Query Engine is responsible for optimizing and executing query plan given a basic SQL plan provided by SQL layer and now such work is mainly delegated to DataFusion.

In addition to the basic functions of SQL, HoraeDB also defines some customized query protocols and optimization rules for some specific query plans by utilizing the extensibility provided by DataFusion. For example, the implementation of PromQL is implemented in this way and read it if you are interested.

Pluggable Table Engine

module path: https://github.com/apache/incubator-horaedb/tree/main/table_engine

Table Engine is actually a storage engine for managing tables in HoraeDB and the pluggability of Table Engine is a core design of HoraeDB which matters in achieving our long-term target, e.g supporting handle log or tracing workload by implementing new storage engines. HoraeDB will have multiple kinds of Table Engine for different workloads and the most appropriate one should be chosen as the storage engine according to the workload pattern.

Now the requirements for a Table Engine are:

  • Manage all the shared resources under the engine:
    • Memory
    • Storage
    • CPU
  • Manage metadata of tables such as table schema and table options;
  • Provide Table instances which provides read and write methods;
  • Take responsibilities for creating, opening, dropping and closing Table instance;
  • ....

Actually the things that a Table Engine needs to process are a little complicated. And now in HoraeDB only one Table Engine called Analytic is provided and does a good job in processing analytical workload, but it is not ready yet to handle the timeseries workload (we plan to enhance it for a better performance by adding some indexes which help handle timeseries workload).

The following part gives a description about details of Analytic Table Engine.

WAL

module path: https://github.com/apache/incubator-horaedb/tree/main/wal

The model of HoraeDB processing data is WAL + MemTable that the recent written data is written to WAL first and then to MemTable and after a certain amount of data in MemTable is accumulated, the data will be organized in a query-friendly form to persistent devices.

Now three implementations of WAL are provided for standalone and distributed mode:

  • For standalone mode, WAL is based on RocksDB and data is persisted on the local disk.
  • For distributed mode, WAL is required as a distributed component and to be responsible for durability of the newly written data, so now we provide an implementation based on OceanBase.
  • For distributed mode, in addition to OceanBase, we also provide a more lightweight implementation based on Apache Kafka.

MemTable

module path: https://github.com/apache/incubator-horaedb/tree/main/analytic_engine/src/memtable

For WAL can't provide efficient data retrieval, the newly written data is also stored in Memtable for efficient data retrieval, after a certain amount of data is reached, HoraeDB organizes the data in MemTable into a query-friendly storage format (SST) and stores it to the persistent device.

The current implementation of MemTable is based on agatedb's skiplist. It allows concurrent reads and writes and can control memory usage based on Arena.

Flush

module path: https://github.com/apache/incubator-horaedb/blob/main/analytic_engine/src/instance/flush_compaction.rs

What Flush does is that when the memory usage of MemTable reaches the threshold, some MemTables are selected for flushing into query-friendly SSTs saved on persistent device.

During the flushing procedure, the data will be divided by a certain time range (which is configured by table option Segment Duration), and any SST is ensured that the timestamps of the data in it are in the same Segment. Actually this is also a common operation in most timeseries databases which organizes data in the time dimension to speed up subsequent time-related operations, such as querying data over a time range and assisting purge data outside the TTL.

Compaction

module path: https://github.com/apache/incubator-horaedb/tree/main/analytic_engine/src/compaction

The data of MemTable is flushed as SSTs, but the file size of recently flushed SST may be very small. And too small or too many SSTs lead to the poor query performance. Therefore, Compaction is then introduced to rearrange the SSTs so that the multiple smaller SST files can be compacted into a larger SST file.

Manifest

module path: https://github.com/apache/incubator-horaedb/tree/main/analytic_engine/src/meta

Manifest takes responsibilities for managing tables' metadata of Analytic Engine including:

  • Table schema and table options;
  • The sequence number where the newest flush finishes;
  • The information of all the SSTs belonging to the table.

Now the Manifest is based on WAL and Object Storage. The newly written updates on the Manifest are persisted as logs in WAL, and in order to avoid infinite expansion of Manifest (actually every Flush leads to an update), Snapshot is also introduced to clean up the history of metadata updates, and the generated Snapshot will be saved to Object Storage.

Object Storage

module path: https://github.com/apache/incubator-horaedb/tree/main/components/object_store

The SST generated by Flush needs to be persisted and the abstraction of the persistent storage device is ObjectStore including multiple implementations:

The distributed architecture of HoraeDB separates storage and computing, which requires Object Store needs to be a highly available and reliable service independent of HoraeDB. Therefore, storage systems like Amazon S3, Alibaba Cloud OSS is a good choice and in the future implementations on storage systems of some other cloud service providers is planned to provide.

SST

module path: https://github.com/apache/incubator-horaedb/tree/main/analytic_engine/src/sst

SST is actually an abstraction that can have multiple specific implementations. The current implementation is based on Parquet, which is a column-oriented data file format designed for efficient data storage and retrieval.

The format of SST is very critical for retrieving data and is also the most important part to perform well in handling both timeseries and analytic workloads. At present, our Parquet-based implementation is good at processing analytic workload but is poor at processing timeseries workload. In our roadmap, we will explore more storage formats in order to achieve a good performance in both workloads.

Space

module path: https://github.com/apache/incubator-horaedb/blob/main/analytic_engine/src/space.rs

In Analytic Engine, there is a concept called space and here is an explanation for it to resolve some ambiguities when read source code. Actually Analytic Engine does not have the concept of catalog and schema and only provides two levels of relationship: space and table. And in the implementation, the schema id (which should be unique across all catalogs) on the upper layer is actually mapped to space id.

The space in Analytic Engine serves mainly for isolation of resources for different tenants, such as the usage of memory.

Critical Path

After a brief introduction to some important modules of HoraeDB, we will give a description for some critical paths in code, hoping to provide interested developers with a guide for reading the code.

Query

plaintext
┌───────┐      ┌───────┐      ┌───────┐
│       │──1──▶│       │──2──▶│       │
│Server │      │  SQL  │      │Catalog│
│       │◀─10──│       │◀─3───│       │
└───────┘      └───────┘      └───────┘
                │    ▲
               4│   9│
                │    │
                ▼    │
┌─────────────────────────────────────┐
│                                     │
│             Interpreter             │
│                                     │
└─────────────────────────────────────┘
                           │    ▲
                          5│   8│
                           │    │
                           ▼    │
                   ┌──────────────────┐
                   │                  │
                   │   Query Engine   │
                   │                  │
                   └──────────────────┘
                           │    ▲
                          6│   7│
                           │    │
                           ▼    │
 ┌─────────────────────────────────────┐
 │                                     │
 │            Table Engine             │
 │                                     │
 └─────────────────────────────────────┘

Take SELECT SQL as an example. The figure above shows the query procedure and the numbers in it indicates the order of calling between the modules.

Here are the details:

  • Server module chooses a proper rpc module (it may be HTTP, gRPC or mysql) to process the requests according the protocol used by the requests;
  • Parse SQL in the request by the parser;
  • With the parsed sql and the information provided by catalog/schema module, DataFusion can generate the logical plan;
  • With the logical plan, the corresponding Interpreter is created and logical plan will be executed by it;
  • For the logical plan of normal Select SQL, it will be executed through SelectInterpreter;
  • In the SelectInterpreter the specific query logic is executed by the Query Engine:
    • Optimize the logical plan;
    • Generate the physical plan;
    • Optimize the physical plan;
    • Execute the physical plan;
  • The execution of physical plan involves Analytic Engine:
    • Data is obtained by read method of Table instance provided by Analytic Engine;
    • The source of the table data is SST and Memtable, and the data can be filtered by the pushed down predicates;
    • After retrieving the table data, Query Engine will complete the specific computation and generate the final results;
  • SelectInterpreter gets the results and feeds them to the protocol module;
  • After the protocol layer converts the results, the server module responds to the client with them.

The following is the flow of function calls in version v1.2.2:

text
/* Detailed source-code truncated for AI context efficiency. */
  1. The received request will be forwarded to handle_sql after various protocol conversions, and since the request may not be processed by this node, it may need to be forwarded to maybe_forward_sql_query to handle the forwarding logic.
  2. After constructing the ForwardRequest in maybe_forward_sql_query, call forward
  3. After constructing the RouteRequest in forward, call route
  4. Use route to get the destination node endpoint and return to forward.
  5. Call forward_with_endpoint to forward the request
  6. return forward
  7. return maybe_forward_sql_query
  8. return handle_sql
  9. If this is a Local request, call fetch_sql_query_output to process it
  10. Call parse_sql to parse sql into Statment
  11. return fetch_sql_query_output
  12. Call statement_to_plan with Statment
  13. Construct Planner with ctx and Statment, and call the statement_to_plan method of Planner
  14. The planner will call the corresponding planner method for the requested category, at this point our sql is a query and will call sql_statement_to_plan
  15. Call sql_statement_to_datafusion_plan , which will generate the datafusion object, and then call SqlToRel::sql_statement_to_plan
  16. The generated logical plan is returned from SqlToRel::sql_statement_to_plan
  17. return
  18. return
  19. return
  20. Call execute_plan_involving_partition_table (in the default configuration) for subsequent optimization and execution of this logical plan
  21. Call build_interpreter to generate Interpreter
  22. return
  23. Call Interpreter's interpreter_execute_plan method for logical plan execution.
  24. The corresponding execute function is called, at this time the sql is a query, so the execute of the SelectInterpreter will be called
  25. call execute_logical_plan , which will call build_df_session_ctx to generate the optimizer
  26. build_df_session_ctx will use the config information to generate the corresponding context, first using datafusion and some custom optimization rules (in logical_optimize_rules()) to generate the logical plan optimizer, using apply_adapters_for_physical_optimize_rules to generate the physical plan optimizer
  27. return optimizer
  28. Call optimize_plan, using the optimizer just generated to first optimize the logical plan and then the physical plan
  29. Return to optimized physical plan
  30. execute physical plan
  31. returned after execution
  32. After collecting the results of all slices, return
  33. return
  34. return
  35. return
  36. Return to the upper layer for network protocol conversion and finally return to the request sender

Write

plaintext
┌───────┐      ┌───────┐      ┌───────┐
│       │──1──▶│       │──2──▶│       │
│Server │      │  SQL  │      │Catalog│
│       │◀─8───│       │◀─3───│       │
└───────┘      └───────┘      └───────┘
                │    ▲
               4│   7│
                │    │
                ▼    │
┌─────────────────────────────────────┐
│                                     │
│             Interpreter             │
│                                     │
└─────────────────────────────────────┘
      │    ▲
      │    │
      │    │
      │    │
      │    │       ┌──────────────────┐
      │    │       │                  │
     5│   6│       │   Query Engine   │
      │    │       │                  │
      │    │       └──────────────────┘
      │    │
      │    │
      │    │
      ▼    │
 ┌─────────────────────────────────────┐
 │                                     │
 │            Table Engine             │
 │                                     │
 └─────────────────────────────────────┘

Take INSERT SQL as an example. The figure above shows the query procedure and the numbers in it indicates the order of calling between the modules.

Here are the details:

  • Server module chooses a proper rpc module (it may be HTTP, gRPC or mysql) to process the requests according the protocol used by the requests;
  • Parse SQL in the request by the parser;
  • With the parsed sql and the catalog/schema module, DataFusion can generate the logical plan;
  • With the logical plan, the corresponding Interpreter is created and logical plan will be executed by it;
  • For the logical plan of normal INSERT SQL, it will be executed through InsertInterpreter;
  • In the InsertInterpreter, write method of Table provided Analytic Engine is called:
    • Write the data into WAL first;
    • Write the data into MemTable then;
  • Before writing to MemTable, the memory usage will be checked. If the memory usage is too high, the flush process will be triggered:
    • Persist some old MemTables as SSTs;
    • Store updates about the new SSTs and the flushed sequence number of WAL to Manifest;
    • Delete the corresponding WAL entries;
  • Server module responds to the client with the execution result.

File: content/en/docs/design/clustering.md

title: "Introduction to Architecture of HoraeDB Cluster"
---Note: Some of the features mentioned in the article have not yet been implemented.

Overview

text
/* Detailed source-code truncated for AI context efficiency. */

The diagram above describes the architecture of a HoraeDB cluster, where some key concepts need to be explained:

  • HoraeMeta Cluster: Takes responsibilities for managing the metadata and resource scheduling of the cluster;
  • Shard(L)/Shard(F): Leader shard and follower shard consisting of multiple tables;
  • HoraeDB: One HoraeDB instance consisting of multiple shards;
  • WAL Service: Write-ahead log service for storing new-written real-time data;
  • Object Storage: Object storage service for storing SST converted from memtable;

From the architecture diagram above, it can be concluded that the compute and storage are separated in the HoraeDB cluster, which makes it easy to implement useful distributed features, such as elastic autoscaling of compute/storage resources, high availability, load balancing, and so on.

Let's dive into some of the key components mentioned above before explaining how these features are implemented.

Shard

Shard is the basic scheduling unit in the cluster, which consists of a group of tables. And the tables in a shard share the same region for better storage locality in the WAL Service, and because of this, it is efficient to recover the data of all tables in the shard by scanning the entire WAL region. For most of implementations of WAL Service, without the shard concept, it costs a lot to recover the table data one by one due to massive random IO, and this case will deteriorate sharply when the number of tables grows to a certain level.

A specific role, Leader or Follower, should be assigned to a shard. A pair of leader-follower shards share the same set of tables, and the leader shard can serve the write and query requests from the client while the follower shard can only serve the read-only requests, and must synchronize the newly written data from the WAL service in order to provide the latest snapshot for data retrieval. Actually, the follower is not needed if the high availability is not required, while with at least one follower, it takes only a short time to resume service by simply switching the Follower to Leader when the HoraeDB instance on which the leader shard exists crashes.

The diagram below concludes the relationship between HoraeDB instance, Shard, Table. As shown in the diagram, the leader and follower shards are interleaved on the HoraeDB instance.

plaintext
┌─HoraeDB Instance0──────┐     ┌─HoraeDB Instance1──────┐
│  ┌─Shard0(L)────────┐  │     │  ┌─Shard0(F)────────┐  │
│  │ ┌────┬────┬────┐ │  │     │  │ ┌────┬────┬────┐ │  │
│  │ │ T0 │ T1 │ T2 │ │  │     │  │ │ T0 │ T1 │ T2 │ │  │
│  │ └────┴────┴────┘ │  │     │  │ └────┴────┴────┘ │  │
│  └──────────────────┘  │     │  └──────────────────┘  │
│                        │     │                        │
│  ┌─Shard1(F)────────┐  │     │  ┌─Shard1(L)────────┐  │
│  │ ┌────┬────┬────┐ │  │     │  │ ┌────┬────┬────┐ │  │
│  │ │ T0 │ T1 │ T2 │ │  │     │  │ │ T0 │ T1 │ T2 │ │  │
│  │ └────┴────┴────┘ │  │     │  │ └────┴────┴────┘ │  │
│  └──────────────────┘  │     │  └──────────────────┘  │
└────────────────────────┘     └────────────────────────┘

Since Shard is the basic scheduling unit, it is natural to introduce some basic shard operations:

  • Create/Drop table to/from a shard;
  • Open/Close a shard;
  • Split one shard into two shards;
  • Merge two shards into one shard;
  • Switch the role of a shard;

With these basic shard operations, some complex scheduling logic can be implemented, e.g. perform an expansion by splitting one shard into two shards and migrating one of them to the new HoraeDB instance.

HoraeMeta

HoraeMeta is implemented by embedding an ETCD inside to ensure consistency and takes responsibilities for cluster metadata management and scheduling.

The cluster metadata includes:

  • Table information, such as table name, table ID, and which cluster the table belongs to;
  • The mapping between table and shard and between shard and HoraeDB instance;
  • ...

As for the cluster scheduling work, it mainly includes:

  • Receiving the heartbeats from the HoraeDB instances and determining the online status of these registered instances;
  • Assigning specific role shards to the registered HoraeDB instances;
  • Participating in table creation by assigning a unique table ID and the most appropriate shard to the table;
  • Performing load balancing through shard operations according to the load information sent with the heartbeats;
  • Performing expansion through shard operations when new instances are registered;
  • Initiating failover through shard operations when old instances go offline;

Route

In order to avoid the overhead of forwarding requests, the communication between clients and the HoraeDB instances is peer-to-peer, that is to say, the client should retrieve routing information from the server before sending any specific write/query requests.

Actually, the routing information is decided by the HoraeMeta, but clients are only allowed the access to it through the HoraeDB instances rather than HoraeMeta, to avoid potential performance issues on the HoraeMeta.

WAL Service & Object Storage

In the HoraeDB cluster, WAL Service and Object Storage exist as separate distributed systems featured with HA, data replication and scalability. Current distributed implementations for WAL Service includes Kafka and OBKV (access OceanBase by its table api), and the implementations for Object Storage include popular object storage services, such as AWS S3, Azure object storage and Aliyun OSS.

The two components are similar in that they are introduced to serve as the underlying storage layer for separating compute and storage, while the difference between two components is obvious that WAL Service is used to store the newly written data from the real-time write requests whose individual size is small but quantity is large, and Object Storage is used to store the read-friendly data files (SST) organized in the background, whose individual size is large and aggregate size is much larger.

The two components make it much easier to implement the horaedb cluster, which features horizontal scalability, high availability and load balancing.

Scalability

Scalability is an important feature for a distributed system. Let's take a look at to how the horizontal scalability of the HoraeDB cluster is achieved.

First, the two storage components (WAL Service and Object Storage) should be horizontally scalable when deciding on the actual implementations for them, so the two storage services can be expanded separately if the storage capacity is not sufficient.

It will be a little bit complex when discussing the scalability of the compute service. Basically, these cases will bring the capacity problem:

  • Massive queries on massive tables;
  • Massive queries on a single large table;
  • Massive queries on a normal table;

For the first case, it is easy to achieve horizontal scalability just by assigning shards that are created or split from old shards to expanded HoraeDB instances.

For the second case, the table partitioning is proposed and after partitioning, massive queries are distributed across multiple HoraeDB instances.

And the last case is the most important and the most difficult. Actually, the follower shard can handle part of the queries, but the number of follower shards is limited by the throughput threshold of the synchronization from the WAL regions. As shown in the diagram below, a pure compute shard can be introduced if the followers are not enough to handle the massive queries. Such a shard is not required to synchronize data with the leader shard, and retrieves the newly written data from the leader/follower shard only when the query comes. As for the SSTs required by the query, they will be downloaded from Object Storage and cached afterwards. With the two parts of the data, the compute resources are fully utilized to execute the CPU-intensive query plan. As we can see, such a shard can be added with only a little overhead (retrieving some data from the leader/follower shard when it needs), so to some extent, the horizontal scalability is achieved.

plaintext
┌HoraeDB─────┬┬─┐
                            ┌──newly written─│  │  │TableN││ │
                            ▼                └──Shard(L/F)┴┴─┘
┌───────┐  Query  ┌HoraeDB─────┬┬─┐
│client │────────▶│  │  │TableN││ │
└───────┘         └──Shard─────┴┴─┘          ┌───────────────┐
                            ▲                │    Object     │
                            └───old SST──────│    Storage    │
                                             └───────────────┘

High Availability

Assuming that WAL service and Object Storage are highly available, the high availability of the HoraeDB cluster can be achieved by such a procedure:

  • When detecting that the heartbeat is broken, HoraeMeta determines that the HoraeDB instance is offline;
  • The follower shards whose paired leader shards exist on the offline instance are switched to leader shards for fast failover;
  • A slow failover can be achieved by opening the crashed shards on another instance if such follower shards don't exist.
text
/* Detailed source-code truncated for AI context efficiency. */

Load Balancing

HoraeMeta collects the instance load information contained in the received heartbeats to create a load overview of the whole cluster, according to which the load balancing can be implemented as an automatic mechanism:

  • Pick a shard on a low-load instance for the newly created table;
  • Migrate a shard from a high-load instance load to another low-load instance;
  • Split the large shard on the high-load instance and migrate the split shards to other low-load instances;

File: content/en/docs/design/compaction_offload.md

title: "Compaction Offload"

Note: This feature is still in development.

This chapter discusses compaction offload, which is designed to separate the compaction workload from the local horaedb nodes and delegate it to external compaction nodes.

Overview

plaintext
┌─────────────────────────────────────────────────────────────────────────┐
│                                                                         │
│                           HoraeMeta Cluster                             │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘
    ▲                ▲                           |                   |
    │                │1.Fetch compaction         │(Monitor compaction│
    │                │  node  info               │node)              │
    |                │                           ▼                   ▼
┌────────────┐  ┌────────────┐              ┌────────────┐   ┌────────────┐
│            │  │            │2.Offload Task│            │   │            │
│  HoraeDB   │  │  HoraeDB   │  ─────────▶  │ Compaction │   │ Compaction │
│            │  │            │  ◀─────────  │ Node       │   │ Node       │
└────────────┘  └────────────┘ 4.Ret Result └────────────┘   └────────────┘
    |                |                            |                 |
    │  5.Update the  │                            │    3.Compact    │
    │    SSTable     │                            │                 │
    ▼                ▼                            ▼                 ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                                                ┌─────────────────────┐  │
│                            Object Storage      │ Temporary Workspace │  │
│                                                └─────────────────────┘  │
└─────────────────────────────────────────────────────────────────────────┘

The diagram above describes the architecture of cluster for compaction offload, where some key concepts need to be explained:

  • Compaction Node: Takes responsibility to handle offloaded compaction tasks. The compaction node receives the compaction task and performs the actual merging of SSTables, then sends back the task result to HoraeDB.
  • HoraeMeta Cluster: HoraeMeta acts as a compaction nodes manager in the compaction offload scenario. It monitors the compaction nodes cluster and schedule the compaction nodes.

The procedure of remote compaction based above architecture diagram is:

  1. HoraeDB servers fetch the information of suitable compaction nodes from the HoraeMeta.
  2. HoraeDB submit the compaction task to the remote compaction node, according to the information fetched from HoraeMeta.
  3. Compaction node executes the task and write results to the temporary workspace.
  4. Compaction node sends compaction results back to HoraeDB.
  5. HoraeDB receives the result, installs the data in temporary workspace and purges compaction input files.

The architecture above makes it easy to implement some wonderful features like load balancing and high availability. Let's dive into the key components in the architecture and talking about how these features are implemented.

Compaction Node

Compaction Node runs the main logic of compaction. It is implemented based on HoraeDB and distinguished by:

  • NodeType: A config parameter used to distinguished the HoraeDB and CompactionNode. This info would be sent to HoraeMeta through heartbeat.

The compaction service is implemented as grpc service.

HoraeMeta

HoraeMeta manages the compaction nodes cluster with CompactionNodeManager, which takes responsibilities for compaction nodes metadata management and scheduling.

The compaction nodes metadata includes:

  • Compaction node information, such as node name, node state;
  • A compaction node name list, used as the key to access compaction node info, for better scheduling with round-robin strategy;
  • ...

As for the compaction nodes scheduling work, it mainly includes:

  • Receiving the heartbeats from the compaction node and determining the online status of these registered nodes.
  • Performing load balancing according to the compaction nodes cluster info.
  • Providing the info of suitable compaction node for HoraeDB when remote compaction execution is needed.

Load Balancing

Load Balancing is critical for compaction nodes cluster to make their overall processing more efficient. The effect of load balancing mainly based on the schedule algorithm for compaction nodes impl in CompactionNodeManager.

(ps: The current implementation of schedule algorithm is round-robin strategy for easiness.)

The main process for the schedule algorithm based on real load is:

  • HoraeMeta collects the compaction nodes load information through the heartbeats to create a load overview of the compaction nodes cluster.
  • Pick a compaction node with low load according to the load overview.

High Availability

The fault tolerance of above architecture can be achieved by such a procedure:

  • When detecting that the heartbeat is broken, HoraeMeta determines that the compaction node is offline.
  • When HoraeMeta can not provide suitable compaction node for HoraeDB or compaction node doesn't return the task result successfully, HoraeDB would switches to run compaction task locally.

File: content/en/docs/design/storage.md

title: "Storage"

The storage engine mainly provides the following two functions：

  1. Persistence of data
  2. Under the premise of ensuring the correctness of the data, organize the data in the most reasonable way to meet the query needs of different scenarios.

This document will introduce the internal implementation of the storage engine in HoraeDB. Readers can refer to the content here to explore how to use HoraeDB efficiently.

Overall Structure

HoraeDB is a distributed storage system based on the share-nothing architecture.

Data between different servers is isolated from each other and does not affect each other. The storage engine in each stand-alone machine is a variant of log-structured merge-tree, which is optimized for time-series scenarios. The following figure shows its core components:

Write Ahead Log (WAL)

A write request will be written to

  1. memtable in memory
  2. WAL in durable storage

Since memtable is not persisted to the underlying storage system in real time, so WAL is required to ensure the reliability of the data in memtable.

On the other hand, due to the design of the distributed architecture, WAL itself is required to be highly available. Now there are following implementations in HoraeDB:

Memtable

Memtable is a memory data structure used to hold recently written table data. Different tables have its corresponding memtable.

Memtable is read-write by default (aka active), and when the write reaches some threshold, it will become read-only and be replaced by a new memtable.

The read-only memtable will be flushed to the underlying storage system in SST format by background thread. After flush is completed, the read-only memtable can be destroyed, and the corresponding data in WAL can also be deleted.

Sorted String Table（SST）

SST is a persistent format for data, which is stored in the order of primary keys of table. Currently, HoraeDB uses parquet format for this.

For HoraeDB, SST has an important option: segment_duration, only SST within the same segment can be merged, which is benefical for time-series data. And it is also convenient to eliminate expired data.

In addition to storing the original data, the statistical information of the data will also be stored in the SST to speed up the query, such as the maximum value, the minimum value, etc.

Compactor

Compactor can merge multiple small SST files into one, which is used to solve the problem of too many small files. In addition, Compactor will also delete expired data and duplicate data during the compaction. In future, compaction maybe add more task, such as downsample.

The current compaction strategy in HoraeDB reference Cassandra:

Manifest

Manifest records metadata of table, SST file, such as: the minimum and maximum timestamps of the data in an SST.

Due to the design of the distributed architecture, the manifest itself is required to be highly available. Now in HoraeDB, there are mainly the following implementations：

  • WAL
  • ObjectStore

ObjectStore

ObjectStore is place where data (i.e. SST) is persisted.

Generally speaking major cloud vendors should provide corresponding services, such as Alibaba Cloud's OSS and AWS's S3.

Explore Related Technologies & Topics

View All 9,111+ Repositories →

Discover similar high-velocity repositories, agent skills, and OpenAPI specifications across the ecosystem.

#big-data Browse big-data catalog → #cloud Browse cloud catalog → #content Browse content catalog → #couchdb Browse couchdb catalog → #database Browse database catalog → #erlang Browse erlang catalog → ⚡ MCP Servers Model Context Protocol → 🧠 Agent Skills Prompt & rule files → 🔌 OpenAPI Specs REST & Swagger schemas →