subql

SQL subscription dispatch engine for Change Data Capture fanout.

subql dispatches CDC row events to consumers based on SQL WHERE subscriptions. Predicates compile once. Equivalent SQL is deduplicated across subscribers, then hybrid indexes (equality, range, IS NULL, and a fallback set) prune candidates before the VM evaluates. The VM honors SQL three-valued logic (TRUE, FALSE, UNKNOWN) and routes session-bound and durable subscriptions through one path. With the std feature, predicate state persists to durable shards with background merge. WAL input is pluggable: PgOutput, wal2json v1 and v2, Debezium, and Maxwell parsers feed the same path, including table-level TRUNCATE. Streaming aggregates cover COUNT(*), COUNT(col), SUM(col), AVG(col), and the variance/standard-deviation family (VAR_POP, VAR_SAMP, STDDEV_POP, STDDEV_SAMP). Any SQL sqlparser accepts (multiple dialects) works as a predicate.

Quick Start

With the diesel-typed feature, a subscription is a diesel query. Diesel checks the columns, the types of the values, and the comparison between them at compile time, and subql takes the placeholder SQL and serialized binds that diesel's own backend serializer produces. The predicate never has to exist as a string a caller could mistype.

# #[cfg(feature = "diesel-typed")] {
use diesel::prelude::*;
use sql_traits::structs::ParserDB;
use sqlparser::dialect::PostgreSqlDialect;
use subql::backend::{Postgres, Value};
use subql::testing::TestEvent;
use subql::{catalog_helpers, DefaultIds, SubscriptionEngine};

diesel::table! {
    orders (id) {
        id -> Int4,
        amount -> Int4,
        status -> Text,
    }
}

let catalog = ParserDB::parse::<PostgreSqlDialect>(
    "CREATE TABLE orders (id INT PRIMARY KEY, amount INT, status TEXT);",
)?;
let orders_id = catalog_helpers::table_id::<Postgres, _>(&catalog, "orders").unwrap();
let mut engine: SubscriptionEngine<TestEvent<Postgres>, DefaultIds, ParserDB> =
    SubscriptionEngine::new(catalog, PostgreSqlDialect {});

engine.register_select_typed::<diesel::pg::Pg, _>(
    42,
    &orders::table.filter(orders::amount.gt(100)),
)?;

let event = TestEvent::<Postgres>::insert(
    orders_id,
    vec![Value::Int(1), Value::Int(250), Value::String("paid".into())],
)
.with_pk_columns([0u16]);

let notifs = engine.consumers(&event)?;
assert_eq!(notifs.inserted(), vec![42]);
# }
# Ok::<(), Box<dyn std::error::Error>>(())

Placeholder SQL plus typed binds is the engine's own contract, and register_select_typed is a producer for it. It renders the query exactly as diesel would send it and decodes each backend's serialized binds into typed values, so every subscription keeps the SQL the database can re-run when the engine hands the answer back to a re-read. A raw-text subscription (SubscriptionRequest::new(42, "SELECT * FROM orders WHERE amount > 100")) enters the same compile, index, and dedup path. SQLite and MySQL join through the diesel-typed-sqlite and diesel-typed-mysql features.

For reconnect catchup, match_rows(&event) on SubscriptionEngine and AutoResolvingEngine matches every consumer an event reaches in one pass through the table's index, and matches_consumer(&event, consumer_id) evaluates only that consumer's bound row predicates, returning a ConsumerMatch with scalar inserted, updated and deleted flags. Both read current membership admissions and report per-subscription unanswered cells, evaluation failures and the checkpoint. Neither reads connectors, folds aggregates, moves memberships, stamps activity, or reports narrowings, so replaying an event after later ones leaves the engine as it was.

Caller-bound read rules

With membership-term, with_translator intersects caller-bound row subscriptions with the table's complete classified public read rule. Served::read_rule_folded reports whether the intersection is served in process. Rules that cannot be classified or served keep query-only matching and require the caller's visibility check.

The fold admits no rows when the table has row security and no permissive read policy.

Identity and subject-set comparisons use separate term slots, including when they compare one column. Empty subject sets admit nobody through the set comparison. describe_terms exposes seed reads for membership-table rules, whose current admissions, granting subjects, and membership watches persist in durable shards.

unregister_query accepts the original query interest, and complete owner transfers report both audiences. Missing old-image cells remain unanswered and require a read before a transition can be settled.

Streaming Aggregates

Alongside row-match subscriptions, register an aggregate instead of a SELECT *. Diesel spells COUNT(*), COUNT(col), SUM(col) and AVG(col) as .count(), diesel::dsl::count(col), sum(col) and avg(col). The variance/stddev family (VAR_POP/VAR_SAMP/STDDEV_POP/STDDEV_SAMP) has no diesel built-in and arrives as SQL text through register. The engine keeps the running value and reports it whenever it moves, so the caller stores nothing and folds nothing.

A registration answers with an aggregate_bootstrap, a runnable query for the starting numbers. Run it, then pass an AggregateSeedInstall to Install::install with the decoded row and the fence the read reports, which the shipped Postgres connectors take from the read's own snapshot. The fence is what lets the engine drop every change the read already saw, including one the stream delivers after the install, rather than counting it twice. Until the numbers land the subscription reports nothing, and a read the engine cannot line up against what it folded is refused with a AggregateInstallError so the caller can reset_aggregate and read again.

Aggregate subscribers never appear in consumers() output, and vice versa. UPDATE deltas need both old and new row images, so a source that omits old images (before / old) gets an error for update events. A TRUNCATE needs nothing from the caller: the table is empty afterwards, so the engine empties the value itself and reports it.

# #[cfg(feature = "diesel-typed")] {
use diesel::prelude::*;
use sql_traits::structs::ParserDB;
use sqlparser::dialect::PostgreSqlDialect;
use subql::backend::{Postgres, Value};
use subql::testing::TestEvent;
use subql::{catalog_helpers, AggValue, DefaultIds, SubscriptionEngine};

diesel::table! {
    orders (id) {
        id -> Int4,
        amount -> Int4,
        status -> Text,
    }
}

let catalog = ParserDB::parse::<PostgreSqlDialect>(
    "CREATE TABLE orders (id INT PRIMARY KEY, amount INT, status TEXT);",
)
.expect("the DDL parses");
let orders_id =
    catalog_helpers::table_id::<Postgres, _>(&catalog, "orders").expect("orders is cataloged");
let mut engine: SubscriptionEngine<TestEvent<Postgres>, DefaultIds, ParserDB> =
    SubscriptionEngine::new(catalog, PostgreSqlDialect {});

// Live count of active orders, and their running total, both for consumer 42.
let counted = engine
    .register_select_typed::<diesel::pg::Pg, _>(
        42,
        &orders::table.filter(orders::status.eq("active")).count(),
    )
    .expect("the count registers");
let totalled = engine
    .register_select_typed::<diesel::pg::Pg, _>(
        42,
        &orders::table
            .filter(orders::status.eq("active"))
            .select(diesel::dsl::sum(orders::amount)),
    )
    .expect("the sum registers");

// Starting numbers over an empty table. Nothing has been folded yet, so this
// read cannot have raced a change and needs no fence.
subql::Install::install(
    &mut engine,
    counted.subscription_id,
    subql::AggregateSeedInstall {
        // An empty table: a NULL total over no contributing rows, and a
        // contributor count of zero.
        rows: vec![vec![Value::Null, Value::Int(0)]],
        fence: None,
    },
)
.expect("the count's numbers land");
subql::Install::install(
    &mut engine,
    totalled.subscription_id,
    subql::AggregateSeedInstall {
        rows: vec![vec![Value::Int(0)]],
        fence: None,
    },
)
.expect("the sum's numbers land");

let event = TestEvent::<Postgres>::insert(
    orders_id,
    vec![Value::Int(1), Value::Int(250), Value::String("active".into())],
)
.with_pk_columns([0u16]);

// One report per subscription, because one consumer's two aggregates are two
// separate values.
let mut updates = engine.aggregate_updates(&event).expect("the event folds");
updates.sort_by_key(|update| update.subscription);
assert_eq!(updates[0].subscription, counted.subscription_id);
assert_eq!(updates[0].consumer, 42);
assert_eq!(
    updates[0].change,
    subql::AggregateValueChange::Set(subql::AggregateResultValue::Folded(
        AggValue::CountStar(1),
    )),
);
assert_eq!(
    updates[1].change,
    subql::AggregateValueChange::Set(subql::AggregateResultValue::Folded(
        // `amount` is an `INT`, whose sum is a `bigint` on PostgreSQL.
        AggValue::Sum(Some(subql::NumericValue::Integer(250))),
    )),
);
# }

Aggregate variants

SQL	AggValue variant	Notes
SELECT COUNT(*) FROM t WHERE ...	CountStar(i64)	+/-1 per matching row
SELECT COUNT(col) FROM t WHERE ...	CountColumn(i64)	skips NULL cells
SELECT SUM(col) FROM t WHERE ...	Sum(Option<NumericValue>)	exact, in the type the engine sums into: bigint, numeric/DECIMAL, integer or double. Infinity and NaN fold as the engine folds them. None when no row contributes, which is what every engine answers
SELECT AVG(col) FROM t WHERE ...	Avg(Option<NumericValue>)	the engine's own division of the exact total by the count: numeric on PostgreSQL, DECIMAL on MySQL, a real on SQLite. None when no row contributes
SELECT VAR_POP(col) FROM t WHERE ...	VarPop(Option<f64>)	None until one row contributes, and zero at exactly one
SELECT VAR_SAMP(col) FROM t WHERE ...	VarSamp(Option<f64>)	None until two rows contribute
SELECT STDDEV_POP(col) FROM t WHERE ...	StddevPop(Option<f64>)	the square root of VAR_POP
SELECT STDDEV_SAMP(col) FROM t WHERE ...	StddevSamp(Option<f64>)	the square root of VAR_SAMP

Type validation

Column types come from the SQL DDL parsed into ParserDB. When a column's type can be determined (e.g. INT, REAL, TEXT), the engine will not fold SUM or AVG over a non-numeric column (Bool, String). Such a query still registers, on a tier that re-reads it against the database, and Registered::not_served_because carries the reason it is not folded in process.

use sql_traits::structs::ParserDB;
use sqlparser::dialect::PostgreSqlDialect;
use subql::backend::Postgres;
use subql::testing::TestEvent;
use subql::{DefaultIds, SubscriptionEngine, SubscriptionRequest};

let catalog = ParserDB::parse::<PostgreSqlDialect>(
    "CREATE TABLE products (price REAL, name TEXT, id INT PRIMARY KEY);",
)?;
let mut engine: SubscriptionEngine<TestEvent<Postgres>, DefaultIds, ParserDB> =
    SubscriptionEngine::new(catalog, PostgreSqlDialect {});

// Accepted, price is Float:
engine.register(SubscriptionRequest::new(
    1, "SELECT SUM(price) FROM products WHERE id > 0",
))?;

// Registered, but not folded in process: name is String, so the answer comes
// from re-reading rather than from the change stream.
let rows = engine.register(SubscriptionRequest::new(
    1, "SELECT SUM(name) FROM products WHERE id > 0",
))?;
assert!(rows.served().is_none());
assert!(rows.not_served_because.is_some());

# Ok::<(), Box<dyn std::error::Error>>(())

This example is text on purpose, because it is the case the typed path cannot produce. Diesel rejects sum over a Text column, so a caller on register_select_typed learns of the mistake from the compiler before subql is ever involved.

Replay

ReplayDetector maintains one single-table selected result exactly across complete source transactions from retained state, with no database read, and DurableEngine runs it per instance behind a quota, an acknowledgement ledger, and caller-owned atomic persistence. The contracts are in REPLAY_STATE.md and DURABLE_REPLAY.md.