orsx

This workspace contains a Rust library (orsx) and a proc-macro crate (orsx-macros).

The current implementation covers:

1. Schema-driven Postgres migrations (including an online rewrite path for large tables).
2. Columnar retrieval into ColumnarBatch (COPY BINARY and row-wise).
3. Numeric vector compression stored as BYTEA with a small envelope.
4. Flattened wide schemas for “processor outputs” (recursive flatten inside an inline mod, deterministic column ids/order, schema hash, and a deterministic binder/visitor).

This README is intended to describe the repository as it exists right now, including current limitations.

• No ORM and no query builder. You write SQL.
• No cross-database support (Postgres only).
• No automatic “struct ↔ row” mapping for columnar reads (columnar reads return ColumnarBatch, not Vec<MyStruct>).

• orsx/: library (public API)
• orsx-macros/: proc macros (#[derive(OrsxMigrate)], #[derive(OrsxColumnar)], #[derive(OrsxFlatten)], #[orsx::orsx_flatten_module])

Add the crate and use sqlx for connections (orsx re-exports sqlx):

[dependencies]
orsx = { path = "./orsx" }
tokio = { version = "1", features = ["full"] }

You define a table schema in Rust via #[derive(OrsxMigrate)]. At runtime, orsx::Migrations:

• creates tables that do not exist,
• applies “safe ALTER” changes when possible (e.g. add a nullable column),
• otherwise performs an online rewrite:
  • creates a shadow table with the desired schema,
  • installs a trigger that records changed primary keys into a changelog table,
  • backfills data from the original table into the shadow table in chunks,
  • applies changelog catch-up rounds,
• takes a short ACCESS EXCLUSIVE lock to drain remaining changes and swap tables,
• keeps a backup table with the original data.

“Zero-loss-by-backup” here means: when a rewrite happens, the old table is preserved as a backup table (not dropped).

The current online rewrite implementation has constraints that are enforced in code:

• Online rewrite requires either:
  • exactly one primary key column, or
  • MigrationConfig.enable_migration_key = true (adds and uses __orsx_mig_id BIGINT as the rewrite key).
• Introspection assumes the table is in the public schema.
• Constraint handling is limited (single-column PK/unique are tracked; other constraints are not fully modeled).
• Some diffs intentionally trigger rewrite (type changes, column position changes, drop column, etc.).

use orsx::prelude::*;

#[derive(OrsxMigrate)]
#[orsx_table("my_table")]
struct MyTable {
    #[orsx_column(primary_key)]
    id: String,
    name_: String,
    pwt: f64,
}

#[tokio::main]
async fn main() -> Result<()> {
    let db_url = std::env::var("DATABASE_URL").expect("DATABASE_URL");
    let pool = sqlx::PgPool::connect(&db_url).await?;

    let dummy = MyTable { id: "x".into(), name_: "n".into(), pwt: 0.0 };
    Migrations::init(&pool, &[(dummy, None)]).await?;

    Ok(())
}

The migration behavior is controlled by orsx::migrations::config::MigrationConfig:

• enforce_column_order: if true, Postgres physical column order must match the Rust spec order; mismatches become rewrite-required.
• enforce_exact_columns: if true, the live table must contain exactly the columns in the spec (no extras).
• allow_destructive_drops: only relevant when enforce_exact_columns=true; if true, extra DB columns are removed from the live table via rewrite, but the backup table retains the original columns/data.
• allow_column_renames: if true, fields annotated with #[orsx_column(rename_from = "...")] can be renamed via ALTER TABLE ... RENAME COLUMN ....

Example:

use orsx::migrations::config::MigrationConfig;
use orsx::prelude::*;

let cfg = MigrationConfig {
    enforce_column_order: true,
    enforce_exact_columns: true,
    allow_destructive_drops: true,
    ..MigrationConfig::default()
};

Migrations::init_with_config(&pool, &[(dummy, None)], &cfg).await?;

If you want online rewrite for a table spec that does not have exactly one PK column, enable:

• MigrationConfig.enable_migration_key = true

This adds and maintains a BIGINT column named __orsx_mig_id and uses it as the rewrite key.

use orsx::migrations::config::MigrationConfig;
use orsx::prelude::*;

let cfg = MigrationConfig {
    enable_migration_key: true,
    ..MigrationConfig::default()
};

Migrations::init_with_config(&pool, &[(dummy, None)], &cfg).await?;

There are three related mechanisms:

• Primary key: #[orsx_column(primary_key)] (single-column primary key is required for online rewrite).
• Single-column unique: #[orsx_column(unique)]
• Indexes:
  • single-column: #[orsx_column(index ...)]
  • multi-column: table-level index(...) declarations inside #[orsx_table(...)]

#[derive(orsx::OrsxMigrate)]
#[orsx_table("users")]
struct User {
    #[orsx_column(primary_key)]
    id: String,

    #[orsx_column(index)] // btree index on email
    email: String,

    #[orsx_column(index(type = "gin"))] // gin index
    payload: String,
}

#[derive(orsx::OrsxMigrate)]
#[orsx_table("users")]
struct User {
    #[orsx_column(primary_key)]
    id: String,

    #[orsx_column(unique)]
    email: String,
}

On existing tables, ORSX enforces “unique” by creating a unique index concurrently (idempotent).

Table-level index(...) declarations live inside #[orsx_table(...)]:

#[derive(orsx::OrsxMigrate)]
#[orsx_table(
  "users",
  index(columns("tenant_id", "email"), unique),
  index(columns("tenant_id", "created_at"))
)]
struct User {
    #[orsx_column(primary_key)]
    id: String,
    tenant_id: String,
    email: String,
    created_at: orsx::Timestamp,
}

Notes:

• type="btree"|"gin"|"gist"|"hash" is supported (default is btree).
• name="..." is optional; if omitted, ORSX derives a deterministic, table-specific index name at runtime.

When applying indexes on existing tables:

• SQL uses IF NOT EXISTS, and
• ORSX also checks for an equivalent existing index by semantics (method + uniqueness + ordered column list).

This prevents creating duplicates if the database already has the same index under a different name.

Notes:

• Partial indexes (... WHERE ...) and expression indexes (... ((expr)) ...) are not treated as equivalent to plain column indexes.

Reference tests:

• orsx/tests/migrations_indexes_idempotency.rs

You provide:

• a SELECT ... query,
• a ColumnarSchema describing the expected columns (order matters),

and you get a ColumnarBatch with:

• typed fixed-width buffers (Vec<i64>, Vec<u64> bits for f64, etc.),
• varlen buffers as offsets + a single Vec<u8> data blob (with a helper to coalesce),
• a validity bitmap per column to represent NULLs.

There are two readers:

• CopyBinaryBatchReader: uses COPY (SELECT ...) TO STDOUT (FORMAT BINARY) and parses the stream.
• RowWiseBatchReader: uses sqlx::query(select_sql).fetch(...) and try_get per cell, but still fills a ColumnarBatch.

There is also:

• ColumnarBatchReader + ColumnarReaderMode::Auto(...): chooses COPY vs row-wise based on expected query shape (still returns ColumnarBatch either way).

ColumnarType supports:

• Bool, I16, I32, I64
• F32, F64 (stored as IEEE754 bits)
• Uuid (16 bytes)
• TimestampTzMicros (i64 microseconds since Unix epoch)
• Utf8 (raw bytes + optional UTF-8 validation)
• Bytes (raw bytes)
• JsonbText (JSON/JSONB, stored as UTF-8 JSON text; COPY BINARY strips JSONB version byte)

Unsupported types should be treated as “not implemented yet” rather than “silently coerced”.

If you want to scan JSONB without parsing into a typed struct, use sqlx::types::JsonValue (or sqlx::types::Json<...>) in your Rust struct; #[derive(orsx::OrsxColumnar)] maps it to ColumnarType::JsonbText.

use sqlx::types::JsonValue;
use orsx::columnar::OrsxColumnar;

#[derive(orsx::OrsxColumnar)]
struct MyTable {
    id: i64,
    payload: JsonValue, // JSONB column
}

let schema = MyTable::columnar_schema()?;

Notes:

• JsonbText stores the JSON value as UTF-8 text bytes.
• JSONB does not preserve key ordering or input formatting; the stored bytes are the JSON text received from the Postgres binary protocol.

#[derive(orsx::OrsxColumnar)] generates:

• MyTable::columnar_schema() -> Result<ColumnarSchema>
• MyTable::COL_<FIELD> index constants

use orsx::columnar::{ColumnarBatch, ColumnarBatchReader, ColumnarReaderMode, OrsxColumnar};

#[derive(orsx::OrsxColumnar)]
struct MyTable {
    name_: String,
    pwt: f64,
}

async fn read_batch(conn: &mut sqlx::PgConnection) -> orsx::Result<ColumnarBatch> {
    let schema = MyTable::columnar_schema()?;
    let mut batch = ColumnarBatch::new(schema.clone(), 100_000)?;

    let sql = "SELECT name_, pwt FROM my_table ORDER BY name_";
    let mut reader = ColumnarBatchReader::new_select_unchecked(
        conn,
        sql,
        schema,
        ColumnarReaderMode::Auto(Default::default()),
    )
    .await?;

    let _rows = reader.next_batch_into(&mut batch).await?;
    Ok(batch)
}

Notes:

• The *_unchecked name is intentional: ORSX does not parse or sanitize your SQL. Use parameter binding for values.
• For the row-wise reader, select_sql must outlive the reader (pass a long-lived &str, not a temporary String).

Direct COPY BINARY:

use orsx::columnar::{ColumnarBatch, CopyBinaryBatchReader};

let schema = MyTable::columnar_schema()?;
let mut batch = ColumnarBatch::new(schema.clone(), 100_000)?;

let mut reader = CopyBinaryBatchReader::new_select_unchecked(conn, "SELECT name_, pwt FROM my_table", schema).await?;
let _rows = reader.next_batch_into(&mut batch).await?;

Direct row-wise:

use orsx::columnar::{ColumnarBatch, RowWiseBatchReader};

let schema = MyTable::columnar_schema()?;
let mut batch = ColumnarBatch::new(schema.clone(), 100_000)?;

let mut reader = RowWiseBatchReader::new_select_unchecked(conn, "SELECT name_, pwt FROM my_table", schema).await?;
let _rows = reader.next_batch_into(&mut batch).await?;

If you use RowWiseBatchReader directly, you can opt into a one-time preflight check (performed on the first row):

• validate_column_count: row.columns().len() must match ColumnarSchema::len()
• validate_column_names: returned column names must match ColumnarField.name (when Some)
• validate_type_compatible: returned SQL types must be compatible with the schema's decode types (sqlx::Type<Postgres>::compatible)

use orsx::columnar::{ColumnarBatch, RowWiseBatchReader, RowWiseBatchReaderConfig};

let schema = MyTable::columnar_schema()?;
let mut batch = ColumnarBatch::new(schema.clone(), 100_000)?;

let sql = "SELECT name_, pwt FROM my_table";
let mut reader = RowWiseBatchReader::new_select_unchecked(conn, sql, schema.clone())
    .await?
    .with_config(RowWiseBatchReaderConfig {
        validate_column_count: true,
        validate_column_names: true,
        validate_type_compatible: true,
    });

let _rows = reader.next_batch_into(&mut batch).await?;

Limitations:

• Empty result sets cannot be preflighted (no first row), so the preflight does not run.
• Even with validate_type_compatible, the final authority remains try_get during decoding (driver-level rules).

If you want preflight while using ColumnarBatchReader (including Auto(...)), use:

• ColumnarBatchReader::new_select_unchecked_with_row_wise_config(..., Some(row_wise_cfg))

Example:

use orsx::columnar::{ColumnarBatch, ColumnarBatchReader, ColumnarReaderMode, RowWiseBatchReaderConfig};

let schema = MyTable::columnar_schema()?;
let mut batch = ColumnarBatch::new(schema.clone(), 100_000)?;

let cfg = RowWiseBatchReaderConfig {
    validate_column_count: true,
    validate_column_names: true,
    validate_type_compatible: true,
};

let mut reader = ColumnarBatchReader::new_select_unchecked_with_row_wise_config(
    conn,
    "SELECT name_, pwt FROM my_table",
    schema,
    ColumnarReaderMode::Auto(Default::default()),
    Some(cfg),
).await?;

let _rows = reader.next_batch_into(&mut batch).await?;

let name_offsets = batch.var_chunks(MyTable::COL_NAME_).unwrap().0;
let mut name_data = Vec::new();
batch.coalesce_var_into(MyTable::COL_NAME_, &mut name_data)?;

let pwt_bits = batch.fixed_f64_bits(MyTable::COL_PWT).unwrap();
let pwt0 = f64::from_bits(pwt_bits[0]);

// For row i:
let i = 123usize;
let start = name_offsets[i] as usize;
let end = name_offsets[i + 1] as usize;
let name_i = std::str::from_utf8(&name_data[start..end]).unwrap();

orsx::columnar::encode_orsxcol_v1_into encodes a ColumnarBatch into a versioned byte buffer. decode_orsxcol_v1(_into) decodes and validates it.

This is intended for “send batch over the wire / store in cache” use cases.

For very wide “processor output” structs (hundreds to thousands of columns), ORSX can generate:

• a Postgres table schema (OrsxMigrate::spec()),
• a columnar schema (OrsxColumnar::columnar_schema()),
• deterministic column ids and ordering (COLUMNS_IN_ORDER, METRIC_COLUMNS_IN_ORDER),
• a deterministic schema hash (SCHEMA_HASH),
• a deterministic write-path binder (visit_values_in_order(...) + PgArgumentsVisitor).

This is intended to avoid hand-writing and maintaining large INSERT/UPSERT bind lists.

On stable Rust, proc-macros cannot read the invoking source file via proc_macro::Span (that API is unstable), so recursive flattening must work from the tokens the macro receives.

#[orsx::orsx_flatten_module] requires an inline module body (mod outputs { ... }), so it can parse all related structs in that module and recurse without filesystem reads.

For flat structs (no #[orsx_family] recursion), #[derive(OrsxFlatten)] is available. For recursive flatten, use the module macro.

• Root type:
  • #[orsx_processor_id("...")] (required; also participates in the schema hash)
• Family fields (nested structs):
  • #[orsx_family(prefix = "ma_")]
• Leaf fields:
  • #[orsx_column(skip)] to exclude a field from the flattened schema
  • #[orsx_column(id = "custom_id")] to override the generated id for a leaf

The macro consumes these marker attributes and removes them from the emitted items (so downstream code does not need to register them as “known” attributes).

• Non-metric (provenance) fields come first, in root declaration order.
• Flattened metric columns are generated with canonical ids and then sorted lexicographically by id.

This ordering is exposed as:

• COLUMNS_IN_ORDER (provenance + metric)
• METRIC_COLUMNS_IN_ORDER (metric only)

use orsx::prelude::*;

#[orsx::orsx_flatten_module]
mod outputs {
    #[derive(Clone)]
    pub struct MovingAverages {
        pub ema_9: f64,
        pub ema_21: f64,
    }

    #[derive(Clone)]
    pub struct Oscillators {
        pub rsi_14: f64,
    }

    #[orsx_processor_id("proc_a")]
    #[derive(Clone)]
    pub struct Out {
        pub pair: String,
        pub e_ms: i64,

        #[orsx_family(prefix = "ma_")]
        pub ma: MovingAverages,

        #[orsx_family(prefix = "osc_")]
        pub osc: Oscillators,
    }
}

The macro also generates:

• Out::visit_values_in_order(&self, visitor: &mut impl OrsxValueVisitor) -> Result<()>

ORSX provides a PgArgumentsVisitor to bind into sqlx::postgres::PgArguments:

use orsx::prelude::*;
use sqlx::postgres::PgArguments;

let out = outputs::Out { /* ... */ };

let mut args = PgArguments::default();
{
    let mut v = orsx::PgArgumentsVisitor::new(&mut args);
    out.visit_values_in_order(&mut v)?;
}

Compressed<T> stores a numeric vector as BYTEA with an envelope:

• magic/version
• codec id + element type id
• element count + uncompressed byte length
• CRC32 of the compressed payload
• payload bytes

This is not generic “data compression”; it is a narrow mechanism for numeric vectors.

Example (insert + select):

use orsx::{Compressed, CompressedWorkspace};
use sqlx::Row;

let v = Compressed(vec![1.0_f64, 2.0, 3.0]);
let mut ws = CompressedWorkspace::default();
let mut bytes = Vec::new();
v.encode_envelope_into(&mut bytes, &mut ws)?;

sqlx::query("INSERT INTO my_vecs (id, payload) VALUES ($1, $2)")
    .bind("row1")
    .bind(bytes)
    .execute(&pool)
    .await?;

let raw: Vec<u8> = sqlx::query("SELECT payload FROM my_vecs WHERE id = $1")
    .bind("row1")
    .fetch_one(&pool)
    .await?
    .try_get(0)?;

let decoded = Compressed::<f64>::decode_envelope(&raw)?;
assert_eq!(decoded.as_slice(), &[1.0, 2.0, 3.0]);

If you prefer, you can bind Compressed<T> directly (it implements SQLx Type/Encode/Decode for Postgres):

use orsx::Compressed;

sqlx::query("INSERT INTO my_vecs (id, payload) VALUES ($1, $2)")
    .bind("row1")
    .bind(Compressed(vec![1.0_f64, 2.0, 3.0]))
    .execute(&pool)
    .await?;

let decoded: Compressed<f64> = sqlx::query_scalar("SELECT payload FROM my_vecs WHERE id = $1")
    .bind("row1")
    .fetch_one(&pool)
    .await?;
assert_eq!(decoded.as_slice(), &[1.0, 2.0, 3.0]);

This example shows one possible “end-to-end” flow. It is intentionally explicit about SQL and schema.

use orsx::prelude::*;
use orsx::columnar::{ColumnarBatch, ColumnarBatchReader, ColumnarReaderMode, OrsxColumnar, encode_orsxcol_v1_into};

#[derive(OrsxMigrate, orsx::OrsxColumnar)]
#[orsx_table("wf_items")]
struct Item {
    #[orsx_column(primary_key)]
    id: String,
    name_: String,
    pwt: f64,
}

#[tokio::main]
async fn main() -> Result<()> {
    let db_url = std::env::var("DATABASE_URL").expect("DATABASE_URL");
    let pool = sqlx::PgPool::connect(&db_url).await?;

    // 1) Migrate (create or update schema)
    let dummy = Item { id: "x".into(), name_: "n".into(), pwt: 0.0 };
    Migrations::init(&pool, &[(dummy, None)]).await?;

    // 2) Write some rows (raw SQL)
    sqlx::query("INSERT INTO wf_items (id, name_, pwt) VALUES ($1,$2,$3) ON CONFLICT (id) DO UPDATE SET name_ = EXCLUDED.name_, pwt = EXCLUDED.pwt")
        .bind("1")
        .bind("alice")
        .bind(1.25_f64)
        .execute(&pool)
        .await?;

    // 3) Columnar read
    let mut conn = pool.acquire().await?;
    let schema = Item::columnar_schema()?;
    let mut batch = ColumnarBatch::new(schema.clone(), 100_000)?;

    let sql = "SELECT name_, pwt FROM wf_items ORDER BY id";
    let mut reader = ColumnarBatchReader::new_select_unchecked(
        &mut conn,
        sql,
        schema,
        ColumnarReaderMode::Auto(Default::default()),
    )
    .await?;
    let _rows = reader.next_batch_into(&mut batch).await?;

    // 4) Encode for transport
    let mut out = Vec::new();
    encode_orsxcol_v1_into(&batch, &mut out)?;
    // `out` can be returned from an API or written to disk.

    Ok(())
}

All of these numbers are from protocols/orsx2_evidence/columnar_trials.md and are “release” builds.

From protocols/orsx2_evidence/columnar_trials.md:

• 2026-01-14 14:40:38Z:
  • 100k × 50 cols: COPY → ColumnarBatch 262.405752ms, row-wise → ColumnarBatch 299.598514ms
  • 100k × 500 cols: COPY → ColumnarBatch 2.430023982s, row-wise → ColumnarBatch 2.892875905s
• 2026-01-14 14:41:55Z:
  • 1M × 50 cols: COPY → ColumnarBatch 2.75208442s, row-wise → ColumnarBatch 2.528227132s (row-wise is faster here)
• 2026-01-15 09:23:12Z:
  • 100k × 50 cols: COPY → ColumnarBatch 280.819598ms, row-wise → ColumnarBatch 267.554416ms
  • 100k × 500 cols: COPY → ColumnarBatch 2.471189591s, row-wise → ColumnarBatch 2.798175158s
  • 1M × 50 cols: COPY → ColumnarBatch 2.757631491s, row-wise → ColumnarBatch 2.523348916s

Exact commands used for these trials are recorded in the log entries.

All of these numbers are from protocols/orsx2_evidence/migration_trials.md and are “release” builds.

From protocols/orsx2_evidence/migration_trials.md:

• 2026-01-14T10:48:43Z (UUID PK, 1,000,000 seeded + 100,000 writer inserts):
  • cutover lock: ~1012ms (budget 5000ms)
  • backfill: ~21.324s (rows reported: 1,100,000)
  • total online rewrite: ~26.014s
• 2026-01-14T11:53:05Z (strict order/exact enforced on 1M rows, forces rewrite):
  • strict migration: ~7.70s vs default alter ~34.8ms

These are workload- and hardware-dependent; they are meant as evidence of current behavior, not a guarantee.

Bench results for the generated binder/visitor are recorded in:

• protocols/orsx2_evidence/bench_results.md

To run the current microbenchmarks locally:

• cargo bench -p orsx --bench flatten

Most integration tests require a running Postgres.

Environment variable used by tests:

• ORSX_TEST_DATABASE_URL (defaults to postgresql://orsx:orsx@localhost:15432/orsx2_test)

Useful commands:

• Unit tests (no DB): cargo test -p orsx --lib
• DB correctness tests (require Postgres):
  • cargo test -p orsx --tests (runs all integration tests; requires Postgres)
  • cargo test -p orsx --test columnar_copy_binary --release
  • cargo test -p orsx --test columnar_jsonb
  • cargo test -p orsx --test migrations_strict_correctness
  • cargo test -p orsx --test migrations_indexes_idempotency
• Perf / large-table tests (ignored by default; require Postgres and time):
  • cargo test -p orsx --test columnar_perf_trials --release -- --ignored --nocapture
  • cargo test -p orsx --test migrations_online_big_uuid --release -- --ignored --nocapture

Notes:

• Ignored tests are intentionally excluded from the default suite; they are stress/perf workloads and may require a clean test database.
• Some large-table tests create the uuid-ossp extension (CREATE EXTENSION IF NOT EXISTS "uuid-ossp").

Suggested validation set (choose based on what you changed):

• cargo test -p orsx --lib
• cargo test -p orsx --tests (requires Postgres)
• cargo bench -p orsx --bench flatten (optional)

This repo can use cargo-husky to install git hooks from .cargo-husky/hooks/.

Current hooks:

• pre-commit: cargo fmt --check, cargo clippy -D warnings, cargo test --workspace --lib
• pre-push: cargo test --workspace --lib

To install/update hooks, run a command that builds dev-dependencies at least once (for example: cargo test -p orsx --lib).