diff --git a/Cargo.lock b/Cargo.lock index 1c5a3bc8..98636c16 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2932,6 +2932,7 @@ dependencies = [ "async-channel", "async-trait", "backon", + "bumpalo", "bytes", "chrono", "clap", diff --git a/Cargo.toml b/Cargo.toml index f3f63eef..55fea161 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -46,6 +46,7 @@ astral-tokio-tar = { version = "0.6", default-features = false } async-channel = "2.5" async-trait = "0.1" backon = { version = "1", features = ["tokio-sleep"] } +bumpalo = "3" bytes = "1" chrono = { version = "0.4", default-features = false, features = ["clock"] } clap = { version = "4", features = ["derive", "env"] } diff --git a/docker/Dockerfile b/docker/Dockerfile index 89d0e509..ad3c7037 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -49,8 +49,9 @@ RUN make -C pgext clean \ FROM postgres:${PG_MAJOR}-alpine AS runtime # bash: entrypoint shebang. ca-certificates: TLS to clickhouse. -# liblz4/libzstd already present as PG deps. -RUN apk add --no-cache bash ca-certificates +# postgis: so the daemon-owned shadow PG can render the schema's geography +# column via typoutput. liblz4/libzstd already present as PG deps. +RUN apk add --no-cache bash ca-certificates postgis COPY --from=rust-builder /usr/local/bin/walshadow-stream /usr/local/bin/walshadow-stream COPY --from=ext-builder /src/pgext/walshadow.so \ diff --git a/docker/Dockerfile.source b/docker/Dockerfile.source new file mode 100644 index 00000000..80732f34 --- /dev/null +++ b/docker/Dockerfile.source @@ -0,0 +1,8 @@ +# Source Postgres for the stress schema: stock postgres:18 + PostGIS, so the +# gist dump's geography column loads unchanged. +FROM postgres:18-bookworm +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + postgresql-18-postgis-3 \ + postgresql-18-pgvector \ + && rm -rf /var/lib/apt/lists/* diff --git a/docs/destination-tables.md b/docs/destination-tables.md index 894cedae..54b66de2 100644 --- a/docs/destination-tables.md +++ b/docs/destination-tables.md @@ -64,7 +64,12 @@ Common mappings include: | `time` | `Time64(6)` | | `timestamp`, `timestamptz` | `DateTime64(..., 'UTC')` | | `uuid` | `UUID` | -| `json`, `jsonb`, `inet`, `cidr`, `interval`, arrays, unknown types | `String` | +| `json`, `jsonb` | `JSON` | +| `hstore` | `Map(String, Nullable(String))` | +| `vector`, `halfvec` (pgvector) | `Array(Float32)` | +| `geography`, `geometry` (PostGIS) | `String` (WKT) | +| `[]` arrays | `Array(Nullable())`; unknown elem → `Array(Nullable(String))` | +| `inet`, `cidr`, `interval`, unknown types | `String` | Nullable source columns become `Nullable(...)` unless used as ClickHouse sort keys. ClickHouse deployments using PostgreSQL `time` columns must enable diff --git a/plans/INDEX.md b/plans/INDEX.md index 2d1d286f..3b38b3d6 100644 --- a/plans/INDEX.md +++ b/plans/INDEX.md @@ -52,9 +52,10 @@ components. User workflows and supported behavior live under [future/INDEX.md](future/INDEX.md) collects design docs for unbuilt work: runtime-config signals and net-new knobs, two-phase commit, sequence-state replication, cross-table ordering, CH-bounce recovery, -parked operational polish. Once built, keep behavior in code and tests, move -user-facing consequences into `docs/`, and retain only rationale or invariants -which code cannot express +greenfield tier-3 oracle (throwaway bootstrap PG), oracle emitting CH-native +blocks, parked operational polish. Once built, keep behavior in code and tests, +move user-facing consequences into `docs/`, and retain only rationale or +invariants which code cannot express ## Architecture diagrams diff --git a/plans/bootstrap.md b/plans/bootstrap.md index 598ca152..bc821ba8 100644 --- a/plans/bootstrap.md +++ b/plans/bootstrap.md @@ -53,7 +53,11 @@ for rendered diagram. Five clusters top→bottom: [emitter.md](emitter.md)). One synthetic ack seq per rfn flip; `tail.finish` seals partial batches and waits all seqs durable before handoff. Metrics-only runs (no `--ch-config`) instead drain - through `drain_backfill` into a counting `TupleObserver` + through `drain_backfill` into a counting `TupleObserver`. + Bridge-routed tier-3 values (jsonb, arrays, hstore, …) can't be + resolved here — the shadow/bridge don't exist until after bootstrap — + so they currently land empty; in-tree types (geography, vector) are + fine. Fix in [future/greenfield_oracle.md](future/greenfield_oracle.md) 4. **Shadow handoff** — `BootstrapOutcome { start, end }` returned; daemon writes `standby.signal` and calls `materialize_conf` to replace shadow's config files. Config includes walshadow settings, diff --git a/plans/emitter.md b/plans/emitter.md index 4704466c..11c97326 100644 --- a/plans/emitter.md +++ b/plans/emitter.md @@ -356,7 +356,10 @@ today is `String`, anything else dies cleanly at `append` `pk_member = true` strips `Nullable(_)` wrap because CH refuses `Nullable` in `ORDER BY`. User-visible matrix lives in [`docs/destination-tables.md`](../docs/destination-tables.md#default-type-mapping), -hard-coded by `base_type_for` +hard-coded by `base_type_for`. Dynamic-OID types (hstore, pgvector, arrays, +PostGIS) are matched on `RelAttr.type_name`, not OID; CH forbids `Nullable` +over `Array`/`Map` so those stay bare (a NULL source value lands as an empty +one), while `Nullable(JSON)` is allowed. `numeric` needs `1 ≤ p ≤ 76` for `Decimal`; `p = 0`, scale outside `0 ≤ s ≤ p`, or unconstrained `numeric` (which can carry NaN/±Inf) fall diff --git a/plans/future/INDEX.md b/plans/future/INDEX.md index 19356c59..f5717ce0 100644 --- a/plans/future/INDEX.md +++ b/plans/future/INDEX.md @@ -16,6 +16,8 @@ rationale under `plans/` only when code cannot express it * [failover.md](failover.md) — beyond the switchover crossing in [../failover.md](../failover.md): unplanned promotion (transaction-state fence, overwrite contrecord), slotless pause windows, timeline-aware archive and base-backup replay * [sync_commit_witness.md](sync_commit_witness.md) — walshadow as RPO=0 durability standby * [two_phase_commit.md](two_phase_commit.md) — `XLOG_XACT_PREPARE` handling and gxid-keyed buffer +* [greenfield_oracle.md](greenfield_oracle.md) — resolve bridge-routed tier-3 (jsonb/arrays/hstore) during greenfield bootstrap via a throwaway PG (walshadow module + source extensions) torn down after; OID-matching constraint (restore-from-backup vs resolve-by-name) +* [oracle_native_blocks.md](oracle_native_blocks.md) — bridge/pgext emits CH-native column bytes instead of `typoutput` text, batching `DECODE` from row-at-a-time to a column-major list of rows (matches CH's columnar block, amortizes the round trip); removes the emitter-side composite re-parse (`ColumnBuf::{Array,Map,Json}`, text parsers, `NodeArena`); does not solve greenfield * [ch_bounce_recovery.md](ch_bounce_recovery.md) — deeper re-emit-from-spill on retry-budget exhaustion * [pinned_ddl_baseline.md](pinned_ddl_baseline.md) — schema-event outcome must be a function of config + baseline, not cache warmth: CH-existence / persisted-baseline options for cross-restart consistency, drop detection across downtime, opt-in mapping vs republish * [coverage100.md](coverage100.md) — drive `cargo llvm-cov` line coverage toward 100%: tiered work list (pure units → fixtures → live e2e → hard tail) diff --git a/plans/future/greenfield_oracle.md b/plans/future/greenfield_oracle.md new file mode 100644 index 00000000..afc3dc96 --- /dev/null +++ b/plans/future/greenfield_oracle.md @@ -0,0 +1,57 @@ +# greenfield_oracle — a bridge for tier-3 during bootstrap + +## Problem + +Greenfield bootstrap ([../bootstrap.md](../bootstrap.md)) page-walks a +base/object-store backup and drains rows through `pipeline::bootstrap::drain` +**before the shadow PG and its bridge exist** (bridge is created after +`run_bootstrap` returns). So bridge-routed tier-3 values — jsonb, arrays, +hstore, tsvector, ranges, domains — have no PG to render them and land empty. +In-tree types (geography, vector; see [../oracle.md](../oracle.md)) are fine. + +`BackupSource` is a page reader, not a running PG, so there is nothing to +decode on-disk tier-3 Datums during the drain. + +## Approach + +Stand up a **throwaway Postgres with the walshadow module + the source's +extensions**, point the bridge/oracle at it for the bootstrap drain, then tear +it down once bootstrap completes. `resolve_decoded_heap` already takes an +`Option<&Oracle>`; wire this oracle in for the greenfield drain and the whole +tier-3 path resolves exactly like live. + +Lifecycle: create → (extensions/module ready) → pass `Some(oracle)` to +`bootstrap::drain` → drain → drop the temp PG + its datadir/socket before the +real shadow is materialized for streaming. + +## The OID-matching constraint (the hard part) + +`ws_decode_datum_text` renders a Datum by running the type's `typoutput`, +looked up by **OID**. Built-in tier-3 OIDs are stable across clusters (jsonb +3802, `int4[]` 1007, …) so a fresh `initdb` + `CREATE EXTENSION` handles them. +But **extension type OIDs are assigned at `CREATE EXTENSION` time and differ +per cluster** — a fresh temp PG's `hstore`/`geography`/`vector` OID won't match +the source OID carried in the on-disk bytes, so typoutput lookup misfires. + +Two ways to satisfy it: + +- **Restore the temp PG from the backup** (it then carries the source catalog, + OIDs match) — essentially a short-lived shadow. Reuses the base-backup we + already fetched; heaviest but exact. Overlaps with the Option-A framing in + the earlier analysis. +- **Resolve by type name, not OID** — extend the bridge `DECODE` protocol to + carry the type name; the worker looks up `typoutput` via + `regtype`/`pg_type.typname` in the temp PG (which has the same-named + extensions installed). Lets a plain `initdb` + extensions work regardless of + OID drift. Smaller PG, but a protocol + worker change. + +## Open questions + +- Which extensions to install: derive from the source catalog (types actually + present) vs a fixed set; fail-soft when one isn't available. +- Cost/timing: temp-PG spin-up vs bootstrap duration; only worth it when tier-3 + columns exist in the mapped set. +- Interaction with restart/resume: bootstrap re-runs must recreate/tear down + the temp PG idempotently. +- Does not change the emitter contract; it only makes `Some(oracle)` available + earlier. Orthogonal to [oracle_native_blocks.md](oracle_native_blocks.md). diff --git a/plans/future/oracle_native_blocks.md b/plans/future/oracle_native_blocks.md new file mode 100644 index 00000000..8793598b --- /dev/null +++ b/plans/future/oracle_native_blocks.md @@ -0,0 +1,78 @@ +# oracle_native_blocks — bridge emits CH-native, not text + +## Problem + +Today bridge-routed tier-3 values round-trip through **text**: the worker runs +`typoutput` → `ColumnValue::Text`, and the emitter then **re-parses** that text +back into ClickHouse's columnar form. The emitter-side machinery that does this +re-parse — `ColumnBuf::{Array,Map,Json}`, `encode_array`/`encode_map`/ +`encode_json` (with `parse_pg_array_1d` / `parse_vector_list` / hstore + JSON +parsers), and the `NodeArena` nested-`ColumnBuilder` builder — is +decode-to-text-then-parse-back, which is wasteful and brittle (PG text quoting, +CH `JSON` object-only + null-slot rules, etc.). + +## Approach + +Have the resolver produce ClickHouse column data directly, so the emitter +splices it in without a text detour. The extension links a CH serializer +(`clickhouse-c` / `pg-clickhouse-c`); the `DECODE` op carries the **target CH +type** per item and returns native column bytes instead of a `typoutput` +string. The emitter keeps its columnar assembly for scalars but drops the +composite text parsers. + +## Batch the DECODE (row-at-a-time → list of rows) + +Today the bridge is **per-tuple**: `resolve_pending_tuple` bundles one row's +pending columns into a single `DECODE` request, and the decode pool calls it +once per new/old tuple — one socket round trip per row. That's fine for +scattered text but wrong-shaped for native output: a ClickHouse column is +**many rows of one type laid out together**, so a row-at-a-time worker can't +build a column and every row pays a round trip. + +Change the unit of resolution from one row to a **list of rows** (a +column-major batch): + +- `DECODE` takes, per pending column, the target CH type + the raw on-disk + Datum for *every row in the batch* (nulls marked); the worker decodes down a + column and returns that column's native bytes (values + null map, plus + offsets for Array/Map) in one shot. +- Resolve at **batch granularity, not tuple granularity**. The batcher already + accumulates per-table row batches (`InsertBatch`/chunk — see + [../emitter.md](../emitter.md)); hand a whole table-chunk to the bridge and + get back native columns ready to append to the block. This amortizes the + round trip over the batch and matches CH's columnar layout end-to-end. +- Ordering/back-pressure: one in-flight batch request per table-chunk keeps the + existing seq/ack accounting; size the batch to the inserter's block size. + +## Scope / cost + +- **pgext** (`pgext/decode.c`, `worker.c`, `walshadow.h`): link the CH + serializer; `DECODE` request becomes column-batched (target CH type + a list + of raw Datums per column); response returns the serialized native column + (values + null map + offsets). Reimplement PG-Datum → CH-native for the + tier-3 matrix in C. +- **Protocol/coupling**: a PG-side component must now know the destination CH + type (it knows nothing about ClickHouse today) — carry it in the request or a + negotiated intermediate. +- **Rust** (`ops/bridge.rs`, `ops/oracle.rs`, `emit/…`): `Bridge::decode` + takes a batch and returns whole native columns; resolution moves from the + per-tuple call in `emit/pipeline/decode.rs` to batch granularity alongside + the batcher. Delete the emitter composite `ColumnBuf` variants + text parsers + + `NodeArena`; scalar `ColumnBuf`/`build_column` stays. +- Estimated large + higher-risk (new C dep in the PGXS `.so`, native-format + edge cases across many types, batched-protocol reframe). + +## Does NOT solve greenfield + +This only changes *where* serialization happens; it still needs a live PG with +the walshadow module to do the decoding. Greenfield has none until after +bootstrap — that gap is [greenfield_oracle.md](greenfield_oracle.md), and the +two are independent. + +## Alternative considered + +In-tree Rust decoders for PG array/hstore/jsonb on-disk binary (no bridge at +all) — self-contained and greenfield-friendly, but re-implements PG's +varlena/alignment/null-bitmap/JEntry formats byte-exact, which is its own large +correctness surface. Native-from-the-extension reuses PG's own rendering and is +preferred where a PG is available. diff --git a/plans/oracle.md b/plans/oracle.md index 0ceb014e..0168bd11 100644 --- a/plans/oracle.md +++ b/plans/oracle.md @@ -36,14 +36,34 @@ Why these: disambiguation lives at type-OID level not body bytes (on-disk vs wire confusion surfaced here historically) +## In-tree extension types + +PostGIS `geography`/`geometry` and pgvector `vector`/`halfvec` are rendered +in-tree from their on-disk bytes by `render_ext_columns` (`src/ops/oracle.rs`), +matched on `RelAttr.type_name` (dynamic OIDs). geography 2-D points → +WKT `POINT(x y)` (`gserialized_point_to_wkt`); vector → `[a,b,c]` +(`vector_to_text`). No bridge round-trip, so these resolve even where the +shadow worker is unavailable (e.g. greenfield bootstrap). + ## Bridge-routed Tier 3 -`jsonb`, arrays, `tsvector`, ranges, domains. Heap decoder emits +`jsonb`, arrays, `hstore`, `tsvector`, ranges, domains. Heap decoder emits [`ColumnValue::PgPending { type_oid, raw }`](../src/heap_decoder.rs); [`resolve_pending_tuple`](../src/oracle.rs) collects every pending column of a tuple into one `DECODE` request to shadow's bridge worker, swaps `PgPending` for `Text` on each item that rendered +## Shared resolve step + +`resolve_decoded_heap(oracle, attrs, decoded)` runs `render_ext_columns` then +(if an oracle is present) `resolve_pending_tuple` over a heap's new/old tuples. +Both the live decode pool (`emit/pipeline/decode.rs`) and the object-store / +COPY backfill paths call it, so backfilled rows resolve identically to +streamed ones. **Greenfield bootstrap is the exception**: it runs before the +shadow/bridge exist, so bridge-routed types there resolve to nothing (in-tree +types still work) — the fix is +[future/greenfield_oracle.md](future/greenfield_oracle.md). + Two alternatives considered (insert + select round-trip; `SELECT $1::bytea::::text`) require reconstructing wire format from on-disk format — same codec work the worker elides diff --git a/plans/vector_qbit.md b/plans/vector_qbit.md new file mode 100644 index 00000000..7caa124a --- /dev/null +++ b/plans/vector_qbit.md @@ -0,0 +1,138 @@ +# vector_qbit — fix halfvec decode + map pgvector to ClickHouse QBit + +Two coupled changes to the pgvector path: + +1. **Correctness:** `halfvec` is currently mis-decoded (read as `f32`), so it + lands empty. Fix the on-disk decode. +2. **Mapping:** map `vector`/`halfvec` to CH `QBit(elem, dim)` (vector search + type, tunable precision) instead of `Array(Float32)`. + +Both live in the pgvector in-tree render + type-bridge + DDL path; the native +insert path is deliberately left untouched. + +## Part A — fix halfvec decode + +`render_ext_columns` (`src/ops/oracle.rs:141`) routes both `vector` and +`halfvec` to `vector_to_text`, which reads `[dim u16][unused u16][f32 × dim]` +— 4 bytes/element (`src/ops/oracle.rs:168`, `off = 4 + i*4`). Correct for +pgvector `vector` (fp32 elements). Wrong for `halfvec`, whose on-disk layout is +`int16 dim; int16 unused; half x[dim]` — **IEEE binary16, 2 bytes/element**. +Result: wrong offsets + the length guard `raw.len() < 4 + dim*4` fails → +returns `None` → column empties. + +Steps: +- Add `halfvec_to_text(raw)` reading element `i` at `off = 4 + i*2`, 2 bytes LE, + IEEE fp16 → f32, emit `[a,b,c]` (same decimal-list shape `encode_array` / + `parse_vector_list` already consumes). Length guard `raw.len() >= 4 + dim*2`. +- Route `"halfvec"` → `halfvec_to_text`; keep `"vector"` → `vector_to_text` + (`src/ops/oracle.rs:143`). +- fp16→f32: add the `half` crate (not currently a dep) — `half::f16::from_le_bytes([lo,hi]).to_f32()`. + Hand-rolling the bit unpack is the no-dep alternative. +- Tests: known bit patterns — `0x3C00`→1.0, `0x4000`→2.0, `0xC000`→-2.0, + `0x0001` (min subnormal), `0x7BFF` (max finite ≈65504). (pgvector rejects + inf/NaN in halfvec, but decode should not panic on them.) + +This fix stands on its own regardless of Part B. + +## Part B — map vector/halfvec → QBit + +### Target types + +- `vector` (fp32) → `QBit(Float32, dim)` +- `halfvec` (fp16) → `QBit(BFloat16, dim)` — storage-halved, matches the source's + intent; lossy vs IEEE fp16 (bf16 = 8-bit exp / 7-bit mantissa), acceptable for + ANN. Never bit-reinterpreted: decode produces real f32 decimals, the server + rounds on cast. +- No dimension (`typmod <= 0`, unspecified-dim pgvector column) → fall back to + `Array(Float32)` (today's behaviour); QBit requires a fixed dimension. + +`dim` comes from `att.typmod`. **Verify:** pgvector stores the dimension +directly in `atttypmod` (`vector(3)` → typmod 3, no VARHDRSZ offset), same for +`halfvec`. Helper: `fn pgvector_dim(typmod: i32) -> Option` returning +`(typmod > 0).then_some(typmod as u32)`. + +### The DDL-vs-wire split (the mechanism) + +The vendored binding's `Kind` enum has **no QBit variant** +(`clickhouse-c-rs/src/types.rs:16`), so `TypeAst::parse("QBit(...)")` cannot +produce a buildable column, and `ChScalar::from_kind` has no `BFloat16` arm +(`src/emit/ch_emitter.rs:1033`) so `Array(BFloat16)` is unbuildable too. + +Therefore the native insert block stays **`Array(Float32)`** and the server +performs the implicit `Array → QBit` cast on INSERT (confirmed: arrays convert +to QBit when length == dimension; element type auto-converts, incl. +Float32→BFloat16 narrowing). The DDL column type is the only thing that becomes +`QBit(...)`. + +Implementation — keep the wire type, split out a storage/DDL type: +- `base_type_for` (`src/catalog/type_bridge.rs:123`) keeps returning the **wire + type** `Array(Float32)` for `vector`/`halfvec` (emitter, batcher, mapping + `target_type`, block build all untouched — still `Array(Float32)`). +- Add `ddl_column_type(att) -> String` returning `QBit(Float32, dim)` / + `QBit(BFloat16, dim)` for pgvector-with-dim, else the wire type. +- Carry it on the resolved column (add `ddl_type: Option` alongside + `ch_type`, default = `ch_type`). `render_create_table` + (`src/emit/ch_ddl.rs:742`) emits `ddl_type` for the column definition; the + mapping's `target_type` (insert) stays `ch_type`. + +Net: CREATE TABLE says `col QBit(Float32, N)`; every INSERT sends an +`Array(Float32)` block; CH casts. No change to `ch_emitter`, `inserter`, or the +binding. + +### Experimental setting + +QBit is experimental (CH 25.10+). CREATE TABLE (and possibly INSERT) needs the +enabling setting on the session. Push it via the binding's per-query settings +(`chc_query_setting`, `clickhouse-c-rs/src/sys.rs:602`). +**Verify the exact GUC name** (`allow_experimental_qbit_type`) and whether the +insert session needs it, not just DDL. Deployments run CH 26.3 (>= 25.10), so +availability is fine; gate the QBit mapping on CH version to avoid breaking +older targets (fall back to `Array(Float32)`). + +## Decisions / defaults + +- Element type: `vector`→Float32, `halfvec`→BFloat16 (per review). Make it + overridable later via `config_column` if lossless halfvec (Float32) is wanted; + not required for v1. +- Stride: omit (no grouping). +- Existing tables created as `Array(Float32)` are **not** migrated — only newly + auto-created tables get QBit. Recreate to adopt. Note in release notes. + +## Risks / open items + +- **Native `Array → QBit` INSERT (highest risk).** Docs confirm implicit + conversion on INSERT + explicit CAST; PR ClickHouse#91846 added the Array→QBit + cast. Must validate over the **native TCP protocol** (not just SQL VALUES) + end-to-end. If native INSERT does not auto-convert, fallbacks: keep + `Array(Float32)` DDL, or issue an `ALTER ... MODIFY`/materialized cast — decide + only if validation fails. +- pgvector `atttypmod` dimension encoding — verify empirically (docker pgvector + stack) before trusting `pgvector_dim`. +- Exact experimental setting name + whether INSERT needs it. +- Operator TOML mapping path (`render_create_table_from_mapping`, + `src/emit/ch_ddl.rs:772`) uses `c.target_type` for both DDL and insert, so + QBit is **auto-create only** in v1; a TOML `QBit` target_type would break the + insert build. A separate storage-type knob is a follow-up. +- QBit columns are not orderable — never emit into `ORDER BY` (pgvector columns + aren't PKs, so not an issue in practice). + +## Test plan + +- Unit: `halfvec_to_text` bit-pattern tests (Part A). +- Unit: `base_type_for`/`ddl_column_type` — `vector(3)`→wire `Array(Float32)` + + DDL `QBit(Float32, 3)`; `halfvec(4)`→DDL `QBit(BFloat16, 4)`; unspecified-dim + → `Array(Float32)` both. +- E2E on the docker stack (`stress/`, pgvector `embedding vector(3)`): fresh + volumes → walshadow auto-creates `embedding QBit(Float32, 3)`; pgbench insert; + confirm CH stores QBit and a `QBit`-aware distance query returns; add a + `halfvec(N)` column and confirm values round-trip (decode fix + cast). + +## Touch points + +- `src/ops/oracle.rs` — `halfvec_to_text`, route halfvec (Part A). +- `Cargo.toml` — `half` crate (if used). +- `src/catalog/type_bridge.rs:123` — keep wire `Array(Float32)`; add + `ddl_column_type` + `pgvector_dim`. +- `src/emit/ch_ddl.rs` — resolved `ddl_type` field; `render_create_table:742` + uses it; QBit experimental setting on the DDL session. +- CH session settings wiring (`chc_query_setting`) — experimental flag. diff --git a/src/catalog/type_bridge.rs b/src/catalog/type_bridge.rs index 9e28ce95..577649a2 100644 --- a/src/catalog/type_bridge.rs +++ b/src/catalog/type_bridge.rs @@ -24,8 +24,12 @@ //! | `interval` | `String` | | //! | `uuid` | `UUID` | | //! | `inet` / `cidr` | `String` | | -//! | `json` / `jsonb` | `String` | CH `JSON` opt-in via namespace config | -//! | array / unknown | `String` | falls through to PGPending bytes | +//! | `json` / `jsonb` | `JSON` | native CH JSON (string serialization) | +//! | `hstore` | `Map(String, Nullable(String))` | by type name | +//! | `vector` / `halfvec` | `Array(Float32)` | pgvector, by type name | +//! | `geography` / `geometry` | `String` | WKT `POINT(x y)`, rendered at decode | +//! | `[]` | `Array(Nullable())` | 1-D supported elems, else String | +//! | unknown | `String` | falls through to PgPending bytes | //! //! Nullability: `not_null = false` wraps inner in `Nullable(_)` unless //! column is in CH `ORDER BY` (PK columns must stay non-nullable; caller @@ -70,10 +74,12 @@ pub enum BridgeError { /// `pk_member = true` forces non-nullable: CH refuses `Nullable` in `ORDER BY` pub fn map(att: &RelAttr, pk_member: bool) -> Result { let inner = base_type_for(att)?; - let ch_type = if pk_member || att.not_null { - inner.clone() - } else { + let nullable = + !pk_member && !att.not_null && !inner.starts_with("Array(") && !inner.starts_with("Map("); + let ch_type = if nullable { format!("Nullable({inner})") + } else { + inner.clone() }; let default_sql = render_default(att, &inner); Ok(ResolvedColumn { @@ -106,11 +112,45 @@ pub fn base_type_for(att: &RelAttr) -> Result { TIMESTAMPOID | TIMESTAMPTZOID => datetime64_ch_type(att.typmod), UUIDOID => "UUID".into(), INETOID | CIDROID => "String".into(), - JSONOID | JSONBOID => "String".into(), - _ => "String".into(), + JSONOID | JSONBOID => "JSON".into(), + // Extension / array types carry dynamic OIDs, so match on the type + // name: `hstore` → Map, `vector` (pgvector) → Array(Float32), + // `_` (PG array convention) → Array of a supported element, else + // the String fallback (geography stays String, rendered WKT at decode). + _ => { + if att.type_name == "hstore" { + "Map(String, Nullable(String))".into() + } else if att.type_name == "vector" || att.type_name == "halfvec" { + "Array(Float32)".into() + } else if let Some(arr) = array_ch_type(&att.type_name) { + arr + } else { + "String".into() + } + } }) } +/// PG array type name (`_int4`, `_text`, …) → `Array(Nullable())` for +/// supported element types; `None` (→ String fallback) for anything else, +/// including multi-dim-only element names walshadow can't render natively. +fn array_ch_type(type_name: &str) -> Option { + let elem = type_name.strip_prefix('_')?; + let inner = match elem { + "int2" => "Int16", + "int4" => "Int32", + "int8" => "Int64", + "oid" => "UInt32", + "float4" => "Float32", + "float8" => "Float64", + "bool" => "Bool", + "text" | "varchar" | "bpchar" | "name" | "citext" => "String", + "uuid" => "UUID", + _ => "String", + }; + Some(format!("Array(Nullable({inner}))")) +} + /// Render post-`DEFAULT ` fragment from `missing_text` (PG fast-path /// `ALTER TABLE ... ADD COLUMN ... DEFAULT k`). `None` when no fast-path /// default or value not cleanly expressible as CH literal @@ -395,6 +435,88 @@ mod tests { assert_eq!(render_pg_timestamp(i64::MAX), "1970-01-01 00:00:00"); } + fn named_attr(oid: u32, type_name: &str, not_null: bool) -> RelAttr { + let mut a = attr(oid, -1, not_null, None); + a.type_name = type_name.into(); + a + } + + #[test] + fn arrays_map_to_native_ch_array() { + assert_eq!( + base_type_for(&named_attr(1007, "_int4", true)).unwrap(), + "Array(Nullable(Int32))" + ); + assert_eq!( + base_type_for(&named_attr(1016, "_int8", true)).unwrap(), + "Array(Nullable(Int64))" + ); + assert_eq!( + base_type_for(&named_attr(1009, "_text", true)).unwrap(), + "Array(Nullable(String))" + ); + // Unsupported element type keeps array shape with String elements. + assert_eq!( + base_type_for(&named_attr(1231, "_numeric", true)).unwrap(), + "Array(Nullable(String))" + ); + } + + #[test] + fn hstore_maps_to_ch_map() { + assert_eq!( + base_type_for(&named_attr(99999, "hstore", true)).unwrap(), + "Map(String, Nullable(String))" + ); + } + + #[test] + fn json_maps_to_ch_json() { + assert_eq!( + base_type_for(&attr(JSONOID, -1, true, None)).unwrap(), + "JSON" + ); + assert_eq!( + base_type_for(&attr(JSONBOID, -1, true, None)).unwrap(), + "JSON" + ); + assert_eq!( + map(&attr(JSONBOID, -1, true, None), false).unwrap().ch_type, + "JSON" + ); + assert_eq!( + map(&attr(JSONBOID, -1, false, None), false) + .unwrap() + .ch_type, + "Nullable(JSON)" + ); + } + + #[test] + fn vector_maps_to_array_float32() { + assert_eq!( + base_type_for(&named_attr(99998, "vector", false)).unwrap(), + "Array(Float32)" + ); + } + + #[test] + fn composite_types_never_wrapped_in_nullable() { + // nullable source array/hstore → bare Array/Map (CH forbids Nullable) + assert_eq!( + map(&named_attr(1007, "_int4", false), false) + .unwrap() + .ch_type, + "Array(Nullable(Int32))" + ); + assert_eq!( + map(&named_attr(99999, "hstore", false), false) + .unwrap() + .ch_type, + "Map(String, Nullable(String))" + ); + } + #[test] fn int_types_map_to_signed_ch() { assert_eq!( diff --git a/src/emit/ch_emitter.rs b/src/emit/ch_emitter.rs index cddd78cd..7aff37c3 100644 --- a/src/emit/ch_emitter.rs +++ b/src/emit/ch_emitter.rs @@ -27,7 +27,7 @@ use std::num::NonZeroUsize; use std::sync::Arc; use std::time::Duration; -use clickhouse_c::{Allocator, ColumnBuilder, Kind, TypeAst}; +use clickhouse_c::{Allocator, ColumnBuilder, Kind, TypeAst, TypeRef}; #[cfg(test)] use crate::ch::is_retryable; @@ -1035,6 +1035,66 @@ pub(crate) enum ColumnBuf { data: Vec, null_map: Vec, }, + /// `Array()`: `offsets[i]` is the cumulative element count through + /// row `i`; elements accumulate in `inner`. `elem` drives text→wire for + /// each element. Not itself Nullable (CH forbids `Nullable(Array)`): a + /// NULL/absent source value lands as an empty array. + Array { + offsets: Vec, + inner: Box, + elem: ChScalar, + }, + /// `Map(String, )` == `Array(Tuple(keys, vals))`. `offsets[i]` is + /// the cumulative entry count through row `i`. Keys are non-nullable + /// `String`; a NULL/absent source value lands as an empty map. + Map { + offsets: Vec, + keys: Box, + vals: Box, + }, + /// CH `JSON` (string serialization). `null_map` is `Some` for + /// `Nullable(JSON)`. A value that isn't valid JSON text (unresolved jsonb + /// on-disk bytes, non-text) can't go in — it becomes NULL (nullable) or + /// `{}` (non-nullable), never raw bytes CH would reject. + Json { + offsets: Vec, + data: Vec, + null_map: Option>, + }, +} + +/// Scalar CH element kind for array elements: drives parsing a PG array text +/// token into little-endian wire bytes (or a String element). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum ChScalar { + Bool, + I8, + I16, + I32, + I64, + U32, + F32, + F64, + Uuid, + Str, +} + +impl ChScalar { + fn from_kind(kind: Option) -> Option { + Some(match kind? { + Kind::Bool => Self::Bool, + Kind::Int8 => Self::I8, + Kind::Int16 => Self::I16, + Kind::Int32 => Self::I32, + Kind::Int64 => Self::I64, + Kind::UInt32 => Self::U32, + Kind::Float32 => Self::F32, + Kind::Float64 => Self::F64, + Kind::Uuid => Self::Uuid, + Kind::String => Self::Str, + _ => return None, + }) + } } impl ColumnBuf { @@ -1044,7 +1104,43 @@ impl ColumnBuf { /// means varlen on-wire shape; only `String` is handled, other /// varlens fall through to the String arms and die on first `append`. fn new_for_ast(ast: &TypeAst) -> Result { - let view = ast.view(); + Self::new_for_view(ast.view()) + } + + fn new_for_view(view: TypeRef<'_>) -> Result { + match view.kind() { + Some(Kind::Array) => { + let child = view + .child(0) + .ok_or_else(|| EmitterError::Type("Array type with no child".into()))?; + let elem_kind = if child.kind() == Some(Kind::Nullable) { + child.child(0).and_then(|c| c.kind()) + } else { + child.kind() + }; + let elem = ChScalar::from_kind(elem_kind) + .ok_or_else(|| EmitterError::Type("Array element type not encodable".into()))?; + return Ok(Self::Array { + offsets: Vec::new(), + inner: Box::new(Self::new_for_view(child)?), + elem, + }); + } + Some(Kind::Map) => { + let k = view + .child(0) + .ok_or_else(|| EmitterError::Type("Map type with no key".into()))?; + let v = view + .child(1) + .ok_or_else(|| EmitterError::Type("Map type with no value".into()))?; + return Ok(Self::Map { + offsets: Vec::new(), + keys: Box::new(Self::new_for_view(k)?), + vals: Box::new(Self::new_for_view(v)?), + }); + } + _ => {} + } let (nullable, inner) = if view.kind() == Some(Kind::Nullable) { ( true, @@ -1054,6 +1150,13 @@ impl ColumnBuf { } else { (false, view) }; + if inner.kind() == Some(Kind::Json) { + return Ok(Self::Json { + offsets: Vec::new(), + data: Vec::new(), + null_map: nullable.then(Vec::new), + }); + } let elem = inner.elem_size(); Ok(match (nullable, elem) { (false, 0) => Self::String { @@ -1089,6 +1192,31 @@ impl ColumnBuf { data, null_map, } => offsets.len() * 8 + data.len() + null_map.len(), + Self::Array { offsets, inner, .. } => offsets.len() * 8 + inner.approx_size(), + Self::Map { + offsets, + keys, + vals, + } => offsets.len() * 8 + keys.approx_size() + vals.approx_size(), + Self::Json { + offsets, + data, + null_map, + } => offsets.len() * 8 + data.len() + null_map.as_ref().map_or(0, Vec::len), + } + } + + /// Rows appended so far (element count for the inner buffer of an Array/Map). + fn rows(&self) -> usize { + match self { + Self::Fixed { width, bytes } => bytes.len().checked_div(*width).unwrap_or(0), + Self::String { offsets, .. } => offsets.len(), + Self::NullableFixed { null_map, .. } | Self::NullableString { null_map, .. } => { + null_map.len() + } + Self::Array { offsets, .. } + | Self::Map { offsets, .. } + | Self::Json { offsets, .. } => offsets.len(), } } @@ -1112,6 +1240,28 @@ impl ColumnBuf { offsets.push(data.len() as u64); Ok(()) } + // CH Array/Map can't be Nullable; a NULL source value is an empty + // array/map (offset repeats the running element count). + Self::Array { offsets, inner, .. } => { + offsets.push(inner.rows() as u64); + Ok(()) + } + Self::Map { offsets, keys, .. } => { + offsets.push(keys.rows() as u64); + Ok(()) + } + Self::Json { + offsets, + data, + null_map, + } => { + if let Some(nm) = null_map { + nm.push(1); + } + data.extend_from_slice(b"{}"); + offsets.push(data.len() as u64); + Ok(()) + } _ => Err(EmitterError::UnsupportedValue { target_column: String::new(), kind: "NULL for non-Nullable column", @@ -1291,46 +1441,66 @@ impl TableEncoder { } } -/// Innermost leaf node per column: the `Nullable` value slab, or the whole -/// column when not nullable. A `ColumnBuilder` node borrows its slabs and -/// cannot move once a wrapper or the block aliases it, so the caller owns -/// these leaves for the block's lifetime. Buffers stay immutable until -/// `send_data` returns. -pub(crate) fn build_leaves( - bufs: &[ColumnBuf], +/// Build the root `ColumnBuilder` for one column's buffer, recursively for +/// composite (`Array`/`Map`) shapes. A `ColumnBuilder` node borrows its slabs +/// and any child nodes and cannot move once aliased, so every intermediate +/// node is allocated in `bump` (stable address, one shared `'b` lifetime) and +/// lives until `send_data` returns. `n_rows` is the row count at this level +/// (element count for a nested inner buffer). +pub(crate) fn build_column<'b>( + buf: &'b ColumnBuf, n_rows: usize, -) -> Result>, EmitterError> { - bufs.iter() - .map(|buf| match buf { - ColumnBuf::Fixed { width, bytes: data } - | ColumnBuf::NullableFixed { - width, inner: data, .. - } => ColumnBuilder::fixed(data, *width, n_rows).map_err(Into::into), - ColumnBuf::String { offsets, data } - | ColumnBuf::NullableString { offsets, data, .. } => { - ColumnBuilder::string(offsets, data, n_rows).map_err(Into::into) + bump: &'b bumpalo::Bump, +) -> Result, EmitterError> { + Ok(match buf { + ColumnBuf::Fixed { width, bytes } => ColumnBuilder::fixed(bytes, *width, n_rows)?, + ColumnBuf::String { offsets, data } => ColumnBuilder::string(offsets, data, n_rows)?, + ColumnBuf::NullableFixed { + width, + null_map, + inner, + } => { + let leaf = bump.alloc(ColumnBuilder::fixed(inner, *width, n_rows)?); + leaf.nullable(null_map)? + } + ColumnBuf::NullableString { + offsets, + data, + null_map, + } => { + let leaf = bump.alloc(ColumnBuilder::string(offsets, data, n_rows)?); + leaf.nullable(null_map)? + } + ColumnBuf::Array { offsets, inner, .. } => { + let child = bump.alloc(build_column(inner, inner.rows(), bump)?); + child.array(offsets, n_rows)? + } + ColumnBuf::Map { + offsets, + keys, + vals, + } => { + let entries = keys.rows(); + let kb = build_column(keys, entries, bump)?; + let vb = build_column(vals, entries, bump)?; + let children = bump.alloc([kb, vb]); + let ptrs: &mut [*mut clickhouse_c::sys::chc_column; 2] = + bump.alloc([std::ptr::null_mut(); 2]); + let tup = bump.alloc(ColumnBuilder::tuple(&children[..], &mut ptrs[..])?); + tup.array(offsets, n_rows)? + } + ColumnBuf::Json { + offsets, + data, + null_map, + } => match null_map { + Some(nm) => { + let leaf = bump.alloc(ColumnBuilder::string(offsets, data, n_rows)?); + leaf.nullable(nm)? } - }) - .collect() -} - -/// `Nullable` wrapper per column, `None` when the column is not nullable. -/// Each wrapper aliases its leaf in `leaves`, so `leaves` must be fully built -/// (no further pushes) and outlive the wrappers. Pair with [`build_leaves`]: -/// the block appends `roots[i]` when `Some`, else `leaves[i]`. -pub(crate) fn build_roots<'l, 'b: 'l>( - leaves: &'l [ColumnBuilder<'b>], - bufs: &'b [ColumnBuf], -) -> Result>>, EmitterError> { - leaves - .iter() - .zip(bufs) - .map(|(leaf, buf)| match buf { - ColumnBuf::NullableFixed { null_map, .. } - | ColumnBuf::NullableString { null_map, .. } => Ok(Some(leaf.nullable(null_map)?)), - ColumnBuf::Fixed { .. } | ColumnBuf::String { .. } => Ok(None), - }) - .collect() + None => ColumnBuilder::string(offsets, data, n_rows)?, + }, + }) } fn push_fixed(buf: &mut ColumnBuf, le: &[u8]) -> Result<(), EmitterError> { @@ -1600,6 +1770,12 @@ fn encode_value( v: &ColumnValue, decimal: Option, ) -> Result<(), EmitterError> { + match buf { + ColumnBuf::Array { .. } => return encode_array(buf, v), + ColumnBuf::Map { .. } => return encode_map(buf, v), + ColumnBuf::Json { .. } => return encode_json(buf, v), + _ => {} + } match v { ColumnValue::Null => buf.append_null(), ColumnValue::Bool(b) => buf.append_fixed_bytes(&[*b as u8]), @@ -1678,6 +1854,311 @@ fn encode_value( } } +/// Post-oracle text of an array/hstore value, or `None` for a NULL/absent +/// value. Non-text (unresolved bridge) logs and yields `None` (empty). +fn composite_text(v: &ColumnValue) -> Option<&str> { + match v { + ColumnValue::Text(s) | ColumnValue::Name(s) | ColumnValue::Json(s) => Some(s.as_str()), + ColumnValue::Null => None, + _ => { + tracing::warn!( + target: "walshadow::emitter", + "array/map column value not resolved to text; emitting empty", + ); + None + } + } +} + +fn encode_array(buf: &mut ColumnBuf, v: &ColumnValue) -> Result<(), EmitterError> { + let ColumnBuf::Array { + offsets, + inner, + elem, + } = buf + else { + unreachable!("encode_array on non-Array buffer"); + }; + let elem = *elem; + if let Some(text) = composite_text(v) { + // pgvector renders `[1,2,3]`; PG arrays render `{1,2}`. + let parsed = if text.trim_start().starts_with('[') && text.trim_end().ends_with(']') { + parse_vector_list(text) + } else { + parse_pg_array_1d(text) + }; + match parsed { + Ok(elems) => { + for e in &elems { + match e { + Some(tok) => append_scalar_text(inner, elem, tok)?, + None => inner.append_null()?, + } + } + } + Err(()) => tracing::warn!( + target: "walshadow::emitter", + value = %text, + "array value not 1-D/parseable; emitting empty array", + ), + } + } + offsets.push(inner.rows() as u64); + Ok(()) +} + +fn encode_map(buf: &mut ColumnBuf, v: &ColumnValue) -> Result<(), EmitterError> { + let ColumnBuf::Map { + offsets, + keys, + vals, + } = buf + else { + unreachable!("encode_map on non-Map buffer"); + }; + if let Some(text) = composite_text(v) { + match parse_hstore(text) { + Ok(pairs) => { + for (k, val) in &pairs { + keys.append_string_bytes(k.as_bytes())?; + match val { + Some(vv) => vals.append_string_bytes(vv.as_bytes())?, + None => vals.append_null()?, + } + } + } + Err(()) => tracing::warn!( + target: "walshadow::emitter", + value = %text, + "hstore value unparseable; emitting empty map", + ), + } + } + offsets.push(keys.rows() as u64); + Ok(()) +} + +fn encode_json(buf: &mut ColumnBuf, v: &ColumnValue) -> Result<(), EmitterError> { + let ColumnBuf::Json { + offsets, + data, + null_map, + } = buf + else { + unreachable!("encode_json on non-Json buffer"); + }; + if let ColumnValue::Json(s) | ColumnValue::Text(s) | ColumnValue::Name(s) = v + && s.trim_start().starts_with('{') + { + if let Some(nm) = null_map { + nm.push(0); + } + data.extend_from_slice(s.as_bytes()); + } else { + if let Some(nm) = null_map { + nm.push(1); + } + data.extend_from_slice(b"{}"); + } + offsets.push(data.len() as u64); + Ok(()) +} + +/// Append one array element token (already unquoted) to the element buffer, +/// parsing it per the CH element kind. +fn append_scalar_text(buf: &mut ColumnBuf, elem: ChScalar, tok: &str) -> Result<(), EmitterError> { + let bad = |k: &'static str| EmitterError::Type(format!("array element {k}: {tok:?}")); + match elem { + ChScalar::Bool => buf.append_fixed_bytes(&[u8::from(matches!(tok, "t" | "true" | "1"))]), + ChScalar::I8 => { + buf.append_fixed_bytes(&tok.parse::().map_err(|_| bad("i8"))?.to_le_bytes()) + } + ChScalar::I16 => { + buf.append_fixed_bytes(&tok.parse::().map_err(|_| bad("i16"))?.to_le_bytes()) + } + ChScalar::I32 => { + buf.append_fixed_bytes(&tok.parse::().map_err(|_| bad("i32"))?.to_le_bytes()) + } + ChScalar::I64 => { + buf.append_fixed_bytes(&tok.parse::().map_err(|_| bad("i64"))?.to_le_bytes()) + } + ChScalar::U32 => { + buf.append_fixed_bytes(&tok.parse::().map_err(|_| bad("u32"))?.to_le_bytes()) + } + ChScalar::F32 => { + buf.append_fixed_bytes(&tok.parse::().map_err(|_| bad("f32"))?.to_le_bytes()) + } + ChScalar::F64 => { + buf.append_fixed_bytes(&tok.parse::().map_err(|_| bad("f64"))?.to_le_bytes()) + } + ChScalar::Uuid => { + let raw = parse_uuid_text(tok).ok_or_else(|| bad("uuid"))?; + buf.append_fixed_bytes(&crate::decode::codecs::uuid_to_ch_wire(&raw)) + } + ChScalar::Str => buf.append_string_bytes(tok.as_bytes()), + } +} + +fn parse_uuid_text(s: &str) -> Option<[u8; 16]> { + let hex: Vec = s.bytes().filter(|c| *c != b'-').collect(); + if hex.len() != 32 { + return None; + } + let mut out = [0u8; 16]; + for (i, byte) in out.iter_mut().enumerate() { + let h = std::str::from_utf8(&hex[i * 2..i * 2 + 2]).ok()?; + *byte = u8::from_str_radix(h, 16).ok()?; + } + Some(out) +} + +/// A `"..."`-quoted token starting at `start`; returns the unescaped value and +/// the index just past the closing quote. Only `\\` and `\"` are escapes in +/// PG array/hstore output; raw multibyte bytes are copied verbatim. +fn read_quoted(s: &str, start: usize) -> Result<(String, usize), ()> { + let b = s.as_bytes(); + let mut i = start + 1; + let mut out: Vec = Vec::new(); + while i < b.len() { + match b[i] { + b'\\' => { + i += 1; + if i >= b.len() { + return Err(()); + } + out.push(b[i]); + i += 1; + } + b'"' => return String::from_utf8(out).map(|v| (v, i + 1)).map_err(|_| ()), + c => { + out.push(c); + i += 1; + } + } + } + Err(()) +} + +/// Parse PG's canonical 1-D array output `{a,b,"c,d",NULL}` into elements +/// (`None` = NULL). `Err` for multi-dimensional input (nested `{}` or a +/// `[lo:hi]=` dimension prefix), which the caller downgrades to an empty array. +fn parse_pg_array_1d(s: &str) -> Result>, ()> { + let s = s.trim(); + if s.starts_with('[') { + return Err(()); + } + let inner = s + .strip_prefix('{') + .and_then(|x| x.strip_suffix('}')) + .ok_or(())?; + let mut out = Vec::new(); + if inner.is_empty() { + return Ok(out); + } + let b = inner.as_bytes(); + let mut i = 0; + loop { + if b[i] == b'"' { + let (val, ni) = read_quoted(inner, i)?; + out.push(Some(val)); + i = ni; + } else { + let start = i; + while i < b.len() && b[i] != b',' { + if b[i] == b'{' || b[i] == b'}' { + return Err(()); + } + i += 1; + } + let tok = &inner[start..i]; + out.push(if tok == "NULL" { + None + } else { + Some(tok.to_string()) + }); + } + if i >= b.len() { + break; + } + if b[i] == b',' { + i += 1; + continue; + } + return Err(()); + } + Ok(out) +} + +/// Parse a pgvector literal `[1,2,3]` into element tokens. No NULLs/quotes. +fn parse_vector_list(s: &str) -> Result>, ()> { + let inner = s + .trim() + .strip_prefix('[') + .and_then(|x| x.strip_suffix(']')) + .ok_or(())?; + if inner.is_empty() { + return Ok(Vec::new()); + } + Ok(inner + .split(',') + .map(|t| Some(t.trim().to_string())) + .collect()) +} + +/// Parse PG's hstore output `"k"=>"v", "k2"=>NULL` into key/value pairs +/// (`None` = NULL value). Keys are always quoted. +fn parse_hstore(s: &str) -> Result)>, ()> { + let s = s.trim(); + let mut out = Vec::new(); + if s.is_empty() { + return Ok(out); + } + let b = s.as_bytes(); + let mut i = 0; + let skip_ws = |b: &[u8], mut i: usize| { + while i < b.len() && b[i] == b' ' { + i += 1; + } + i + }; + loop { + i = skip_ws(b, i); + if i >= b.len() { + break; + } + if b[i] != b'"' { + return Err(()); + } + let (key, ni) = read_quoted(s, i)?; + i = skip_ws(b, ni); + if i + 1 >= b.len() || b[i] != b'=' || b[i + 1] != b'>' { + return Err(()); + } + i = skip_ws(b, i + 2); + let val = if i < b.len() && b[i] == b'"' { + let (v, ni) = read_quoted(s, i)?; + i = ni; + Some(v) + } else if s[i..].starts_with("NULL") { + i += 4; + None + } else { + return Err(()); + }; + out.push((key, val)); + i = skip_ws(b, i); + if i >= b.len() { + break; + } + if b[i] == b',' { + i += 1; + continue; + } + return Err(()); + } + Ok(out) +} + crate::atomic_stats! { /// CH emitter counters. `fetch_add(_, Relaxed)`; status loop reads /// via `.load(Relaxed)`. @@ -1798,6 +2279,36 @@ impl std::fmt::Debug for ColumnBuf { .field("offsets_len", &offsets.len()) .field("data_len", &data.len()) .finish(), + Self::Array { + offsets, + inner, + elem, + } => f + .debug_struct("Array") + .field("rows", &offsets.len()) + .field("elem", elem) + .field("inner", inner) + .finish(), + Self::Map { + offsets, + keys, + vals, + } => f + .debug_struct("Map") + .field("rows", &offsets.len()) + .field("keys", keys) + .field("vals", vals) + .finish(), + Self::Json { + offsets, + data, + null_map, + } => f + .debug_struct("Json") + .field("rows", &offsets.len()) + .field("data_len", &data.len()) + .field("nullable", &null_map.is_some()) + .finish(), } } } @@ -2127,6 +2638,10 @@ mod tests { ("Nullable(String)", "NullableString"), ("FixedString(7)", "Fixed"), ("Nullable(FixedString(7))", "NullableFixed"), + ("Array(Nullable(Int32))", "Array"), + ("Map(String, Nullable(String))", "Map"), + ("JSON", "Json"), + ("Nullable(JSON)", "Json"), ]; for (name, tag) in cases { let ast = TypeAst::parse(name, alloc).expect("parses"); @@ -2136,11 +2651,151 @@ mod tests { ColumnBuf::String { .. } => "String", ColumnBuf::NullableFixed { .. } => "NullableFixed", ColumnBuf::NullableString { .. } => "NullableString", + ColumnBuf::Array { .. } => "Array", + ColumnBuf::Map { .. } => "Map", + ColumnBuf::Json { .. } => "Json", }; assert_eq!(actual, tag, "{name}"); } } + #[test] + fn encode_json_text_null_and_unresolved() { + let alloc = Allocator::stdlib(); + let ast = TypeAst::parse("Nullable(JSON)", alloc).unwrap(); + let mut buf = ColumnBuf::new_for_ast(&ast).unwrap(); + encode_value(&mut buf, &ColumnValue::Json("{\"a\":1}".into()), None).unwrap(); + encode_value(&mut buf, &ColumnValue::Null, None).unwrap(); + // unresolved jsonb (raw on-disk bytes) must not reach CH as raw → NULL + encode_value( + &mut buf, + &ColumnValue::PgPending { + type_oid: 3802, + raw: vec![1, 0, 0, 0], + }, + None, + ) + .unwrap(); + let ColumnBuf::Json { + offsets, + data, + null_map, + } = &buf + else { + panic!("expected Json buf"); + }; + assert_eq!(data, &br#"{"a":1}{}{}"#[..]); + assert_eq!(offsets, &[7, 9, 11]); + assert_eq!(null_map.as_deref(), Some(&[0u8, 1, 1][..])); + } + + #[test] + fn parse_array_handles_nulls_quotes_and_rejects_multidim() { + assert_eq!( + parse_pg_array_1d("{1,2,3}").unwrap(), + vec![Some("1".into()), Some("2".into()), Some("3".into())] + ); + assert_eq!( + parse_pg_array_1d("{}").unwrap(), + Vec::>::new() + ); + assert_eq!( + parse_pg_array_1d("{a,NULL,\"b,c\"}").unwrap(), + vec![Some("a".into()), None, Some("b,c".into())] + ); + assert_eq!( + parse_pg_array_1d("{\"a\\\"b\"}").unwrap(), + vec![Some("a\"b".into())] + ); + assert!(parse_pg_array_1d("{{1,2},{3,4}}").is_err()); + assert!(parse_pg_array_1d("[1:2]={1,2}").is_err()); + } + + #[test] + fn parse_vector_and_encode_into_array_float32() { + assert_eq!( + parse_vector_list("[1,2,3]").unwrap(), + vec![Some("1".into()), Some("2".into()), Some("3".into())] + ); + assert_eq!( + parse_vector_list("[]").unwrap(), + Vec::>::new() + ); + let alloc = Allocator::stdlib(); + let ast = TypeAst::parse("Array(Float32)", alloc).unwrap(); + let mut buf = ColumnBuf::new_for_ast(&ast).unwrap(); + encode_value(&mut buf, &ColumnValue::Text("[1.5,2.5]".into()), None).unwrap(); + let ColumnBuf::Array { + offsets, + inner, + elem, + } = &buf + else { + panic!("expected Array buf"); + }; + assert_eq!(*elem, ChScalar::F32); + assert_eq!(offsets, &[2]); + assert_eq!(inner.rows(), 2); + } + + #[test] + fn parse_hstore_pairs_and_nulls() { + assert_eq!( + parse_hstore("\"k\"=>\"v\", \"x\"=>NULL").unwrap(), + vec![("k".into(), Some("v".into())), ("x".into(), None)] + ); + assert_eq!( + parse_hstore("").unwrap(), + Vec::<(String, Option)>::new() + ); + assert_eq!( + parse_hstore("\"a=>b\"=>\"c,d\"").unwrap(), + vec![("a=>b".into(), Some("c,d".into()))] + ); + } + + #[test] + fn encode_array_offsets_track_elements_and_null_is_empty() { + let alloc = Allocator::stdlib(); + let ast = TypeAst::parse("Array(Nullable(Int32))", alloc).unwrap(); + let mut buf = ColumnBuf::new_for_ast(&ast).unwrap(); + encode_value(&mut buf, &ColumnValue::Text("{10,20,30}".into()), None).unwrap(); + encode_value(&mut buf, &ColumnValue::Null, None).unwrap(); + encode_value(&mut buf, &ColumnValue::Text("{40}".into()), None).unwrap(); + let ColumnBuf::Array { offsets, inner, .. } = &buf else { + panic!("expected Array buf"); + }; + // cumulative element counts: 3, still 3 (empty), 4 + assert_eq!(offsets, &[3, 3, 4]); + assert_eq!(inner.rows(), 4); + } + + #[test] + fn encode_map_offsets_track_entries() { + let alloc = Allocator::stdlib(); + let ast = TypeAst::parse("Map(String, Nullable(String))", alloc).unwrap(); + let mut buf = ColumnBuf::new_for_ast(&ast).unwrap(); + encode_value( + &mut buf, + &ColumnValue::Text("\"a\"=>\"1\", \"b\"=>NULL".into()), + None, + ) + .unwrap(); + encode_value(&mut buf, &ColumnValue::Null, None).unwrap(); + let ColumnBuf::Map { + offsets, + keys, + vals, + .. + } = &buf + else { + panic!("expected Map buf"); + }; + assert_eq!(offsets, &[2, 2]); + assert_eq!(keys.rows(), 2); + assert_eq!(vals.rows(), 2); + } + #[test] fn decimal_text_scales_to_integer() { fn le_i64(text: &str, scale: i32) -> [u8; 8] { diff --git a/src/emit/pipeline/decode.rs b/src/emit/pipeline/decode.rs index d960040f..50d6d106 100644 --- a/src/emit/pipeline/decode.rs +++ b/src/emit/pipeline/decode.rs @@ -120,6 +120,14 @@ pub async fn decode_and_route( commit_ts, commit_lsn, }; + // Extension types (PostGIS geography → WKT, pgvector → [..]) rendered + // in-tree before the bridge, which can't (or needn't) resolve them. + if let Some(t) = committed.decoded.new.as_mut() { + crate::ops::oracle::render_ext_columns(&rel.attributes, &mut t.columns); + } + if let Some(t) = committed.decoded.old.as_mut() { + crate::ops::oracle::render_ext_columns(&rel.attributes, &mut t.columns); + } if let Some(oracle) = &ctx.oracle { // Resolve PgPending via shadow PG extension if let Some(t) = committed.decoded.new.as_mut() { diff --git a/src/emit/pipeline/inserter.rs b/src/emit/pipeline/inserter.rs index 5b1f2203..67f38316 100644 --- a/src/emit/pipeline/inserter.rs +++ b/src/emit/pipeline/inserter.rs @@ -12,7 +12,7 @@ //! dedups by `_lsn`). Retry-exhaustion is fatal: the watermark can't advance //! without this batch. -use clickhouse_c::{Allocator, BlockBuilder, BoxedAsyncClient, TypeAst}; +use clickhouse_c::{Allocator, BlockBuilder, BoxedAsyncClient, ColumnBuilder, TypeAst}; use tokio::task::JoinHandle; use crate::ch::{ @@ -20,7 +20,7 @@ use crate::ch::{ reconnect_if_idle, with_timeout, }; use crate::config::ResolvedConfig; -use crate::emit::ch_emitter::{EmitterConfig, EmitterStats, build_leaves, build_roots}; +use crate::emit::ch_emitter::{EmitterConfig, EmitterStats, build_column}; use crate::emit::pipeline::Fatal; use crate::emit::pipeline::ack::AckHandle; use crate::emit::pipeline::batcher::{BatchMeta, InsertBatch}; @@ -172,12 +172,14 @@ impl Inserter { .asts .remove(&batch.meta.table_key) .expect("ensure_asts inserted"); + let arena = bumpalo::Bump::new(); let result = 'send: { - let leaves = match build_leaves(&batch.buffers, batch.n_rows) { - Ok(v) => v, - Err(e) => break 'send Err(e), - }; - let roots = match build_roots(&leaves, &batch.buffers) { + let nodes: Vec> = match batch + .buffers + .iter() + .map(|buf| build_column(buf, batch.n_rows, &arena)) + .collect::>() + { Ok(v) => v, Err(e) => break 'send Err(e), }; @@ -188,8 +190,7 @@ impl Inserter { .iter() .enumerate() .try_for_each(|(i, col)| { - let node = roots[i].as_ref().unwrap_or(&leaves[i]); - bb.append(&col.name, asts[i].view(), node) + bb.append(&col.name, asts[i].view(), &nodes[i]) .map_err(Into::into) }); match appended { diff --git a/src/ops/oracle.rs b/src/ops/oracle.rs index 3a291d83..399ccf1b 100644 --- a/src/ops/oracle.rs +++ b/src/ops/oracle.rs @@ -17,6 +17,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use crate::decode::heap_decoder::ColumnValue; use crate::ops::bridge::{Bridge, DecodedItem}; +use crate::schema::RelAttr; crate::atomic_stats! { pub struct OracleStats { @@ -117,6 +118,77 @@ pub async fn resolve_pending_tuple(oracle: &Oracle, columns: &mut [Option]) { + for att in attrs { + if att.dropped { + continue; + } + let Ok(idx) = usize::try_from(att.attnum - 1) else { + continue; + }; + let Some(cell) = columns.get_mut(idx) else { + continue; + }; + let Some(ColumnValue::PgPending { raw, .. }) = cell.as_ref() else { + continue; + }; + let rendered = match att.type_name.as_str() { + "geography" | "geometry" => gserialized_point_to_wkt(raw), + "vector" | "halfvec" => vector_to_text(raw), + _ => None, + }; + if let Some(text) = rendered { + *cell = Some(ColumnValue::Text(text)); + } + } +} + +/// pgvector on-disk `vector`: `[dim u16-le][unused u16-le][f32-le × dim]` → +/// `[a,b,c]`. `None` if the body is short of `dim` floats. +fn vector_to_text(raw: &[u8]) -> Option { + if raw.len() < 4 { + return None; + } + let dim = u16::from_le_bytes(raw[0..2].try_into().ok()?) as usize; + if raw.len() < 4 + dim * 4 { + return None; + } + let mut out = String::from("["); + for i in 0..dim { + if i > 0 { + out.push(','); + } + let off = 4 + i * 4; + let f = f32::from_le_bytes(raw[off..off + 4].try_into().ok()?); + out.push_str(&f.to_string()); + } + out.push(']'); + Some(out) +} + +/// PostGIS on-disk GSERIALIZED → `POINT(x y)` for 2-D points. Layout: +/// `[srid(3) + gflags(1)][geomtype u32-le][…]`; POINT has geomtype 1 and its +/// two `f64`-LE coordinates are the trailing 16 bytes. `None` for non-points. +fn gserialized_point_to_wkt(raw: &[u8]) -> Option { + if raw.len() < 16 { + return None; + } + let geomtype = u32::from_le_bytes(raw[4..8].try_into().ok()?); + if geomtype != 1 { + return None; + } + let n = raw.len(); + let x = f64::from_le_bytes(raw[n - 16..n - 8].try_into().ok()?); + let y = f64::from_le_bytes(raw[n - 8..n].try_into().ok()?); + Some(format!("POINT({x} {y})")) +} + impl OracleStats { pub fn summary(&self) -> String { use std::fmt::Write as _; @@ -139,6 +211,30 @@ impl OracleStats { mod tests { use super::*; + #[test] + fn gserialized_2d_point_renders_wkt() { + // [srid+gflags:4][geomtype=1:4][X f64le][Y f64le] + let mut raw = vec![0u8, 0, 0, 0, 1, 0, 0, 0]; + raw.extend_from_slice(&30.5_f64.to_le_bytes()); + raw.extend_from_slice(&81.25_f64.to_le_bytes()); + assert_eq!( + gserialized_point_to_wkt(&raw).as_deref(), + Some("POINT(30.5 81.25)") + ); + // non-point geomtype → None + raw[4] = 2; + assert_eq!(gserialized_point_to_wkt(&raw), None); + } + + #[test] + fn vector_body_renders_bracket_list() { + let mut raw = vec![3u8, 0, 0, 0]; // dim=3, unused=0 + for f in [1.5f32, 2.0, 3.25] { + raw.extend_from_slice(&f.to_le_bytes()); + } + assert_eq!(vector_to_text(&raw).as_deref(), Some("[1.5,2,3.25]")); + } + #[test] fn stats_summary_skips_zero_buckets() { let s = OracleStats::default(); diff --git a/src/source/transition.rs b/src/source/transition.rs index 09eb4d23..6f5f686b 100644 --- a/src/source/transition.rs +++ b/src/source/transition.rs @@ -298,7 +298,7 @@ impl ForkBarrier { fork_segment, }); } - if !self.shadow_apply_lsn.is_some_and(|a| a >= switch_lsn.get()) { + if self.shadow_apply_lsn.is_none_or(|a| a < switch_lsn.get()) { return Some(ForkWait::ShadowApply { applied: self.shadow_apply_lsn, fork: switch_lsn, diff --git a/tests/runtime_config_e2e.rs b/tests/runtime_config_e2e.rs index 58d0da4b..add6a48b 100644 --- a/tests/runtime_config_e2e.rs +++ b/tests/runtime_config_e2e.rs @@ -515,7 +515,8 @@ async fn opt_in_non_empty_backfills_pre_opt_in_rows() { FROM walshadow_test.inventory WHERE _is_deleted = 0 AND id = 2", ) .expect("ch backfilled row"); - assert_eq!(bolt, "bolt\t12.5\t2024-01-02 03:04:06.000000\t{\"b\": 2}"); + // `meta` is CH `Nullable(JSON)` now: JSON normalizes whitespace ({"b":2}). + assert_eq!(bolt, "bolt\t12.5\t2024-01-02 03:04:06.000000\t{\"b\":2}"); // TOAST-sized values survive COPY. let big = ch