From df78787a4554b64a05ecd40c0763ebeb874a176b Mon Sep 17 00:00:00 2001 From: ConorSwainDI Date: Fri, 24 Jul 2026 11:15:08 +0100 Subject: [PATCH 01/22] Parquet extract module for KDB-X --- di/pqx/init.q | 6 ++ di/pqx/pqx.md | 213 ++++++++++++++++++++++++++++++++++++++ di/pqx/pqx.q | 265 ++++++++++++++++++++++++++++++++++++++++++++++++ di/pqx/test.csv | 65 ++++++++++++ 4 files changed, 549 insertions(+) create mode 100644 di/pqx/init.q create mode 100644 di/pqx/pqx.md create mode 100644 di/pqx/pqx.q create mode 100644 di/pqx/test.csv diff --git a/di/pqx/init.q b/di/pqx/init.q new file mode 100644 index 00000000..2772865b --- /dev/null +++ b/di/pqx/init.q @@ -0,0 +1,6 @@ +qmamba:use`kx.qmamba +arrow:use`kx.arrow + +\l ::pqx.q + +export:([init;extract;default;manifest;checkandconvertcols;estimate;calibrateratio;calcsize;plan;datalookup;datalookuponesym;writefile;tryfn]) diff --git a/di/pqx/pqx.md b/di/pqx/pqx.md new file mode 100644 index 00000000..c3babe97 --- /dev/null +++ b/di/pqx/pqx.md @@ -0,0 +1,213 @@ +# di.pqx + +Converts an in-memory kdb+ table into one or more `.parquet` files via `kx.arrow`. Rows are grouped +by instrument and packed into files close to a configurable target size, splitting any single +oversized instrument across multiple files where required. A manifest recording what was written +(file, instruments, row count, time range, on-disk size) is accumulated in the module's private +`manifest` table; `extract` itself has no return value. + +--- + +## Features + +- Splits a table into one or more `.parquet` files targeting a configurable file size +- Groups rows by instrument so a single instrument's data is not split across files unless it alone exceeds the target size +- Optionally splits any oversized single instrument across multiple files +- Optionally calibrates the raw-to-parquet size ratio with a trial write, or uses a fixed ratio +- Optionally pre-sorts input data by instrument/time before writing +- Writes files sequentially or in parallel (`peach`) +- Accumulates a manifest of every file written, including row counts, instrument lists, time bounds and on-disk size + +--- + +## Dependencies + +| Dependency | Key | Required | Description | +|---|---|---|---| +| logger | `` `log `` | yes | dict with `info`, `warn`, and `error`, each binary `{[c;m]}` where `c` is a symbol context and `m` is a string | + +**Hard dependencies:** `kx.arrow` and `kx.qmamba` — both are loaded automatically (via `use`) when +`di.pqx` is imported, before `pqx.q` itself is loaded. `extract` calls +`` .m.di.0pqx.arrow.pq.writeParquetFromTable `` to perform every write; `kx.qmamba` is loaded but not +currently called anywhere in `pqx.q`. + +The `log` dependency must be passed to `init` inside a dict keyed on `` `log ``. `init` throws +immediately if `log` is absent, is not a dict, or is missing any of `info`/`warn`/`error`. The value +must already conform to the binary `{[c;m]}` contract — `init` performs no adaptation, so a raw +monadic `kx.log` instance must be wrapped by the caller first. Build the dict from `di.log`, or +hand-roll one. + +```q +logger:use`di.log +logdep:`info`warn`error!(logger.info;logger.warn;logger.error) +pqx:use`di.pqx +pqx.init[enlist[`log]!enlist logdep] + +/ or, skipping the by-hand dict: +/ pqx.init[logger.logdict] +``` + +--- + +## Options + +Passed as the `o` dictionary to `extract`, merged over the module's own `default` dict. Any keys +omitted from `o` fall back to the default shown below. + +| Key | Default | Type | Description | +|---|---|---|---| +| `targetsize` | `512*1024*1024` | long | Target size in bytes for each output file | +| `maxfactor` | `1.5` | float | Hard cap on file size, expressed as a multiple of `targetsize` | +| `splitoversized` | `1b` | boolean | Split any single instrument larger than the cap across multiple files | +| `calibrate` | `1b` | boolean | Run a trial write to measure the raw-to-parquet size ratio instead of using `compressionratio` | +| `compressionratio` | `0.30` | float | Raw-to-parquet size ratio used for size estimation when `calibrate` is `0b` | +| `symcol` | `` `sym `` | symbol | Instrument column | +| `timecol` | `` `time `` | symbol | Time column | +| `presort` | `1b` | boolean | Sort input by `` (symcol;timecol) `` before writing | +| `rowgroupbytes` | `128*1024*1024` | long | Reserved for future use — not currently read by the write path | +| `codec` | `` `zstd `` | symbol | Compression codec, upper-cased and applied to the writer's `` `COMPRESSION `` option | +| `complevel` | `3` | long | Reserved for future use — not currently read by the write path | +| `dictcols` | `` `sym`exchange `` | symbol list | Reserved for future use — not currently read by the write path | +| `parallel` | `0b` | boolean | Write files with `peach` instead of `each` | +| `outdir` | `` `:. `` | symbol | Root output directory | +| `filestub` | `"part"` | string | File name stub; files are written as `-NNNNN.parquet` | + +Output files are written to `//date=
/-NNNNN.parquet`. `extract` throws +(`` `di.pqx: no symcol found `` / `` `di.pqx: no timecol found ``) if the merged `symcol`/`timecol` +is not a column of the input table — this check runs unconditionally, regardless of `presort`. + +--- + +## Manifest Schema + +The module's `manifest` table accumulates one row per file written across all `extract` calls. +`extract` does not return this table (or anything else). `` pqx:use`di.pqx `` only captures a +one-time snapshot of `manifest`/`default` at load time — `pqx.manifest` and `pqx.default` do **not** +track later writes or updates. Query the live state via `` .m.di.0pqx.manifest `` instead (the +private namespace every kdb-x module loads into; see `di.depcheck`'s docs for the `` `.m.di.0 `` +convention). + +| Column | Type | Description | +|---|---|---| +| `file` | symbol | Path written | +| `seq` | long | Sequence number within the partition | +| `syms` | symbol list | Instruments contained in the file | +| `nsyms` | long | Count of instruments in the file | +| `rows` | long | Row count | +| `mintime` | timestamp | Minimum time across the file (for pruning) | +| `maxtime` | timestamp | Maximum time across the file | +| `estbytes` | long | Estimated size at plan time | +| `bytes` | long | Actual on-disk size | +| `split` | boolean | `1b` if this file is a chunk of a split oversized instrument | +| `status` | symbol | `` `ok `` or `` `error `` | + +--- + +## Initialisation + +`init[deps]` wires the injected `log` dependency and must be called before the first `extract`. It +does not touch the parquet writer — the `PARQUET_VERSION`/`COMPRESSION` write options are built +fresh inside every `extract` call, and `kx.arrow` is loaded automatically by the module itself (see +Dependencies). + +```q +pqx:use`di.pqx +logdep:`info`warn`error!({[c;m]};{[c;m]};{[c;m]}) +pqx.init[enlist[`log]!enlist logdep] +``` + +--- + +## Exported Functions + +| Function | Description | +|---|---| +| `init[deps]` | Wire the injected `log` dependency. Call once before the first `extract`. | +| `extract[t;tname;dt;o]` | Write a table out to one or more parquet files, appending one row per file to the module's `manifest`. Returns nothing. | + +The remaining exports — `checkandconvertcols`, `estimate`, `calibrateratio`, `calcsize`, `plan`, +`datalookup`, `datalookuponesym`, `writefile`, `tryfn` — are internal pipeline steps of `extract`, +exposed only so `k4unit` can exercise them directly. Call `extract` for normal use. + +### `init[deps]` +Validate the required `log` dependency and store it for use by every other function. + +| Arg | Type | Description | +|---|---|---| +| `deps` | dict | Must contain `` `log `` → `` `info`warn`error!(infofn;warnfn;errfn) `` | + +Throws (prefixed `di.pqx:`) if `deps` is not a dict, `log` is missing, or the log dict lacks any +required key. + +### `extract[t;tname;dt;o]` +Write table `t` out to one or more parquet files under `//date=
/`, appending one +row per file written to the module's `manifest`. `o` is merged over `default` (see Options). The +output directory is created before the size-estimation/calibration step, so a fresh `outdir` works +with the default `calibrate:1b`. + +| Parameter | Type | Description | +|---|---|---| +| `t` | table | Data to write | +| `tname` | symbol | Table name — used in the output path | +| `dt` | date | Partition date — used in the output path | +| `o` | dict | Option overrides, merged over `default` | + +```q +pqx.extract[trade;`trade;2025.07.15;`targetsize`codec!(256*1024*1024;`gzip)] +``` + +--- + +## Usage Example + +```q +// Include pqx module in a process +pqx:use`di.pqx + +// Wire the log dependency (once per process) +logger:use`di.log +pqx.init[logger.logdict] + +// Write `trade` for 2025.07.15, overriding the target file size and codec +pqx.extract[trade;`trade;2025.07.15;`targetsize`codec!(256*1024*1024;`gzip)] + +// extract has no return value - inspect the accumulated manifest directly. +// pqx.manifest is a frozen snapshot from load time - query the live namespace instead: +.m.di.0pqx.manifest + +file seq syms nsyms rows mintime maxtime estbytes bytes split status +-------------------------------------------------------------------------------------------------------------------------------------------------------------- +:./trade/date=2025.07.15/part-00001.parquet 1 `AAPL`MSFT 2 50000 2025.07.15D00:00:00.000000000 2025.07.15D23:59:59.000000000 1153433 1048576 0b ok +``` + +--- + +## Running Tests + +```q +k4unit:use`di.k4unit +k4unit.moduletest`di.pqx +``` + +`test.csv` drives `extract` across default and overridden options — presort on/off, an oversized +instrument with `splitoversized` on and off, a `symcol` override, parallel (`peach`) writes, and a +custom `filestub`/non-default codec — then asserts on the resulting `` .m.di.0pqx.manifest `` rows +and (via `` .m.di.0pqx.arrow.pq.readParquetToTable ``) the files written back to disk. It also +covers the failure paths: a zero-row table, a table missing `symcol`/`timecol`, and an invalid +codec (see Notes). + +--- + +## Notes + +- `pqx.manifest` and `pqx.default` (the values returned by `` pqx:use`di.pqx ``) are snapshots taken + once at load time — they never reflect later writes or changes. Always read `` .m.di.0pqx.manifest `` + for the live manifest. +- `rowgroupbytes`, `complevel`, and `dictcols` are accepted in `default` and any `o` override, but + nothing in the current write path reads them — only `` `PARQUET_VERSION `` (fixed at + `` `V2.LATEST ``) and `` `COMPRESSION `` (from `codec`) are passed to the writer. +- `symcol`/`timecol` presence is validated unconditionally on every `extract` call, even when + `presort` is `0b`. +- A per-file write failure (e.g. an invalid `codec`) is caught and logged, but currently still + causes `extract` to throw rather than recording that file with `` status=`error `` in the manifest + as the schema implies it should — see `test.csv`'s invalid-codec case. diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q new file mode 100644 index 00000000..ca259c39 --- /dev/null +++ b/di/pqx/pqx.q @@ -0,0 +1,265 @@ +/ define default config +default:( + `targetsize`maxfactor`splitoversized`calibrate`compressionratio, + `symcol`timecol`presort`rowgroupbytes`codec`complevel`dictcols, + `parallel`outdir`filestub + )!( + 512*1024*1024; / ~512 MB target file size + 1.5; / hard cap = target * maxfactor + 1b; / split instruments larger than the cap + 1b; / run a calibration write to set the ratio + 0.30; / raw->parquet ratio if not calibrating + `sym; / instrument column + `time; / time column + 1b; / sort by (sym,time) if not already + 128*1024*1024; / ~128 MB row groups + `zstd; / codec + 3; / compression level + `sym`exchange; / dictionary-encode these columns + 0b; / write files via peach + `:.; / output directory + "part" / file name stub + ); + +/ define empty schema for manifest +manifest:([] + file :`symbol$(); / path written + seq :`long$(); / sequence number within the partition + syms :(); / list of instruments in the file + nsyms :`long$(); / count of instruments + rows :`long$(); / row count + mintime :`timestamp$(); / min time across the file (for pruning) + maxtime :`timestamp$(); / max time across the file + estbytes :`long$(); / estimated size at plan time + bytes :`long$(); / actual on-disk size + split :`boolean$(); / true if this file is a chunk of a split oversized instrument + status :`symbol$() / `ok | `error + ); + +/ takes a table and checks whether any symbol or char columns exist in it +/ if so, converts these to strings, as no Parquet datatype equivalent +checkandconvertcols:{[t] + :$[count c:exec c from meta[t] where t in "Ssc"; + ![t;();0b;c!{(string;x)} each c]; + t + ] + }; + +/ for a partition of data, estimates the size of the tables to be saved to disk +/ if calibrate flag is true in o, a test write is carried out +/ returns a table of storage stats for all isntruments and the compression ratio, which may have changed depending on calibration +estimate:{[t;o;writeopt] + cnts:`rowcnt xasc 0!?[t;();enlist[o[`symcol]]!enlist[o[`symcol]];enlist[`rowcnt]!enlist(count;`time)]; //select rowcnt:count time by sym from t; + medsym:cnts @ first where abs[cnt-med[cnt]]=min[abs[cnt-med[cnt:cnts`rowcnt]]]; + bytesperrow:%[-22!t:.z.m.checkandconvertcols t[where t[o[`symcol]]=medsym[o[`symcol]]];medsym`rowcnt]; + + / calibrate compression ratio if option is enabled + if[o`calibrate; + .z.m.loginfo[`pqx;"Calibrating compression ratio"]; + o[`compressionratio]:.z.m.calibrateratio[t;o;writeopt] + ]; + + / return stats and (new) compression ratio + :(update estbyt:rowcnt*bytesperrow*o[`compressionratio] from cnts;o[`compressionratio]) + }; + +/ writes a sample of data to disk and reads its size on disk +/ calcaultes the compression ratio and returns if a new ratio was successfully calculated, otherwise old ratio is maintained +calibrateratio:{[t;o;writeopt] + / remove leading : from outdir + testloc:$[":" ~ first string[o`outdir]; + 1_string[o`outdir],"/testWrite.parquet"; + string[o`outdir],"/testWrite.parquet"]; + + / outputs two items - success flag and any error msg + .z.m.loginfo[`pqx;"Attempting test write of median sym for calibration"]; + res:.z.m.tryfn[`.m.di.0pqx.arrow.pq.writeParquetFromTable;(testloc;t;writeopt)]; + + / if error returned in first item of res, just return old compression ratio + if[not first res; + .z.m.logwarn[`pqx;"Calibration write unsuccessful. Error - ",last res]; + .z.m.logwarn[`pqx;"Returning existing compression ratio"]; + :o`compressionratio + ]; + + .z.m.loginfo[`pqx;"Test write successful"]; + + sizeondisk:hcount hsym `$testloc; + newratio:sizeondisk % -22!t; + + / clean test file + .z.m.loginfo[`pqx;"Cleaning up test file"]; + system "rm ",testloc; + + / return new ratio + .z.m.loginfo[`pqx;"Returning calibrated compression ratio"]; + :newratio + }; + +/ find the estimated size in bytes for each instrument per file to be saved down +/ in the case of a larger instrument being split, return count[seqno] number of instances of estbytes +calcsize:{[tbl;symcol;syms;seqno] + .z.m.loginfo[`pqx;"Getting estimated bytes for planned files"]; + :"j"$count[seqno]#%[sum[?[tbl;enlist(in;symcol;`syms);0b;()]`estbyt];count seqno] + }; + +/ planning function to bucket instruments based on next-fit packing +/ if an instrument can be added to a bucket without that bucket exceeding the target size, it will be added to that bucket, else a new bucket is started +/ large instruments are also split into multiple files if splitoversized flag is true +plan:{[t;o;maxsize] + symstats:t; + plans:(); + + / if split oversized is required, check against maxsize and return a plan entry for each required file + if[o`splitoversized; + .z.m.loginfo[`pqx;"Splitting large instruments"]; + t:update islargerthantargetsize:estbyt>maxsize from t; + oversized:select from t where islargerthantargetsize; + t:t except oversized; + oversized:update numfiles:ceiling[estbyt%maxsize] from oversized; + plans,:enlist each raze {[t;c] t[`numfiles]#enlist t[c]}[;o`symcol] each oversized + ]; + + / next fit function for packing instruments into buckets if they conform to the max size + if[count t; + .z.m.loginfo[`pqx;"Bucketing small instruments"]; + tabs:t[o[`symcol]]; + sizes:t`estbyt; + n:count tabs; + + step:{[maxsize;sizes;state;i] + sz:sizes i; + tot:state 1; + $[(tot+sz)>maxsize; (1+state 0; sz); (state 0; tot+sz)] + }[maxsize;sizes]; + bins: (step\[(0;0);til n])[;0]; + + plans,:value[tabs @ group bins] + ]; + plans:(1 + til count plans)!plans; + + / attach estbytes to plan's output + :update estbytes:.z.m.calcsize[symstats;o`symcol;;]'[syms;seqno] from {`syms`seqno!/: flip (key[x];value[x])} group plans + }; + +/ get lists of indices by file +/ a pass with multiple instruments is assumed to be one file only, hence the return is flattened into one list +datalookup:{[t;symcol;syms;cnt] + $[1 Date: Tue, 18 Aug 2026 11:54:02 +0100 Subject: [PATCH 02/22] Remove dependency on qmamba --- di/pqx/init.q | 5 ++-- di/pqx/pqx.md | 47 +++++++++++++++-------------- di/pqx/pqx.q | 78 ++++++++++++++++++++++++++++++------------------- di/pqx/test.csv | 40 ++++++++++++------------- 4 files changed, 94 insertions(+), 76 deletions(-) diff --git a/di/pqx/init.q b/di/pqx/init.q index 2772865b..6346b748 100644 --- a/di/pqx/init.q +++ b/di/pqx/init.q @@ -1,6 +1,7 @@ -qmamba:use`kx.qmamba +/ KDB-X Parquet extract module to save kdb+ data to parquet storage convention + arrow:use`kx.arrow \l ::pqx.q -export:([init;extract;default;manifest;checkandconvertcols;estimate;calibrateratio;calcsize;plan;datalookup;datalookuponesym;writefile;tryfn]) +export:([init;extract;getdefault;getmanifest;checkandconvertcols;estimate;plan;writefile;tryfn]) diff --git a/di/pqx/pqx.md b/di/pqx/pqx.md index c3babe97..7c0c2fab 100644 --- a/di/pqx/pqx.md +++ b/di/pqx/pqx.md @@ -26,10 +26,15 @@ oversized instrument across multiple files where required. A manifest recording |---|---|---|---| | logger | `` `log `` | yes | dict with `info`, `warn`, and `error`, each binary `{[c;m]}` where `c` is a symbol context and `m` is a string | -**Hard dependencies:** `kx.arrow` and `kx.qmamba` — both are loaded automatically (via `use`) when -`di.pqx` is imported, before `pqx.q` itself is loaded. `extract` calls -`` .m.di.0pqx.arrow.pq.writeParquetFromTable `` to perform every write; `kx.qmamba` is loaded but not -currently called anywhere in `pqx.q`. +**Hard dependency:** `kx.arrow` — loaded automatically (via `use`) when `di.pqx` is imported, before +`pqx.q` itself is loaded. `extract` calls `` .m.di.0pqx.arrow.pq.writeParquetFromTable `` to perform +every write. + +`kx.arrow` must be resolvable on the process's module search path at that point. Where it's +installed as a conda package (e.g. under a `kx.qmamba`-managed root such as +`~/.kx/root/lib/q/mod`), that location needs to already be on `QPATH` — `di.pqx` does not load +`kx.qmamba` itself to arrange this. Confirm `kx.arrow` loads standalone (`` use`kx.arrow ``) in the +target environment before relying on `di.pqx` there. The `log` dependency must be passed to `init` inside a dict keyed on `` `log ``. `init` throws immediately if `log` is absent, is not a dict, or is missing any of `info`/`warn`/`error`. The value @@ -51,8 +56,8 @@ pqx.init[enlist[`log]!enlist logdep] ## Options -Passed as the `o` dictionary to `extract`, merged over the module's own `default` dict. Any keys -omitted from `o` fall back to the default shown below. +Passed as the `o` dictionary to `extract`, merged over the module's own `default` dict (inspectable +via `getdefault[]`). Any keys omitted from `o` fall back to the default shown below. | Key | Default | Type | Description | |---|---|---|---| @@ -81,11 +86,7 @@ is not a column of the input table — this check runs unconditionally, regardle ## Manifest Schema The module's `manifest` table accumulates one row per file written across all `extract` calls. -`extract` does not return this table (or anything else). `` pqx:use`di.pqx `` only captures a -one-time snapshot of `manifest`/`default` at load time — `pqx.manifest` and `pqx.default` do **not** -track later writes or updates. Query the live state via `` .m.di.0pqx.manifest `` instead (the -private namespace every kdb-x module loads into; see `di.depcheck`'s docs for the `` `.m.di.0 `` -convention). +`extract` does not return this table (or anything else) — call `getmanifest[]` to read it. | Column | Type | Description | |---|---|---| @@ -124,10 +125,12 @@ pqx.init[enlist[`log]!enlist logdep] |---|---| | `init[deps]` | Wire the injected `log` dependency. Call once before the first `extract`. | | `extract[t;tname;dt;o]` | Write a table out to one or more parquet files, appending one row per file to the module's `manifest`. Returns nothing. | +| `getdefault[]` | Return the current `default` options dict. | +| `getmanifest[]` | Return the manifest accumulated so far across all `extract` calls. | -The remaining exports — `checkandconvertcols`, `estimate`, `calibrateratio`, `calcsize`, `plan`, -`datalookup`, `datalookuponesym`, `writefile`, `tryfn` — are internal pipeline steps of `extract`, -exposed only so `k4unit` can exercise them directly. Call `extract` for normal use. +The remaining exports — `checkandconvertcols`, `estimate`, `plan`, `writefile`, `tryfn` — are +internal pipeline steps of `extract`, exposed only so `k4unit` can exercise them directly. Call +`extract` for normal use. ### `init[deps]` Validate the required `log` dependency and store it for use by every other function. @@ -171,9 +174,8 @@ pqx.init[logger.logdict] // Write `trade` for 2025.07.15, overriding the target file size and codec pqx.extract[trade;`trade;2025.07.15;`targetsize`codec!(256*1024*1024;`gzip)] -// extract has no return value - inspect the accumulated manifest directly. -// pqx.manifest is a frozen snapshot from load time - query the live namespace instead: -.m.di.0pqx.manifest +// extract has no return value - inspect the accumulated manifest via the getter +pqx.getmanifest[] file seq syms nsyms rows mintime maxtime estbytes bytes split status -------------------------------------------------------------------------------------------------------------------------------------------------------------- @@ -191,18 +193,15 @@ k4unit.moduletest`di.pqx `test.csv` drives `extract` across default and overridden options — presort on/off, an oversized instrument with `splitoversized` on and off, a `symcol` override, parallel (`peach`) writes, and a -custom `filestub`/non-default codec — then asserts on the resulting `` .m.di.0pqx.manifest `` rows -and (via `` .m.di.0pqx.arrow.pq.readParquetToTable ``) the files written back to disk. It also -covers the failure paths: a zero-row table, a table missing `symcol`/`timecol`, and an invalid -codec (see Notes). +custom `filestub`/non-default codec — then asserts on the resulting `getmanifest[]` rows and (via +`` .m.di.0pqx.arrow.pq.readParquetToTable ``) the files written back to disk. It also covers the +failure paths: a zero-row table, a table missing `symcol`/`timecol`, and an invalid codec (see +Notes). --- ## Notes -- `pqx.manifest` and `pqx.default` (the values returned by `` pqx:use`di.pqx ``) are snapshots taken - once at load time — they never reflect later writes or changes. Always read `` .m.di.0pqx.manifest `` - for the live manifest. - `rowgroupbytes`, `complevel`, and `dictcols` are accepted in `default` and any `o` override, but nothing in the current write path reads them — only `` `PARQUET_VERSION `` (fixed at `` `V2.LATEST ``) and `` `COMPRESSION `` (from `codec`) are passed to the writer. diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index ca259c39..803fa376 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -36,20 +36,20 @@ manifest:([] status :`symbol$() / `ok | `error ); -/ takes a table and checks whether any symbol or char columns exist in it -/ if so, converts these to strings, as no Parquet datatype equivalent checkandconvertcols:{[t] + / takes a table and checks whether any symbol or char columns exist in it + / if so, converts these to strings, as no Parquet datatype equivalent :$[count c:exec c from meta[t] where t in "Ssc"; ![t;();0b;c!{(string;x)} each c]; t ] }; -/ for a partition of data, estimates the size of the tables to be saved to disk -/ if calibrate flag is true in o, a test write is carried out -/ returns a table of storage stats for all isntruments and the compression ratio, which may have changed depending on calibration estimate:{[t;o;writeopt] - cnts:`rowcnt xasc 0!?[t;();enlist[o[`symcol]]!enlist[o[`symcol]];enlist[`rowcnt]!enlist(count;`time)]; //select rowcnt:count time by sym from t; + / for a partition of data, estimates the size of the tables to be saved to disk + / if calibrate flag is true in o, a test write is carried out + / returns a table of storage stats for all isntruments and the compression ratio, which may have changed depending on calibration + cnts:`rowcnt xasc 0!?[t;();enlist[o[`symcol]]!enlist[o[`symcol]];enlist[`rowcnt]!enlist(count;o[`timecol])]; /select rowcnt:count time by sym from t medsym:cnts @ first where abs[cnt-med[cnt]]=min[abs[cnt-med[cnt:cnts`rowcnt]]]; bytesperrow:%[-22!t:.z.m.checkandconvertcols t[where t[o[`symcol]]=medsym[o[`symcol]]];medsym`rowcnt]; @@ -63,9 +63,10 @@ estimate:{[t;o;writeopt] :(update estbyt:rowcnt*bytesperrow*o[`compressionratio] from cnts;o[`compressionratio]) }; -/ writes a sample of data to disk and reads its size on disk -/ calcaultes the compression ratio and returns if a new ratio was successfully calculated, otherwise old ratio is maintained calibrateratio:{[t;o;writeopt] + / writes a sample of data to disk and reads its size on disk + / calcaultes the compression ratio and returns if a new ratio was successfully calculated, otherwise old ratio is maintained + / remove leading : from outdir testloc:$[":" ~ first string[o`outdir]; 1_string[o`outdir],"/testWrite.parquet"; @@ -96,17 +97,18 @@ calibrateratio:{[t;o;writeopt] :newratio }; -/ find the estimated size in bytes for each instrument per file to be saved down -/ in the case of a larger instrument being split, return count[seqno] number of instances of estbytes calcsize:{[tbl;symcol;syms;seqno] + / find the estimated size in bytes for each instrument per file to be saved down + / in the case of a larger instrument being split, return count[seqno] number of instances of estbytes .z.m.loginfo[`pqx;"Getting estimated bytes for planned files"]; :"j"$count[seqno]#%[sum[?[tbl;enlist(in;symcol;`syms);0b;()]`estbyt];count seqno] }; -/ planning function to bucket instruments based on next-fit packing -/ if an instrument can be added to a bucket without that bucket exceeding the target size, it will be added to that bucket, else a new bucket is started -/ large instruments are also split into multiple files if splitoversized flag is true plan:{[t;o;maxsize] + / planning function to bucket instruments based on next-fit packing + / if an instrument can be added to a bucket without that bucket exceeding the target size, it will be added to that bucket + / else a new bucket is created + / large instruments are also split into multiple files if splitoversized flag is true symstats:t; plans:(); @@ -142,23 +144,23 @@ plan:{[t;o;maxsize] :update estbytes:.z.m.calcsize[symstats;o`symcol;;]'[syms;seqno] from {`syms`seqno!/: flip (key[x];value[x])} group plans }; -/ get lists of indices by file -/ a pass with multiple instruments is assumed to be one file only, hence the return is flattened into one list datalookup:{[t;symcol;syms;cnt] + / get lists of indices by file + / a pass with multiple instruments is assumed to be one file only, hence the return is flattened into one list $[1type deps; + '"di.pqx: deps must be a dict with a `log key"]; + if[not `log in key deps; + '"di.pqx: log dependency is required; pass `info`warn`error functions keyed on `log"]; + if[99h<>type deps`log; + '"di.pqx: log value must be a dict of `info`warn`error functions"]; + if[not all (`info`warn`error) in key deps`log; + '"di.pqx: log dict must have `info`warn`error keys; got: ",(", " sv string key deps`log)]; + .z.m.loginfo:deps[`log]`info; + .z.m.logwarn:deps[`log]`warn; + .z.m.logerr:deps[`log]`error; }; \ No newline at end of file diff --git a/di/pqx/test.csv b/di/pqx/test.csv index b5a79592..7c811908 100644 --- a/di/pqx/test.csv +++ b/di/pqx/test.csv @@ -11,50 +11,50 @@ before,0,0,q,pqxnotime:([]sym:5#`AAPL;price:1.0 2.0 3.0 4.0 5.0),1,,Table missin before,0,0,q,pqxaltsym:([]sym:`B`A`B`A`A;alt:`X`Y`X`Y`Y;time:2025.07.15D00:00:00.000000005 2025.07.15D00:00:00.000000004 2025.07.15D00:00:00.000000003 2025.07.15D00:00:00.000000002 2025.07.15D00:00:00.000000001;price:1.0 2.0 3.0 4.0 5.0),1,,Table with a second candidate instrument column to probe the symcol option before,0,0,q,pqxsingle:([]sym:enlist`AAPL;time:enlist 2025.07.15D09:30:00.000000000;price:enlist 123.45),1,,Single-row table edge case -true,0,0,q,98h~type .m.di.0pqx.manifest,1,,Manifest is a table before anything is written +true,0,0,q,98h~type pqx.getmanifest[],1,,Manifest is a table before anything is written run,0,0,q,pqx.extract[pqxbasic;`pqxtrade;2025.07.15;enlist[`outdir]!enlist `:pqxout1/],1,,Extract a multi-sym table with default options -true,0,0,q,9~exec sum rows from .m.di.0pqx.manifest where file like "*pqxout1*",1,,All input rows are accounted for across written files -true,0,0,q,3~exec sum nsyms from .m.di.0pqx.manifest where file like "*pqxout1*",1,,All three instruments are accounted for -true,0,0,q,all `ok=exec status from .m.di.0pqx.manifest where file like "*pqxout1*",1,,Every written file reports ok status -true,0,0,q,all 0 Date: Tue, 18 Aug 2026 12:28:45 +0100 Subject: [PATCH 03/22] Fix typos --- di/pqx/pqx.md | 12 ++++++------ di/pqx/pqx.q | 14 +++++++------- di/pqx/test.csv | 4 +++- 3 files changed, 16 insertions(+), 14 deletions(-) diff --git a/di/pqx/pqx.md b/di/pqx/pqx.md index 7c0c2fab..391f8d60 100644 --- a/di/pqx/pqx.md +++ b/di/pqx/pqx.md @@ -194,9 +194,9 @@ k4unit.moduletest`di.pqx `test.csv` drives `extract` across default and overridden options — presort on/off, an oversized instrument with `splitoversized` on and off, a `symcol` override, parallel (`peach`) writes, and a custom `filestub`/non-default codec — then asserts on the resulting `getmanifest[]` rows and (via -`` .m.di.0pqx.arrow.pq.readParquetToTable ``) the files written back to disk. It also covers the -failure paths: a zero-row table, a table missing `symcol`/`timecol`, and an invalid codec (see -Notes). +`` .m.di.0pqx.arrow.pq.readParquetToTable ``) the files written back to disk. It also covers a +zero-row table and a table missing `symcol`/`timecol` (both fail outright), and an invalid `codec` +(degrades gracefully — see Manifest Schema's `status` column). --- @@ -207,6 +207,6 @@ Notes). `` `V2.LATEST ``) and `` `COMPRESSION `` (from `codec`) are passed to the writer. - `symcol`/`timecol` presence is validated unconditionally on every `extract` call, even when `presort` is `0b`. -- A per-file write failure (e.g. an invalid `codec`) is caught and logged, but currently still - causes `extract` to throw rather than recording that file with `` status=`error `` in the manifest - as the schema implies it should — see `test.csv`'s invalid-codec case. +- A per-file write failure (e.g. an invalid `codec`) is caught and logged at `warn`, and that file + is recorded with `` status=`err `` (and `bytes:0`) in the manifest — it does not abort the rest of + `extract`. diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index 803fa376..1c12dd19 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -49,7 +49,7 @@ estimate:{[t;o;writeopt] / for a partition of data, estimates the size of the tables to be saved to disk / if calibrate flag is true in o, a test write is carried out / returns a table of storage stats for all isntruments and the compression ratio, which may have changed depending on calibration - cnts:`rowcnt xasc 0!?[t;();enlist[o[`symcol]]!enlist[o[`symcol]];enlist[`rowcnt]!enlist(count;o[`timecol])]; /select rowcnt:count time by sym from t + cnts:`rowcnt xasc 0!?[t;();enlist[o[`symcol]]!enlist[o[`symcol]];enlist[`rowcnt]!enlist(count;o[`timecol])]; / select rowcnt:count time by sym from t medsym:cnts @ first where abs[cnt-med[cnt]]=min[abs[cnt-med[cnt:cnts`rowcnt]]]; bytesperrow:%[-22!t:.z.m.checkandconvertcols t[where t[o[`symcol]]=medsym[o[`symcol]]];medsym`rowcnt]; @@ -175,7 +175,7 @@ writefile:{[t;o;writeopt;writedir;map] res:.z.m.tryfn[`.m.di.0pqx.arrow.pq.writeParquetFromTable;(path;.z.m.checkandconvertcols t[i];writeopt)]; $[first res; .z.m.loginfo[`pqx;"seqNo ",string[seqno]," write successful"]; - .z.m.logwarn[`pqx;"seqNo ",string[seqno]," write unsuccessful. Error - ",last res]; + .z.m.logwarn[`pqx;"seqNo ",string[seqno]," write unsuccessful. Error - ",last res] ]; :`file`seq`syms`nsyms`rows`mintime`maxtime`estbytes`bytes`split`status!/: flip ( hsym `$path; @@ -188,7 +188,7 @@ writefile:{[t;o;writeopt;writedir;map] estbytes; @[hcount;hsym `$path;0]; split; - `err`ok[first res] + `error`ok[first res] ) }[t;symcol;writeopt;1 Date: Tue, 18 Aug 2026 13:55:01 +0100 Subject: [PATCH 04/22] Fix error in documentation --- di/pqx/pqx.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/di/pqx/pqx.md b/di/pqx/pqx.md index 391f8d60..f9e54986 100644 --- a/di/pqx/pqx.md +++ b/di/pqx/pqx.md @@ -208,5 +208,5 @@ zero-row table and a table missing `symcol`/`timecol` (both fail outright), and - `symcol`/`timecol` presence is validated unconditionally on every `extract` call, even when `presort` is `0b`. - A per-file write failure (e.g. an invalid `codec`) is caught and logged at `warn`, and that file - is recorded with `` status=`err `` (and `bytes:0`) in the manifest — it does not abort the rest of + is recorded with `` status=`error `` (and `bytes:0`) in the manifest — it does not abort the rest of `extract`. From a0d121bb36986d9b38ef6a0e4c725362d2ef3cce Mon Sep 17 00:00:00 2001 From: ConorSwainDI Date: Wed, 19 Aug 2026 07:41:24 +0100 Subject: [PATCH 05/22] Modify comment --- di/pqx/pqx.q | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index 1c12dd19..897fe0b7 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -49,7 +49,7 @@ estimate:{[t;o;writeopt] / for a partition of data, estimates the size of the tables to be saved to disk / if calibrate flag is true in o, a test write is carried out / returns a table of storage stats for all isntruments and the compression ratio, which may have changed depending on calibration - cnts:`rowcnt xasc 0!?[t;();enlist[o[`symcol]]!enlist[o[`symcol]];enlist[`rowcnt]!enlist(count;o[`timecol])]; / select rowcnt:count time by sym from t + cnts:`rowcnt xasc 0!?[t;();enlist[o[`symcol]]!enlist[o[`symcol]];enlist[`rowcnt]!enlist(count;o[`timecol])]; / select rowcnt:count time by sym from t, using appropriate substitutions for time and sym cols medsym:cnts @ first where abs[cnt-med[cnt]]=min[abs[cnt-med[cnt:cnts`rowcnt]]]; bytesperrow:%[-22!t:.z.m.checkandconvertcols t[where t[o[`symcol]]=medsym[o[`symcol]]];medsym`rowcnt]; From 3b38b5cfadefbf6a79b92b6435320de2080e898e Mon Sep 17 00:00:00 2001 From: ConorSwainDI Date: Wed, 19 Aug 2026 11:17:09 +0100 Subject: [PATCH 06/22] Enlist syms in calcsize; Reference timecol opt in manifest records --- di/pqx/pqx.q | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index 897fe0b7..5f0acf48 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -101,7 +101,7 @@ calcsize:{[tbl;symcol;syms;seqno] / find the estimated size in bytes for each instrument per file to be saved down / in the case of a larger instrument being split, return count[seqno] number of instances of estbytes .z.m.loginfo[`pqx;"Getting estimated bytes for planned files"]; - :"j"$count[seqno]#%[sum[?[tbl;enlist(in;symcol;`syms);0b;()]`estbyt];count seqno] + :"j"$count[seqno]#%[sum[?[tbl;enlist(in;symcol;enlist syms);0b;()]`estbyt];count seqno] }; plan:{[t;o;maxsize] @@ -164,13 +164,12 @@ writefile:{[t;o;writeopt;writedir;map] syms:map[`syms]; seqno:map[`seqno]; estbytes:map[`estbytes]; - symcol:o[`symcol]; / build paths for each seqno paths:writedir,/:o[`filestub],/:"-",/:("0"^-5$string[seqno]),\:".parquet"; / data to write, iterated by file - res:raze {[t;symcol;writeopt;split;path;seqno;estbytes;i] + res:raze {[t;o;writeopt;split;path;seqno;estbytes;i] .z.m.loginfo[`pqx;"Writing seqNo ",string[seqno],", path - ",path]; res:.z.m.tryfn[`.m.di.0pqx.arrow.pq.writeParquetFromTable;(path;.z.m.checkandconvertcols t[i];writeopt)]; $[first res; @@ -180,17 +179,17 @@ writefile:{[t;o;writeopt;writedir;map] :`file`seq`syms`nsyms`rows`mintime`maxtime`estbytes`bytes`split`status!/: flip ( hsym `$path; seqno; - enlist distinct[t[i][symcol]]; - count[distinct[t[i][symcol]]]; + enlist distinct[t[i][o`symcol]]; + count[distinct[t[i][o`symcol]]]; count[i]; - first exec min time from t[i]; - first exec max time from t[i]; + first ?[t[i];();();(min;o`timecol)]; + first ?[t[i];();();(max;o`timecol)]; estbytes; @[hcount;hsym `$path;0]; split; `error`ok[first res] ) - }[t;symcol;writeopt;1 Date: Wed, 19 Aug 2026 12:17:55 +0100 Subject: [PATCH 07/22] Changes based off review feedback --- di/pqx/pqx.md | 9 ++++++--- di/pqx/pqx.q | 10 ++++++++-- di/pqx/test.csv | 11 +++++++++-- 3 files changed, 23 insertions(+), 7 deletions(-) diff --git a/di/pqx/pqx.md b/di/pqx/pqx.md index f9e54986..6c446afa 100644 --- a/di/pqx/pqx.md +++ b/di/pqx/pqx.md @@ -78,8 +78,10 @@ via `getdefault[]`). Any keys omitted from `o` fall back to the default shown be | `filestub` | `"part"` | string | File name stub; files are written as `-NNNNN.parquet` | Output files are written to `//date=
/-NNNNN.parquet`. `extract` throws -(`` `di.pqx: no symcol found `` / `` `di.pqx: no timecol found ``) if the merged `symcol`/`timecol` -is not a column of the input table — this check runs unconditionally, regardless of `presort`. +(`` `di.pqx: no symcol found in table `` / `` `di.pqx: no timecol found in table ``) if the merged +`symcol`/`timecol` is not a column of the input table — this check runs unconditionally, regardless +of `presort`. It also throws (`` `di.pqx: cannot extract from empty table ``) if `t` has zero rows, +regardless of `calibrate` — this check runs first, before any other validation. --- @@ -206,7 +208,8 @@ zero-row table and a table missing `symcol`/`timecol` (both fail outright), and nothing in the current write path reads them — only `` `PARQUET_VERSION `` (fixed at `` `V2.LATEST ``) and `` `COMPRESSION `` (from `codec`) are passed to the writer. - `symcol`/`timecol` presence is validated unconditionally on every `extract` call, even when - `presort` is `0b`. + `presort` is `0b`. A zero-row input table is rejected outright, before that check, regardless of + `calibrate`. - A per-file write failure (e.g. an invalid `codec`) is caught and logged at `warn`, and that file is recorded with `` status=`error `` (and `bytes:0`) in the manifest — it does not abort the rest of `extract`. diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index 5f0acf48..6c93a403 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -48,7 +48,7 @@ checkandconvertcols:{[t] estimate:{[t;o;writeopt] / for a partition of data, estimates the size of the tables to be saved to disk / if calibrate flag is true in o, a test write is carried out - / returns a table of storage stats for all isntruments and the compression ratio, which may have changed depending on calibration + / returns a table of storage stats for all instruments and the compression ratio, which may have changed depending on calibration cnts:`rowcnt xasc 0!?[t;();enlist[o[`symcol]]!enlist[o[`symcol]];enlist[`rowcnt]!enlist(count;o[`timecol])]; / select rowcnt:count time by sym from t, using appropriate substitutions for time and sym cols medsym:cnts @ first where abs[cnt-med[cnt]]=min[abs[cnt-med[cnt:cnts`rowcnt]]]; bytesperrow:%[-22!t:.z.m.checkandconvertcols t[where t[o[`symcol]]=medsym[o[`symcol]]];medsym`rowcnt]; @@ -65,7 +65,7 @@ estimate:{[t;o;writeopt] calibrateratio:{[t;o;writeopt] / writes a sample of data to disk and reads its size on disk - / calcaultes the compression ratio and returns if a new ratio was successfully calculated, otherwise old ratio is maintained + / calculates the compression ratio and returns if a new ratio was successfully calculated, otherwise old ratio is maintained / remove leading : from outdir testloc:$[":" ~ first string[o`outdir]; @@ -206,6 +206,12 @@ extract:{[t;tname;dt;o] / override default opts with o where applicable opts:default,o; + / check for count in tables, error out if not + if[not count[t]; + .z.m.logerr[`pqx;err:"di.pqx: cannot extract from empty table"]; + 'err + ]; + / check for instrument and time cols, error out if not if[not opts[`symcol] in cols[t]; .z.m.logerr[`pqx;err:"di.pqx: no symcol found in table"]; diff --git a/di/pqx/test.csv b/di/pqx/test.csv index 5f514eed..d4cf93e2 100644 --- a/di/pqx/test.csv +++ b/di/pqx/test.csv @@ -12,6 +12,12 @@ before,0,0,q,pqxaltsym:([]sym:`B`A`B`A`A;alt:`X`Y`X`Y`Y;time:2025.07.15D00:00:00 before,0,0,q,pqxsingle:([]sym:enlist`AAPL;time:enlist 2025.07.15D09:30:00.000000000;price:enlist 123.45),1,,Single-row table edge case true,0,0,q,98h~type pqx.getmanifest[],1,,Manifest is a table before anything is written +true,0,0,q,(`targetsize`maxfactor`splitoversized`calibrate`compressionratio`symcol`timecol`presort`rowgroupbytes`codec`complevel`dictcols`parallel`outdir`filestub!(512*1024*1024;1.5;1b;1b;0.30;`sym;`time;1b;128*1024*1024;`zstd;3;`sym`exchange;0b;`:.;"part"))~pqx.getdefault[],1,,getdefault returns the documented default options dict + +fail,0,0,q,pqx.init[()],1,,init fails when deps is not a dict +fail,0,0,q,pqx.init[enlist[`nolog]!enlist logdep],1,,init fails when deps has no log key +fail,0,0,q,pqx.init[enlist[`log]!enlist 5],1,,init fails when the log value is not a dict +fail,0,0,q,pqx.init[enlist[`log]!enlist (enlist`info)!enlist {[c;m] }],1,,init fails when the log dict is missing required warn/error keys run,0,0,q,pqx.extract[pqxbasic;`pqxtrade;2025.07.15;enlist[`outdir]!enlist `:pqxout1/],1,,Extract a multi-sym table with default options true,0,0,q,9~exec sum rows from pqx.getmanifest[] where file like "*pqxout1*",1,,All input rows are accounted for across written files @@ -56,7 +62,8 @@ run,0,0,q,pqx.extract[pqxsingle;`pqxtrade;2025.07.24;enlist[`outdir]!enlist `:pq true,0,0,q,1~exec first rows from pqx.getmanifest[] where file like "*pqxout9*",1,,Single-row table produces a single-row file true,0,0,q,1~count select from pqx.getmanifest[] where file like "*pqxout9*",1,,Single-row table produces exactly one file -fail,0,0,q,pqx.extract[pqxempty;`pqxtrade;2025.07.21;enlist[`outdir]!enlist `:pqxoutf1/],1,,Extract fails on a zero-row table - calibration cannot infer a schema from no rows +fail,0,0,q,pqx.extract[pqxempty;`pqxtrade;2025.07.21;enlist[`outdir]!enlist `:pqxoutf1/],1,,Extract fails on a zero-row table - explicit empty-table guard rejects it outright +fail,0,0,q,pqx.extract[pqxempty;`pqxtrade;2025.07.21;`outdir`calibrate`compressionratio!(`:pqxoutf1b/;0b;0.3)],1,,Extract fails on a zero-row table even with calibrate off - the empty-table check is unconditional fail,0,0,q,pqx.extract[pqxnotime;`pqxtrade;2025.07.22;enlist[`outdir]!enlist `:pqxoutf2/],1,,Extract fails on a missing time column with presort on (the default) fail,0,0,q,pqx.extract[pqxnotime;`pqxtrade;2025.07.22;`outdir`presort!(`:pqxoutf3/;0b)],1,,Extract fails on a missing time column even with presort off - the column check is unconditional fail,0,0,q,pqx.extract[([]price:1.0 2.0 3.0);`pqxtrade;2025.07.23;enlist[`outdir]!enlist `:pqxoutf4/],1,,Extract fails when the table has no sym column at all @@ -64,4 +71,4 @@ run,0,0,q,pqx.extract[pqxbasic;`pqxtrade;2025.07.25;`outdir`codec`calibrate`comp true,0,0,q,`error~first exec status from pqx.getmanifest[] where file like "*pqxoutf5*",1,,The failed file is recorded with status=`error instead of aborting extract true,0,0,q,0~first exec bytes from pqx.getmanifest[] where file like "*pqxoutf5*",1,,The failed file has zero on-disk bytes recorded -after,0,0,q,system "rm -rf pqxout1 pqxout2 pqxout3 pqxout4 pqxout5 pqxout6 pqxout7 pqxout8 pqxout9 pqxoutf1 pqxoutf2 pqxoutf3 pqxoutf4 pqxoutf5",1,,Remove parquet output directories written by tests +after,0,0,q,system "rm -rf pqxout1 pqxout2 pqxout3 pqxout4 pqxout5 pqxout6 pqxout7 pqxout8 pqxout9 pqxoutf1 pqxoutf1b pqxoutf2 pqxoutf3 pqxoutf4 pqxoutf5",1,,Remove parquet output directories written by tests From dd5dc01f78cd310595d833af1981cf40dac6bcb2 Mon Sep 17 00:00:00 2001 From: ConorSwainDI Date: Wed, 19 Aug 2026 14:47:25 +0100 Subject: [PATCH 08/22] Change to directory/file delete/create behaviour --- di/pqx/pqx.q | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index 6c93a403..356948e8 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -90,7 +90,7 @@ calibrateratio:{[t;o;writeopt] / clean test file .z.m.loginfo[`pqx;"Cleaning up test file"]; - system "rm ",testloc; + hdel hsym `$testloc; / return new ratio .z.m.loginfo[`pqx;"Returning calibrated compression ratio"]; @@ -182,8 +182,8 @@ writefile:{[t;o;writeopt;writedir;map] enlist distinct[t[i][o`symcol]]; count[distinct[t[i][o`symcol]]]; count[i]; - first ?[t[i];();();(min;o`timecol)]; - first ?[t[i];();();(max;o`timecol)]; + ?[t[i];();();(min;o`timecol)]; + ?[t[i];();();(max;o`timecol)]; estbytes; @[hcount;hsym `$path;0]; split; @@ -235,7 +235,7 @@ extract:{[t;tname;dt;o] / build dir locations - done before estimate/calibrate, as calibration's trial write lands under outdir if[not count key hsym `$writedir:1_string[opts[`outdir]],string[tname],"/date=",string[dt],"/"; - system "mkdir -p ", writedir; + system "mkdir -p \"", writedir,"\""; ]; / get symstats and new compression ratio From 1588e9321ac0f905fd8339377cc1468f50ef7e0e Mon Sep 17 00:00:00 2001 From: ConorSwainDI Date: Wed, 19 Aug 2026 14:51:12 +0100 Subject: [PATCH 09/22] Newline to end file --- di/pqx/pqx.q | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index 356948e8..fd945190 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -285,4 +285,4 @@ init:{[deps] .z.m.loginfo:deps[`log]`info; .z.m.logwarn:deps[`log]`warn; .z.m.logerr:deps[`log]`error; - }; \ No newline at end of file + }; From 5193417faa23ac1064394dd1068c491064aa0215 Mon Sep 17 00:00:00 2001 From: ConorSwainDI Date: Wed, 19 Aug 2026 16:07:16 +0100 Subject: [PATCH 10/22] Remove unnecessary getdefault --- di/pqx/init.q | 2 +- di/pqx/pqx.md | 5 ++--- di/pqx/pqx.q | 5 ----- di/pqx/test.csv | 1 - 4 files changed, 3 insertions(+), 10 deletions(-) diff --git a/di/pqx/init.q b/di/pqx/init.q index 6346b748..ace5fd8b 100644 --- a/di/pqx/init.q +++ b/di/pqx/init.q @@ -4,4 +4,4 @@ arrow:use`kx.arrow \l ::pqx.q -export:([init;extract;getdefault;getmanifest;checkandconvertcols;estimate;plan;writefile;tryfn]) +export:([init;extract;getmanifest;checkandconvertcols;estimate;plan;writefile;tryfn]) diff --git a/di/pqx/pqx.md b/di/pqx/pqx.md index 6c446afa..b8282fd5 100644 --- a/di/pqx/pqx.md +++ b/di/pqx/pqx.md @@ -56,8 +56,8 @@ pqx.init[enlist[`log]!enlist logdep] ## Options -Passed as the `o` dictionary to `extract`, merged over the module's own `default` dict (inspectable -via `getdefault[]`). Any keys omitted from `o` fall back to the default shown below. +Passed as the `o` dictionary to `extract`, merged over the module's own `default` dict. Any keys +omitted from `o` fall back to the default shown below. | Key | Default | Type | Description | |---|---|---|---| @@ -127,7 +127,6 @@ pqx.init[enlist[`log]!enlist logdep] |---|---| | `init[deps]` | Wire the injected `log` dependency. Call once before the first `extract`. | | `extract[t;tname;dt;o]` | Write a table out to one or more parquet files, appending one row per file to the module's `manifest`. Returns nothing. | -| `getdefault[]` | Return the current `default` options dict. | | `getmanifest[]` | Return the manifest accumulated so far across all `extract` calls. | The remaining exports — `checkandconvertcols`, `estimate`, `plan`, `writefile`, `tryfn` — are diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index fd945190..da3df17f 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -260,11 +260,6 @@ extract:{[t;tname;dt;o] manifest,:res }; -getdefault:{[] - / return the current default options dict - :default; - }; - getmanifest:{[] / return the manifest accumulated so far across all extract calls :manifest; diff --git a/di/pqx/test.csv b/di/pqx/test.csv index d4cf93e2..92db4a4a 100644 --- a/di/pqx/test.csv +++ b/di/pqx/test.csv @@ -12,7 +12,6 @@ before,0,0,q,pqxaltsym:([]sym:`B`A`B`A`A;alt:`X`Y`X`Y`Y;time:2025.07.15D00:00:00 before,0,0,q,pqxsingle:([]sym:enlist`AAPL;time:enlist 2025.07.15D09:30:00.000000000;price:enlist 123.45),1,,Single-row table edge case true,0,0,q,98h~type pqx.getmanifest[],1,,Manifest is a table before anything is written -true,0,0,q,(`targetsize`maxfactor`splitoversized`calibrate`compressionratio`symcol`timecol`presort`rowgroupbytes`codec`complevel`dictcols`parallel`outdir`filestub!(512*1024*1024;1.5;1b;1b;0.30;`sym;`time;1b;128*1024*1024;`zstd;3;`sym`exchange;0b;`:.;"part"))~pqx.getdefault[],1,,getdefault returns the documented default options dict fail,0,0,q,pqx.init[()],1,,init fails when deps is not a dict fail,0,0,q,pqx.init[enlist[`nolog]!enlist logdep],1,,init fails when deps has no log key From 651e13e2e854c5a3e271aac37abae3a945e972db Mon Sep 17 00:00:00 2001 From: ConorSwainDI Date: Thu, 20 Aug 2026 11:43:05 +0100 Subject: [PATCH 11/22] Return manifest entries from extract --- di/pqx/pqx.q | 4 +++- di/pqx/test.csv | 10 +++++++++- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index da3df17f..99716707 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -257,7 +257,9 @@ extract:{[t;tname;dt;o] res:raze parallelfn [.z.m.writefile[t;opts;writeopt;writedir]; plans]; .Q.gc[]; - manifest,:res + / attach to global manifest and return stats for this extract + manifest,:res; + :res }; getmanifest:{[] diff --git a/di/pqx/test.csv b/di/pqx/test.csv index 92db4a4a..abf5a678 100644 --- a/di/pqx/test.csv +++ b/di/pqx/test.csv @@ -26,6 +26,14 @@ true,0,0,q,all 00,1,,Sanity check - manifest already held rows from an earlier extract before this call +true,0,0,q,not pqxret1~pqx.getmanifest[],1,,extract's return value is scoped to this call, not the entire accumulated manifest +true,0,0,q,(count pqxret1) Date: Thu, 20 Aug 2026 11:47:41 +0100 Subject: [PATCH 12/22] Changes in docs to reflect new behavhiour --- di/pqx/pqx.md | 25 +++++++++++++++---------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/di/pqx/pqx.md b/di/pqx/pqx.md index b8282fd5..7bc6a627 100644 --- a/di/pqx/pqx.md +++ b/di/pqx/pqx.md @@ -4,7 +4,7 @@ Converts an in-memory kdb+ table into one or more `.parquet` files via `kx.arrow by instrument and packed into files close to a configurable target size, splitting any single oversized instrument across multiple files where required. A manifest recording what was written (file, instruments, row count, time range, on-disk size) is accumulated in the module's private -`manifest` table; `extract` itself has no return value. +`manifest` table; `extract` also returns this same information for the files it just wrote. --- @@ -87,8 +87,9 @@ regardless of `calibrate` — this check runs first, before any other validation ## Manifest Schema -The module's `manifest` table accumulates one row per file written across all `extract` calls. -`extract` does not return this table (or anything else) — call `getmanifest[]` to read it. +The module's `manifest` table accumulates one row per file written across all `extract` calls; call +`getmanifest[]` to read the full accumulated table. `extract` itself returns a table of the same +shape, scoped to only the file(s) written by that call. | Column | Type | Description | |---|---|---| @@ -126,7 +127,7 @@ pqx.init[enlist[`log]!enlist logdep] | Function | Description | |---|---| | `init[deps]` | Wire the injected `log` dependency. Call once before the first `extract`. | -| `extract[t;tname;dt;o]` | Write a table out to one or more parquet files, appending one row per file to the module's `manifest`. Returns nothing. | +| `extract[t;tname;dt;o]` | Write a table out to one or more parquet files, appending one row per file to the module's `manifest`. Returns that same per-file stats table, scoped to this call. | | `getmanifest[]` | Return the manifest accumulated so far across all `extract` calls. | The remaining exports — `checkandconvertcols`, `estimate`, `plan`, `writefile`, `tryfn` — are @@ -145,9 +146,10 @@ required key. ### `extract[t;tname;dt;o]` Write table `t` out to one or more parquet files under `//date=
/`, appending one -row per file written to the module's `manifest`. `o` is merged over `default` (see Options). The -output directory is created before the size-estimation/calibration step, so a fresh `outdir` works -with the default `calibrate:1b`. +row per file written to the module's `manifest` and returning that same set of rows (see Manifest +Schema) scoped to this call only — it does not include rows from any earlier `extract` call. `o` is +merged over `default` (see Options). The output directory is created before the +size-estimation/calibration step, so a fresh `outdir` works with the default `calibrate:1b`. | Parameter | Type | Description | |---|---|---| @@ -173,14 +175,17 @@ logger:use`di.log pqx.init[logger.logdict] // Write `trade` for 2025.07.15, overriding the target file size and codec -pqx.extract[trade;`trade;2025.07.15;`targetsize`codec!(256*1024*1024;`gzip)] +res:pqx.extract[trade;`trade;2025.07.15;`targetsize`codec!(256*1024*1024;`gzip)] -// extract has no return value - inspect the accumulated manifest via the getter -pqx.getmanifest[] +// res holds only the row(s) written by this call +res file seq syms nsyms rows mintime maxtime estbytes bytes split status -------------------------------------------------------------------------------------------------------------------------------------------------------------- :./trade/date=2025.07.15/part-00001.parquet 1 `AAPL`MSFT 2 50000 2025.07.15D00:00:00.000000000 2025.07.15D23:59:59.000000000 1153433 1048576 0b ok + +// getmanifest[] returns the full accumulated table across every extract call so far +pqx.getmanifest[] ``` --- From 9229af53bba73dcb5f35d072adcadcadfc9e788a Mon Sep 17 00:00:00 2001 From: ConorSwainDI Date: Thu, 27 Aug 2026 15:08:28 +0100 Subject: [PATCH 13/22] Implement one sym per file functionality --- di/pqx/pqx.md | 25 ++++++++-- di/pqx/pqx.q | 126 ++++++++++++++++++++++++++++++------------------ di/pqx/test.csv | 31 +++++++++++- 3 files changed, 130 insertions(+), 52 deletions(-) diff --git a/di/pqx/pqx.md b/di/pqx/pqx.md index 7bc6a627..8d59e30a 100644 --- a/di/pqx/pqx.md +++ b/di/pqx/pqx.md @@ -65,6 +65,7 @@ omitted from `o` fall back to the default shown below. | `maxfactor` | `1.5` | float | Hard cap on file size, expressed as a multiple of `targetsize` | | `splitoversized` | `1b` | boolean | Split any single instrument larger than the cap across multiple files | | `calibrate` | `1b` | boolean | Run a trial write to measure the raw-to-parquet size ratio instead of using `compressionratio` | +| `onesymperfile` | `0b` | boolean | Write exactly one file per instrument, ignoring the target-size bucketing/packing logic. Forces `splitoversized` off (see below) | | `compressionratio` | `0.30` | float | Raw-to-parquet size ratio used for size estimation when `calibrate` is `0b` | | `symcol` | `` `sym `` | symbol | Instrument column | | `timecol` | `` `time `` | symbol | Time column | @@ -78,10 +79,17 @@ omitted from `o` fall back to the default shown below. | `filestub` | `"part"` | string | File name stub; files are written as `-NNNNN.parquet` | Output files are written to `//date=
/-NNNNN.parquet`. `extract` throws -(`` `di.pqx: no symcol found in table `` / `` `di.pqx: no timecol found in table ``) if the merged -`symcol`/`timecol` is not a column of the input table — this check runs unconditionally, regardless -of `presort`. It also throws (`` `di.pqx: cannot extract from empty table ``) if `t` has zero rows, -regardless of `calibrate` — this check runs first, before any other validation. +(`` `di.pqx: input keys not recognised - ... ``) if `o` contains any key not present in the module's +`default` dict — this is the very first check `extract` performs, before the empty-table check. It +also throws (`` `di.pqx: no symcol found in table `` / `` `di.pqx: no timecol found in table ``) if +the merged `symcol`/`timecol` is not a column of the input table — this check runs unconditionally, +regardless of `presort`. It also throws (`` `di.pqx: cannot extract from empty table ``) if `t` has +zero rows, regardless of `calibrate` — this check runs before the column checks but after the input +key check. + +When `onesymperfile` is `1b`, each instrument is written to its own file regardless of `targetsize` +bucketing, and `splitoversized` is forced to `0b` for that call (an oversized single instrument is +still written to one file, not split, even if `splitoversized:1b` is also passed in `o`). --- @@ -105,6 +113,15 @@ shape, scoped to only the file(s) written by that call. | `split` | boolean | `1b` if this file is a chunk of a split oversized instrument | | `status` | symbol | `` `ok `` or `` `error `` | +In addition to the in-memory `manifest`, each `extract` call writes this same per-call stats table +to a `manifest` sidecar file directly under the partition directory (i.e. +`//date=
/manifest`), serialized with `set`/readable back with `get`. A repeat +`extract` call into the same partition overwrites the sidecar with just that call's rows, rather +than accumulating across calls — the sidecar mirrors `extract`'s return value, not `getmanifest[]`. +Writing the sidecar is best-effort: if it fails (for example a permissions issue, or something else +already occupying that path) a warning is logged but `extract` still returns normally and still +updates the in-memory `manifest`. + --- ## Initialisation diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index 99716707..0e556af1 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -1,6 +1,6 @@ / define default config default:( - `targetsize`maxfactor`splitoversized`calibrate`compressionratio, + `targetsize`maxfactor`splitoversized`calibrate`onesymperfile`compressionratio, `symcol`timecol`presort`rowgroupbytes`codec`complevel`dictcols, `parallel`outdir`filestub )!( @@ -8,6 +8,7 @@ default:( 1.5; / hard cap = target * maxfactor 1b; / split instruments larger than the cap 1b; / run a calibration write to set the ratio + 0b; / enforce a single instrument per file 0.30; / raw->parquet ratio if not calibrating `sym; / instrument column `time; / time column @@ -98,50 +99,59 @@ calibrateratio:{[t;o;writeopt] }; calcsize:{[tbl;symcol;syms;seqno] - / find the estimated size in bytes for each instrument per file to be saved down - / in the case of a larger instrument being split, return count[seqno] number of instances of estbytes + / find the estimated size in bytes for file to be saved down by querying symstats .z.m.loginfo[`pqx;"Getting estimated bytes for planned files"]; - :"j"$count[seqno]#%[sum[?[tbl;enlist(in;symcol;enlist syms);0b;()]`estbyt];count seqno] + :sum[?[tbl;enlist(in;symcol;enlist syms);0b;()]`estbyt] }; plan:{[t;o;maxsize] / planning function to bucket instruments based on next-fit packing + / if one sym per file, just output individual buckets for each sym + / else / if an instrument can be added to a bucket without that bucket exceeding the target size, it will be added to that bucket / else a new bucket is created / large instruments are also split into multiple files if splitoversized flag is true symstats:t; plans:(); - / if split oversized is required, check against maxsize and return a plan entry for each required file - if[o`splitoversized; - .z.m.loginfo[`pqx;"Splitting large instruments"]; - t:update islargerthantargetsize:estbyt>maxsize from t; - oversized:select from t where islargerthantargetsize; - t:t except oversized; - oversized:update numfiles:ceiling[estbyt%maxsize] from oversized; - plans,:enlist each raze {[t;c] t[`numfiles]#enlist t[c]}[;o`symcol] each oversized - ]; + / if one sym per file, enlist each sym to assign to individual buckets + / else move to oversized and packing logic + $[o`onesymperfile; + [.z.m.loginfo[`pqx;"Enforcing one sym per file"]; + plans,:enlist each t[o[`symcol]] + ]; + / if split oversized is required, check against maxsize and return a plan entry for each required file + [if[o`splitoversized; + .z.m.loginfo[`pqx;"Splitting large instruments"]; + t:update islargerthantargetsize:estbyt>maxsize from t; + oversized:select from t where islargerthantargetsize; + t:t except oversized; + oversized:update numfiles:ceiling[estbyt%maxsize] from oversized; + plans,:enlist each raze {[t;c] t[`numfiles]#enlist t[c]}[;o`symcol] each oversized + ]; - / next fit function for packing instruments into buckets if they conform to the max size - if[count t; - .z.m.loginfo[`pqx;"Bucketing small instruments"]; - tabs:t[o[`symcol]]; - sizes:t`estbyt; - n:count tabs; - - step:{[maxsize;sizes;state;i] - sz:sizes i; - tot:state 1; - $[(tot+sz)>maxsize; (1+state 0; sz); (state 0; tot+sz)] - }[maxsize;sizes]; - bins: (step\[(0;0);til n])[;0]; - - plans,:value[tabs @ group bins] + / next fit function for packing instruments into buckets if they conform to the max size + if[count t; + .z.m.loginfo[`pqx;"Bucketing small instruments"]; + tabs:t[o[`symcol]]; + sizes:t`estbyt; + n:count tabs; + + step:{[maxsize;sizes;state;i] + sz:sizes i; + tot:state 1; + $[(tot+sz)>maxsize; (1+state 0; sz); (state 0; tot+sz)] + }[maxsize;sizes]; + bins: (step\[(0;0);til n])[;0]; + + plans,:value[tabs @ group bins] + ] + ] ]; - plans:(1 + til count plans)!plans; + plans:flip `seqno`syms!((1 + til count plans);plans); - / attach estbytes to plan's output - :update estbytes:.z.m.calcsize[symstats;o`symcol;;]'[syms;seqno] from {`syms`seqno!/: flip (key[x];value[x])} group plans + / attach estbytes by file to plan's output + :0!`syms xgroup update estbytes:"j"$.z.m.calcsize[symstats;o`symcol;;]'[syms;seqno]%count seqno by syms from plans }; datalookup:{[t;symcol;syms;cnt] @@ -161,35 +171,35 @@ datalookuponesym:{[t;symcol;sym;cnt] writefile:{[t;o;writeopt;writedir;map] / writes data to disk in Parquet format / returns stats to be inserted into the manifest - syms:map[`syms]; - seqno:map[`seqno]; - estbytes:map[`estbytes]; + + syms:map`syms; + map:flip `syms _ map; / build paths for each seqno - paths:writedir,/:o[`filestub],/:"-",/:("0"^-5$string[seqno]),\:".parquet"; - + paths:writedir,/:o[`filestub],/:"-",/:("0"^-5$string[map`seqno]),\:".parquet"; + / data to write, iterated by file - res:raze {[t;o;writeopt;split;path;seqno;estbytes;i] - .z.m.loginfo[`pqx;"Writing seqNo ",string[seqno],", path - ",path]; + res:raze {[t;o;writeopt;split;syms;map;path;i] + .z.m.loginfo[`pqx;"Writing seqNo ",string[map`seqno],", path - ",path]; res:.z.m.tryfn[`.m.di.0pqx.arrow.pq.writeParquetFromTable;(path;.z.m.checkandconvertcols t[i];writeopt)]; $[first res; - .z.m.loginfo[`pqx;"seqNo ",string[seqno]," write successful"]; - .z.m.logwarn[`pqx;"seqNo ",string[seqno]," write unsuccessful. Error - ",last res] + .z.m.loginfo[`pqx;"seqNo ",string[map`seqno]," write successful"]; + .z.m.logwarn[`pqx;"seqNo ",string[map`seqno]," write unsuccessful. Error - ",last res] ]; :`file`seq`syms`nsyms`rows`mintime`maxtime`estbytes`bytes`split`status!/: flip ( hsym `$path; - seqno; - enlist distinct[t[i][o`symcol]]; - count[distinct[t[i][o`symcol]]]; + map`seqno; + enlist[syms]; + count[syms]; count[i]; ?[t[i];();();(min;o`timecol)]; ?[t[i];();();(max;o`timecol)]; - estbytes; + map`estbytes; @[hcount;hsym `$path;0]; split; `error`ok[first res] ) - }[t;o;writeopt;1manifestcountbeforef6,1,,extract still appends to the accumulated in-memory manifest despite the sidecar write failure + +run,0,0,q,pqx.extract[pqxbasic;`pqxtrade;2025.07.29;`outdir`onesymperfile!(`:pqxout12/;1b)],1,,Extract a multi-sym table with onesymperfile on +true,0,0,q,3~count select from pqx.getmanifest[] where file like "*pqxout12*",1,,onesymperfile produces one file per instrument instead of bucketing them together +true,0,0,q,all 1=exec nsyms from pqx.getmanifest[] where file like "*pqxout12*",1,,Every file written under onesymperfile contains exactly one instrument +true,0,0,q,9~exec sum rows from pqx.getmanifest[] where file like "*pqxout12*",1,,All input rows are still accounted for across the per-instrument files +true,0,0,q,not any exec split from pqx.getmanifest[] where file like "*pqxout12*",1,,Files are not flagged as split when written one instrument per file + +run,0,0,q,pqx.extract[pqxoversized;`pqxtrade;2025.07.30;`targetsize`maxfactor`calibrate`compressionratio`outdir`splitoversized`onesymperfile!(1000;1.1;0b;0.30;`:pqxout13/;1b;1b)],1,,onesymperfile takes priority over splitoversized for an oversized single instrument +true,0,0,q,1~count select from pqx.getmanifest[] where file like "*pqxout13*",1,,onesymperfile forces splitoversized off, so the oversized instrument is written to a single file +true,0,0,q,20000~first exec rows from pqx.getmanifest[] where file like "*pqxout13*",1,,The single file still contains every row of the oversized instrument +true,0,0,q,not first exec split from pqx.getmanifest[] where file like "*pqxout13*",1,,The single file is not flagged as split, since splitting was suppressed by onesymperfile + +fail,0,0,q,pqx.extract[pqxbasic;`pqxtrade;2025.07.31;`outdir`badoption!(`:pqxoutf7/;1)],1,,Extract fails when passed an option key that isn't recognised + +after,0,0,q,system "rm -rf pqxout1 pqxout1b pqxout2 pqxout3 pqxout4 pqxout5 pqxout6 pqxout7 pqxout8 pqxout9 pqxout10 pqxout11 pqxout12 pqxout13 pqxoutf1 pqxoutf1b pqxoutf2 pqxoutf3 pqxoutf4 pqxoutf5 pqxoutf6 pqxoutf7",1,,Remove parquet output directories written by tests From f3db6581cdef0878ce7730d8c764731a39b47ff4 Mon Sep 17 00:00:00 2001 From: ConorSwainDI Date: Fri, 28 Aug 2026 13:33:08 +0100 Subject: [PATCH 14/22] Add dictcol functionality --- di/pqx/init.q | 2 +- di/pqx/pqx.md | 46 +++++++++++++++++++++-- di/pqx/pqx.q | 99 +++++++++++++++++++++++++++++++++++++++++++------ di/pqx/test.csv | 38 ++++++++++++++++++- 4 files changed, 167 insertions(+), 18 deletions(-) diff --git a/di/pqx/init.q b/di/pqx/init.q index ace5fd8b..c83e69be 100644 --- a/di/pqx/init.q +++ b/di/pqx/init.q @@ -4,4 +4,4 @@ arrow:use`kx.arrow \l ::pqx.q -export:([init;extract;getmanifest;checkandconvertcols;estimate;plan;writefile;tryfn]) +export:([init;extract;getmanifest;readfile;checkandconvertcols;estimate;plan;writefile;tryfn]) diff --git a/di/pqx/pqx.md b/di/pqx/pqx.md index 8d59e30a..fbc64e5e 100644 --- a/di/pqx/pqx.md +++ b/di/pqx/pqx.md @@ -15,6 +15,7 @@ oversized instrument across multiple files where required. A manifest recording - Optionally splits any oversized single instrument across multiple files - Optionally calibrates the raw-to-parquet size ratio with a trial write, or uses a fixed ratio - Optionally pre-sorts input data by instrument/time before writing +- Optionally partitions output into Hive-style `=/` subdirectories via `dictcols`, one file per combination, ahead of the size-based bucketing above - the partitioned columns are dropped from the on-disk data and can be reconstructed from the path with `readfile` - Writes files sequentially or in parallel (`peach`) - Accumulates a manifest of every file written, including row counts, instrument lists, time bounds and on-disk size @@ -73,7 +74,7 @@ omitted from `o` fall back to the default shown below. | `rowgroupbytes` | `128*1024*1024` | long | Reserved for future use — not currently read by the write path | | `codec` | `` `zstd `` | symbol | Compression codec, upper-cased and applied to the writer's `` `COMPRESSION `` option | | `complevel` | `3` | long | Reserved for future use — not currently read by the write path | -| `dictcols` | `` `sym`exchange `` | symbol list | Reserved for future use — not currently read by the write path | +| `dictcols` | `` `symbol$() `` | symbol list | Virtual, path-only partition columns. When non-empty, takes precedence over `onesymperfile`/`splitoversized`: one file is written per distinct combination of these columns' values, forcing both of those off for the call | | `parallel` | `0b` | boolean | Write files with `peach` instead of `each` | | `outdir` | `` `:. `` | symbol | Root output directory | | `filestub` | `"part"` | string | File name stub; files are written as `-NNNNN.parquet` | @@ -91,6 +92,18 @@ When `onesymperfile` is `1b`, each instrument is written to its own file regardl bucketing, and `splitoversized` is forced to `0b` for that call (an oversized single instrument is still written to one file, not split, even if `splitoversized:1b` is also passed in `o`). +When `dictcols` is non-empty, it takes precedence over both `onesymperfile` and `splitoversized` +(both are forced to `0b` for that call, regardless of what was passed in `o`), and `targetsize`/ +`maxfactor` bucketing does not apply: every distinct combination of the `dictcols` columns' values +becomes exactly one output file, however large. `extract` throws (`` `di.pqx: not all dictcols +found in table ``) if any `dictcols` column is not present in the input table. Output paths gain one +`=/` segment per `dictcols` column, Hive-style, e.g. +`//date=
/exchange=NASDAQ/-NNNNN.parquet` — this path is the only place +a file's combination is recorded; the manifest schema itself is unchanged. The `dictcols` columns +themselves are dropped from the on-disk data before writing (their value is already fixed by the +path, so keeping them in every row would just be redundant storage) — use `readfile` (see below) to +read a file back with its `dictcols` values (and the `date` partition) reattached as columns. + --- ## Manifest Schema @@ -146,6 +159,7 @@ pqx.init[enlist[`log]!enlist logdep] | `init[deps]` | Wire the injected `log` dependency. Call once before the first `extract`. | | `extract[t;tname;dt;o]` | Write a table out to one or more parquet files, appending one row per file to the module's `manifest`. Returns that same per-file stats table, scoped to this call. | | `getmanifest[]` | Return the manifest accumulated so far across all `extract` calls. | +| `readfile[path;readopt]` | Read a single file back, reattaching its `dictcols`/`date` values reconstructed from its path (see Options). | The remaining exports — `checkandconvertcols`, `estimate`, `plan`, `writefile`, `tryfn` — are internal pipeline steps of `extract`, exposed only so `k4unit` can exercise them directly. Call @@ -179,6 +193,28 @@ size-estimation/calibration step, so a fresh `outdir` works with the default `ca pqx.extract[trade;`trade;2025.07.15;`targetsize`codec!(256*1024*1024;`gzip)] ``` +### `readfile[path;readopt]` +Read a single parquet file back via `` .m.di.0pqx.arrow.pq.readParquetToTable ``, then reattach any +values that `extract` stripped from the on-disk data and encoded only in the file's path — the +`date=
` partition segment (reconstructed as a date) and any `dictcols` combination segments +(each reconstructed as a symbol). Every `col=value` segment found in `path` becomes a column in the +returned table, broadcast as a constant across every row. `path` may be a plain string or an hsym, +with or without a leading colon; `readopt` is passed straight through to the underlying reader (e.g. +`` (0#`)!() `` to read every column). + +| Parameter | Type | Description | +|---|---|---| +| `path` | string or symbol | Path to a single file, as recorded in a manifest `file` value | +| `readopt` | dict | Options passed through to `` .m.di.0pqx.arrow.pq.readParquetToTable `` | + +```q +f:1_string first exec file from pqx.getmanifest[] where file like "*exchange=NASDAQ*" +pqx.readfile[f;(0#`)!()] +``` + +Only useful for a file written with `dictcols` set, or to recover the `date` — a file written without +`dictcols` has nothing to reconstruct beyond `date`, since no other columns were stripped from it. + --- ## Usage Example @@ -225,9 +261,11 @@ zero-row table and a table missing `symcol`/`timecol` (both fail outright), and ## Notes -- `rowgroupbytes`, `complevel`, and `dictcols` are accepted in `default` and any `o` override, but - nothing in the current write path reads them — only `` `PARQUET_VERSION `` (fixed at - `` `V2.LATEST ``) and `` `COMPRESSION `` (from `codec`) are passed to the writer. +- `rowgroupbytes` and `complevel` are accepted in `default` and any `o` override, but nothing in the + current write path reads them — only `` `PARQUET_VERSION `` (fixed at `` `V2.LATEST ``) and + `` `COMPRESSION `` (from `codec`) are passed to the writer. `dictcols` (see Options) is read by + `plan`/`estimate`/`writefile` for file planning and output paths, but has no relation to Parquet + dictionary encoding despite the name. - `symcol`/`timecol` presence is validated unconditionally on every `extract` call, even when `presort` is `0b`. A zero-row input table is rejected outright, before that check, regardless of `calibrate`. diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index 0e556af1..af629e6e 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -16,7 +16,7 @@ default:( 128*1024*1024; / ~128 MB row groups `zstd; / codec 3; / compression level - `sym`exchange; / dictionary-encode these columns + `symbol$(); / virtual (path-only) partition columns, taking precedence over onesymperfile/splitoversized 0b; / write files via peach `:.; / output directory "part" / file name stub @@ -50,9 +50,12 @@ estimate:{[t;o;writeopt] / for a partition of data, estimates the size of the tables to be saved to disk / if calibrate flag is true in o, a test write is carried out / returns a table of storage stats for all instruments and the compression ratio, which may have changed depending on calibration - cnts:`rowcnt xasc 0!?[t;();enlist[o[`symcol]]!enlist[o[`symcol]];enlist[`rowcnt]!enlist(count;o[`timecol])]; / select rowcnt:count time by sym from t, using appropriate substitutions for time and sym cols + / group by dictcols as well as symcol, so a sym occurring under multiple dictcols combinations gets its own row + gcols:distinct o[`dictcols],o[`symcol]; + cnts:`rowcnt xasc 0!?[t;();gcols!gcols;enlist[`rowcnt]!enlist(count;o[`timecol])]; / select rowcnt:count time by gcols from t, using appropriate substitutions for time and sym/dictcols medsym:cnts @ first where abs[cnt-med[cnt]]=min[abs[cnt-med[cnt:cnts`rowcnt]]]; - bytesperrow:%[-22!t:.z.m.checkandconvertcols t[where t[o[`symcol]]=medsym[o[`symcol]]];medsym`rowcnt]; + mask:min each flip {[t;medsym;x] t[x]=medsym[x]}[t;medsym] each gcols; / row matches medsym's full (dictcols,symcol) combination, not just its sym + bytesperrow:%[-22!t:.z.m.checkandconvertcols t[where mask];medsym`rowcnt]; / calibrate compression ratio if option is enabled if[o`calibrate; @@ -112,6 +115,17 @@ plan:{[t;o;maxsize] / else a new bucket is created / large instruments are also split into multiple files if splitoversized flag is true symstats:t; + + / dictcols take precedence over everything else - one file per distinct combination. + / computed directly here (rather than through the shared bucket-then-recompute tail below) because + / symstats can carry multiple rows per sym once dictcols grouping is active (see estimate), and the + / tail's calcsize call would double count a sym's bytes across its different dictcols combinations + if[count o`dictcols; + .z.m.loginfo[`pqx;"Enforcing one file per dictcols combination"]; + grp:0!?[t;();(o`dictcols)!o`dictcols;`syms`estbytes!((o`symcol);(sum;`estbyt))]; + :update seqno:enlist each 1+til count grp, estbytes:enlist each "j"$estbytes from grp + ]; + plans:(); / if one sym per file, enlist each sym to assign to individual buckets @@ -154,18 +168,19 @@ plan:{[t;o;maxsize] :0!`syms xgroup update estbytes:"j"$.z.m.calcsize[symstats;o`symcol;;]'[syms;seqno]%count seqno by syms from plans }; -datalookup:{[t;symcol;syms;cnt] +datalookup:{[t;symcol;syms;cnt;mask] / get lists of indices by file / a pass with multiple instruments is assumed to be one file only, hence the return is flattened into one list + / mask restricts to rows belonging to this file's dictcols combination (all 1b when dictcols is unset) $[11 sym doesn't survive a second functional amend). + / guarded on count: a functional delete of an EMPTY column list against a `p#-attributed symcol (set by + / presort, on by default) silently returns a zero-row table instead of being the no-op it should be + t:$[count o`dictcols;![t;();0b;o`dictcols];t]; + + map:flip (`syms,o[`dictcols]) _ map; + idx:.z.m.datalookup[t;o`symcol;syms;count map;combomask]; / build paths for each seqno paths:writedir,/:o[`filestub],/:"-",/:("0"^-5$string[map`seqno]),\:".parquet"; - + / data to write, iterated by file res:raze {[t;o;writeopt;split;syms;map;path;i] .z.m.loginfo[`pqx;"Writing seqNo ",string[map`seqno],", path - ",path]; @@ -199,11 +235,28 @@ writefile:{[t;o;writeopt;writedir;map] split; `error`ok[first res] ) - }[t;o;writeopt;1 Date: Mon, 31 Aug 2026 13:02:05 +0100 Subject: [PATCH 15/22] Virtual table functionality --- di/pqx/init.q | 4 ++- di/pqx/pqx.md | 84 +++++++++++++++++++++++++++++++++++++++++++------ di/pqx/pqx.q | 44 +++++++++++++++++++------- di/pqx/test.csv | 38 +++++++++++++++++++++- 4 files changed, 148 insertions(+), 22 deletions(-) diff --git a/di/pqx/init.q b/di/pqx/init.q index c83e69be..d690cd3c 100644 --- a/di/pqx/init.q +++ b/di/pqx/init.q @@ -1,7 +1,9 @@ / KDB-X Parquet extract module to save kdb+ data to parquet storage convention arrow:use`kx.arrow +pq:use`kx.pq +pqt:use`kx.pq.t \l ::pqx.q -export:([init;extract;getmanifest;readfile;checkandconvertcols;estimate;plan;writefile;tryfn]) +export:([init;extract;getmanifest;checkandconvertcols;estimate;plan;writefile;tryfn;buildvirtualtable;castvirtualcol]) diff --git a/di/pqx/pqx.md b/di/pqx/pqx.md index fbc64e5e..0440040c 100644 --- a/di/pqx/pqx.md +++ b/di/pqx/pqx.md @@ -18,6 +18,7 @@ oversized instrument across multiple files where required. A manifest recording - Optionally partitions output into Hive-style `=/` subdirectories via `dictcols`, one file per combination, ahead of the size-based bucketing above - the partitioned columns are dropped from the on-disk data and can be reconstructed from the path with `readfile` - Writes files sequentially or in parallel (`peach`) - Accumulates a manifest of every file written, including row counts, instrument lists, time bounds and on-disk size +- Builds a queryable virtual table over a directory of previously-written `.parquet` files, without reading any of their data up front - `date`/`dictcols` values are reconstructed straight from each file's path --- @@ -27,15 +28,22 @@ oversized instrument across multiple files where required. A manifest recording |---|---|---|---| | logger | `` `log `` | yes | dict with `info`, `warn`, and `error`, each binary `{[c;m]}` where `c` is a symbol context and `m` is a string | -**Hard dependency:** `kx.arrow` — loaded automatically (via `use`) when `di.pqx` is imported, before -`pqx.q` itself is loaded. `extract` calls `` .m.di.0pqx.arrow.pq.writeParquetFromTable `` to perform -every write. +**Hard dependencies:** `kx.arrow` and `kx.pq` — both loaded automatically (via `use`) when `di.pqx` +is imported, before `pqx.q` itself is loaded. `extract`/`readfile` call +`` .m.di.0pqx.arrow.pq.writeParquetFromTable ``/`` .m.di.0pqx.arrow.pq.readParquetToTable `` to +perform every write/read; `buildvirtualtable` calls `` .m.di.0pqx.pq.pq `` (per-file virtual table) +and `` .m.di.0pqx.pqt.mkP `` (from `` kx.pq.t ``, loaded separately as `pqt`) to compose the +multi-file view. -`kx.arrow` must be resolvable on the process's module search path at that point. Where it's +Both modules must be resolvable on the process's module search path at that point. Where either is installed as a conda package (e.g. under a `kx.qmamba`-managed root such as `~/.kx/root/lib/q/mod`), that location needs to already be on `QPATH` — `di.pqx` does not load -`kx.qmamba` itself to arrange this. Confirm `kx.arrow` loads standalone (`` use`kx.arrow ``) in the -target environment before relying on `di.pqx` there. +`kx.qmamba` itself to arrange this. Confirm `kx.arrow`, `kx.pq`, and `kx.pq.t` each load standalone +(`` use`kx.arrow ``, `` use`kx.pq ``, `` use`kx.pq.t ``) in the target environment before relying on +`di.pqx` there. `kx.arrow`'s bundled `libcurl` may also need a newer OpenSSL than the system default +on the `LD_LIBRARY_PATH` — if `use\`kx.arrow` fails with an `OPENSSL_*` symbol-version error, point +`LD_LIBRARY_PATH` at the `lib` directory the conda-installed `kx.arrow`/`kx.pq` packages ship +alongside it. The `log` dependency must be passed to `init` inside a dict keyed on `` `log ``. `init` throws immediately if `log` is absent, is not a dict, or is missing any of `info`/`warn`/`error`. The value @@ -137,6 +145,53 @@ updates the in-memory `manifest`. --- +## Virtual Tables + +`buildvirtualtable` opens every `.parquet` file under `//` as a single queryable +table, without reading any row data up front - it's the read-side counterpart to `extract`'s output +layout. Each file's `date=
/` segment (and any `=/` `dictcols` segments) is +reconstructed from its path into a virtual column, exactly mirroring what `extract`/`writefile` +stripped from the on-disk data on the way in. Filtering on those virtual columns (e.g. +`` select from vt where date=2025.07.15,exch=`NASDAQ `` ) prunes to just the matching files rather +than scanning everything. + +The returned value is not a regular in-memory kdb+ table - it's a functional/composed object from +`kx.pq.t`'s `mkP`. `select` works on it directly; `exec`/`meta`/`cols` do not work applied directly +to it, only to a `select` (or `meta`) result taken from it first (e.g. +`` exec c from meta vt `` to list columns, `` exec count i from select from vt `` to count rows, +not `` count vt `` or `` cols vt ``). + +Genuine on-disk columns keep whatever type they were written with - notably, character/symbol +columns come back as **strings**, not symbols, since Parquet has no native symbol type (see +`checkandconvertcols`). Only the path-reconstructed `date`/`dictcols` columns come back typed as a +real date/symbol. + +### `buildvirtualtable[hdbdir;tname;datecol;dictcols]` +Find every `.parquet` file under `//` and compose them into one virtual table, +partitioned by the `date=
/` segment and any `dictcols` segments found in each file's path. + +| Parameter | Type | Description | +|---|---|---| +| `hdbdir` | symbol (hsym) | Root directory - matches `extract`'s `outdir` | +| `tname` | symbol | Table name - matches `extract`'s `tname` | +| `datecol` | symbol | Name to give the reconstructed date partition column (need not be literally `` `date `` ) | +| `dictcols` | symbol list | Names of any `dictcols` path segments to reconstruct, in path order - matches `extract`'s `dictcols`. Pass `` `symbol$() `` if the data was written without `dictcols` | + +```q +vt:pqx.buildvirtualtable[`:./;`trade;`date;enlist`exchange] +select from vt where date=2025.07.15,exchange=`NASDAQ +``` + +Building the table succeeds even if no files match (an empty view); querying that empty view then +fails, rather than silently returning zero rows. `dictcols` here does not need to match a prior +`extract` call exactly - it only needs to match the path segments actually present under +`hdbdir/tname/date=.../`. Passing a `dictcols` name that collides with a genuine on-disk column +(rather than one `extract` actually stripped to the path) produces unreliable results — the two +columns are not distinguished internally, unlike on the write side where `writefile` always strips +the real column first. + +--- + ## Initialisation `init[deps]` wires the injected `log` dependency and must be called before the first `extract`. It @@ -160,10 +215,11 @@ pqx.init[enlist[`log]!enlist logdep] | `extract[t;tname;dt;o]` | Write a table out to one or more parquet files, appending one row per file to the module's `manifest`. Returns that same per-file stats table, scoped to this call. | | `getmanifest[]` | Return the manifest accumulated so far across all `extract` calls. | | `readfile[path;readopt]` | Read a single file back, reattaching its `dictcols`/`date` values reconstructed from its path (see Options). | +| `buildvirtualtable[hdbdir;tname;datecol;dictcols]` | Compose every file under `hdbdir/tname/` into one queryable virtual table, with `date`/`dictcols` reconstructed from each file's path (see Virtual Tables). | -The remaining exports — `checkandconvertcols`, `estimate`, `plan`, `writefile`, `tryfn` — are -internal pipeline steps of `extract`, exposed only so `k4unit` can exercise them directly. Call -`extract` for normal use. +The remaining exports — `checkandconvertcols`, `estimate`, `plan`, `writefile`, `tryfn`, +`castvirtualcol` — are internal pipeline steps of `extract`/`buildvirtualtable`, exposed only so +`k4unit` can exercise them directly. Call `extract`/`buildvirtualtable` for normal use. ### `init[deps]` Validate the required `log` dependency and store it for use by every other function. @@ -257,6 +313,13 @@ custom `filestub`/non-default codec — then asserts on the resulting `getmanife zero-row table and a table missing `symcol`/`timecol` (both fail outright), and an invalid `codec` (degrades gracefully — see Manifest Schema's `status` column). +It also builds `buildvirtualtable` views over several of those same `extract` outputs — no +`dictcols`, a single `dictcols` level, and two `dictcols` levels (one overlapping `symcol` itself) — +asserting row counts, virtual-column types, and that filtering on a virtual column prunes to the +right file(s). `castvirtualcol` is exercised directly for both the datecol and non-datecol cases, +including a datecol not literally named `` `date `` . A directory with no matching files is covered +too: building the view over it succeeds, but querying it then fails. + --- ## Notes @@ -272,3 +335,6 @@ zero-row table and a table missing `symcol`/`timecol` (both fail outright), and - A per-file write failure (e.g. an invalid `codec`) is caught and logged at `warn`, and that file is recorded with `` status=`error `` (and `bytes:0`) in the manifest — it does not abort the rest of `extract`. +- `buildvirtualtable` has only been tested pointed at a single `date=` partition at a time; pointing + it at `hdbdir/tname/` directories that span multiple dates is expected to work (the `date` segment + is reconstructed the same way as any other level) but isn't covered by `test.csv` yet. diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index af629e6e..772db3fd 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -198,18 +198,17 @@ writefile:{[t;o;writeopt;writedir;map] writedir:writedir,$[count segs;("/" sv segs),"/";""]; combomask:$[count segs;min each flip {[t;x;y] t[x]=y}[t]'[key combo;value combo];count[t]#1b]; - / compute row indices against the original t - datalookup needs t[symcol], and symcol may itself be one - / of the dictcols columns - then apply those same indices to a separately-stripped table for the write. - / dictcols values are fixed for the whole file (per combomask above) and encoded in its path already, so - / they are dropped once, up front, rather than duplicated into every row of the on-disk data; dropping - / columns from an already row-indexed slice instead breaks the arrow write whenever a file holds more - / than one instrument (datalookup's index shape for >1 sym doesn't survive a second functional amend). - / guarded on count: a functional delete of an EMPTY column list against a `p#-attributed symcol (set by - / presort, on by default) silently returns a zero-row table instead of being the no-op it should be - t:$[count o`dictcols;![t;();0b;o`dictcols];t]; - map:flip (`syms,o[`dictcols]) _ map; + + / row indices must be found before dictcols columns are dropped - datalookup needs t[symcol], and symcol + / may itself be one of the dictcols columns. dictcols values are fixed for the whole file (per combomask + / above) and encoded in its path already, so t is stripped of them in place afterwards - once here, not + / duplicated into every row of the on-disk data, and not dropped later from an already row-indexed slice + / (which breaks the arrow write whenever a file holds more than one instrument). guarded on count: a + / functional delete of an EMPTY column list against a `p#-attributed symcol (set by presort, on by + / default) silently returns a zero-row table instead of being the no-op it should be idx:.z.m.datalookup[t;o`symcol;syms;count map;combomask]; + t:$[count o`dictcols;![t;();0b;o`dictcols];t]; / build paths for each seqno paths:writedir,/:o[`filestub],/:"-",/:("0"^-5$string[map`seqno]),\:".parquet"; @@ -235,7 +234,7 @@ writefile:{[t;o;writeopt;writedir;map] split; `error`ok[first res] ) - }[tw;o;writeopt;1 Date: Mon, 31 Aug 2026 15:14:38 +0100 Subject: [PATCH 16/22] Type check on dictcols --- di/pqx/init.q | 2 +- di/pqx/pqx.q | 9 ++++++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/di/pqx/init.q b/di/pqx/init.q index d690cd3c..bf2cdf6e 100644 --- a/di/pqx/init.q +++ b/di/pqx/init.q @@ -6,4 +6,4 @@ pqt:use`kx.pq.t \l ::pqx.q -export:([init;extract;getmanifest;checkandconvertcols;estimate;plan;writefile;tryfn;buildvirtualtable;castvirtualcol]) +export:([init;extract;getmanifest;checkandconvertcols;estimate;plan;writefile;readfile;tryfn;buildvirtualtable;castvirtualcol]) diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index 772db3fd..4e96a382 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -261,7 +261,7 @@ buildvirtualtable:{[hdbdir;tname;datecol;dictcols] / of their data - partition values (the date segment and any dictcols segments) are reconstructed from / each file's hive-style path, mirroring how writefile strips those same columns from the on-disk data path:` sv hdbdir,tname; - files:([] file:system"find \"",(1 _ string path),"\" -name \"*.parquet\""); + files:([] file:system"find \"",(1 _ string path),"\" -name \"*.parquet\" | sort"); files:update split:"/" vs/:file from files; lv:1+count where "/"=string path; levels:(),datecol,dictcols; @@ -324,14 +324,17 @@ extract:{[t;tname;dt;o] ]; / dictcols take precedence over onesymperfile/splitoversized - one file per combination - if[count opts`dictcols; + if[count opts[`dictcols]; if[not all opts[`dictcols] in cols[t]; .z.m.logerr[`pqx;err:"di.pqx: not all dictcols found in table"]; 'err ]; .z.m.loginfo[`pqx;"dictcols requested, turning off onesymperfile and splitoversized"]; opts[`onesymperfile]:0b; - opts[`splitoversized]:0b + opts[`splitoversized]:0b; + if[-11h=type opts`dictcols; + opts[`dictcols]:enlist opts`dictcols + ] ]; / if presort, sort by sym then time From 2a70c96b5b23dc4145d3da2b259d4e38f119c2fd Mon Sep 17 00:00:00 2001 From: ConorSwainDI Date: Tue, 1 Sep 2026 08:27:09 +0100 Subject: [PATCH 17/22] Minor change remove key check --- di/pqx/pqx.q | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index 4e96a382..31fd8816 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -299,11 +299,9 @@ extract:{[t;tname;dt;o] opts:default,o; / if one sym per file requested, turn off splitoversized - if[`onesymperfile in key o; - if[o`onesymperfile; - .z.m.loginfo[`pqx;"onesymperfile requested, turning off splitoversized"]; - opts[`splitoversized]:0b - ] + if[o`onesymperfile; + .z.m.loginfo[`pqx;"onesymperfile requested, turning off splitoversized"]; + opts[`splitoversized]:0b ]; / check for count in tables, error out if not From 1ef2a37e6e0b2f926cc51aa399055f6bf20f66b7 Mon Sep 17 00:00:00 2001 From: ConorSwainDI Date: Tue, 1 Sep 2026 10:42:59 +0100 Subject: [PATCH 18/22] Rename dictcols to virtualcols --- di/pqx/pqx.md | 51 ++++++++++++++++++------------------- di/pqx/pqx.q | 68 ++++++++++++++++++++++++------------------------- di/pqx/test.csv | 54 +++++++++++++++++++-------------------- 3 files changed, 86 insertions(+), 87 deletions(-) diff --git a/di/pqx/pqx.md b/di/pqx/pqx.md index 0440040c..7fdd2616 100644 --- a/di/pqx/pqx.md +++ b/di/pqx/pqx.md @@ -15,10 +15,10 @@ oversized instrument across multiple files where required. A manifest recording - Optionally splits any oversized single instrument across multiple files - Optionally calibrates the raw-to-parquet size ratio with a trial write, or uses a fixed ratio - Optionally pre-sorts input data by instrument/time before writing -- Optionally partitions output into Hive-style `=/` subdirectories via `dictcols`, one file per combination, ahead of the size-based bucketing above - the partitioned columns are dropped from the on-disk data and can be reconstructed from the path with `readfile` +- Optionally partitions output into Hive-style `=/` subdirectories via `virtualcols`, one file per combination, ahead of the size-based bucketing above - the partitioned columns are dropped from the on-disk data and can be reconstructed from the path with `readfile` - Writes files sequentially or in parallel (`peach`) - Accumulates a manifest of every file written, including row counts, instrument lists, time bounds and on-disk size -- Builds a queryable virtual table over a directory of previously-written `.parquet` files, without reading any of their data up front - `date`/`dictcols` values are reconstructed straight from each file's path +- Builds a queryable virtual table over a directory of previously-written `.parquet` files, without reading any of their data up front - `date`/`virtualcols` values are reconstructed straight from each file's path --- @@ -82,7 +82,7 @@ omitted from `o` fall back to the default shown below. | `rowgroupbytes` | `128*1024*1024` | long | Reserved for future use — not currently read by the write path | | `codec` | `` `zstd `` | symbol | Compression codec, upper-cased and applied to the writer's `` `COMPRESSION `` option | | `complevel` | `3` | long | Reserved for future use — not currently read by the write path | -| `dictcols` | `` `symbol$() `` | symbol list | Virtual, path-only partition columns. When non-empty, takes precedence over `onesymperfile`/`splitoversized`: one file is written per distinct combination of these columns' values, forcing both of those off for the call | +| `virtualcols` | `` `symbol$() `` | symbol list | Virtual, path-only partition columns. When non-empty, takes precedence over `onesymperfile`/`splitoversized`: one file is written per distinct combination of these columns' values, forcing both of those off for the call | | `parallel` | `0b` | boolean | Write files with `peach` instead of `each` | | `outdir` | `` `:. `` | symbol | Root output directory | | `filestub` | `"part"` | string | File name stub; files are written as `-NNNNN.parquet` | @@ -100,17 +100,17 @@ When `onesymperfile` is `1b`, each instrument is written to its own file regardl bucketing, and `splitoversized` is forced to `0b` for that call (an oversized single instrument is still written to one file, not split, even if `splitoversized:1b` is also passed in `o`). -When `dictcols` is non-empty, it takes precedence over both `onesymperfile` and `splitoversized` +When `virtualcols` is non-empty, it takes precedence over both `onesymperfile` and `splitoversized` (both are forced to `0b` for that call, regardless of what was passed in `o`), and `targetsize`/ -`maxfactor` bucketing does not apply: every distinct combination of the `dictcols` columns' values -becomes exactly one output file, however large. `extract` throws (`` `di.pqx: not all dictcols -found in table ``) if any `dictcols` column is not present in the input table. Output paths gain one -`=/` segment per `dictcols` column, Hive-style, e.g. +`maxfactor` bucketing does not apply: every distinct combination of the `virtualcols` columns' values +becomes exactly one output file, however large. `extract` throws (`` `di.pqx: not all virtualcols +found in table ``) if any `virtualcols` column is not present in the input table. Output paths gain one +`=/` segment per `virtualcols` column, Hive-style, e.g. `//date=
/exchange=NASDAQ/-NNNNN.parquet` — this path is the only place -a file's combination is recorded; the manifest schema itself is unchanged. The `dictcols` columns +a file's combination is recorded; the manifest schema itself is unchanged. The `virtualcols` columns themselves are dropped from the on-disk data before writing (their value is already fixed by the path, so keeping them in every row would just be redundant storage) — use `readfile` (see below) to -read a file back with its `dictcols` values (and the `date` partition) reattached as columns. +read a file back with its `virtualcols` values (and the `date` partition) reattached as columns. --- @@ -149,7 +149,7 @@ updates the in-memory `manifest`. `buildvirtualtable` opens every `.parquet` file under `//` as a single queryable table, without reading any row data up front - it's the read-side counterpart to `extract`'s output -layout. Each file's `date=
/` segment (and any `=/` `dictcols` segments) is +layout. Each file's `date=
/` segment (and any `=/` `virtualcols` segments) is reconstructed from its path into a virtual column, exactly mirroring what `extract`/`writefile` stripped from the on-disk data on the way in. Filtering on those virtual columns (e.g. `` select from vt where date=2025.07.15,exch=`NASDAQ `` ) prunes to just the matching files rather @@ -163,19 +163,19 @@ not `` count vt `` or `` cols vt ``). Genuine on-disk columns keep whatever type they were written with - notably, character/symbol columns come back as **strings**, not symbols, since Parquet has no native symbol type (see -`checkandconvertcols`). Only the path-reconstructed `date`/`dictcols` columns come back typed as a +`checkandconvertcols`). Only the path-reconstructed `date`/`virtualcols` columns come back typed as a real date/symbol. -### `buildvirtualtable[hdbdir;tname;datecol;dictcols]` +### `buildvirtualtable[hdbdir;tname;datecol;virtualcols]` Find every `.parquet` file under `//` and compose them into one virtual table, -partitioned by the `date=
/` segment and any `dictcols` segments found in each file's path. +partitioned by the `date=
/` segment and any `virtualcols` segments found in each file's path. | Parameter | Type | Description | |---|---|---| | `hdbdir` | symbol (hsym) | Root directory - matches `extract`'s `outdir` | | `tname` | symbol | Table name - matches `extract`'s `tname` | | `datecol` | symbol | Name to give the reconstructed date partition column (need not be literally `` `date `` ) | -| `dictcols` | symbol list | Names of any `dictcols` path segments to reconstruct, in path order - matches `extract`'s `dictcols`. Pass `` `symbol$() `` if the data was written without `dictcols` | +| `virtualcols` | symbol list | Names of any `virtualcols` path segments to reconstruct, in path order - matches `extract`'s `virtualcols`. Pass `` `symbol$() `` if the data was written without `virtualcols` | ```q vt:pqx.buildvirtualtable[`:./;`trade;`date;enlist`exchange] @@ -183,9 +183,9 @@ select from vt where date=2025.07.15,exchange=`NASDAQ ``` Building the table succeeds even if no files match (an empty view); querying that empty view then -fails, rather than silently returning zero rows. `dictcols` here does not need to match a prior +fails, rather than silently returning zero rows. `virtualcols` here does not need to match a prior `extract` call exactly - it only needs to match the path segments actually present under -`hdbdir/tname/date=.../`. Passing a `dictcols` name that collides with a genuine on-disk column +`hdbdir/tname/date=.../`. Passing a `virtualcols` name that collides with a genuine on-disk column (rather than one `extract` actually stripped to the path) produces unreliable results — the two columns are not distinguished internally, unlike on the write side where `writefile` always strips the real column first. @@ -214,8 +214,8 @@ pqx.init[enlist[`log]!enlist logdep] | `init[deps]` | Wire the injected `log` dependency. Call once before the first `extract`. | | `extract[t;tname;dt;o]` | Write a table out to one or more parquet files, appending one row per file to the module's `manifest`. Returns that same per-file stats table, scoped to this call. | | `getmanifest[]` | Return the manifest accumulated so far across all `extract` calls. | -| `readfile[path;readopt]` | Read a single file back, reattaching its `dictcols`/`date` values reconstructed from its path (see Options). | -| `buildvirtualtable[hdbdir;tname;datecol;dictcols]` | Compose every file under `hdbdir/tname/` into one queryable virtual table, with `date`/`dictcols` reconstructed from each file's path (see Virtual Tables). | +| `readfile[path;readopt]` | Read a single file back, reattaching its `virtualcols`/`date` values reconstructed from its path (see Options). | +| `buildvirtualtable[hdbdir;tname;datecol;virtualcols]` | Compose every file under `hdbdir/tname/` into one queryable virtual table, with `date`/`virtualcols` reconstructed from each file's path (see Virtual Tables). | The remaining exports — `checkandconvertcols`, `estimate`, `plan`, `writefile`, `tryfn`, `castvirtualcol` — are internal pipeline steps of `extract`/`buildvirtualtable`, exposed only so @@ -252,7 +252,7 @@ pqx.extract[trade;`trade;2025.07.15;`targetsize`codec!(256*1024*1024;`gzip)] ### `readfile[path;readopt]` Read a single parquet file back via `` .m.di.0pqx.arrow.pq.readParquetToTable ``, then reattach any values that `extract` stripped from the on-disk data and encoded only in the file's path — the -`date=
` partition segment (reconstructed as a date) and any `dictcols` combination segments +`date=
` partition segment (reconstructed as a date) and any `virtualcols` combination segments (each reconstructed as a symbol). Every `col=value` segment found in `path` becomes a column in the returned table, broadcast as a constant across every row. `path` may be a plain string or an hsym, with or without a leading colon; `readopt` is passed straight through to the underlying reader (e.g. @@ -268,8 +268,8 @@ f:1_string first exec file from pqx.getmanifest[] where file like "*exchange=NAS pqx.readfile[f;(0#`)!()] ``` -Only useful for a file written with `dictcols` set, or to recover the `date` — a file written without -`dictcols` has nothing to reconstruct beyond `date`, since no other columns were stripped from it. +Only useful for a file written with `virtualcols` set, or to recover the `date` — a file written without +`virtualcols` has nothing to reconstruct beyond `date`, since no other columns were stripped from it. --- @@ -314,7 +314,7 @@ zero-row table and a table missing `symcol`/`timecol` (both fail outright), and (degrades gracefully — see Manifest Schema's `status` column). It also builds `buildvirtualtable` views over several of those same `extract` outputs — no -`dictcols`, a single `dictcols` level, and two `dictcols` levels (one overlapping `symcol` itself) — +`virtualcols`, a single `virtualcols` level, and two `virtualcols` levels (one overlapping `symcol` itself) — asserting row counts, virtual-column types, and that filtering on a virtual column prunes to the right file(s). `castvirtualcol` is exercised directly for both the datecol and non-datecol cases, including a datecol not literally named `` `date `` . A directory with no matching files is covered @@ -326,9 +326,8 @@ too: building the view over it succeeds, but querying it then fails. - `rowgroupbytes` and `complevel` are accepted in `default` and any `o` override, but nothing in the current write path reads them — only `` `PARQUET_VERSION `` (fixed at `` `V2.LATEST ``) and - `` `COMPRESSION `` (from `codec`) are passed to the writer. `dictcols` (see Options) is read by - `plan`/`estimate`/`writefile` for file planning and output paths, but has no relation to Parquet - dictionary encoding despite the name. + `` `COMPRESSION `` (from `codec`) are passed to the writer. `virtualcols` (see Options) is read by + `plan`/`estimate`/`writefile` for file planning and output paths only. - `symcol`/`timecol` presence is validated unconditionally on every `extract` call, even when `presort` is `0b`. A zero-row input table is rejected outright, before that check, regardless of `calibrate`. diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index 31fd8816..b9fa83c9 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -1,7 +1,7 @@ / define default config default:( `targetsize`maxfactor`splitoversized`calibrate`onesymperfile`compressionratio, - `symcol`timecol`presort`rowgroupbytes`codec`complevel`dictcols, + `symcol`timecol`presort`rowgroupbytes`codec`complevel`virtualcols, `parallel`outdir`filestub )!( 512*1024*1024; / ~512 MB target file size @@ -50,11 +50,11 @@ estimate:{[t;o;writeopt] / for a partition of data, estimates the size of the tables to be saved to disk / if calibrate flag is true in o, a test write is carried out / returns a table of storage stats for all instruments and the compression ratio, which may have changed depending on calibration - / group by dictcols as well as symcol, so a sym occurring under multiple dictcols combinations gets its own row - gcols:distinct o[`dictcols],o[`symcol]; - cnts:`rowcnt xasc 0!?[t;();gcols!gcols;enlist[`rowcnt]!enlist(count;o[`timecol])]; / select rowcnt:count time by gcols from t, using appropriate substitutions for time and sym/dictcols + / group by virtualcols as well as symcol, so a sym occurring under multiple virtualcols combinations gets its own row + gcols:distinct o[`virtualcols],o[`symcol]; + cnts:`rowcnt xasc 0!?[t;();gcols!gcols;enlist[`rowcnt]!enlist(count;o[`timecol])]; / select rowcnt:count time by gcols from t, using appropriate substitutions for time and sym/virtualcols medsym:cnts @ first where abs[cnt-med[cnt]]=min[abs[cnt-med[cnt:cnts`rowcnt]]]; - mask:min each flip {[t;medsym;x] t[x]=medsym[x]}[t;medsym] each gcols; / row matches medsym's full (dictcols,symcol) combination, not just its sym + mask:min each flip {[t;medsym;x] t[x]=medsym[x]}[t;medsym] each gcols; / row matches medsym's full (virtualcols,symcol) combination, not just its sym bytesperrow:%[-22!t:.z.m.checkandconvertcols t[where mask];medsym`rowcnt]; / calibrate compression ratio if option is enabled @@ -116,13 +116,13 @@ plan:{[t;o;maxsize] / large instruments are also split into multiple files if splitoversized flag is true symstats:t; - / dictcols take precedence over everything else - one file per distinct combination. + / virtualcols take precedence over everything else - one file per distinct combination. / computed directly here (rather than through the shared bucket-then-recompute tail below) because - / symstats can carry multiple rows per sym once dictcols grouping is active (see estimate), and the - / tail's calcsize call would double count a sym's bytes across its different dictcols combinations - if[count o`dictcols; - .z.m.loginfo[`pqx;"Enforcing one file per dictcols combination"]; - grp:0!?[t;();(o`dictcols)!o`dictcols;`syms`estbytes!((o`symcol);(sum;`estbyt))]; + / symstats can carry multiple rows per sym once virtualcols grouping is active (see estimate), and the + / tail's calcsize call would double count a sym's bytes across its different virtualcols combinations + if[count o`virtualcols; + .z.m.loginfo[`pqx;"Enforcing one file per virtualcols combination"]; + grp:0!?[t;();(o`virtualcols)!o`virtualcols;`syms`estbytes!((o`symcol);(sum;`estbyt))]; :update seqno:enlist each 1+til count grp, estbytes:enlist each "j"$estbytes from grp ]; @@ -171,7 +171,7 @@ plan:{[t;o;maxsize] datalookup:{[t;symcol;syms;cnt;mask] / get lists of indices by file / a pass with multiple instruments is assumed to be one file only, hence the return is flattened into one list - / mask restricts to rows belonging to this file's dictcols combination (all 1b when dictcols is unset) + / mask restricts to rows belonging to this file's virtualcols combination (all 1b when virtualcols is unset) $[1 Date: Tue, 1 Sep 2026 11:44:42 +0100 Subject: [PATCH 19/22] Change manifest output to JSON --- di/pqx/pqx.q | 30 +++++++++++++++++------------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index b9fa83c9..e0c8ed6e 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -24,17 +24,18 @@ default:( / define empty schema for manifest manifest:([] - file :`symbol$(); / path written - seq :`long$(); / sequence number within the partition - syms :(); / list of instruments in the file - nsyms :`long$(); / count of instruments - rows :`long$(); / row count - mintime :`timestamp$(); / min time across the file (for pruning) - maxtime :`timestamp$(); / max time across the file - estbytes :`long$(); / estimated size at plan time - bytes :`long$(); / actual on-disk size - split :`boolean$(); / true if this file is a chunk of a split oversized instrument - status :`symbol$() / `ok | `error + file :`symbol$(); / path written + seq :`long$(); / sequence number within the partition + virtualcols :(); / list of virtual columns + syms :(); / list of instruments in the file + nsyms :`long$(); / count of instruments + rows :`long$(); / row count + mintime :`timestamp$(); / min time across the file (for pruning) + maxtime :`timestamp$(); / max time across the file + estbytes :`long$(); / estimated size at plan time + bytes :`long$(); / actual on-disk size + split :`boolean$(); / true if this file is a chunk of a split oversized instrument + status :`symbol$() / `ok | `error ); checkandconvertcols:{[t] @@ -221,9 +222,10 @@ writefile:{[t;o;writeopt;writedir;map] .z.m.loginfo[`pqx;"seqNo ",string[map`seqno]," write successful"]; .z.m.logwarn[`pqx;"seqNo ",string[map`seqno]," write unsuccessful. Error - ",last res] ]; - :`file`seq`syms`nsyms`rows`mintime`maxtime`estbytes`bytes`split`status!/: flip ( + :`file`seq`virtualcols`syms`nsyms`rows`mintime`maxtime`estbytes`bytes`split`status!/: flip ( hsym `$path; map`seqno; + enlist o`virtualcols; enlist[syms]; count[syms]; count[i]; @@ -382,11 +384,13 @@ extract:{[t;tname;dt;o] / write down manifest to partition - best-effort, does not abort the extract call if it fails .z.m.loginfo[`pqx;"Writing down manifest file for table: ",string[tname],"; date: ",string dt]; - manwrite:.z.m.tryfn[set;(hsym `$writedir,"manifest";res)]; + manwrite:.z.m.tryfn[0:;(hsym `$writedir,"manifest.json";enlist .j.j res)]; if[not first manwrite; .z.m.logwarn[`pqx;"di.pqx: error writing manifest to disk: ",last manwrite] ]; + .z.m.loginfo[`pqx;"Extract complete table: ",string[tname]," data for date: ",string dt]; + / attach to global manifest and return stats for this extract manifest,:res; :res From 361466d0f4291782acc0248cc60d94b9f7cec5d0 Mon Sep 17 00:00:00 2001 From: ConorSwainDI Date: Tue, 1 Sep 2026 14:23:23 +0100 Subject: [PATCH 20/22] Docs changes --- di/pqx/pqx.md | 40 +++++++++++++++++++++++++++------------- di/pqx/pqx.q | 4 +++- di/pqx/test.csv | 23 +++++++++++++++++------ 3 files changed, 47 insertions(+), 20 deletions(-) diff --git a/di/pqx/pqx.md b/di/pqx/pqx.md index 7fdd2616..692e8c9b 100644 --- a/di/pqx/pqx.md +++ b/di/pqx/pqx.md @@ -106,9 +106,9 @@ When `virtualcols` is non-empty, it takes precedence over both `onesymperfile` a becomes exactly one output file, however large. `extract` throws (`` `di.pqx: not all virtualcols found in table ``) if any `virtualcols` column is not present in the input table. Output paths gain one `=/` segment per `virtualcols` column, Hive-style, e.g. -`//date=
/exchange=NASDAQ/-NNNNN.parquet` — this path is the only place -a file's combination is recorded; the manifest schema itself is unchanged. The `virtualcols` columns -themselves are dropped from the on-disk data before writing (their value is already fixed by the +`//date=
/exchange=NASDAQ/-NNNNN.parquet` — the file's combination is also +recorded directly in the manifest's `virtualcols` column (see Manifest Schema below). The `virtualcols` +columns themselves are dropped from the on-disk data before writing (their value is already fixed by the path, so keeping them in every row would just be redundant storage) — use `readfile` (see below) to read a file back with its `virtualcols` values (and the `date` partition) reattached as columns. @@ -124,6 +124,7 @@ shape, scoped to only the file(s) written by that call. |---|---|---| | `file` | symbol | Path written | | `seq` | long | Sequence number within the partition | +| `virtualcols` | symbol list | The `virtualcols` option value in effect for this call - the list of columns partitioned into this file's path, not the combination's values themselves (see Options) | | `syms` | symbol list | Instruments contained in the file | | `nsyms` | long | Count of instruments in the file | | `rows` | long | Row count | @@ -135,13 +136,20 @@ shape, scoped to only the file(s) written by that call. | `status` | symbol | `` `ok `` or `` `error `` | In addition to the in-memory `manifest`, each `extract` call writes this same per-call stats table -to a `manifest` sidecar file directly under the partition directory (i.e. -`//date=
/manifest`), serialized with `set`/readable back with `get`. A repeat -`extract` call into the same partition overwrites the sidecar with just that call's rows, rather -than accumulating across calls — the sidecar mirrors `extract`'s return value, not `getmanifest[]`. -Writing the sidecar is best-effort: if it fails (for example a permissions issue, or something else -already occupying that path) a warning is logged but `extract` still returns normally and still -updates the in-memory `manifest`. +to a `manifest.json` sidecar file directly under the partition directory (i.e. +`//date=
/manifest.json`), serialized to a single line of JSON with `.j.j` and +written with `0:`. A repeat `extract` call into the same partition overwrites the sidecar with just +that call's rows, rather than accumulating across calls — the sidecar mirrors `extract`'s return +value, not `getmanifest[]`. Writing the sidecar is best-effort: if it fails (for example a +permissions issue, or something else already occupying that path) a warning is logged but `extract` +still returns normally and still updates the in-memory `manifest`. + +Read it back with `` .j.k first read0 hsym `$"//date=
/manifest.json" ``. JSON has +no native date/timestamp/symbol type, so the round trip is not type-preserving: `file`, `syms`, +`status` and `virtualcols` come back as plain strings (cast back with `` `$ ``) and `mintime`/`maxtime` +come back as ISO-8601 strings rather than timestamps; numeric columns (`seq`, `nsyms`, `rows`, +`estbytes`, `bytes`) come back as floats rather than longs (cast back with `` "j"$ ``) and `split` +comes back as a native JSON boolean unchanged. --- @@ -289,9 +297,9 @@ res:pqx.extract[trade;`trade;2025.07.15;`targetsize`codec!(256*1024*1024;`gzip)] // res holds only the row(s) written by this call res -file seq syms nsyms rows mintime maxtime estbytes bytes split status --------------------------------------------------------------------------------------------------------------------------------------------------------------- -:./trade/date=2025.07.15/part-00001.parquet 1 `AAPL`MSFT 2 50000 2025.07.15D00:00:00.000000000 2025.07.15D23:59:59.000000000 1153433 1048576 0b ok +file seq virtualcols syms nsyms rows mintime maxtime estbytes bytes split status +------------------------------------------------------------------------------------------------------------------------------------------------------------------------ +:./trade/date=2025.07.15/part-00001.parquet 1 `symbol$() `AAPL`MSFT 2 50000 2025.07.15D00:00:00.000000000 2025.07.15D23:59:59.000000000 1153433 1048576 0b ok // getmanifest[] returns the full accumulated table across every extract call so far pqx.getmanifest[] @@ -320,6 +328,12 @@ right file(s). `castvirtualcol` is exercised directly for both the datecol and n including a datecol not literally named `` `date `` . A directory with no matching files is covered too: building the view over it succeeds, but querying it then fails. +It also covers the `manifest.json` sidecar directly — parsing it back with `.j.k`/`read0` and casting +its columns against `extract`'s returned per-call stats table (see Manifest Schema), that a repeat +`extract` call into the same partition overwrites rather than accumulates the sidecar, and that a +sidecar write failure (something else already occupying that path) still lets `extract` return +normally and update the in-memory `manifest`. + --- ## Notes diff --git a/di/pqx/pqx.q b/di/pqx/pqx.q index e0c8ed6e..04873175 100644 --- a/di/pqx/pqx.q +++ b/di/pqx/pqx.q @@ -301,7 +301,9 @@ extract:{[t;tname;dt;o] opts:default,o; / if one sym per file requested, turn off splitoversized - if[o`onesymperfile; + / read from opts (always has the key, via default), not o - indexing a missing key on o throws + / or is spuriously truthy depending on the types of whatever other keys the caller did pass + if[opts`onesymperfile; .z.m.loginfo[`pqx;"onesymperfile requested, turning off splitoversized"]; opts[`splitoversized]:0b ]; diff --git a/di/pqx/test.csv b/di/pqx/test.csv index e6ced831..40dace42 100644 --- a/di/pqx/test.csv +++ b/di/pqx/test.csv @@ -80,17 +80,28 @@ true,0,0,q,`error~first exec status from pqx.getmanifest[] where file like "*pqx true,0,0,q,0~first exec bytes from pqx.getmanifest[] where file like "*pqxoutf5*",1,,The failed file has zero on-disk bytes recorded run,0,0,q,pqxret10:pqx.extract[pqxbasic;`pqxtrade;2025.07.26;enlist[`outdir]!enlist `:pqxout10/],1,,Extract a table to exercise the on-disk manifest sidecar -true,0,0,q,"not ()~key hsym `$""pqxout10/pqxtrade/date=2025.07.26/manifest""",1,,extract writes a manifest sidecar file into the partition directory -true,0,0,q,"pqxret10~get hsym `$""pqxout10/pqxtrade/date=2025.07.26/manifest""",1,,The on-disk manifest sidecar matches extract's returned per-call stats table +true,0,0,q,"not ()~key hsym `$""pqxout10/pqxtrade/date=2025.07.26/manifest.json""",1,,extract writes a manifest.json sidecar file into the partition directory +run,0,0,q,pqxsidecar10:.j.k first read0 hsym `$"pqxout10/pqxtrade/date=2025.07.26/manifest.json",1,,Parse back the JSON manifest sidecar - numeric columns come back as floats and symbol/date/timestamp columns as strings +true,0,0,q,1~count pqxsidecar10,1,,The sidecar has the same row count as extract's returned per-call stats table +true,0,0,q,("j"$exec seq from pqxsidecar10)~exec seq from pqxret10,1,,seq round-trips through JSON once cast back from float to long +true,0,0,q,("j"$exec nsyms from pqxsidecar10)~exec nsyms from pqxret10,1,,nsyms round-trips through JSON once cast back from float to long +true,0,0,q,("j"$exec rows from pqxsidecar10)~exec rows from pqxret10,1,,rows round-trips through JSON once cast back from float to long +true,0,0,q,("j"$exec bytes from pqxsidecar10)~exec bytes from pqxret10,1,,bytes round-trips through JSON once cast back from float to long +true,0,0,q,(exec split from pqxsidecar10)~exec split from pqxret10,1,,split is a native JSON boolean and round-trips without casting +true,0,0,q,(`$exec status from pqxsidecar10)~exec status from pqxret10,1,,status round-trips through JSON once cast back from string to symbol +true,0,0,q,(`$exec file from pqxsidecar10)~exec file from pqxret10,1,,file round-trips through JSON once cast back from string to symbol +true,0,0,q,(`$first exec syms from pqxsidecar10)~first exec syms from pqxret10,1,,syms round-trips through JSON once each instrument name is cast back from string to symbol +true,0,0,q,0=count first exec virtualcols from pqxsidecar10,1,,virtualcols is empty in the sidecar for a call made without virtualcols set run,0,0,q,pqxret11a:pqx.extract[pqxbasic;`pqxtrade;2025.07.27;enlist[`outdir]!enlist `:pqxout11/],1,,First extract call into a partition that will be written to twice run,0,0,q,pqxret11b:pqx.extract[pqxsingle;`pqxtrade;2025.07.27;enlist[`outdir]!enlist `:pqxout11/],1,,Second extract call into the same partition with different data -true,0,0,q,"pqxret11b~get hsym `$""pqxout11/pqxtrade/date=2025.07.27/manifest""",1,,The manifest sidecar reflects only the most recent call's stats -true,0,0,q,"not pqxret11a~get hsym `$""pqxout11/pqxtrade/date=2025.07.27/manifest""",1,,Repeat extract calls into the same partition overwrite the sidecar rather than appending to it +run,0,0,q,pqxsidecar11:.j.k first read0 hsym `$"pqxout11/pqxtrade/date=2025.07.27/manifest.json",1,,Parse back the JSON manifest sidecar after the second call into the same partition +true,0,0,q,("j"$exec rows from pqxsidecar11)~exec rows from pqxret11b,1,,The manifest.json sidecar reflects only the most recent call's stats +true,0,0,q,not ("j"$exec rows from pqxsidecar11)~exec rows from pqxret11a,1,,Repeat extract calls into the same partition overwrite the sidecar rather than appending to it -run,0,0,q,system "mkdir -p pqxoutf6/pqxtrade/date=2025.07.28/manifest",1,,Pre-create a directory at the manifest sidecar's path so the sidecar write itself fails +run,0,0,q,system "mkdir -p pqxoutf6/pqxtrade/date=2025.07.28/manifest.json",1,,Pre-create a directory at the manifest.json sidecar's path so the sidecar write itself fails run,0,0,q,manifestcountbeforef6:count pqx.getmanifest[],1,,Snapshot the accumulated manifest row count before the manifest-write-failure case -run,0,0,q,pqxretf6:pqx.extract[pqxbasic;`pqxtrade;2025.07.28;enlist[`outdir]!enlist `:pqxoutf6/],1,,Extract still succeeds even though writing its manifest sidecar fails +run,0,0,q,pqxretf6:pqx.extract[pqxbasic;`pqxtrade;2025.07.28;enlist[`outdir]!enlist `:pqxoutf6/],1,,Extract still succeeds even though writing its manifest.json sidecar fails true,0,0,q,9~exec sum rows from pqxretf6,1,,extract still returns full per-file stats despite the sidecar write failure true,0,0,q,all `ok=exec status from pqxretf6,1,,Every data file write still succeeds despite the later sidecar write failure true,0,0,q,(count pqx.getmanifest[])>manifestcountbeforef6,1,,extract still appends to the accumulated in-memory manifest despite the sidecar write failure From 64f395c80494de0e1e3e18d9085b532cd9a28216 Mon Sep 17 00:00:00 2001 From: ConorSwainDI Date: Tue, 1 Sep 2026 14:37:23 +0100 Subject: [PATCH 21/22] Docs changes --- di/pqx/test.csv | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/di/pqx/test.csv b/di/pqx/test.csv index 40dace42..395d65c6 100644 --- a/di/pqx/test.csv +++ b/di/pqx/test.csv @@ -99,6 +99,16 @@ run,0,0,q,pqxsidecar11:.j.k first read0 hsym `$"pqxout11/pqxtrade/date=2025.07.2 true,0,0,q,("j"$exec rows from pqxsidecar11)~exec rows from pqxret11b,1,,The manifest.json sidecar reflects only the most recent call's stats true,0,0,q,not ("j"$exec rows from pqxsidecar11)~exec rows from pqxret11a,1,,Repeat extract calls into the same partition overwrite the sidecar rather than appending to it +run,0,0,q,pqxret17:pqx.extract[pqxoversized;`pqxtrade;2025.08.06;`targetsize`maxfactor`calibrate`compressionratio`outdir`splitoversized!(1000;1.1;0b;0.30;`:pqxout17/;1b)],1,,Extract an oversized instrument split across multiple files to exercise a multi-row manifest sidecar +true,0,0,q,1 Date: Tue, 1 Sep 2026 14:43:09 +0100 Subject: [PATCH 22/22] Add single row assumption --- di/pqx/test.csv | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/di/pqx/test.csv b/di/pqx/test.csv index 395d65c6..bb655c1c 100644 --- a/di/pqx/test.csv +++ b/di/pqx/test.csv @@ -90,7 +90,7 @@ true,0,0,q,("j"$exec bytes from pqxsidecar10)~exec bytes from pqxret10,1,,bytes true,0,0,q,(exec split from pqxsidecar10)~exec split from pqxret10,1,,split is a native JSON boolean and round-trips without casting true,0,0,q,(`$exec status from pqxsidecar10)~exec status from pqxret10,1,,status round-trips through JSON once cast back from string to symbol true,0,0,q,(`$exec file from pqxsidecar10)~exec file from pqxret10,1,,file round-trips through JSON once cast back from string to symbol -true,0,0,q,(`$first exec syms from pqxsidecar10)~first exec syms from pqxret10,1,,syms round-trips through JSON once each instrument name is cast back from string to symbol +true,0,0,q,(`$first exec syms from pqxsidecar10)~first exec syms from pqxret10,1,,syms round-trips through JSON once each instrument name is cast back from string to symbol (single row case) true,0,0,q,0=count first exec virtualcols from pqxsidecar10,1,,virtualcols is empty in the sidecar for a call made without virtualcols set run,0,0,q,pqxret11a:pqx.extract[pqxbasic;`pqxtrade;2025.07.27;enlist[`outdir]!enlist `:pqxout11/],1,,First extract call into a partition that will be written to twice