Raw capture — raw/flare-db.md

url: https://github.com/flare-db/flare-db title: FlareDB author: Ganesh Sivakumar date_fetched: 2026-07-08 date_published: 2025

FlareDB — Full Repo Analysis

Overview

FlareDB is an Apache Beam-native streaming database written in Rust. It accepts Beam pipeline jobs (submitted via a Java SDK runner), fuses the pipeline graph into executable stages, and executes them either natively (for core transforms like Impulse and GroupByKey) or by delegating to a Java Beam SDK harness. PCollections are persisted in an embedded Arrow-native LSM-tree store (Tonbo).

Version: 0.1.8. License: Apache 2.0. Single author: Ganesh Sivakumar.

Project Structure

flare-db/
  flaredb/             # core engine (Rust)
    src/
      main.rs          # entry: starts 6 gRPC servers on 127.0.0.1:8099
      lib.rs           # module declarations, DEFAULT_API_SERVICE_URL
      engine/
        mod.rs
        executor.rs    # StageExecutor: BFS traversal of executable graph
        store.rs       # FlareElementStore, BeamRecord, Arrow schema derivations
        coders.rs      # Beam wire-format coders (varint, KV, iterable, windowed value)
        harness/       # Beam Fn API servers
          control.rs   # FlareControlService: register/process bundle gRPC
          data.rs      # FlareDataService: element streaming with multiplexer
          log.rs       # FlareLogService
          state.rs     # FlareStateService
      fusion/
        fuser.rs       # GreedyPipelineFuser: greedy fusion algorithm
        pipeline.rs    # QueryablePipeline, ExecutableGraph, FusedPipeline
        stage.rs       # ExecutableStage, CollectionConsumers, SiblingKey
        refs.rs        # SideInputRef, TimerRef, UserStateRef
      transforms/
        mod.rs         # FlareTransform trait, from_urn() dispatch
        impluse.rs     # Impulse: produces single empty-byte[] element
        gbk.rs         # GroupByKey: in-memory HashMap-based grouping
      jobservice/
        server.rs      # FlareJobService: Beam Job API gRPC implementation
        job.rs         # Job, JobStore, fuse_pipeline()
        artifact.rs    # ArtifactStore
        urns.rs        # All Beam URN constants
      utils/
        errors.rs      # Error types
        macros.rs      # check_argument! macro
        path.rs        # Directory layout helpers
        visualization.rs # DOT graph output
  flare-cli/           # CLI tool (Rust)
    src/main.rs        # flare init/up/down commands
  beam-model-rs/       # Generated protobuf Rust types for Beam model
  runner-sdk/java/     # Java runner SDK (FlareRunner, FlarePipelineJob)
  beam/model/          # Apache Beam protobuf definitions (vendored)
  example/wordcount/   # WordCount example pipeline

Code size: flaredb/src ~4,000 lines of hand-written Rust; beam-model-rs ~17,000 lines of generated protobuf types; flare-cli ~630 lines total.

Full Architecture

Entry Point (main.rs)

flare_up() starts 6 gRPC servers on a single tonic Server::builder():

  1. JobServiceServer — Beam Job API: PrepareJob, RunJob (plus stubs for GetJobs, GetState, Cancel, etc.)
  2. ArtifactStagingServiceServer — Receives staged pipeline artifacts
  3. BeamFnControlServer — Register process bundle descriptors + send process bundle instructions
  4. BeamFnDataServer — Bidirectional element streaming between harness and runner
  5. BeamFnLoggingServer — SDK harness log forwarding
  6. BeamFnStateServer — State API (stub, not yet implemented)

All on one port (127.0.0.1:8099) — a single-node deployment model.

Pipeline Ingestion Flow

  1. Java SDK FlareRunner.run(pipeline) translates Beam pipeline to proto, calls PrepareJob gRPC
  2. FlareJobService.prepare() receives pipeline proto, calls Job::new() which: a. Builds a QueryablePipeline (petgraph-based graph of PTransform/PCollection nodes) b. Runs GreedyPipelineFuser.fuse_pipeline() to group transforms into ExecutableStages c. Builds an ExecutableGraph (DAG of Worker nodes + Runner nodes with ConsumerMetaData edges)
  3. FlareJobService.run() spawns a Java FnHarness process, waits for gRPC connection, then calls StageExecutor.execute_pipeline()

StageExecutor (executor.rs)

Executes the ExecutableGraph in a BFS traversal:

Root (Runner: Impulse) → Worker (fused SDK stages) → Runner (GroupByKey) → ...

For Worker nodes (SDK stages): 1. Register the ProcessBundleDescriptor with the harness via control channel 2. Send ProcessBundleRequest to start the bundle 3. Spawn a background task that reads the input PCollection from the store, encodes elements through the Beam coder pipeline, and sends them to the harness via the data channel 4. Spawn a concurrent task that receives output elements from the harness, decodes them, and writes to the store 5. tokio::select! between harness response and decode task completion, with 60s timeout

For Runner nodes (native transforms): 1. Build a ProcessBundleDescriptor with the transform's spec, PCollections, coders, etc. 2. Register the bundle with the harness 3. Call runner_transform.execute(ctx) directly — reads from store, processes, writes back

Pipeline Fusion (fuser.rs)

GreedyPipelineFuser — the most architecturally interesting module:

  1. Root discovery: Finds all transforms with no inputs, separates into runner (unfused) and SDK (fusible candidates)
  2. Sibling grouping: Consumers of the same PCollection are grouped by (PCollection, Environment) key. Within each key, sibling groups are formed where every member is mutually compatible
  3. Greedy stage fusion: From each sibling group, a GreedyStageFuser grows a stage by walking forward through the pipeline graph, fusing compatible PCollections into the stage until it hits a materialization boundary
  4. Fusion boundaries — a PCollection MUST materialize (write to store) when ANY consumer: - Has state or timers (ParDo with state_specs/timer_family_specs) - Has side inputs - Runs in a different environment - Is a runner-implemented transform (GroupByKey, Impulse) - Is a GBK, CreateView, or Splittable DoFn process element
  5. Deduplication: After fusion, ensure_single_producer() detects PCollections with multiple producers and injects synthetic Flatten transforms to merge partial outputs

After fusion, the pipeline is a FusedPipeline with: - sdk_stages: IndexSet<ExecutableStage> — fused SDK transforms - runner_stages: IndexSet<PTransformNode> — runner-implemented transforms that break fusion

Storage Engine (store.rs)

FlareElementStore is the core state layer:

Beam Coder System (coders.rs)

Hand-written Rust implementations of Beam's standard wire-format coders:

Harness Communication (harness/)

Each Beam Fn API endpoint follows the same pattern:

  1. Server struct holds an Arc<Inner> with outgoing: Mutex<Option<Receiver>> and incoming: Mutex<Option<Streaming>>
  2. When the harness connects, the gRPC handler takes the outgoing Receiver (wrapping it in ReceiverStream) and stores the incoming stream
  3. A paired "channel" struct holds the sender end and provides typed methods (send_elements, register_bundle, etc.)

The data channel is the most complex: it has an ElementStreamMultiplexer that routes elements from the harness back to the correct stage consumer by DataKey {instruction_id, transform_id}. A background task (stream_elements) drains the tonic stream and fans out to per-key unbounded channels.

Runner Transforms

Impulse (impluse.rs): Produces a single PrimitiveValue::Bytes(vec![]) element. This is the Beam root transform — every pipeline starts here.

GroupByKey (gbk.rs): Scans the input PCollection (expecting KV records), builds a HashMap<PrimitiveValue, IterableValue> grouping values by key, and writes GBK records. Pure Rust, no harness involved. This is the critical optimization — avoiding a shuffle through the harness.

CLI (flare-cli/src/main.rs)

Three commands: - flare init — Downloads FlareDB binary + Java harness JAR from GitHub releases, creates ~/.flaredb/{bin,instances} - flare up — Spawns FlareDB process with instance UUID, writes state.json, polls for port 8099 - flare down — SIGTERM → wait → SIGKILL, removes state.json

Java Runner SDK

FlareRunner extends Beam's PipelineRunner<FlarePipelineJob>: 1. Translates pipeline to proto via PipelineTranslation.toProto() 2. Sends PrepareJobRequest → gets preparation_id + staging token 3. Stages artifacts via Beam's ArtifactStagingService 4. Sends RunJobRequest → blocks until job completes

Key Techniques

  1. Greedy fusion with deduplication: The fusion algorithm mirrors the Beam portability framework's approach in Java but reimplemented in Rust with petgraph. The dedup pass injecting synthetic Flattens for multi-producer PCollections is a correctness mechanism often missed by naive fusion implementations.

  2. PCollection-as-database: Unlike most Beam runners that hold PCollections in memory or on disk as opaque blobs, FlareDB stores each PCollection as a typed Arrow RecordBatch in an LSM-tree with schema derived from the actual element types. This means PCollections are queryable and survive beyond pipeline execution.

  3. Schema auto-derivation: derive_schema_from_records() inspects the first batch of elements and determines the Arrow schema. It enforces type homogeneity across batches (e.g., no mixing String and Int64 elements in the same PCollection).

  4. Dual execution model: Runner transforms execute directly in Rust (no serialization, no harness overhead), while SDK transforms go through the Java harness. The fusion algorithm treats runner transforms as materialization boundaries — SDK transforms fuse until they hit a runner transform, then materialize.

  5. WindowedValue wrapping: All elements crossing the runner-SDK boundary are wrapped in WindowedValue (timestamp + windows + pane info). Currently all windows are GlobalWindow, but the codec infrastructure supports general windowing.

Design Decisions & Trade-offs

What they optimized for

What they sacrificed

Interesting trade-offs

Comparison to Related Systems

Dependencies