agora-etl-rs 0.2.0

Rust acceleration layer for agora-etl — high-throughput record dispatch

agora-etl-rs

agora-etl-rs is the optional Rust acceleration layer for agora-etl.

It speeds up proven hot paths in the Agora runtime while preserving the same pipeline semantics, recovery behavior, and public builder model. The goal is not to create a second runtime. The goal is to make the existing runtime cheaper to execute.

agora-etl 0.4.x discovers this package automatically when it is installed. Most teams should interact with it through agora-etl, not by wiring the Rust classes directly.

What Ships in 0.2.0

agora-etl-rs 0.2.0 provides capability metadata plus native implementations for selected runtime primitives:

• record_buffer
• metrics_accumulator
• linear_batch_buffer
• checkpoint_state
• memory_tracer
• sync_builtin_chain_executor
• csv_arrow_writer
• jsonl_arrow_writer
• jsonl_arrow_reader

These capabilities are consumed by agora-etl 0.4.x through its acceleration facade and surfaced in explain output, run summaries, and agora doctor.

When to Install It

Install agora-etl-rs when a pipeline is CPU-sensitive enough that Python-side buffering, metrics, or batch bookkeeping show up in profiles, or when a run must fail fast if acceleration is unavailable.

Do not install it expecting different runtime semantics. agora-etl-rs should make hot paths faster, not change how a pipeline behaves.

Requirements

• Python >= 3.11
• agora-etl >= 0.4.0,<1
• A supported CPython runtime

Installation

Install both packages explicitly:

pip install agora-etl agora-etl-rs

Or use the convenience extra from agora-etl:

pip install "agora-etl[rs]"

Verifying the Extension

Check that the native module is importable:

from agora_rs import RUST_AVAILABLE, capabilities, is_available, version

print(RUST_AVAILABLE)   # True when the extension loaded
print(is_available())   # True when native bindings are active
print(version())        # "0.2.0"
print(capabilities())   # tuple[str, ...]

When the extension is unavailable, agora_rs still exposes compatibility stubs so package metadata can be inspected safely. In that state:

• RUST_AVAILABLE is False
• is_available() returns False
• capabilities() returns ()

Using It Through agora-etl

agora-etl-rs is selected by agora-etl through the acceleration boundary in agora 0.4.x.

Declarative configuration:

[performance]
acceleration = "auto"      # auto | off | required
profile = "balanced"       # balanced | throughput | low_latency

Code-level configuration:

from agora import DeliveryConfig, Pipeline

bound = Pipeline(source).build(
    sink,
    config=DeliveryConfig(
        acceleration_mode="required",
        performance_profile="throughput",
    ),
)

Interpretation:

• auto uses compatible Rust paths when installed
• off forces pure-Python execution for debugging or comparison
• required fails before source or sink open if acceleration is unavailable

Useful places to inspect the chosen fast path:

explain = pipeline.explain()
print(explain.acceleration.to_dict())

summary = await pipeline.run()
print(summary.runtime.to_dict())

From the CLI:

agora doctor --config agora.toml
agora doctor --json --config agora.toml

Public Python Surface

Most users only need:

• agora_rs.__version__
• agora_rs.version()
• agora_rs.capabilities()
• agora_rs.is_available()
• agora_rs.RUST_AVAILABLE

Advanced users and internal integration tests may also use the native classes directly:

• RecordBuffer
• MetricsAccumulator
• LinearBatchBuffer
• CheckpointState
• InMemoryTracer
• SyncBuiltinChainExecutor
• CsvArrowWriter
• JsonlArrowWriter
• read_jsonl_record_batches(...)

That surface is lower-level than agora-etl and should be treated as an acceleration substrate, not a replacement for the main builder API.

What It Accelerates

Area	Python path	Rust path
Source prefetch buffering	asyncio.Queue style coordination	bounded record buffer with native synchronization
Runtime metrics counting	Python-side per-record updates	batched native accumulation
Linear write batching	Python list and bookkeeping objects	native batch buffer
Checkpoint cadence checks	Python object bookkeeping	native checkpoint state helper
In-memory tracing helpers	Python span storage	native span container
Built-in sync middleware chain	Python dispatch loop	native chain executor
CSV / JSONL Arrow file paths	Python fallback path	native writer / reader helpers

Not every run will activate every primitive. agora-etl decides that per lane, per writer shape, and per performance profile.

Building From Source

This package lives inside the Agora monorepo.

git clone https://github.com/thanhtham010891/agora-etl.git
cd agora-etl/packages/agora-rs

Create a virtual environment and install build tooling:

python3.11 -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install "maturin>=1.0,<2" "pytest>=8.0" "pytest-asyncio>=0.24"

Build and install the extension into the active environment:

maturin develop --release

Build distributable artifacts:

maturin build --release --out dist
maturin sdist --out dist

Testing and Quality Checks

Run the Python test suite:

pytest tests/ -v

Run Rust formatting and linting:

cargo fmt --check
cargo clippy --all-targets --all-features -- -D warnings

cargo test is not the primary test entrypoint here because the package uses pyo3 abi3 bindings and the meaningful coverage lives in the Python-driven integration tests after a maturin build.

Release Notes

The release workflow expects the GitHub release tag to match the package version exactly:

v0.2.0

Artifacts published by CI include:

• platform wheels
• source distribution

License

Apache-2.0. See LICENSE and NOTICE.