diff --git a/di/pqx/init.q b/di/pqx/init.q new file mode 100644 index 00000000..ace5fd8b --- /dev/null +++ b/di/pqx/init.q @@ -0,0 +1,7 @@ +/ KDB-X Parquet extract module to save kdb+ data to parquet storage convention + +arrow:use`kx.arrow + +\l ::pqx.q + +export:([init;extract;getmanifest;checkandconvertcols;estimate;plan;writefile;tryfn]) diff --git a/di/pqx/pqx.md b/di/pqx/pqx.md new file mode 100644 index 00000000..7bc6a627 --- /dev/null +++ b/di/pqx/pqx.md @@ -0,0 +1,219 @@ +# 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` also returns this same information for the files it just wrote. + +--- + +## 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 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 +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 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. + +--- + +## Manifest Schema + +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 | +|---|---|---| +| `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 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 +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` 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 | +|---|---|---| +| `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 +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 + +// getmanifest[] returns the full accumulated table across every extract call so far +pqx.getmanifest[] +``` + +--- + +## 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 `getmanifest[]` rows and (via +`` .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). + +--- + +## 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. +- `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`. +- 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 new file mode 100644 index 00000000..99716707 --- /dev/null +++ b/di/pqx/pqx.q @@ -0,0 +1,285 @@ +/ 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 + ); + +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 + ] + }; + +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 + 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]) + }; + +calibrateratio:{[t;o;writeopt] + / writes a sample of data to disk and reads its size on disk + / 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]; + 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"]; + hdel hsym `$testloc; + + / return new ratio + .z.m.loginfo[`pqx;"Returning calibrated compression ratio"]; + :newratio + }; + +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;enlist syms);0b;()]`estbyt];count seqno] + }; + +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:(); + + / 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 + }; + +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; + }; diff --git a/di/pqx/test.csv b/di/pqx/test.csv new file mode 100644 index 00000000..abf5a678 --- /dev/null +++ b/di/pqx/test.csv @@ -0,0 +1,81 @@ +action,ms,bytes,lang,code,repeat,minver,comment +before,0,0,q,pqx:use`di.pqx,1,,Load module +before,0,0,q,logdep:`info`warn`error!(3#{[c;m] }),1,,No-op log dependency for tests +before,0,0,q,pqx.init[enlist[`log]!enlist logdep],1,,Wire the required log dependency + +before,0,0,q,pqxbasic:([]sym:`AAPL`MSFT`GOOG`AAPL`MSFT`GOOG`AAPL`MSFT`GOOG;time:2025.07.15D09:30:00.000000000+1000000000*til 9;price:100.0 200.0 300.0 101.0 201.0 301.0 102.0 202.0 302.0;size:10 20 30 40 50 60 70 80 90),1,,Deterministic multi-sym table for general checks +before,0,0,q,pqxunsorted:([]sym:`B`A`B`A`A;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,,Two-sym table with times out of order for presort checks +before,0,0,q,pqxempty:0#pqxbasic,1,,Zero-row table for the empty-table edge case +before,0,0,q,pqxoversized:([]sym:20000#`AAPL;time:2025.07.15D00:00:00.000000000+til 20000;price:20000?100.0),1,,Single-instrument table sized to exceed a small target file size +before,0,0,q,pqxnotime:([]sym:5#`AAPL;price:1.0 2.0 3.0 4.0 5.0),1,,Table missing the time column +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 pqx.getmanifest[],1,,Manifest is a table before anything is written + +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 +true,0,0,q,3~exec sum nsyms from pqx.getmanifest[] where file like "*pqxout1*",1,,All three instruments are accounted for +true,0,0,q,all `ok=exec status from pqx.getmanifest[] where file like "*pqxout1*",1,,Every written file reports ok status +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)