repository·main·Indexed 19 days ago
https://github.com/electric-sql/d2tsA TypeScript implementation of Differential Dataflow for building type-safe, high-performance incremental data pipelines. It enables efficient computations over changing data streams by only recomputing modified parts of the data. Features include a rich set of pipeline operators (map, filter, join, reduce, count), D2QL for SQL-like query structures, and integration with ElectricSQL ShapeStreams for real-time database change processing. Also includes D2Mini, a simplified version for Incremental View Maintenance (IVM) without multi-dimensional versioning.
D2Mini is a minimal implementation of the D2TS dataflow graph library. It is a simplified version designed for Incremental View Maintenance (IVM) based on Differential Dataflow, but without the complexities of multi-dimensional versioning.
While the API is almost identical to @electric-sql/d2ts, D2Mini simplifies the workflow by removing the requirement to:
A MultiSet is a map of values to their multiplicity, used to represent changes (inserts and deletes) to a collection.
1.-1.[key, value]. This is essential for joins and grouping operations.Example of a keyed MultiSet for comments keyed by userId:
// MultiSet of new "comments" keyed by userId
const multiSet = new MultiSet<[string, Comment]>([
[['321', { id: '1', text: 'Hello, world!', userId: '321' }], 1],
[['123', { id: '2', text: 'Hello, world!', userId: '123' }], 1],
])
// Representing an insert and a delete
const changes = new MultiSet<[string, Comment]>([
[['321', { id: '1', text: 'Hello, world!', userId: '321' }], 1], // Insert
[['123', { id: '2', text: 'Hello, world!', userId: '123' }], -1], // Delete
])The integration connects real-time database changes to an incremental computation graph using the following pattern:
users, issues, comments) from a single Electric instance.electricStreamToD2Input: A helper function used to bridge ElectricSQL streams into D2TS inputs.D2TS relies on several specialized data structures to manage differential dataflow operations:
MultiSet: Used to represent collections where elements can have multiplicities.Version: Manages partially ordered versions to track data evolution.Antichain: Used for managing frontiers within the dataflow.Index: Responsible for storing versioned operator state.D2TS provides two categories of operators:
src/operators/.src/sqlite/operators/.D2QL queries are defined as TypeScript objects. The structure mimics SQL syntax using specific keys for selection, source tables, filtering, and joins.
Key components include:
select: An array containing column references (e.g., '@column') or objects for aliasing (e.g., { alias: '@column' }).from: The name of the primary table.where: An array representing conditions, typically in the format ['@column', 'operator', 'value'].join: An array of join objects specifying the type (e.g., 'inner', 'left'), the from table, and the on condition.const query = {
select: ['@column1', { alias: '@column2' }],
from: 'table_name',
where: ['@column', '=', 'value'],
join: [{
type: 'inner',
from: 'other_table',
on: ['@table.column', '=', '@other_table.column']
}]
};D2TS uses versions to track data state and frontiers to represent the lower bound of future data.
Versions are a lattice of integers. While you can use multidimensional versions, most use cases only require a single integer.
v() helper function to create Version objects efficiently. This ensures object reuse for faster equality checks.sendData and sendFrontier.An Antichain is a set of disjoint versions representing the frontier. You can create one using the Antichain constructor or pass a single integer directly to sendFrontier.
// Creating a version
const version = v(1)
const multiVersion = v([1, 2])
// Sending data with a simple integer version
input.sendData(1, new MultiSet([[1, 1]]))
// Creating an Antichain (frontier)
const frontier = new Antichain([v(1), v([2])])
// Sending a frontier with a simple integer
input.sendFrontier(1)The execution of a D2TS dataflow graph involves several coordinated components:
src/graph.ts and src/D2.ts.To use D2Mini, you follow a lifecycle of creating a graph, defining inputs, building a pipeline of operators, finalizing the graph, sending data as changes, and running the computation.
MultiSet, which represents a change to the data rather than the state itself. An element with a multiplicity of 1 represents an Insert, while a multiplicity of -1 represents a Delete.map, filter, and debug applied to a stream.new D2() graph.graph.newInput<T>()..pipe().graph.finalize() to lock the graph structure.input.sendData(multiSet).graph.run().import { D2, map, filter, debug, MultiSet } from '@electric-sql/d2ts'
// 1. Create a new D2 graph
const graph = new D2()
// 2. Create an input stream
const input = graph.newInput<number>()
// 3. Build a pipeline
const output = input.pipe(
map((x) => x + 5),
filter((x) => x % 2 === 0),
debug('output'),
)
// 4. Finalize the pipeline
graph.finalize()
// 5. Send data (MultiSet represents changes: [value, multiplicity])
input.sendData(
new MultiSet([
[1, 1], // Insert 1
[2, 1], // Insert 2
[3, 1], // Insert 3
]),
)
// 6. Process the data
graph.run()To use D2TS in your project, install the package using npm:
npm install @electric-sql/d2tsYou can execute the D2QL example files using tsx. Replace EXAMPLE_FILENAME.ts with the specific file you wish to run from the examples/d2ql/ directory.
npx tsx examples/d2ql/EXAMPLE_FILENAME.tsFor larger datasets or when persistence is required, several operators support a SQLite backend. You can pass a database wrapper as the final argument to the operator.
Supported operators with SQLite:
consolidate()count()distinct()join()map()reduce()Operators will automatically create necessary tables and indexes. It is recommended to use the same database instance for all operators to ensure state is stored in a single location.
To use better-sqlite3, wrap it with BetterSQLite3Wrapper.
import Database from 'better-sqlite3'
// ...
const sqlite = new Database('./my_database.db')
const db = new BetterSQLite3Wrapper(sqlite)
const output = input.pipe(consolidate(db))