Krishiv：Rust 原生混合计算引擎，统一批处理 SQL、流式管道与增量视图维护

Krishiv is a Rust-native hybrid compute engine that unifies batch SQL, streaming pipelines, and incremental view maintenance under one Apache Arrow / DataFusion runtime. The same engine runs embedded in your process, as a single-node daemon, or as a distributed cluster.

⚠️ Pre-release — not for production use Krishiv is under active development and has not yet reached its first stable release. APIs, storage formats, and wire protocols may change between versions. We recommend using Krishiv for evaluation, prototyping, and development purposes only. Please wait for the first stable release before deploying to production.

Install

Docker (recommended for getting started)

docker pull ghcr.io/krishivai/krishiv:latest
docker run --rm -it ghcr.io/krishivai/krishiv:latest sql --query "SELECT 42 AS answer"

Or run a single-node daemon with Flight SQL on :50051:

docker run -d --name krishiv -p 50051:50051 ghcr.io/krishivai/krishiv:latest local start

Rust (crates.io)

[dependencies]
krishiv = "0.1"

For library use, add the specific crates you need:

[dependencies]
krishiv-api     = "0.1"   # Session, DataFrame, IncrementalDataFrame
krishiv-delta   = "0.1"   # DeltaBatch, IVM operators
krishiv-connectors = { version = "0.1", features = ["iceberg"] }

Python (PyPI)

pip install krishiv

With optional extras:

pip install "krishiv[arrow]"       # PyArrow + Pandas
pip install "krishiv[iceberg]"     # Iceberg lakehouse support
pip install "krishiv[all]"         # everything

Quick Start

Batch SQL

Rust (Session is sync-friendly; every method has an _async twin)

use krishiv_api::Session;

fn main() -> krishiv_api::Result<()> {
    let session = Session::new();
    session.register_record_batches("orders", vec![orders_batch])?;

    let result = session
        .sql("SELECT status, COUNT(*) AS n FROM orders GROUP BY status")?
        .collect()?;
    println!("{}", result.pretty()?);
    Ok(())
}

Python

import pyarrow as pa
import krishiv as ks

session = ks.Session()
session.register_record_batches("orders", [pa.record_batch(
    {"status": ["a", "b", "a"], "amount": [10.0, 25.0, 5.0]})])

result = session.sql("SELECT status, COUNT(*) AS n FROM orders GROUP BY status").collect()
print(result.pretty())

CLI

krishiv sql --query "SELECT 1 AS value"
krishiv explain --analyze --query "SELECT 1 AS value"

Streaming

import krishiv as ks

session = ks.Session()
events = session.sql("SELECT 1 AS n, 100 AS val, 1000 AS ts UNION ALL SELECT 1, 200, 2000")

windowed = (
    events.to_streaming()
    .with_event_time("ts")
    .key_by("n")
    .tumbling_window(1000)      # ms
)

async for batch in await windowed.execute_stream_async():
    print(batch)

Windowed SQL (TUMBLE / HOP / SESSION) over an unbounded source compiles to the same operators; krishiv stream runs it from the CLI.

Incremental View Maintenance (IVM)

import pyarrow as pa
import krishiv as ks

session = ks.Session()
job = session.ivm("sales_pipeline")

class Revenue(ks.Schema):
    region: str
    total: float

job.register_view(
    "revenue",
    "SELECT region, SUM(amount) AS total FROM sales GROUP BY region",
    Revenue,
    is_materialized=True,
)

# Tick 1 — inserts
job.feed("sales", ks.DeltaBatch.from_inserts(
    pa.RecordBatch.from_pydict({"region": ["us", "eu"], "amount": [100.0, 200.0]})))
print(job.step().total_output_rows, job.snapshot("revenue"))

# Tick 2 — an update is a retraction plus an insertion
before = pa.RecordBatch.from_pydict({"region": ["us"], "amount": [100.0]})
after = pa.RecordBatch.from_pydict({"region": ["us"], "amount": [250.0]})
job.feed("sales", ks.DeltaBatch.from_update(before, after))
job.step()
print(job.snapshot("revenue"))          # us=250, eu=200

ckpt = job.checkpoint()                 # bytes; job.restore(ckpt) later

examples/ has complete programs for every mode (docs/guides/running-examples.md).

Deployment Modes

What's Inside

• Apache Arrow columnar memory — zero-copy between operators
• DataFusion SQL engine — full SELECT, JOIN, GROUP BY, window functions
• Iceberg-first lakehouse — catalog integration, Parquet read/write, snapshot isolation
• Exactly-once semantics — for certified source/sink/checkpoint combinations
• Pluggable connectors — Kafka, S3, Parquet, Iceberg (Delta and Hudi experimental)
• Durable state — RocksDB-backed keyed state with TTL and checkpoint/restore

Crate Map

Building from Source

# Check everything compiles
cargo check --workspace

# Run tests
cargo test --workspace --exclude krishiv-python

# Build single-node binary
cargo build --release -p krishiv --features single-node

# Build distributed + Kubernetes binary
cargo build --release -p krishiv --features full

Docker build

# Fast local image (pre-built binaries)
docker build -f deploy/docker/Dockerfile.fast -t krishiv:local .

# Production image (multi-stage, ~50MB)
docker build -f deploy/docker/Dockerfile.prod -t krishiv:prod .

Documentation

The architecture set in docs/architecture/ covers every feature and every crate:

• Overview · Execution modes · SQL engine · Planning and optimization
• Scheduler and coordinator · Executor and data plane · Shuffle · State, checkpoints, savepoints
• Streaming · Incremental view maintenance · Connectors and lakehouse · Public interfaces
• Security · Observability · Deployment and durability · Configuration
• Performance · Testing and quality · Compatibility and versioning

Also: Engine contracts, Connector SDK, Roadmap, Compatibility, Contributing.

Krishiv is licensed under the Apache License 2.0.