 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/image_85642ae4b3.png) 

## Background

Let’s begin by thinking about the projects you are working on. How would you be willing to:

- Run a full unit test suite of your project locally?
- Start a cluster and run a full integration test suite locally?
- Run performance testing (e.g., benchmarks) of your changes?

... when you’ve changed some code and are preparing to submit a PR?

Back when we started the RisingWave project, developers always complain about the slow test speed and bad test experience. They’re indeed painful for developers! 

**Unit tests were super slow.**
We had a workspace of 10+ crates and 700 unit test cases. Running a full test suite with cargo test took a long time, not to mention the time spent on compiling with code coverage.

**Integration tests were very hard.**
RisingWave is a cloud-native streaming database from day one, and a minimal RisingWave cluster requires a compute node, a meta node, and a frontend node, which need to be started with many parameters and configs. Therefore, we had a long bash script to start a cluster, and we had to wait 30 seconds before we could run integration tests. Sometimes, if developers forgot to stop the components they had previously started, it would take several minutes to figure out why the e2e test yielded no output. With more and more components (e.g., multiple compute nodes, MinIO, Kafka) added to the integration test suite, few developers were even willing to start clusters by themselves. We would simply submit a PR with our fingers crossed, hoping the change would pass e2e tests on CI. Furthermore, Every time we made changes to the command-line interface, we had to hack the startup scripts and let other developers know the changes so that everyone could still correctly set up RisingWave on their local machine. At that time, CI failure rates were high. There was no way of detecting whether a component was correctly set up (e.g. by gRPC health check) — we just used `sleep 5s`. Sometimes the e2e tests just failed because the TCP port was closed or the service was not fully initialized.

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/risingwave_dependency_a0a179a42e.png) 

**Performance tests were painful.**
Developers had to do everything manually — compile RisingWave, send binaries to EC2, write command-line arguments by ourselves, check if every service started, configure the Grafana dashboard, run benchmarks, and finally, clean up everything. Anyway, no one knew whether their PRs would affect performance because (nearly) no one would do performance tests.

## RiseDev

I got tired of those problems, so I initiated RiseDev, the RisingWave developers’ tool, at the end of December 2021. I initially thought about naming it RiseLAB, as developers will do a lot of “experiments” in the “lab” of “RisingWave.” Nevertheless, to avoid name conflicts with some well-known brands, we later renamed it to RiseDev before open-source.

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/risedev_git_log_32c7a020eb.png) 

Two weeks after RiseDev was introduced, all developers at Singularity Data were using it for daily development. It’s super easy to use: with one line of command `./risedev d`, it will start a RisingWave cluster within 4 seconds. It also offers several other options to tweak, but we won’t go deeper about the internals of RiseDev today. 

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/risedev_4_seconds_launch_b45c6c7dcf.png) 

Now let's see how RiseDev can simplify the development process and improve the developers' experiences. 

### Fast Unit Tests

Rust developers generally use `cargo test` to invoke unit tests. Let’s recap [how unit tests are compiled and run](https://doc.rust-lang.org/rustc/tests/index.html) with `cargo test`:

- All crates within the workspace will be compiled with a special config `test`, so that code inside `#[cfg(test)]` will be compiled.
- Each crate (library) will produce a binary linked with `libtest`, which executes unit tests.
- `cargo test` will invoke each unit test binary one by one to complete the unit test process.

The problem is with the last step — indeed, those unit test binaries can run in parallel! That’s where [cargo-nextest](https://github.com/nextest-rs/nextest) comes into play. It runs unit tests from different crates in parallel, so as to improve test speed. Furthermore, it integrates with [libtest-mimic](https://docs.rs/libtest-mimic/latest/libtest_mimic/) well. We have some file-based tests, where the test harness will scan test files and generate unit tests dynamically using libtest-mimic. Nextest supports such use cases well. After migrating to nextest, we could run all test cases within 10 seconds:

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/risedev_parallel_test_c86ca1c855.png) 

One issue with nextest at that time was the lack of coverage tools. Luckily, I found [cargo-llvm-cov](https://github.com/taiki-e/cargo-llvm-cov) and [sent a PR](https://github.com/taiki-e/cargo-llvm-cov/pull/144) to add integration with nextest. Now, with a single command `cargo llvm-cov nextest`, we can run unit tests with coverage reports using nextest. This solution also improves build speed compared with `cargo tarpaulin`, the coverage tool we used before.

### Easy Cluster Setup

The core functionality of RiseDev is composing and starting a test cluster. RiseDev implements a flexible yaml configuration engine that enables developers to compose the cluster as they want. Let’s take a look at the usage examples below.

By default, RiseDev only starts RisingWave in in-memory mode: a meta node storing metadata in memory, a compute node storing states in memory, and a frontend node. The cluster configuration is stored in `risedev.yml` at the root of the GitHub repository. The default config is like this:

```
risedev:
  default:
    - use: meta-node
    - use: compute-node
    - use: frontend
```

And with a single command `./risedev d`, all these components start one by one. RiseDev ensures each component is fully initialized before starting the next one by doing health checks to avoid starting a misconfigured cluster.

```
✅ tmux: session risedev
✅ prepare: all previous services have been stopped
✅ meta-node-5690: api grpc://127.0.0.1:5690/, dashboard http://127.0.0.1:5691/
✅ compute-node-5688: api grpc://127.0.0.1:5688/
✅ frontend-4566: api postgres://127.0.0.1:4566/
✅ playground: done bootstrapping with config default
---- summary of startup time ----
meta-node-5690: 0.83s
compute-node-5688: 1.37s
frontend-4566: 0.83s
-------------------------------
```

RiseDev invokes [tmux](https://github.com/tmux/tmux) command to spawn processes in the background, which provides developers maximum control of every component. Developers can attach to the tmux console and manage each running process within the cluster.

Now when a developer wants to persist state store data on disk, they can modify the `risedev.yml` in this way:

```
risedev:
  default:
    - use: minio
    - use: meta-node
    - use: compute-node
    - use: frontend
    - use: compactor
```

That’s where the magic happens — RiseDev automatically provides available services. Now that we have `minio` in cluster profile, RiseDev automatically picks it up when starting the compute node.

```
✅ tmux: session risedev
✅ prepare: all previous services have been stopped
✅ minio: api http://127.0.0.1:9301/, console http://127.0.0.1:9400/
✅ meta-node-5690: api grpc://127.0.0.1:5690/, dashboard http://127.0.0.1:5691/
✅ compute-node-5688: api grpc://127.0.0.1:5688/
✅ frontend-4566: api postgres://127.0.0.1:4566/
✅ compactor-6660: compactor 127.0.0.1:6660
✅ playground: done bootstrapping with config default
---- summary of startup time ----
minio: 0.38s
meta-node-5690: 0.82s
compute-node-5688: 1.41s
frontend-4566: 0.83s
compactor-6660: 0.69s
-------------------------------
```

When starting the services, RiseDev automatically creates buckets and configures policies for MinIO. And while initiating the compute node, RiseDev adds a command-line argument `--state-store hummock+minio://...` so that the compute node uses MinIO for storing data. After all, a two-line change enables data persistence of RisingWave!

It’s common that developers may want to test a 3-node cluster and ensure their changes work under a distributed environment. RiseDev also supports this with this flexible yaml config engine.

```
risedev:
  default:
    - use: minio
    - use: meta-node
    - use: compute-node
      port: 5687
    - use: compute-node
      port: 5688
    - use: compute-node
      port: 5689
    - use: frontend
    - use: compactor
```

And when developers want to test RisingWave with external data sources like Kafka, and view metrics in Grafana, then can simply run `./risedev d full`. RiseDev automatically configures everything! Just navigate to `localhost:3001` and we can find the metrics we want:

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/risedev_grafana_22225bc82e.png) 

Sounds cool, right? After starting the cluster, we can now invoke `sqllogictest`, a SQL test framework to run integration tests originating from the SQLite project. We’re using [sqllogictest-rs](https://github.com/risinglightdb/sqllogictest-rs), a reimplementation of sqllogictest in the Rust programming language, maintained by the [RisingLightDB community](https://github.com/risinglightdb). We’ve contributed a lot of features to make it easier and more convenient to use.

Putting together, RiseDev provides a flexible configuration framework to start a RisingWave cluster. It runs health checks, configures services, and generates command-line arguments for all components. With minor changes, we can compose RisingWave cluster of our needs and run integration tests!

### Painless Performance Tests

When developing RisingWave, we care a lot about performance — how to ingest data faster, how to read from and write to S3 more efficiently, etc. But for a long time, we had no idea how each commit would affect performance. After some investigations, we set up some standard RisingWave AMIs on EC2 with all required tools installed. Developers only need to create an EC2 from the template, and they can immediately start benchmarking their changes with RiseDev — no need to install the Rust compiler or other dependencies by themselves.

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/risedev_AMI_c552712745.png) 

But single-node setups are only for quick verifications — all components share the same disk and CPUs, and dataset size is generally small. What we eventually need is a full distributed RisingWave setup with a large amount of data to benchmark.

We’ve developed two separate tools to set up a test environment and deploy RisingWave.

**Setting up Benchmark Environment in One Command**

Thanks to Terraform, we can describe our benchmark infrastructure on AWS using HCL code. Inside Singularity Data, we have a private Terraform module to start EC2 for each RisingWave component, connect them in one VPC, and set up an S3 bucket and ECR (elastic container registry). Every EC2 will get an IAM role with the necessary permissions to access each experiment’s S3 bucket and ECR without any manual configuration.


**Deploying RisingWave**

RiseDev already can start in a cluster locally, but can we deploy it in a remote cluster? The answer is YES! RiseDev supports generating docker-compose config and deploying RisingWave using docker-compose. It will read the deployment information from Terraform and generate a docker-compose config on each EC2. For example, we can instruct RiseDev to deploy one compute node to the EC2 named `rw-compute-0`:

```
    - use: compute-node
      id: compute-node-0
      listen-address: "0.0.0.0"
      address: ${dns-host:rw-compute-0}
```

And with `./risedev compose-deploy <profile>`, RiseDev will generate a docker-compose config for that node. Then, we can execute `./risedev apply-compose-deploy` to initiate the deployment process. Within one minute, developers can get their cluster up and running!

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/risedev_docker_compose_d08c1970f2.png) 

## Summary

RiseDev, the developers’ tool for RisingWave, has revolutionized the development process of RisingWave. Developers can easily use RiseDev to run local tests and performance tests. It automates the process of starting and deploying a full RisingWave cluster. It has made RisingWave’s development process much smoother and more efficient.

There is much more to explore in RisingWave and RiseDev, like *jaeger* tracing, auto Grafana dashboard generating, using deterministic testing to accelerate time-based tests, etc. We can’t cover them all in one article. We’ve constantly been improving the development experience, so developers will find it easy to work on the RisingWave project.

---

\*_Note 1: RiseDev deployment is only used for testing and should not be used in production environments. Please check out our docs on how to use RisingWave for [testing purposes](https://www.risingwave.dev/docs/latest/get-started/), how to deploy RisingWave on [Kubernetes](https://github.com/singularity-data/risingwave-operator), and stay tuned for RisingWave Cloud in our [roadmap](https://singularity-data.com/blog/on-the-way-to-democratized-stream-processing-risingwaves-roadmap)._

\*_Note 2: We have not published any official benchmark numbers of RisingWave — the performance varies under different use cases, and you should try it yourself!_

---

**About the Author:**

Chi is an engineering intern at Singularity Data. He is interested in database kernel development. At the same time, he also cares about the development experience, and tries to make every developer of RisingWave happy with coding. Chi is currently a master’s student at Carnegie Mellon University. He obtained his Bachelor’s degree in computer science from Shanghai Jiao Tong University.


Building the development environment manually for RisingWave is painful. We have developed RiseDev, the developer tool, to drastically simplify the development process. It automates the process of starting and deploying a full RisingWave cluster and makes the development process much smoother and more efficient.

Developer Experience Counts: Behind the Scenes of RiseDev

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Blog_datajoint_1_beeb81ea08.png) 

Disclaimer: This blog focuses on the design in the released version [v0.1.10](https://github.com/risingwavelabs/risingwave/releases/tag/v0.1.10). RisingWave is actively evolving, and we expect any technical details to be outdated at any time.  

In database systems, we can speed up join computations by creating indexes. For example, Postgres supports [Index Scans](https://www.postgresql.org/docs/current/indexes-index-only-scans.html) to speed up joins by scanning the index instead of the original data. However, "index" is difficult to implement in stream processing systems, mainly due to two reasons:  

- **The pattern of accessing states in streaming systems is usually "one-read-one-write."** Stream processing systems deal with unbounded data, which requires persisting intermediate states in the state storage and recovering from checkpoints in the case of system failures. These intermediate states are usually only read and written by a single operator with a single degree of parallelism and are not shared. 

- **An index is not a stream of data flowing in the dataflow graph but something that is queried by an operator.** Generally speaking, the streaming data flows through the dataflow, and operators can always obtain data produced by upstream operators in the dataflow. The stream operators only perform operations based on the data in the received stream and their local intermediate states. On the other hand, the index is something that needs to be actively queried (rather than flowing on the graph). The operator needs to actively obtain the index to get the final result. 

[Differential Dataflow](https://timelydataflow.github.io/differential-dataflow/) (DD) implements an indexing scheme, referred to as *[arrangement](https://arxiv.org/abs/1812.02639)*, in a streaming system. An arrangement can be interpreted as an MVCC index that records the mapping of ``(index key, time) → value``. An arrangement built for streaming data is essentially an index, where downstream operators can efficiently query all values corresponding to a certain time point through the index key.  

On top of arrangements, DD introduces [Delta Join](https://materialize.com/blog/joins-in-materialize/), a special kind of join operator in its streaming system. As long as all streams that need to be joined are indexed on the join key, Delta Join can utilize the index to generate the final join result directly without any intermediate state. That is saying if two joins cover the same stream, which is indexed, the delta join operators for two joins can access the same arrangement on that stream directly, thus saving the storage and maintenance cost. In contrast, a regular join operator needs to persist the records of the left and right streams, respectively. Furthermore, if it is a multi-way join, the record of each binary join also needs to be persisted.  

It might seem satisfying that with arrangement and delta joins, we can share the state in a streaming system and save the cost of duplicated states. But this approach does not directly apply to RisingWave for mainly two reasons:  

- Arrangement join is implemented as in-memory MVCC, in which each record is timestamped, and index persistence is not considered. However, in RisingWave, streams are not timestamped.  

- Delta join in DD only works in a single instance, and RisingWave has been a distributed system from day one.  

In RisingWave, we support state sharing by revising the delta join implementation. The core idea is to explore the epoch-based MVCC capability of the underlying cloud-native state storage. In this blog, I first recap RisingWave's data persistence process on its cloud-native storage, then introduce the implementation of shared state, and finally, I detail the implementation of delta join in RisingWave. 

 
# Checkpointing in RisingWave 

Checkpointing is the process in which stream processing systems persist their intermediate states. In RisingWave, checkpointing is completely asynchronous. The checkpointing procedure does not affect the flow of data in the streaming system. The operator can continuously process the streaming data without being affected by the progress of the checkpointing.  You can check this [document](https://github.com/risingwavelabs/risingwave/blob/main/docs/checkpoint.md) for more details. Here, we briefly recapped the parts of the checkpoint process that may be relevant to implementing shared states. 

 

The data of RisingWave's state storage engine "Hummock" mainly exists in two places: the shared buffer on each compute node and object storage services (which can be MinIO, S3, etc.). Shared buffer organizes data by checkpoint epoch; object storage service uses an LSM tree to organize data. Hummock stores snapshots at a certain point in time. The intermediate state calculated by the operator in an epoch is stored in the flush buffer inside the operator. 

 

Let's take a simple query as an example: 

```
CREATE MATERIALIZED VIEW mv_1 AS 

  SELECT * FROM t1, t2 

    WHERE t1.v1 = t2.v2 
```

This query yields two `TableScan` operators — `Join operator` and  `Materialize operator` — in the streaming engine. The `TableScan` operator injects the changes from internal tables (or external data sources) to downstream operators in real time; the `Join` operator is used to compute the match from two streams satisfying the join condition `t1.v1 = t2.v2`; and the `Materialize` operator materializes the final result into storage, which can be queried by user-issued arbitrary queries. Suppose both tables `t1 and t2` have only one column `v1 and v2`, respectively. At this point, the user inserts two rows in `1,2` in `t1`. 

  ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Picture1_c4b7df6217.png) 

After the two rows arrive at the `Join` operator, the operator will query the cache on whether these two rows have a match in `t2`. If there is a cache miss, then the operator will instead search the state store. The state store will scan the shared buffer as well as the data on the shared storage and merge the flush buffers to get the corresponding two lines in `t2`.

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Picture2_0230bf1674.png)

After finding the match, the two new lines in `t1` will be written to the flush buffer of the operator. At the same time, the matched data will be sent to downstream operators in the dataflow graph.
 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Picture3_efa80c09e9.png) 

Then the downstream `Materialize` operator will write the two new lines of matched results to its own flush buffer.
 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Picture4_52a17e82d2.png) 

At this point, the system decides to start a checkpointing process.  RisingWave's meta-service will inject barriers into all data sources to start the checkpointing process. In this example, there are two sources, `Table 1` and `Table 2`. When the system injects a barrier with epoch = 5, the data from epoch = 4 will be written to the state store by the operator, and the operator that receives the barrier will next process the data with epoch = 5.
 
 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Picture5_e283410fad.png) 

The `Join` operator aligns two barriers from its two upstreams. After receiving the barrier, the operator will write the data in the flush buffer to the shared buffer in the form of key-value pairs, and the operator can start processing the next batch of data.

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Picture6_9e47a927c0.png) 

After the barrier flows through the `Materialize` operator, the materialized result will also be written into the shared buffer.

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Picture6_9e47a927c0.png) 

And once the barrier flows through the entire graph, each computing node encodes the data in its shared buffer into SST files in the background and uploads it to the shared storage layer. When all nodes are uploaded, the checkpoint is declared complete. At this point, the shared buffer will clear the data that has been checkpointed, and the data on the shared storage will also be visible to the user's batch query.

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Picture8_df93cabbe2.png) 

At this juncture, when users execute a query, e.g., `SELECT * FROM mv_1`, they can see the latest data.

Therefore, for an operator, checkpointing is simply encoding the data and writing it to the shared buffer without waiting for any I/O operations. In the background, the computing node will upload the data written to the shared buffer to the shared storage and complete the checkpoint process asynchronously. The regular computation of the stream operator will not be affected at all throughout the whole checkpoint process.

## Shared state and index

After introducing RisingWave's barrier-based checkpointing process, we will now discuss how to implement a shared state based on such a design.

### Index

In RisingWave, the only state that can be shared is the "index," which is a special kind of materialized view.

```
CREATE TABLE table(v1 int, v2 int, v3 int);
CREATE INDEX idx on table(v1, v2); 
```

Here `CREATE INDEX` is equivalent to:

```
CREATE MATERIALIZED VIEW idx AS
  SELECT * FROM table ORDER BY v1, v2; 
```
In RisingWave, the index is a specialized materialized view sorted according to the index key specified by the user. A materialized view is essentially key-value pairs ordered by certain rules. If an operator wants to query all records satisfying the condition `v1 = 1, v2 = 2`, it only needs to sequentially scan all records starting with `[1, 2]` in the indexed view. Indexed materialized views are constructed using the Arrange operator.

Next, we discuss how to implement shared indexes - allowing multiple downstream operators to access the same index (shared state) at the same time. In RisingWave, most of the state is "one-read-one-write." For example, the join operator will only read and write the join key in its local computation. Data is placed on shared storage, mainly facilitating the adjustment of parallelism (i.e., scale-in, scale-out). The following points should be considered to achieve a "one-write-multiple-read" state:


- How to ensure that the operator can read a consistent state? If an intermediate state is in the process of writing, there is a problem with the access consistency of this state, and some special handling is required.
- How does reading shared state reduce the impact on stream operators? If every access to the shared state requires accessing the remote storage or querying the upstream operator through RPC, it will greatly affect the performance of streaming computation.

### Single node shared state

We start with discussing the case of sharing states on a single node. Let's first assume that there is an operator called `Read` that needs to read the shared state. The `Read` operator happens to be on the same compute node as the `Arrange` operator.

At this point, the `Arrange` operator receives the barrier.

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Picture9_39004d9c8e.png) 

The `Arrange` operator writes the data with epoch = 1 into the shared buffer and sends the barrier to its downstream operators.

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Picture10_4f855308a1.png) 

When the `Read` operator receives the barrier, we can safely say that the upstream operator has written its state into the shared buffer. Therefore, in the single-node scenario, the downstream operator can directly read the epoch = 1 data that the `Arrange` operator writes into the shared buffer. After receiving the read request of epoch = 1, the storage will merge the data of epoch < 1 on the shared storage and the data of epoch = 1 in the shared buffer to get a complete state snapshot (with epoch = 1) of the `Arrange` operator in the state storage.

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Picture12_9b43f3e0be.png)

### Shared state across nodes

Things are more tricky when it comes to multiple nodes. If the `Arrange` and  `Read` operators are not located on the same node, the `Read` operator needs to obtain the states written by the upstream `Arrange` operator remotely in some way. RisingWave achieves this by message replication: RisingWave replicates the update from the `Arrange` operator to the downstream node. The shared state is written to the shared storage only by the upstream operator, and the downstream operator only writes it to its local shared buffer after receiving the message. After the checkpoint is completed, the downstream node discards its local data and obtains a complete snapshot directly from the shared storage. Here is an example to describe this process in detail:

The `Arrange` and `Read` operators are located on two different nodes, `Worker 1` and `Worker 2`, respectively. In the downstream node `Worker 2`, we create a "shadow operator" `Replicate` for the `Arrange` operator. After the `Replicate` operator receives the data sent by the upstream `Arrange`, it replicates the data and writes it to the shared buffer on `Worker 2`.
 
 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Picture13_8feb50336d.png)

The `Read` operator receives a barrier, indicating that the data for its local computation can be accessed on the node where it is located. As shown in the figure below, the data read by the `Read` operator is actually the combination of the replicated data in the shared buffer of `Worker 2` and the data on the shared storage. At this point, the data written by the `Arrange` operator exists in the shared buffer of `Worker 1` and `Worker 2` at the same time. 

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Picture14_1d24d21e67.png) 

During the checkpointing process, only `Worker 1`, i.e., the node where `Arrange` is located, will upload the states in its local shared buffer to the shared storage. `Worker 2` at the downstream will directly discard the "shadow data" replicated by the `Replicate` operator in its local memory. At this point, if the `Read` operator initiates a request to read epoch = 1, it can read all the data from the shared storage.

## Delta Join


[Delta Join](https://materialize.com/blog/joins-in-materialize/) is an implementation of joins without maintaining intermediate states. Materialize implements delta join based on differential dataflow, in which each record has its own timestamp. In RisingWave, there is no native support for timestamping each record. The data on the shared storage are organized based on the barrier(epoch)-based checkpointing. To support delta join in RisingWave, we need to revise its implementation. 

### Mathematical principles

After supporting the shared state, we can reuse the state of indexes in delta join. 

For stream processing systems, given the query A⋈B (assuming natural join, i.e., equal joins), the streaming engine actually needs to compute Δ(A⋈B) in each epoch. According to the equivalence rules of relational algebra, we have:

`Δ(A⋈B)=[ΔA⋈(B∪ΔB)]∪(ΔB∪A)`

The rule also generalized to multi-way joins. For example, the delta Δ(A⋈B⋈C) of a three-way join query can be expanded into:

```
Δ(A⋈B⋈C)=[ΔA⋈(B∪ΔB)⋈(C∪ΔC)]
 ∪[ΔB⋈A⋈(C∪ΔC)]
 ∪[ΔC⋈A⋈B]
```
It is easy to find that the update of N-way joins can always be expressed as a ‘normal form’: the union of N terms, in which:

- Each clause is a production of n items.
- The first item is ΔT.
- Each table appears only once.
- Except for the first item, each table appears as either T or T+ΔT.

In RisingWave, we can build a dataflow graph to represent such delta join computation in the streaming engine. Here T+ΔT can be read from the index of the current epoch, and
T can be read from the index of the previous epoch. For example, for a three-way join, using the above formula, we can construct the following dataflow graph:

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/Picture15_cb5a887290.png) 

Each line of operators corresponds to exactly one term in the normal form. 

We implement a special `Union` operator, which merges the delta stream from three upstream ``Lookup`` executors in order. Firstly ΔC, then ΔB, finally ΔA. Otherwise, the order of the two causally related records will be reversed, resulting in erroneous results.

### Lookup operator implementation
 

What the `Lookup` operator does is essentially stream-table join. One of its inputs is a stream, and the other is an index (it can be a `Replicate`, or it can be directly connected to `Arrange`). 

 

If the ``Lookup`` operator accesses the index of the ****current**** epoch, it will wait until the barrier on the `Arrange` side arrives before starting the computation. As shown in the figure below, before receiving the ready barrier, the ``Lookup`` operator will cache the data on the stream; after receiving the barrier, it will immediately start processing the buffered data and find the corresponding matches from the index. 

  ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/figure_4_1_4d752a04ed.png) 

If the `Lookup` operator accesses the index of the **last** epoch, since the Lookup operator always aligns the barrier, when processing the current epoch data, the barrier of the previous epoch must have been received. Therefore, the index of the last epoch must have been ready. In this mode, when `Lookup` receives a message, it matches the message from the index directly. 

 ![](https://singularity-strapi-prod.s3.ap-southeast-1.amazonaws.com/figure_4_2_d2023e272b.png) 

### Representation of Delta Join in the Optimizer
 

The dataflow graph of delta join is derived from `LogicalMultiWayJoin`. If there is an index matching the join operation, the optimizer will convert the logical multi-way join plan into a delta join dataflow graph. In the optimizer, the representation of the Lookup operator is similar to that of `HashJoin`, and it can perform optimizations such as predicate pushdown just like other operators. In particular, when generating N-way delta join plans, we implement a delta join solver to search for the best `Lookup` order. The solver uses a greedy algorithm to select the order of the tables in each line of join to find a plan with the minimum number of shuffles. 

## Summary
 

State sharing and join optimization in RisingWave are taking shape. We implement the shared state and indexes of the stream processing system based on our cloud-native MVCC state storage. On top of that, we implement index-based join optimization in stream processing. There are still many optimizations that can be done in the future, such as supporting secondary indexes, shared caches,  more join types, and much more. We will discuss these topics in the future. 

 

[RisingWave](https://github.com/risingwavelabs/risingwave) is an SQL-based open-source cloud-native streaming database implemented in Rust. It is backed by [RisingWave Labs](http://risingwave-labs.com/). Join our [Slack community](https://join.slack.com/t/risingwave-community/shared_invite/zt-120rft0mr-d8uGk3d~NZiZAQWPnElOfw), follow us on [Twitter](https://twitter.com/SingularityData) and [Linkedin](https://www.linkedin.com/company/singularity-data), and subscribe to our [Youtube channel](https://www.youtube.com/channel/UCsHwdyBRxBpmkA5RRd0YNEA) to learn more about RisingWave.  
 


In database systems, we can speed up join computations by creating indexes. Here we examine RisingWave's data persistence process on its cloud-native storage, the implementation of shared state, and delta join.

Shared Indexes and Joins in Streaming Databases 

Chi Zhang

Community Contributor @ RisingWave Labs

Yingjun Wu

CEO @ RisingWave Labs

As a distributed SQL streaming database, RisingWave is well-optimized for stream processing. Let's take a look at whether it is meant for your stream processing application.

Is RisingWave the Right Streaming Database for Your Stream Processing Application?

Rayees Pasha

Head of Product @ RisingWave Labs

Jovana

Community Manager @ RisingWave Labs

We are excited to announce the launch of Streaming Stories. This platform brings together stories from users around the world; all centered around their experiences with stream processing.

RisingWave Labs Launching a Webinar Series: Streaming Stories

Jon Chuang

Software Engineer

Zhidong Guo

Storage Intern

Siyuan Wang

In this blog, we'll look at three techniques in RisingWave that leverage the power of S3-compatible object stores to achieve better performance.

Object Store Optimizations in RisingWave 

In this blog, our CEO and Founder, Yingjun Wu, will combine his software development and customer engagement experiences to discuss the past, present, and future of stream processing and streaming databases from a practitioner’s perspective.

Rethinking Stream Processing and Streaming Databases

Sooter Saalu

Community Contributor

Materialized views are database objects that contain the results of your operations, calculations, or queries. They exist as a locally saved representation of your data that you can use and reuse without recalculating or computing data values. Using materialized views saves processing time and cost and allows you to front-load your frequently used or most time-consuming queries.  This article explores materialized views in modern-day databases, their use cases, why they are important, how they are connected with data analytics, and their limitations.  

How to Update Materialized Views in Real Time

Since its official open-source release on GitHub under the Apache 2.0 license on April 8, 2022, RisingWave has grown into a popular project with over 100 contributors and 3,700 stars. Let's take a step back and reflect on our open-source community.

RisingWave Community Retrospective and a Look Forward 

Somrita Ganchoudhuri

Marketing Communications Manager @ RisingWave Labs

Let's take a look back at 2022! In this blog, we focus on five key updates from 2022.

A Look Back at 2022: 5 Key Highlights

Tushar Malik

RisingWave Community Contributor

In a nutshell, RisingWave is a distributed, cloud-native streaming database. It is built from the ground up to ingest and process data from streaming messaging queues in a horizontally scalable way. I discovered the RisingWave Code Camp, a mentorship program for junior developers to advance their skills in software and database systems, through a friend who'd met the team at Kubecon EU. Unfamiliar with Rust and admittedly daunted at the prospect of working on a database, I was hesitant to apply but hey, I needed an excuse to learn Rust, and I was intrigued by the idea of working with the amazing people building RisingWave! 

Building a Source Connector: A RisingWave Code Camp Experience 

Tianyi Zhuang

Software Engineer @ RisingWave Labs

Seven months ago, we open-sourced RisingWave, a cloud-native streaming database written in Rust. In this blog, I will introduce how RisingWave manages the streaming state in our Hummock cloud storage engine.

State Management for Cloud Native Streaming: Getting to the Core

What’s next for real-time data systems? Let’s take a look at what startups are working on!

Future of Real-time Data Systems: A Recap of Current’22

The Cloud Native Computing Foundation brings Kubecon, its flagship conference, to promote education and advancement of cloud native computing. Kubernetes, Envoy, and Prometheus are just a few of the rapidly expanding cloud-native projects that the Cloud Native Computing Foundation supports, oversees, and directs as a member of the Linux Foundation. This time it's happening from October 24 – 28, 2022, in Detroit, Michigan.

Kubecon + CloudNativeCon: Top 3 reasons for attending it

We’re excited to announce that we have secured $36M in Series A funding to meet surging demand for streaming databases that enable users to make real-time decisions. This brings the total amount raised to more than $40 million. We plan to use this funding to accelerate our mission of democratizing stream processing technology that until now has been available to only a select few organizations.

RisingWave Labs Raises $36M in Series A Funding for its Stream Processing Platform 

Recently, RisingWave Labs' team attended Current 2022, where over 2,000 people from around the globe gathered to see what is new and exciting in the world of data streaming technologies. There was a wide range of sessions and events available — attendees could get a taste of what is happening in the world of streaming.

Current 2022: 4 themes you should not miss

Hey, we are at Booth #ST9. Welcome to stop by and say hi! 

Current 2022: The Next Generation of Kafka Summit

We believe focusing on messaging is just as important as the focus on product differentiation. Keeping this in mind, we are changing our company name to RisingWave Labs. After all, RisingWave is at the heart of our innovation. And this change will allow us to stay close to our product and the RisingWave community.

We are RisingWave Labs now!

Shunjie Ding

RisingWave is a cloud-native streaming database. It is designed to fully leverage the elasticity provided by the cloud. However, deploying and maintaining RisingWave on the cloud is still a challenging technical problem. To tackle the problem, we developed the RisingWave operator, an extension that features the operations of managing RisingWave on Kubernetes. It supports out-of-the-box restarting, scaling, and upgrading operations on a RisingWave cluster. The RisingWave operator is now open-sourced and it is one step further toward the cloud era for RisingWave.

One Step Towards Cloud: The RisingWave Operator

Everyone is talking about the modern data stack (MDS) nowadays. I am a data system person. I started building core database systems in the big data era, and have witnessed the birth and prosperity of cloud computing over the last decade. But the first time I came across the term “modern data stack,” I felt confused - is it just yet another buzzword that the cloud service vendors created to attract people’s eyeballs? There are so many articles online, but most of them are quite markety and salesy. After running a startup building core systems in the modern data stack domain for a while, I would like to share my thoughts. In this article, I will explain “modern data stack” to you in simple terms, and discuss why modern data stack can really matter in companies.

Everything You Wanted to Know but were Afraid to Ask about Modern Data Stack (part I)

Emily Le

Technical Writer @ RisingWave Labs

As companies collect increasing amounts of data daily, it is crucial to query and analyze the data to make timely decisions quickly. To support such business scenarios, OLAP and stream processing are two popular techniques that promise to deliver analytical results with low latency. However, since both techniques claim to target real-time analysis, it can be quite confusing when trying to understand the differences between them and when to use them. What is OLAP? What is stream processing? How are they different from each other? This blog answers these questions by comprehensively comparing OLAP and stream processing.

Real-time analytics: how is OLAP different from stream processing?

Heng Ma

Tao Wu

Product Manager @ RisingWave Labs

Social media platforms like Twitter process a high volume of messages every second. To gain insights from social media platform data in real-time, companies need to adopt stream processing in their tech stack. However, this can be a struggle as many stream processing frameworks are complicated to design and implement. This blog presents how RisingWave can seamlessly work with Apache Pulsar to provide a real-time data streaming solution, making real-time applications much easier to build and maintain.

Tutorial: Pulsar + RisingWave for Fast Twitter Event Processing

Open-source mascots have been widely considered synonyms for their projects/products. So, it should come as no surprise that we thought RisingWave too, deserves a mascot that will add an element of fun to the often-intimidating world of stream processing. Meet Wavy, a squid that won our hearts and will help us keep RisingWave a welcoming and friendly community.

Hello World, Meet Wavy

The amount of streaming data has grown explosively over the past few years. A lot of businesses realize that they need to move to stream processing, but they have a hard time figuring out the route to take. 

Build and Maintain Real-time Applications Faster and Easier with Redpanda and RisingWave

Two months ago, we open-sourced RisingWave, a cloud-native streaming database. RisingWave is developed on the mission to democratize stream processing — to make stream processing simple, affordable, and accessible.

On the Way to Democratized Stream Processing: RisingWave’s Roadmap

As an early-stage database startup, we entirely deleted our C++ codebase after 7 months of development and rewrote everything from scratch in Rust programming language. Here's how we arrived at a decision and why we think this was one of the best decisions ever.

Building a Cloud Database from Scratch: Why We Moved from C++ to Rust

Two weeks ago, we open-sourced RisingWave, a cloud-native streaming database. And in no time, the project gained lots of attention from the open-source community and became the #1 trending project written in Rust!

Is RisingWave the Next Apache Flink?

Stream processing is a well-studied topic. Over the last few decades, researchers and practitioners have devoted substantial efforts to developing fast, scalable, and reliable streaming systems tailored for real-time analytics over streaming data.