CSV, Parquet, SQLite, and Kafka I/O

Default CSV read

With just a path, Ibex reads the header row, infers Int64, Float64, or string/categorical columns, and returns a DataFrame.

import "csv";

let trades = read_csv("data/trades.csv");

Null tokens

Pass a comma-separated null specification. <empty> means empty fields should be treated as nulls too.

import "csv";

let train = read_csv("train.csv", "<empty>,NA");

Custom delimiter

This is useful for semicolon-separated files or other non-comma dialects. Quoted commas are still preserved correctly.

import "csv";

let raw = read_csv("quotes.txt", "", ";");

Headerless files

When has_header is false, Ibex synthesizes col1, col2, and so on. This is what the 1BRC benchmark file uses.

import "csv";

let raw = read_csv("measurements.txt", "", ";", false)
    [select { station = col1, temp = col2 }];

What the demo does

The bundled Kafka demo uses Redpanda as a local Kafka-compatible broker. A synthetic tick producer writes JSON ticks into the ticks topic, and two Ibex stream jobs consume that same topic:

• one job groups by symbol and venue and sends live summaries to ws://127.0.0.1:8765
• one job resamples ticks into 5-second OHLC bars and sends them to ws://127.0.0.1:8766

The dashboard at demo/kafka/ws_dashboard.html opens both WebSocket feeds and shows them as two tabs.

Start the demo

# shell
scripts/demo-kafka.sh
scripts/run-kafka-dashboard.sh

# optional terminal viewers
python3 demo/kafka/ws_client.py
python3 demo/kafka/ohlc_ws_client.py

The Docker stack starts Redpanda plus the random tick producer. The launcher script starts two Ibex REPL processes in the background and keeps them running until you stop it.

The Ibex side

import "kafka";
import "websocket";

ws_listen(8765);

let summary = Stream {
    source = kafka_recv(
        "localhost:19092",
        "ticks",
        "ibex-demo-summary",
        "ts:timestamp,symbol:str,venue:str,price:f64,size:i64",
        "poll_timeout_ms=100;consumer.auto.offset.reset=latest;consumer.session.timeout.ms=6000"
    ),
    transform = [by { symbol, venue }, select {
        trades = count(),
        last_px = last(price),
        total_size = sum(size)
    }],
    sink = ws_send(8765)
};

The OHLC job in examples/kafka_ohlc.ibex uses the same source but switches the transform to [resample 5s, by { symbol }, select { ... }] and writes to ws://127.0.0.1:8766.

Kafka-specific notes

• The schema string is explicit and required: use name:type entries like "ts:timestamp,symbol:str,venue:str,price:f64,size:i64".
• For live dashboards, prefer consumer.auto.offset.reset=latest so the stream starts at current traffic instead of replaying old messages.
• poll_timeout_ms is only the wait between polls. It is not a lifetime timeout for the stream.
• For per-message live streams, str is a better demo key type than cat; it avoids per-message categorical dictionary churn.

Read Parquet

Parquet is the cleaner choice when you control the data pipeline and want strong type preservation, compact files, and faster reloads than CSV.

import "parquet";

let prices = read_parquet("data/prices.parquet");

Write Parquet

write_parquet returns the number of rows written. String and categorical columns are stored as UTF-8. Dates and timestamps are preserved as typed Parquet columns.

import "parquet";

let n = write_parquet(prices, "out/prices.parquet");

Minimal SQLite query

read_adbc takes a driver, a database URI or path, and a SQL query. Passing the SQLite driver shared library path is the most portable shape when you are using a conda environment.

When using import "adbc" in the REPL, start Ibex with --plugin-path ./build-release/tools so it can find both adbc.ibex and adbc.so.

import "adbc";

let trades = read_adbc(
    "/path/to/libadbc_driver_sqlite.so",
    "/tmp/trades.sqlite",
    "select symbol, qty, px from trades order by qty desc"
);

1M-row measurements showcase

The helper script below loads the first 1,000,000 rows of the 1BRC benchmark file into SQLite. After that, the Ibex query shape is the same aggregation you would run directly over CSV.

# shell
python3 scripts/import_measurements_sqlite.py \
    --input examples/measurements.txt \
    --limit 1000000 \
    --output /tmp/measurements_1m.sqlite

LD_LIBRARY_PATH=~/envs/ibex/lib:$LD_LIBRARY_PATH \
./build-release/tools/ibex --plugin-path ./build-release/tools

# ibex
import "adbc";

let measurements = read_adbc(
    "~/envs/ibex/lib/libadbc_driver_sqlite.so",
    "/tmp/measurements_1m.sqlite",
    "select station, temp from measurements"
);

measurements[select {
    min_temp = min(temp),
    avg_temp = mean(temp),
    max_temp = max(temp)
}, by station, order station];