Chain Time Streaming with Amp, Part 1

Introduction

Amp is a blockchain-native database. Being a database, all data is represented as tables and rows. Being blockchain-native, it must also be a streaming system, ingesting new blocks efficiently and keeping data consistent across forks and chain reorganizations.

This blog series explores the streaming side of Amp, with this first post establishing the theory behind it and what it means to process data in chain time. We will also dive into the special case of streaming joins.

Streams and timestamps

Blockchains are always mining new blocks, so if you want to analyze blockchain data, you will need a streaming system. Traditional streaming systems have two notions of time:

• Event time: the timestamp at which a source generated an event
• Processing time: the timestamp at which the system received the event

The short story is that users expect their system to respect event time, but due to events arriving late or out of order, processing time often imposes cutoffs or reordering of data. This is a common cause of inconsistency and non-determinism in streaming systems.

Blockchain consensus gives a direct solution to the problem of consistent ordering. There is consensus on a global block order, and within each block on the order of their transactions. Even so, blockchains introduce their own particularities: native notions of time and progress such as block numbers and block hashes, but more critically, the possibility of chain reorgs and forks. Enter chain time.

Chain time

A traditional streaming system would have timestamps as its hardcoded representation of time. While blocks may have a timestamp attached to them, they are not sufficient. A blockchain database needs to keep data consistent across forks and reorgs, and therefore block numbers and block hashes are crucial to reason about ordering and forking of the chain.

Amp bridges block numbers and hashes, respectively, to the concepts of watermark and cursor, which are more familiar to a streaming data analyst. Block numbers are how we typically reason about blockchain progress, while block hashes are used to unambiguously address a position in a DAG of forks.

So, to follow the transition from blockchain data to a streaming system, we can update our mental models with these analogies: block numbers are watermarks; block hashes are cursors.

Streaming query execution

Amp is a SQL database, but it must also behave as a streaming system. Raw blockchain data is represented by tables such as blocks, transactions and logs. Rows are continuously being appended to them.

Processing is done in microbatches. Within each microbatch, SQL execution actually happens: Amp applies the SQL statement to each microbatch, projecting columns and filtering rows by the WHERE clause. Amp uses DataFusion, an embedded Rust SQL engine for high-throughput processing. Microbatches are processed sequentially, but within each microbatch execution can be parallelized by DataFusion. In real deployments, a single microbatch can span hundreds of thousands of blocks.

Every row receives a watermark, which here is simply the associated block number. Each microbatch is also associated with a final watermark and cursor. In this way, chain time is threaded from the source raw data through layers of stream processing. The transformed data inherits the watermark and cursor of its source.

Resumable Queries

One powerful feature enabled by chain time cursors is resumable streaming queries. A long-running query is subject to network disconnections and timeouts. To avoid being notified twice by the same transaction across reconnects, we use the cursor. In a streaming gRPC query, Amp exposes the cursor inside the app_metadata field. By storing the last received cursor client-side, the query can be resumed by passing it in the amp-resume request header.

Will it stream?

Projections and filters can trivially be applied block-by-block, or even row-by-row, and the end result will be the same as if we had applied it over the entire table all at once. This property is very convenient for streaming, as we can process a new batch of data in isolation from the historical data.

However, not all SQL operations are created equal for streaming. Projections and filters are stateless and compose naturally with microbatch execution. Stateful operations like GROUP BY and JOIN are a different class of problem. Keeping the high-throughput properties of Amp while adding support for this feature is a real technical challenge.

Streaming joins

The essential relational operator in SQL is the join. In the blockchain domain, it appears constantly: joining transactions to logs, transfers to addresses, contracts to their events. For non-streaming queries, our DataFusion SQL execution layer can directly process it. But DataFusion is designed for high-throughput columnar execution of batch queries, not for stateful streaming.

To solve this problem, we needed to go back to research fundamentals. We can more formally state the problem as:

Given two tables T and S, and a pre-computed join output T ⋈ S, we must efficiently compute the join update Δ(T ⋈ S), given ΔT and ΔS.

Here ⋈ denotes the join and Δ denotes the delta, i.e. the set of new or changed rows. Crucially, an inner join is an append-only operation: it only adds rows to its output over time. This problem is widely studied in database research, particularly in the IVM (incremental view maintenance) literature. The join update rule is:

Δ(T ⋈ S) = (ΔT ⋈ S) ∪ (T ⋈ ΔS) ∪ (ΔT ⋈ ΔS)

Relative to the naive approach of recomputing the join from scratch every time, the efficiency gain is clear. All joins involve the delta rows, which have a smaller cardinality relative to the entire history. Bringing this back to Amp, we can compute each of the three terms using DataFusion, mapping the deltas directly to Amp's microbatch execution model.

Conclusion

Blockchains introduce challenges that traditional stream processing systems were not designed to handle. Chain time is the model Amp uses to process blockchain streams reliably.

If you are building systems that depend on real-time blockchain data, request a demo to see how chain-time streaming works in practice.