From df78787a4554b64a05ecd40c0763ebeb874a176b Mon Sep 17 00:00:00 2001 From: ConorSwainDI Date: Fri, 24 Jul 2026 11:15:08 +0100 Subject: [PATCH 01/12] 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/12] 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/12] 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/12] 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/12] 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/12] 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/12] 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/12] 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/12] 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/12] 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/12] 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/12] 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[] ``` ---