dfe-loader

Loads Kafka and gRPC streams into ClickHouse, batching small writes into large inserts. In a DFE deployment it is the last hop of the core data path, taking what dfe-receiver sends over Kafka or direct gRPC and landing it in the tables dfe-schemas defines.

What it gives you

• Routing to db.table from fields in each record, so one loader can feed many tables.
• Columns read from the live table's system.columns: matching fields become typed columns and the rest stays in _json, with no schema compiled in.
• RowBinary inserts by default, so ClickHouse does no JSON parsing on the way in.
• Per-table buffers flushed on row count, byte size or age, whichever comes first.
• At-least-once delivery by default: Kafka offsets commit once per flush, never past a row not yet in ClickHouse or the DLQ. kafka.acknowledgements.enabled: false commits at receipt instead, and a row only a DLQ could take is dropped and counted in dfe_loader_rows_lost_total when no DLQ is working.
• A batch with a bad row is split until only the bad rows are dead-lettered, and a schema drift error refreshes the cached schema and retries.
• Optional GeoIP, reputation and risk columns at load time.

Quick start

The first success needs no broker and no datastore:

cargo build
./target/debug/dfe-loader --config config.example.yaml config-check

That prints [ok] configuration is valid and the resolved config, and opens no connections. --config is a global option, so it goes BEFORE the subcommand -- config-check --config ... is rejected by the argument parser.

To move real data, bring up a local broker and datastore:

docker compose -f docker-compose.dev.yaml up -d
cargo run -- --config config.dev.yaml

Metrics land on http://localhost:9090/metrics, with /livez and /readyz alongside. docker compose -f docker-compose.dev.yaml down -v takes it down.

What it does

dfe-loader reads records off Kafka or a direct gRPC listener, works out which ClickHouse table each one belongs to, and inserts batches. The per-message work is kept cheap and the expensive work deferred to the flush:

Stage	What happens
Parse	JSON only. SIMD parse, payload kept as Arc<[u8]>. A record that is not JSON is dead-lettered
Route	db.table from configured fields, resolved BEFORE flattening
Promote	Schema read from system.columns, matching fields become typed columns, the rest stays in _json
Enrich	Optional GeoIP, reputation and risk columns
Buffer	Per-table, flushed on row count, byte size or age
Insert	RowBinary by default, so ClickHouse skips JSON parsing. JSONEachRow is the fallback
Commit	Kafka offsets commit once per flush cycle, never past a row not yet in ClickHouse or the DLQ -- at-least-once

JSON is the only payload format. MessagePack, supported in DFE/XDR 2.0 and 2.1, is deprecated in DFE 2.2 and no longer accepted: the JSON path (SIMD parsing with sonic-rs, zstd on the wire) is fast enough that MessagePack gave no CPU saving.

Failures degrade rather than stall: a batch with a bad row is binary-split so only the bad rows are dead-lettered, a batch whose insert fails is held and retried with backoff while the commit stays below it, and a schema drift error invalidates the cached schema and retries.

Configuration essentials

One YAML file, named by --config. The chart mounts it at /etc/dfe/loader.yaml. config.example.yaml is the annotated reference, and every setting has a default, so an empty config boots.

Environment overrides use the DFE_LOADER prefix in two forms, and the difference matters: DFE_LOADER_KAFKA_GROUP_ID is the flat single-underscore set, a fixed contract with dfe-engine, while DFE_LOADER__CLICKHOUSE__DATABASE uses double underscores to nest into any config key.

Setting	Default	Note
clickhouse.hosts	localhost:8123	HTTP port family
clickhouse.protocol	http	native (9000, 9440 with TLS) also works, for rowbinary inserts
clickhouse.database	dfe	The connection database, separate from routing.default_db
clickhouse.insert_format	rowbinary	jsoneachrow is the fallback
clickhouse.tls.enabled	unset	The ONLY honoured TLS key. Mount a private CA into the container trust store rather than naming cert files
kafka.group	dfe-loader
routing.default_db / default_table	dfe / main	Where a record lands when no routing field matches
buffer.flush_rows / flush_bytes / flush_age_secs	20000 / 67108864 (64 MiB) / 5	Per-table flush triggers, each greater than 0
metrics.address	0.0.0.0:9090

Full cascade and per-setting behaviour: docs/configuration.md.

Documentation

docs/README.md is the index.

Topic	Doc
Layers and the dynamic insert	docs/architecture.md
Config cascade and behaviour	docs/configuration.md
Running, metrics, hot-reload, DLQ	docs/operations.md
Insert formats, schema cache, types, TLS, DDL directives	docs/clickhouse/
Routing, coercion, enrichment, capture modes	docs/pipeline/
Kafka, gRPC, the common header	docs/transport/
Container and chart publishing	docs/deployment/
Test layout and what needs infra	tests/TESTING.md

CONTRIBUTING.md, SECURITY.md, COMMERCIAL.md.

Issues

Open bugs: this repo | whole DFE suite

License

BUSL-1.1, licensor HYPERI PTY LIMITED. Each version converts to Apache 2.0 on the third anniversary of its first public distribution. See LICENSE and COMMERCIAL.md.

Context

What this is

dfe-loader writes rows into ClickHouse -- the last hop of the DFE core data path and the only component that writes there.

NOT a transform stage, NOT a schema manager. Its only outbound topic is the DLQ. dfe-engine creates the destination tables, and the loader holds records pending schema rather than issuing DDL. The dynamic insert comes from dfe-clickhouse's dynamic module, because the clickhouse crate keeps insertion compile-time typed.

Where things live

Path	Holds
src/main.rs, src/config/	CLI entry, config structs, Config::validate(), flat env
src/pipeline/	Orchestrator and per-message processor -- the hot path
src/routing/, src/transform/, src/enrich/	Routing, coercion and capture, enrichment
src/buffer/, src/clickhouse/	Per-table buffers; query client, schema cache, inserter
tests/	smoke.rs, integration/, e2e/ -- see tests/TESTING.md

Commands that prove a change

make check                                  # hyperi-ci check -- the full local gate
cargo nextest run --lib                     # unit, no infra
cargo nextest run --test smoke              # startup smoke, no infra
cargo nextest run --test integration_tests  # integration, some cases need ClickHouse
cargo nextest run -- --ignored              # e2e, needs Kafka AND ClickHouse

Green lies three ways: skip_if_no_clickhouse!() and skip_if_no_kafka!() return early and PASS when the datastore or broker is absent, the e2e suite is #[ignore] so a plain run never touches the real insert path, and default = [] compiles a feature-gated test out while CI still reports green. ci.yml sets paths-ignore for docs/** and **.md, so a docs-only push runs no jobs.

What tends to bite

Don't	Do	Why
Bump scalo and leave release.helm.library behind	Move release.helm.library in .hyperi-ci.yaml to the same scalo version	The release assembles the chart from the emitted contract on that scalo-service version, and a scalo-service release ships the schema for only the contract version its scalo release writes
Gate a test behind a cargo feature without adding it to .hyperi-ci.yaml	Add default,<feature> to the feature-set list	default = [] (Cargo.toml:304), so it compiles out of every CI run and CI still reports green. helm_contract.rs:64 asserts transport-memory and testcontainers stay listed
Inject a nested config key as DFE_LOADER_SECTION_FIELD	DFE_LOADER__SECTION__FIELD	figment strips exactly DFE_LOADER_, so it arrived as _kafka.sasl.username, matched no field and was dropped silently. Pods ran with no SASL and an empty ClickHouse password (tests/integration/config_reachability.rs)
Set clickhouse.protocol: native with insert_format: json_each_row	rowbinary over native, or http for JSONEachRow	JSONEachRow is an HTTP body format. validate() rejects the pair by name
Name clickhouse.tls.ca_cert_file, cert_file, key_file or skip_verify	Set tls.enabled, mount the CA into the trust store	Only enabled reaches a client. The rest parsed and did nothing, so validate() now fails naming them
Widen the cel range past scalo's	Keep it on >=0.14.5, <0.15	cel::Program crosses the scalo boundary. Wider resolves two semver-incompatible cel crates and Program stops being the same type (Cargo.toml:46-51)
Take clickhouse from git or a [patch]	The crates.io release	deny.toml allows no git source, and dfe-clickhouse builds against the published clickhouse only
Mechanically sync dfe-engine's loader validation to Config::validate()	Read both, keep the divergence	dfe-engine scopes the broker check to the Kafka transport and adds a grpc.listen check this side lacks. A blind sync rejects valid gRPC-only configs at author time

Where this sits

Inbound, declared in dfe-infra/suite.yaml:

• scalo-rs -> dfe-loader (cargo-dep) -- Cargo.toml:36 takes scalo by range for transports, config cascade, CLI, metrics, deployment contract and DLQ. A scalo release reaches this repo here.

• scalo-rs -> dfe-loader (generated-file, lockstep) -- Dockerfile is emitted by scalo::deployment::generate_dockerfile() at schema version 4. Regenerate and commit the diff. Outbound, so what a change here can break:

• dfe-loader -> dfe-infra (image-pin, lockstep) -- dfe-infra/versions.yaml pins this image and the thin chart released here, and Argo deploys that published chart.

• dfe-loader -> dfe-engine (mirrored-logic) -- dfe-engine hand-reimplements this repo's config validation in plugins_builtin/loader.py to catch errors when it authors a config. Nothing is copied, so no script compares them.

Runtime, not build edges: dfe-engine's dfe-schema entry point creates every destination table before the loader can insert, and producers reach the loader over Kafka (*_land / *_load topics) or direct gRPC (default topic main_land) -- in the suite that producer is dfe-receiver.