From 880473e3b75ef1a6cc8420ac9beb0a1999e9f551 Mon Sep 17 00:00:00 2001 From: Olly99999 Date: Mon, 6 Jul 2026 17:39:14 +0100 Subject: [PATCH 1/4] Initial Di.merge --- di/merge/deps.q | 5 ++ di/merge/init.q | 8 ++ di/merge/merge.md | 148 ++++++++++++++++++++++++++++++++ di/merge/merge.q | 209 ++++++++++++++++++++++++++++++++++++++++++++++ di/merge/test.csv | 137 ++++++++++++++++++++++++++++++ 5 files changed, 507 insertions(+) create mode 100644 di/merge/deps.q create mode 100644 di/merge/init.q create mode 100644 di/merge/merge.md create mode 100644 di/merge/merge.q create mode 100644 di/merge/test.csv diff --git a/di/merge/deps.q b/di/merge/deps.q new file mode 100644 index 00000000..a7fb14c9 --- /dev/null +++ b/di/merge/deps.q @@ -0,0 +1,5 @@ +/ hard module dependencies and their minimum versions, validated by di.depcheck +/ di.merge has no hard dependencies - its only runtime dependency (log) is injected via +/ init as a dictionary of functions, and the parted column(s) needed for a merge are +/ passed in by the caller (see di.sort) rather than read from sort config +deps:(`$())!(); diff --git a/di/merge/init.q b/di/merge/init.q new file mode 100644 index 00000000..f6726307 --- /dev/null +++ b/di/merge/init.q @@ -0,0 +1,8 @@ +/ load core functionality into the module +\l ::merge.q + +/ module version - compared against dependants' minimum requirements by di.depcheck. Removed for now until PR merged +/version:"0.1.0"; + +/ public api - only the functions intended to be called externally are exported +export:([init;getpartchunks;mergebypart;mergebycol;mergehybrid;trackpartition;clearpartsizes;getpartsizes;checkpartitiontype;checkenumerabletype;getextrapartitions;getfirstcharpartitions]) diff --git a/di/merge/merge.md b/di/merge/merge.md new file mode 100644 index 00000000..9fa12156 --- /dev/null +++ b/di/merge/merge.md @@ -0,0 +1,148 @@ +# di.merge + +On-disk data-merging utilities for the write-down flow, extracted from TorQ's `code/common/merge.q`. During intraday write-down a process (typically a WDB) writes data to disk in temporary *partition segments* to keep memory flat; at end-of-day those segments are merged into the final HDB partition. `di.merge` performs that merge - whole-partition, column-by-column, or a hybrid of the two chosen per partition by a configurable row-count / byte-size limit - and tracks the size of each segment as it is written to drive that decision. + +--- + +## Features + +- Merge on-disk partition segments into a destination partition without holding the whole partition in memory: whole-partition (`mergebypart`), column-by-column (`mergebycol`), or a per-partition hybrid (`mergehybrid`) +- Track each segment's row count and byte-size estimate as it is written (`trackpartition`) and use it to size merge batches (`getpartchunks`) and pick the merge method +- Re-sort a segment by its parted column(s) when the `p#` attribute cannot otherwise be applied, so the merged partition ends up correctly parted +- Batch sizing bounded by both a data limit (rows or bytes, per `mergebybytelimit`) and a maximum partition count per batch (`partlimit`) +- Helpers to validate and enumerate a table's parted columns (`checkpartitiontype`, `checkenumerabletype`, `getextrapartitions`, `getfirstcharpartitions`) +- No hard module dependencies - logging is injected via `init`, and the parted column(s) a merge needs are passed in by the caller rather than read from sort config, so `di.merge` stays independent of `di.sort` + +--- + +## Dependencies + +| Dependency | Key | Required | Description | +|---|---|---|---| +| logger | `` `log `` | yes | `info`/`warn`/`error` - binary `{[c;m]}` where `c` is a symbol context and `m` is a string. A `kx.log` instance is accepted directly and auto-wrapped | + +**Hard dependency:** none - the only runtime dependency (`log`) is injected via `init`, so any module exporting the contracted signatures can be supplied. + +The parted (`` `p# ``) column(s) for a table - `extrapartitiontype` in the function signatures below - are **passed in by the caller**. TorQ's original `merge.q` read these from the `.sort.params` global; on extraction that coupling was removed so `di.merge` does not depend on `di.sort`. The caller (e.g. `di.wdb`) obtains the parted columns from its sort configuration and passes them to `mergebypart` / `mergehybrid`. + +**Logging contract.** Internally the module calls the logger as binary `.z.m.log[\`info][\`merge;"msg"]` (`{[c;m]}` - context symbol + message). You may pass either a `kx.log` instance (its monadic `{[msg]}` functions are detected and auto-wrapped, folding the context tag into the message as `"merge: ..."`) or a custom `` `info`warn`error `` dict of `{[c;m]}` functions, used as-is. + +--- + +## Initialisation + +`init[deps]` takes a single dictionary combining the injected `log` dependency with any configuration overrides. Config keys are optional - omit any and the module falls back to the default; unrecognised keys are ignored. `init` throws immediately if `deps` is not a dictionary or `log` is missing or malformed. + +| Key | Default | Description | +|---|---|---| +| `` `mergebybytelimit `` | `0b` | `0b` sizes batches / picks the merge method by row count, `1b` by byte-size estimate | +| `` `partlimit `` | `1000` | maximum number of partitions merged together in a single batch | +| `` `log `` | *(required)* | logger providing `info`/`warn`/`error` (a `kx.log` instance is auto-wrapped) | + +```q +merge:use`di.merge +merge.init[enlist[`log]!enlist kxlog] +/ or with config overrides: +merge.init[`mergebybytelimit`partlimit`log!(1b;500;kxlog)] +``` + +--- + +## Exported Functions + +### `init[deps]` +Wire config + the injected `log` dependency (one dict). Must be called before anything else. + +### `trackpartition[ptdir;rowcount;bytes]` +Accumulate the row count and byte-size estimate for a freshly-written segment, keyed by partition directory. Call each time data is written to a segment. +```q +merge.trackpartition[`:tmp/2025.01.01/trade; count data; -22!data] +``` + +### `getpartsizes[]` +Return the current tracked segment sizes (keyed on `ptdir`), e.g. to sync to sort-worker processes. + +### `clearpartsizes[]` +Drop all tracked segment sizes. Call once end-of-day merging is complete. + +### `getpartchunks[partdirs;mergelimit]` +Split `partdirs` into batches to be merged together, each within `mergelimit` (rows or bytes per `mergebybytelimit`) and no larger than `partlimit` partitions. Returns a list of batches (each a list of partition directories). +```q +merge.getpartchunks[partdirs; 1000000] +``` + +### `mergebypart[extrapartitiontype;dest;partchunks]` +Merge one batch of whole partition segments into destination partition `dest`, re-sorting by the parted column(s) `extrapartitiontype` first if the `p#` attribute cannot otherwise be applied. Typically iterated over the output of `getpartchunks`. +```q +merge.mergebypart[`sym; ` sv dest,`] each merge.getpartchunks[partdirs;lim] +``` + +### `mergebycol[tableinfo;dest;segment]` +Merge one `segment` into `dest` a column at a time, holding at most a single column in memory. `tableinfo` is `(tablename;schema)`; the schema supplies the column list. +```q +merge.mergebycol[(`trade;schema); dest] each partdirs +``` + +### `mergehybrid[extrapartitiontype;tableinfo;dest;partdirs;mergelimit]` +Merge `partdirs` using whichever method fits each: whole-partition for those within `mergelimit`, column-by-column for any single partition over it (creating the `.d` file if column-by-column merging produced none). + +### `checkpartitiontype[tablename;extrapartitiontype]` +Error-log any parted column named for the table that is absent from it. + +### `checkenumerabletype[tablename;extrapartitiontype]` +Confirm every parted column has an enumerable type (`h`/`i`/`j`/`s`) so it can key a partition; error-logs otherwise. + +### `getextrapartitions[tablename;extrapartitiontype]` +Return the distinct combinations of the parted column values - one per partition directory. + +### `getfirstcharpartitions[tablename;extrapartitiontype]` +Return the partition values grouped by the first character of the (single) parted column. + +--- + +## Partition-size store schema + +`getpartsizes[]` returns the store, keyed on `ptdir`: + +| Column | Type | Description | +|---|---|---| +| ptdir | `symbol` | partition directory of the segment | +| rowcount | `long` | accumulated row count written to that segment | +| bytes | `long` | accumulated byte-size estimate of that segment | + +--- + +## Usage Example + +```q +kxlog:use`kx.log +merge:use`di.merge +merge.init[enlist[`log]!enlist kxlog.createLog[]] + +/ as each segment is written down, record its size +merge.trackpartition[seg; count data; -22!data] + +/ at end-of-day, merge the segments for a table into its hdb partition +partdirs:merge.getpartsizes[][`ptdir] +merge.mergehybrid[`sym; (`trade;schema); dest; partdirs; 1000000] +merge.clearpartsizes[] +``` + +--- + +## Running Tests + +```q +k4unit:use`di.k4unit +k4unit.moduletest`di.merge +``` + +The test suite injects a no-op binary mock logger (and a capturing logger for the log-path assertions). It covers: dependency validation (non-dict deps / missing / non-dict / incomplete `log` all throw, with the `di.merge` error prefix); config application via row-count vs byte-size batching; `trackpartition`/`getpartsizes`/`clearpartsizes`; the `getpartchunks` batching and `partlimit` splitting logic; and end-to-end `mergebypart`, `mergebycol` and `mergehybrid` against real on-disk segments written to a scratch directory (cleaned up afterwards). + +--- + +## Notes + +- Extracted from TorQ `code/common/merge.q`. The only behavioural change on extraction is the removal of the `.sort.params` lookup (`getextrapartitiontype`): the parted column(s) are now a parameter (`extrapartitiontype`) supplied by the caller, keeping the module standalone. +- `init` must be called before any other function (it wires the logger). +- The merge functions operate on on-disk paths (`get`/`set`/`upsert` against file symbols); they do not manage the segment or destination directory lifecycle - that remains the caller's responsibility. diff --git a/di/merge/merge.q b/di/merge/merge.q new file mode 100644 index 00000000..75ddbb2a --- /dev/null +++ b/di/merge/merge.q @@ -0,0 +1,209 @@ +/ merge module for kdb-x +/ on-disk data-merging utilities for the write-down (wdb) flow: intraday data is +/ written to disk in temporary partition segments, then merged into the final hdb +/ partition - either whole-partition, column-by-column, or a hybrid of the two chosen +/ per partition by a row-count / byte-size limit +/ merging keeps memory flat - segments are read and upserted a batch at a time rather +/ than held in memory and written once at end-of-day +/ the parted (`p#) column(s) for a table are supplied by the caller as extrapartitiontype; +/ the module never reads sort config itself, so it stays independent of di.sort +/ config and dependencies are passed to init in a single dictionary: config keys (see +/ merge.md) are optional with defaults; log is required and init errors immediately if it +/ is missing - see merge.md +/ module-local state convention: read config bare, mutate via .z.m, and access injected +/ dependencies via .z.m at every call site + +/ ============================================================ +/ module state and defaults +/ ============================================================ + +/ row count and byte-size estimate of each on-disk segment as it is written, keyed by +/ partition directory - drives the batch sizing and merge-method decisions below +partsizes:([ptdir:`symbol$()] rowcount:`long$(); bytes:`long$()); + +/ configuration defaults - overridden by the config dictionary passed to init +mergebybytelimit:0b; / 0b = size batches / choose method by row count, 1b = by byte-size estimate +partlimit:1000; / maximum number of partitions merged together in a single batch + +/ ============================================================ +/ internal helpers +/ ============================================================ + +/ normalise an injected logger to the binary {[c;m]} contract +normlog:{[logdict] + / detect a kx.log instance by its marker keys and wrap its monadic level functions into + / binary {[c;m]}, folding the context tag into the message; a plain {[c;m]} dict passes through + $[any `getlvl`sinks`fmts in key logdict; + `info`warn`error!( + {[fn;c;m] fn[string[c],": ",m]}[logdict`info;]; + {[fn;c;m] fn[string[c],": ",m]}[logdict`warn;]; + {[fn;c;m] fn[string[c],": ",m]}[logdict`error;]); + logdict] + }; + +/ validate and wire the injected dependencies from the deps dict +setdeps:{[deps] + / log is the only dependency - required, and must provide info/warn/error (a kx.log + / instance is auto-wrapped by normlog); validate before writing any module state + if[99h<>type deps; + '"di.merge: deps must be a dictionary of config and injected dependencies - see merge.md"]; + if[not `log in key deps; + '"di.merge: log dependency is required; pass `info`warn`error (or a kx.log logger) keyed on `log"]; + if[99h<>type deps`log; + '"di.merge: log must be a dict of info/warn/error functions (or a kx.log logger)"]; + lg:normlog deps`log; + if[not all `info`warn`error in key lg; + '"di.merge: log must provide info/warn/error; got: ",", " sv string key lg]; + .z.m.log:lg; + }; + +/ apply recognised config overrides from the deps dict, defaulting where a key is absent +setconfig:{[deps] + cfg:$[99h=type deps;deps;()!()]; + .z.m.mergebybytelimit:$[`mergebybytelimit in key cfg;cfg`mergebybytelimit;0b]; + .z.m.partlimit:$[`partlimit in key cfg;cfg`partlimit;1000]; + }; + +/ merge a single column from a segment into the destination partition, logging on failure +mergeonecol:{[dest;segment;col] + / filepaths to the destination column and the matching column in the segment + destcol:` sv dest,col; + destdata:get segcol:` sv segment,col; + .z.m.log[`info][`merge;"merging ",(string segcol)," to ",string destcol]; + .[upsert;(destcol;destdata); + {[dc;e] .z.m.log[`error][`merge;"failed to save data to ",(string dc)," with error : ",e]}[destcol;]]; + }; + +/ ============================================================ +/ public api - partition-size tracking +/ ============================================================ + +/ accumulate the row count and byte-size estimate for a freshly-written segment +trackpartition:{[ptdir;rowcount;bytes] + / call each time data is written to a segment; keyed by partition directory + .z.m.partsizes[ptdir]+:(rowcount;bytes); + }; + +/ drop all tracked segment sizes - call once end-of-day merging is complete +clearpartsizes:{ + .z.m.partsizes:0#partsizes; + }; + +getpartsizes:{partsizes}; / current tracked segment sizes (e.g. to sync to sort-worker processes) + +/ ============================================================ +/ public api - merging +/ ============================================================ + +/ split the partition directories into batches to be merged together +getpartchunks:{[partdirs;mergelimit] + / each batch stays within mergelimit (row count or byte estimate per mergebybytelimit) and + / holds no more than partlimit partitions + / tracked sizes for just the partitions we are merging + t:select from partsizes where ptdir in partdirs; + / the measure we accumulate against the limit + r:$[mergebybytelimit;exec bytes from t;exec rowcount from t]; + / running-total scan resets to the current partition whenever adding it would breach the limit + l:(where r={$[ztype chunks;chunks:(,/)chunks]; + .z.m.log[`info][`resort;"checking that the contents of this subpartition conform"]; + / can the p# attribute be applied as-is? if not, the data must be re-sorted by the parted column + pattrtest:@[{@[x;y;`p#];0b}[chunks;];extrapartitiontype;{1b}]; + if[pattrtest; + .z.m.log[`info][`resort;"re-sorting contents of subpartition"]; + chunks:xasc[extrapartitiontype;chunks]; + .z.m.log[`info][`resort;"the p attribute can now be applied"]; + ]; + .z.m.log[`info][`merge;"upserting ",(string count chunks)," rows to ",string dest]; + / append the merged rows to permanent storage, logging (not throwing) on failure + .[upsert;(dest;chunks); + {[e;d;p] .z.m.log[`error][`merge;"failed to merge to ",string[d]," from segments ",(", " sv string p)," Error is - ",string e]}[;dest;partchunks]]; + }; + +/ merge one segment into the destination partition a column at a time +mergebycol:{[tableinfo;dest;segment] + / holds at most a single column in memory rather than the whole partition + .z.m.log[`info][`merge;"upserting columns from ",(string segment)," to ",string dest]; + mergeonecol[dest;segment;] each cols tableinfo[1]; + }; + +/ merge the given partitions using whichever method fits each one +mergehybrid:{[extrapartitiontype;tableinfo;dest;partdirs;mergelimit] + / whole-partition for those within the limit, column-by-column for any single partition over it + overlimit:$[mergebybytelimit; + exec ptdir from partsizes where ptdir in partdirs,bytes>mergelimit; + exec ptdir from partsizes where ptdir in partdirs,rowcount>mergelimit]; + if[(count overlimit)<>count partdirs; + partdirs:partdirs except overlimit; + .z.m.log[`info][`merge;"merging ",(", " sv string partdirs)," by whole partition"]; + mergebypart[extrapartitiontype;` sv dest,`]'[getpartchunks[partdirs;mergelimit]]; + ]; + if[0<>count overlimit; + .z.m.log[`info][`merge;"merging ",(", " sv string overlimit)," column by column"]; + mergebycol[tableinfo;dest]'[overlimit]; + / column-by-column merge writes no .d file - create one if none exists yet + if[()~key ` sv dest,`.d; + .z.m.log[`info][`merge;"creating file ",string ` sv dest,`.d]; + (` sv dest,`.d) set cols tableinfo[1]; + ]; + ]; + }; + +/ ============================================================ +/ public api - parted-column checks and partition enumeration +/ ============================================================ + +/ error-log any parted column named for the table that is absent from it +checkpartitiontype:{[tablename;extrapartitiontype] + $[count colsnotintab:extrapartitiontype where not extrapartitiontype in cols get tablename; + .z.m.log[`error][`checkpart;"parted columns ",(", " sv string colsnotintab)," are defined in sort.csv but not present in ",(string tablename)," table"]; + .z.m.log[`info][`checkpart;"all parted columns defined in sort.csv are present in ",(string tablename)," table"]]; + }; + +/ confirm every parted column has an enumerable type (h/i/j/s) so it can key a partition +checkenumerabletype:{[tablename;extrapartitiontype] + $[all extrapartitiontype in exec c from meta[tablename] where t in "hijs"; + .z.m.log[`info][`checkenumerable;"all columns do have an enumerable type in ",(string tablename)," table"]; + .z.m.log[`error][`checkenumerable;"not all columns ",string[extrapartitiontype]," do have an enumerable type in ",(string tablename)," table"]]; + }; + +/ distinct combinations of the parted column values - one per partition directory +getextrapartitions:{[tablename;extrapartitiontype] + / functional form of: select distinct extrapartitiontype from tablename + value each ?[tablename;();1b;extrapartitiontype!extrapartitiontype] + }; + +/ partition values grouped by the first character of the (single) parted column +getfirstcharpartitions:{[tablename;extrapartitiontype] + raze each value (?[tablename;();();(distinct;first extrapartitiontype)]) group ?[tablename;();();({first each string x};(distinct;first extrapartitiontype))] + }; + +/ ============================================================ +/ initialisation +/ ============================================================ + +init:{[deps] + / initialise from a single dictionary holding config overrides and the injected log dependency + / config keys are optional and fall back to defaults: + / `mergebybytelimit - 0b (row count) or 1b (byte-size estimate) - default 0b + / `partlimit - max partitions merged together in a batch - default 1000 + / dependency is required: + / `log - a logger providing info/warn/error (a kx.log instance is auto-wrapped) + / example: + / merge.init[enlist[`log]!enlist kxlog] + setdeps deps; + setconfig deps; + .z.m.log[`info][`merge;"di.merge initialised"]; + }; diff --git a/di/merge/test.csv b/di/merge/test.csv new file mode 100644 index 00000000..19043ced --- /dev/null +++ b/di/merge/test.csv @@ -0,0 +1,137 @@ +action,ms,bytes,lang,code,repeat,minver,comment +comment,,,,,,,Setup - load module, inject a silent mock logger and a capturing logger, create a scratch dir +before,0,0,q,merge:use`di.merge,1,1,load the merge module +before,0,0,q,os:use`di.os,1,1,load os for portable directory management in the on-disk tests +before,0,0,q,mylog:`info`warn`error!({[c;m]};{[c;m]};{[c;m]}),1,1,silent binary log mock - {[c;m]} is the internal contract +before,0,0,q,"caplog:([] fn:`symbol$();ctx:`symbol$();msg:())",1,1,capture table for the log-path assertions +before,0,0,q,logcap:`info`warn`error!({[c;m]`caplog upsert(`info;c;m)};{[c;m]`caplog upsert(`warn;c;m)};{[c;m]`caplog upsert(`error;c;m)}),1,1,capturing binary logger recording level, context and message +before,0,0,q,merge.init[enlist[`log]!enlist mylog],1,1,initialise with the silent logger +before,0,0,q,.test.root:`:mergetesttmp,1,1,scratch directory root for the on-disk merges +before,0,0,q,{if[os.isdir x;os.deldir x]} .test.root,1,1,remove any leftover scratch dir from a previous run +before,0,0,q,os.mkdir .test.root,1,1,create the scratch dir +before,0,0,q,".test.splay:{[d;t] (` sv d,`) set t}",1,1,helper to write a table as a splayed partition directory +before,0,0,q,".test.slash:{` sv x,`}",1,1,helper to append the trailing slash mergebypart expects on dest + +comment,,,,,,,init - dependency injection validation (log is required) +fail,0,0,q,merge.init[(::)],1,1,init rejects :: as deps +fail,0,0,q,merge.init[42],1,1,init rejects a non-dict deps value +fail,0,0,q,merge.init[()!()],1,1,init rejects empty dict (log missing) +fail,0,0,q,merge.init[enlist[`other]!enlist mylog],1,1,init rejects a dict missing the log key +fail,0,0,q,merge.init[enlist[`log]!enlist 42],1,1,init rejects a non-dict log value +fail,0,0,q,merge.init[enlist[`log]!enlist(`info`warn!(mylog`info;mylog`warn))],1,1,init rejects a log dict missing the error level +true,0,0,q,"(8#@[merge.init;(::);{x}])~""di.merge""",1,1,error message is prefixed di.merge +run,0,0,q,merge.init[enlist[`log]!enlist mylog],1,1,re-init with a valid log dep + +comment,,,,,,,trackpartition accumulates row count and bytes per partition directory +run,0,0,q,merge.clearpartsizes[],1,1,start from an empty store +true,0,0,q,0=count merge.getpartsizes[],1,1,store is empty +run,0,0,q,merge.trackpartition[`:t1;5;500],1,1,record a segment write +run,0,0,q,merge.trackpartition[`:t1;3;100],1,1,record a second write to the same segment +true,0,0,q,8=first exec rowcount from merge.getpartsizes[] where ptdir=`:t1,1,1,row counts accumulate +true,0,0,q,600=first exec bytes from merge.getpartsizes[] where ptdir=`:t1,1,1,byte estimates accumulate +run,0,0,q,merge.clearpartsizes[],1,1,clear the store +true,0,0,q,0=count merge.getpartsizes[],1,1,clearpartsizes empties the store + +comment,,,,,,,getpartchunks batches partitions up to the row-count limit +run,0,0,q,merge.clearpartsizes[],1,1,reset the store +run,0,0,q,merge.trackpartition[`:d1;100;1000],1,1,track partition d1 +run,0,0,q,merge.trackpartition[`:d2;100;1000],1,1,track partition d2 +run,0,0,q,merge.trackpartition[`:d3;100;1000],1,1,track partition d3 +true,0,0,q,merge.getpartchunks[`:d1`:d2`:d3;250]~(`:d1`:d2;enlist`:d3),1,1,d1+d2 fit one batch of 250 rows, d3 starts a new one + +comment,,,,,,,getpartchunks batches by byte estimate when mergebybytelimit is set +run,0,0,q,merge.init[`mergebybytelimit`log!(1b;mylog)],1,1,switch to byte-size batching +run,0,0,q,merge.clearpartsizes[],1,1,reset the store +run,0,0,q,merge.trackpartition[`:b1;1;100],1,1,track b1 with a small row count but 100 bytes +run,0,0,q,merge.trackpartition[`:b2;1;100],1,1,track b2 +run,0,0,q,merge.trackpartition[`:b3;1;100],1,1,track b3 +true,0,0,q,merge.getpartchunks[`:b1`:b2`:b3;250]~(`:b1`:b2;enlist`:b3),1,1,batching uses bytes not row count (row count alone would fit all three) +run,0,0,q,merge.init[enlist[`log]!enlist mylog],1,1,restore row-count batching (default) + +comment,,,,,,,getpartchunks splits a batch that exceeds partlimit partitions +run,0,0,q,merge.init[`partlimit`log!(2;mylog)],1,1,cap batches at two partitions +run,0,0,q,merge.clearpartsizes[],1,1,reset the store +run,0,0,q,merge.trackpartition[`:e1;1;0],1,1,track e1 +run,0,0,q,merge.trackpartition[`:e2;1;0],1,1,track e2 +run,0,0,q,merge.trackpartition[`:e3;1;0],1,1,track e3 +true,0,0,q,merge.getpartchunks[`:e1`:e2`:e3;1000000]~(`:e1`:e2;enlist`:e3),1,1,a huge data limit would take all three but partlimit splits them 2+1 +run,0,0,q,merge.init[enlist[`log]!enlist mylog],1,1,restore the default partlimit + +comment,,,,,,,mergebycol appends each segment's columns to the destination partition +run,0,0,q,.test.cschema:([]px:`float$();sz:`long$()),1,1,column schema +run,0,0,q,.test.ctinfo:(`quote;.test.cschema),1,1,tableinfo pair (name;schema) +run,0,0,q,.test.cdest:`:mergetesttmp/cdest,1,1,destination partition directory +run,0,0,q,.test.cseg1:`:mergetesttmp/cseg1,1,1,segment 1 +run,0,0,q,.test.cseg2:`:mergetesttmp/cseg2,1,1,segment 2 +run,0,0,q,.test.splay[.test.cdest;([]px:1 2f;sz:10 20)],1,1,seed the destination with two rows +run,0,0,q,.test.splay[.test.cseg1;([]px:3 4f;sz:30 40)],1,1,write segment 1 +run,0,0,q,.test.splay[.test.cseg2;([]px:enlist 5f;sz:enlist 50)],1,1,write segment 2 (single row) +run,0,0,q,merge.mergebycol[.test.ctinfo;.test.cdest;] each (.test.cseg1;.test.cseg2),1,1,merge both segments column by column +true,0,0,q,(get .test.cdest)~([]px:1 2 3 4 5f;sz:10 20 30 40 50),1,1,destination holds the seed rows then each segment appended in order + +comment,,,,,,,mergebypart joins whole segments and upserts them to the destination +run,0,0,q,.test.pdest:`:mergetesttmp/pdest,1,1,destination partition directory +run,0,0,q,.test.pseg1:`:mergetesttmp/pseg1,1,1,segment 1 +run,0,0,q,.test.pseg2:`:mergetesttmp/pseg2,1,1,segment 2 +run,0,0,q,.test.splay[.test.pdest;([]px:1 2f;sz:10 20)],1,1,seed the destination +run,0,0,q,.test.splay[.test.pseg1;([]px:3 4f;sz:30 40)],1,1,write segment 1 +run,0,0,q,.test.splay[.test.pseg2;([]px:enlist 5f;sz:enlist 50)],1,1,write segment 2 +run,0,0,q,merge.mergebypart[`$();.test.slash .test.pdest;(.test.pseg1;.test.pseg2)],1,1,merge one batch of both whole segments (no parted column) +true,0,0,q,(get .test.pdest)~([]px:1 2 3 4 5f;sz:10 20 30 40 50),1,1,segments joined and appended to the destination + +comment,,,,,,,mergebypart re-sorts by the parted column when p# cannot be applied directly +run,0,0,q,.test.rdest:`:mergetesttmp/rdest,1,1,destination partition directory +run,0,0,q,.test.rseg1:`:mergetesttmp/rseg1,1,1,segment 1 +run,0,0,q,.test.rseg2:`:mergetesttmp/rseg2,1,1,segment 2 +run,0,0,q,.test.splay[.test.rdest;([]grp:`long$();px:`float$())],1,1,empty destination +run,0,0,q,.test.splay[.test.rseg1;([]grp:1 2;px:1 2f)],1,1,segment 1 - grp interleaves with segment 2 +run,0,0,q,.test.splay[.test.rseg2;([]grp:1 2;px:3 4f)],1,1,segment 2 +run,0,0,q,merge.mergebypart[enlist`grp;.test.slash .test.rdest;(.test.rseg1;.test.rseg2)],1,1,merge with grp as the parted column - joined data is not parted so must be re-sorted +true,0,0,q,(get .test.rdest)~([]grp:1 1 2 2;px:1 3 2 4f),1,1,rows re-sorted by grp so the p attribute can be applied + +comment,,,,,,,mergehybrid merges under-limit partitions whole and over-limit ones column by column +run,0,0,q,.test.hschema:([]px:`float$();sz:`long$()),1,1,column schema +run,0,0,q,.test.htinfo:(`trade;.test.hschema),1,1,tableinfo pair +run,0,0,q,.test.hdest:`:mergetesttmp/hdest,1,1,destination partition directory +run,0,0,q,.test.h1:`:mergetesttmp/h1,1,1,under-limit segment +run,0,0,q,.test.h2:`:mergetesttmp/h2,1,1,over-limit segment +run,0,0,q,.test.splay[.test.hdest;([]px:`float$();sz:`long$())],1,1,empty destination +run,0,0,q,.test.splay[.test.h1;([]px:10 11f;sz:1 2)],1,1,write the under-limit segment +run,0,0,q,.test.splay[.test.h2;([]px:20 21 22f;sz:3 4 5)],1,1,write the over-limit segment +run,0,0,q,merge.clearpartsizes[],1,1,reset the store +run,0,0,q,merge.trackpartition[.test.h1;2;0],1,1,h1 is small (2 rows) +run,0,0,q,merge.trackpartition[.test.h2;100;0],1,1,h2 is large (100 rows) - over the limit below +run,0,0,q,merge.mergehybrid[`$();.test.htinfo;.test.hdest;(.test.h1;.test.h2);10],1,1,merge with a limit of 10: h1 whole, h2 column by column +true,0,0,q,(get .test.hdest)~([]px:10 11 20 21 22f;sz:1 2 3 4 5),1,1,h1 merged whole first then h2 merged column by column +run,0,0,q,merge.clearpartsizes[],1,1,clear the store + +comment,,,,,,,getextrapartitions returns the distinct parted-column combinations +run,0,0,q,qsym:([]sym:`c`a`b`a`c;px:1 2 3 4 5),1,1,table with a repeating symbol column +true,0,0,q,3=count merge.getextrapartitions[`qsym;enlist`sym],1,1,one entry per distinct sym value +true,0,0,q,(asc raze merge.getextrapartitions[`qsym;enlist`sym])~`a`b`c,1,1,the three distinct sym values + +comment,,,,,,,parted-column checks route through the injected logger +run,0,0,q,merge.init[enlist[`log]!enlist logcap],1,1,init with the capturing logger +run,0,0,q,qtbl:([]a:1 2;b:3 4;f:1.0 2.0),1,1,table with long and float columns +run,0,0,q,caplog:0#caplog,1,1,reset the capture +run,0,0,q,merge.checkpartitiontype[`qtbl;enlist`c],1,1,check a parted column absent from the table +true,0,0,q,`error in exec fn from caplog,1,1,missing parted column is error-logged +run,0,0,q,caplog:0#caplog,1,1,reset the capture +run,0,0,q,merge.checkpartitiontype[`qtbl;enlist`a],1,1,check a parted column present in the table +true,0,0,q,`info in exec fn from caplog,1,1,present parted column is info-logged +run,0,0,q,caplog:0#caplog,1,1,reset the capture +run,0,0,q,merge.checkenumerabletype[`qtbl;enlist`f],1,1,check a non-enumerable (float) parted column +true,0,0,q,`error in exec fn from caplog,1,1,non-enumerable parted column is error-logged +run,0,0,q,caplog:0#caplog,1,1,reset the capture +run,0,0,q,merge.checkenumerabletype[`qtbl;enlist`a],1,1,check an enumerable (long) parted column +true,0,0,q,`info in exec fn from caplog,1,1,enumerable parted column is info-logged + +comment,,,,,,,init routes its own message through the injected logger +run,0,0,q,caplog:0#caplog,1,1,reset the capture +run,0,0,q,merge.init[enlist[`log]!enlist logcap],1,1,re-init with the capturing logger +true,0,0,q,`merge in exec ctx from caplog,1,1,init logged under the merge context +true,0,0,q,any caplog[`msg] like "di.merge initialised",1,1,init emitted its initialisation message +run,0,0,q,merge.init[enlist[`log]!enlist mylog],1,1,restore the silent logger + +comment,,,,,,,teardown - remove the scratch directory +run,0,0,q,{if[os.isdir x;os.deldir x]} .test.root,1,1,delete the scratch dir From e1b658f304742b71c3d0891786f25567dc1efdde Mon Sep 17 00:00:00 2001 From: alowrydi Date: Thu, 20 Aug 2026 16:17:35 +0100 Subject: [PATCH 2/4] Refactor di.merge to current module conventions Rebuilds the scaffolding around the initial draft to match conventions that landed since it was written (di.eodtime, di.depcheck, di.kafka, di.servers, di.heartbeat), while keeping the original merge algorithm (getpartchunks, mergebypart, mergebycol, mergehybrid) - a faithful, line-checked port of TorQ's code/common/merge.q, verified against both real callers (wdb.q, tickerlogreplay.q) - untouched. Bug fix: mergebypart's failure handler double-converted an already-string error message via a redundant string call, which corrupted the log message and made the handler itself throw a second, uncaught error - so a merge failure that should be logged and skipped instead crashed the caller. Fixed and covered by a regression test. The bug predates this refactor (present in the legacy TorQ source too) and was only surfaced by deliberately forcing an upsert failure, since no happy-path test exercised that branch. Two deliberate design decisions: - init now preserves tracked-but-unmerged partition sizes across a re-init (e.g. a live config reload) instead of wiping them - partsizes is orthogonal to the deps a re-init typically changes, and silent data loss is worse than leaving it alone. init logs explicitly when it preserves existing tracked partitions. - mergebycol's column read is intentionally left unprotected (unlike mergebypart's now-guarded upsert): a partial-column failure mid-merge would leave the destination silently inconsistent (some columns updated, others stale), which is worse than failing loudly. Scaffolding changes: - requireinit guard on every exported function except init/getapimeta - normlog/kx.log auto-detection removed; init does strict binary-dict validation requiring info+error only (merge.q never calls warn) - init validation inlined, no setdeps/setconfig split - VERSION file added, version exported, read defensively - deps.q removed - standalone module, no hard deps - getapimeta added for di.api registration - stale sort.csv wording removed from checkpartitiontype's log messages - new: syncpartsizes, giving the receive side of the legacy partsizes IPC fan-out a real, guarded function instead of a raw upsert message - state access normalised to .z.m throughout 148 k4unit assertions, including real on-disk-segment coverage for all three merge paths, the requireinit guard, version/getapimeta shape, syncpartsizes, the re-init preservation behaviour, and the mergebypart error-handler fix. --- di/merge/VERSION | 1 + di/merge/deps.q | 5 - di/merge/init.q | 19 +++- di/merge/merge.md | 124 ++++++++++++++++++++++--- di/merge/merge.q | 230 ++++++++++++++++++++++++++++++---------------- di/merge/test.csv | 80 ++++++++++++++-- 6 files changed, 351 insertions(+), 108 deletions(-) create mode 100644 di/merge/VERSION delete mode 100644 di/merge/deps.q diff --git a/di/merge/VERSION b/di/merge/VERSION new file mode 100644 index 00000000..6c6aa7cb --- /dev/null +++ b/di/merge/VERSION @@ -0,0 +1 @@ +0.1.0 \ No newline at end of file diff --git a/di/merge/deps.q b/di/merge/deps.q deleted file mode 100644 index a7fb14c9..00000000 --- a/di/merge/deps.q +++ /dev/null @@ -1,5 +0,0 @@ -/ hard module dependencies and their minimum versions, validated by di.depcheck -/ di.merge has no hard dependencies - its only runtime dependency (log) is injected via -/ init as a dictionary of functions, and the parted column(s) needed for a merge are -/ passed in by the caller (see di.sort) rather than read from sort config -deps:(`$())!(); diff --git a/di/merge/init.q b/di/merge/init.q index f6726307..89aaff43 100644 --- a/di/merge/init.q +++ b/di/merge/init.q @@ -1,8 +1,21 @@ / load core functionality into the module \l ::merge.q -/ module version - compared against dependants' minimum requirements by di.depcheck. Removed for now until PR merged -/version:"0.1.0"; +/ module version, read from the VERSION file rather than hardcoded, so a release bump touches one +/ plain-text file. read module-relative (`:::` resolves to di/merge) and BEFORE the export line, +/ since export:([...]) evaluates each name. +/ NB `version` must STAY in the export: di.depcheck resolves a dependency's version from the export +/ dict (checkdepversion) and classes a missing one as a FAILURE - which makes di.depcheck.init throw +/ for any process loading a module that declares this one as a hard dependency +/ trim, and fail LOUD on a missing/unreadable/empty VERSION, rather than a bare `first read0`: +/ a raw OS error names no module, and read0 strips the line terminator but NOT trailing spaces - so a +/ padded file yields a padded version, which di.depcheck compares as a STRING and silently fails +/ every dependent module's check. an empty value is worse still: it reads to depcheck as +/ "exports no version", i.e. the exact failure the VERSION file was added to prevent +version:@[{trim first read0 x};`:::VERSION;{'"di.merge: VERSION file missing or unreadable"}]; +if[0=count version;'"di.merge: VERSION file is empty"]; / public api - only the functions intended to be called externally are exported -export:([init;getpartchunks;mergebypart;mergebycol;mergehybrid;trackpartition;clearpartsizes;getpartsizes;checkpartitiontype;checkenumerabletype;getextrapartitions;getfirstcharpartitions]) +export:([init;checkpartitiontype;checkenumerabletype;getextrapartitions;getfirstcharpartitions; + getpartchunks;mergebypart;mergebycol;mergehybrid;trackpartition;clearpartsizes; + getpartsizes;syncpartsizes;version;getapimeta]) diff --git a/di/merge/merge.md b/di/merge/merge.md index 9fa12156..47c92a99 100644 --- a/di/merge/merge.md +++ b/di/merge/merge.md @@ -8,6 +8,7 @@ On-disk data-merging utilities for the write-down flow, extracted from TorQ's `c - Merge on-disk partition segments into a destination partition without holding the whole partition in memory: whole-partition (`mergebypart`), column-by-column (`mergebycol`), or a per-partition hybrid (`mergehybrid`) - Track each segment's row count and byte-size estimate as it is written (`trackpartition`) and use it to size merge batches (`getpartchunks`) and pick the merge method +- Sync a partsizes table received from a peer process into local tracked state (`syncpartsizes`) - see "Cross-process partition-size sync" below - Re-sort a segment by its parted column(s) when the `p#` attribute cannot otherwise be applied, so the merged partition ends up correctly parted - Batch sizing bounded by both a data limit (rows or bytes, per `mergebybytelimit`) and a maximum partition count per batch (`partlimit`) - Helpers to validate and enumerate a table's parted columns (`checkpartitiontype`, `checkenumerabletype`, `getextrapartitions`, `getfirstcharpartitions`) @@ -19,33 +20,47 @@ On-disk data-merging utilities for the write-down flow, extracted from TorQ's `c | Dependency | Key | Required | Description | |---|---|---|---| -| logger | `` `log `` | yes | `info`/`warn`/`error` - binary `{[c;m]}` where `c` is a symbol context and `m` is a string. A `kx.log` instance is accepted directly and auto-wrapped | +| logger | `` `log `` | yes | binary `{[c;m]}` functions keyed `` `info`error `` - `c` is a symbol context, `m` is a string. `di.merge` never calls `warn`, confirmed against every log call site in the legacy source, so it is not required (extra keys, e.g. from `di.log`'s `logdict`, are accepted and ignored) | **Hard dependency:** none - the only runtime dependency (`log`) is injected via `init`, so any module exporting the contracted signatures can be supplied. -The parted (`` `p# ``) column(s) for a table - `extrapartitiontype` in the function signatures below - are **passed in by the caller**. TorQ's original `merge.q` read these from the `.sort.params` global; on extraction that coupling was removed so `di.merge` does not depend on `di.sort`. The caller (e.g. `di.wdb`) obtains the parted columns from its sort configuration and passes them to `mergebypart` / `mergehybrid`. +The parted (`` `p# ``) column(s) for a table - `extrapartitiontype` in the function signatures below - are **passed in by the caller**. TorQ's original `merge.q` read these from the `.sort.params` global (populated from `sort.csv`); on extraction that coupling was removed so `di.merge` does not depend on `di.sort`. The caller (e.g. `di.wdb`) obtains the parted columns from its own sort configuration and passes them to `mergebypart` / `mergehybrid` / the check functions. -**Logging contract.** Internally the module calls the logger as binary `.z.m.log[\`info][\`merge;"msg"]` (`{[c;m]}` - context symbol + message). You may pass either a `kx.log` instance (its monadic `{[msg]}` functions are detected and auto-wrapped, folding the context tag into the message as `"merge: ..."`) or a custom `` `info`warn`error `` dict of `{[c;m]}` functions, used as-is. +**Logging contract.** Internally the module calls the logger through `.z.m.loginfo[\`ctx;"msg"]` / `.z.m.logerr[\`ctx;"msg"]` (binary `{[c;m]}` - context symbol + message). Pass a plain `` `info`error `` (or fuller) dict of `{[c;m]}` functions - `di.log`'s `logdict` is a ready-made example. No auto-detection or adaptation is performed; the dict must already conform. --- ## Initialisation -`init[deps]` takes a single dictionary combining the injected `log` dependency with any configuration overrides. Config keys are optional - omit any and the module falls back to the default; unrecognised keys are ignored. `init` throws immediately if `deps` is not a dictionary or `log` is missing or malformed. +`init[deps]` takes a single dictionary combining the injected `log` dependency with any configuration overrides. Config keys are optional - omit any and the module falls back to the default; unrecognised keys are ignored. `init` throws immediately, before any state is wired, if `deps` is not a dictionary or `log` is missing, not a dict, or missing `info`/`error`. There is no default logger and no silent fallback. | Key | Default | Description | |---|---|---| | `` `mergebybytelimit `` | `0b` | `0b` sizes batches / picks the merge method by row count, `1b` by byte-size estimate | | `` `partlimit `` | `1000` | maximum number of partitions merged together in a single batch | -| `` `log `` | *(required)* | logger providing `info`/`warn`/`error` (a `kx.log` instance is auto-wrapped) | +| `` `log `` | *(required)* | logger providing at least `info`/`error` | ```q merge:use`di.merge -merge.init[enlist[`log]!enlist kxlog] +logging:use`di.log +merge.init[logging.logdict] / or with config overrides: -merge.init[`mergebybytelimit`partlimit`log!(1b;500;kxlog)] +merge.init[logging.logdict,`mergebybytelimit`partlimit!(1b;500)] ``` +Every exported function except `init` and `getapimeta` calls `init` must have run first - see "requireinit guard" below. + +--- + +## requireinit guard + +Every exported function except `init`/`getapimeta` refuses to run before `init` has wired the +dependencies, throwing a clear `` "di.merge: : init must be called before any other +function" `` instead of failing on an unset logger with a confusing raw error. The guard probes +`.z.m.loginfo`, the one piece of module state with no load-time default anywhere in the file - +unlike `partsizes`/`mergebybytelimit`/`partlimit`, which all start with real, valid-looking +defaults whether `init` has run or not and so cannot reliably signal "not yet initialised". + --- ## Exported Functions @@ -65,6 +80,9 @@ Return the current tracked segment sizes (keyed on `ptdir`), e.g. to sync to sor ### `clearpartsizes[]` Drop all tracked segment sizes. Call once end-of-day merging is complete. +### `syncpartsizes[t]` +Upsert a `getpartsizes[]`-shaped table `t`, typically received over IPC from a peer process, into local tracked state. See "Cross-process partition-size sync" below. + ### `getpartchunks[partdirs;mergelimit]` Split `partdirs` into batches to be merged together, each within `mergelimit` (rows or bytes per `mergebybytelimit`) and no larger than `partlimit` partitions. Returns a list of batches (each a list of partition directories). ```q @@ -87,7 +105,7 @@ merge.mergebycol[(`trade;schema); dest] each partdirs Merge `partdirs` using whichever method fits each: whole-partition for those within `mergelimit`, column-by-column for any single partition over it (creating the `.d` file if column-by-column merging produced none). ### `checkpartitiontype[tablename;extrapartitiontype]` -Error-log any parted column named for the table that is absent from it. +Error-log any parted column supplied for the table that is absent from it. ### `checkenumerabletype[tablename;extrapartitiontype]` Confirm every parted column has an enumerable type (`h`/`i`/`j`/`s`) so it can key a partition; error-logs otherwise. @@ -98,6 +116,24 @@ Return the distinct combinations of the parted column values - one per partition ### `getfirstcharpartitions[tablename;extrapartitiontype]` Return the partition values grouped by the first character of the (single) parted column. +### `version` +Module version string (from the `VERSION` file), for `di.depcheck` to check against dependants' minimum-version requirements. + +### `getapimeta[]` +This module's API metadata, one row per callable API function, for `di.torq` to register with `di.api`. `init`/`getapimeta` are framework plumbing and are not registered. + +--- + +## Two callers, not fully symmetric + +Both legacy callers (`wdb.q`, `tickerlogreplay.q`) call `checkpartitiontype`, `getextrapartitions`, +`getpartchunks`, `mergebypart`, `mergebycol` and `mergehybrid`. **`checkenumerabletype` and +`getfirstcharpartitions` are wdb.q-only** - they back wdb's `partbyenum` and `partbyfirstchar` +writedown modes, which `tickerlogreplay.q`'s simpler `partandmerge` replay mode has no equivalent +of. A future consumer that only exercises the tickerlogreplay-style call pattern (as `di.wdb` will +initially, most likely) should not assume the full API surface is exercised end-to-end by that +usage alone - `checkenumerabletype`/`getfirstcharpartitions` need their own coverage. + --- ## Partition-size store schema @@ -110,14 +146,75 @@ Return the partition values grouped by the first character of the (single) parte | rowcount | `long` | accumulated row count written to that segment | | bytes | `long` | accumulated byte-size estimate of that segment | +`getpartchunks[partdirs;mergelimit]` silently drops any requested `partdirs` entry that has not +been `trackpartition`'d - it filters against `getpartsizes[]`, so an untracked partition is simply +absent from every batch rather than raising an error. Callers must ensure every partition they mean +to merge has been tracked first. + +--- + +## Two deliberate design decisions from adversarial testing + +A deliberate adversarial pass beyond the k4unit happy-path suite surfaced two behaviours that needed +a conscious decision rather than either a silent fix or a silently-shipped gap. Both are resolved: + +**`init` preserves tracked-but-unmerged partition sizes across a re-init.** Calling `init` again with +valid deps (e.g. a live config reload via `di.torq`) does **not** wipe `.z.m.partsizes` - any segments +`trackpartition`'d since the last `clearpartsizes[]` survive. `partsizes` is orthogonal to the +`log`/`mergebybytelimit`/`partlimit` deps a re-init is typically changing, and silently discarding +tracked-but-unmerged segment sizes is a worse failure mode than leaving them alone - a re-init that +happens to land between `trackpartition` calls and the next merge should not cause data to go +unmerged with no trace. When `init` finds pre-existing tracked partitions it says so explicitly: +`` "di.merge initialised, N segment(s) already tracked, preserved" `` - so the decision is visible in +the log rather than something a future debugger has to discover by reading source. A fresh, first-ever +`init` (nothing tracked yet) logs the plain `"di.merge initialised"` with no such claim. + +**`mergebycol` deliberately keeps failing uncaught on a missing/corrupt segment column; `mergebypart` +does not.** These are not equivalent failure modes, so making them match would not obviously be the +safer choice - it might be the wrong one. `mergebypart` fails per whole partition: if one partition's +upsert fails, `dest` simply doesn't get that partition's data yet, which is incomplete but still +internally consistent. `mergebycol` merges one column at a time into the *same* `dest`; if a swallowed +failure let it carry on past a bad column, `dest` would end up with some columns reflecting the new +data and others silently stale - a genuinely worse, silently-inconsistent partition, not just a +delayed merge. So `mergebycol`'s column read is intentionally left unprotected: a missing or +unreadable segment column throws straight out of `mergebycol`/`mergehybrid`'s column-by-column path, +by design, rather than risking a half-updated destination. (This also happens to match the legacy +TorQ source and Olly's draft, both of which have the same unprotected-read structure - but the reason +to keep it here is the partial-column-write risk above, not merely that it matches prior behaviour.) + +--- + +## Cross-process partition-size sync + +Legacy `wdb.q` fans `.merge.partsizes` out to sort workers via raw async IPC in two places - +`endofdaymerge` (targeting `.z.pd[]`, the process's own worker handles) and `informsortandreload` +(targeting discovered peer sort/reload processes) - both conditionally guarded on the merge method +being `part` or `hybrid`. There was no symmetric receive-side function: a receiving process +evaluated the raw `(upsert;`.merge.partsizes;y)` tuple directly, which only worked if it had +already loaded `merge.q` so the table existed with the right schema - an undocumented, load-order- +dependent contract. + +`syncpartsizes[t]` gives the receive side a real function to go through instead. This is a genuine +improvement over the legacy pattern, not just a faithfulness gap-fill: because `syncpartsizes` is +guarded by `requireinit`, the receiving process must now have called `di.merge.init` first - the +"must have already loaded merge.q" precondition becomes explicit and checked rather than implicit. + +```q +/ sender side (e.g. di.wdb, once merge functionality lands there) +(neg h) (`.merge.syncpartsizes; merge.getpartsizes[]) + +/ receiver side +merge.syncpartsizes[t] / upserts wholesale into local tracked state +``` + --- ## Usage Example ```q -kxlog:use`kx.log +logging:use`di.log merge:use`di.merge -merge.init[enlist[`log]!enlist kxlog.createLog[]] +merge.init[logging.logdict] / as each segment is written down, record its size merge.trackpartition[seg; count data; -22!data] @@ -137,12 +234,13 @@ k4unit:use`di.k4unit k4unit.moduletest`di.merge ``` -The test suite injects a no-op binary mock logger (and a capturing logger for the log-path assertions). It covers: dependency validation (non-dict deps / missing / non-dict / incomplete `log` all throw, with the `di.merge` error prefix); config application via row-count vs byte-size batching; `trackpartition`/`getpartsizes`/`clearpartsizes`; the `getpartchunks` batching and `partlimit` splitting logic; and end-to-end `mergebypart`, `mergebycol` and `mergehybrid` against real on-disk segments written to a scratch directory (cleaned up afterwards). +The test suite injects a no-op binary mock logger (and a capturing logger for the log-path assertions). It covers: dependency validation (non-dict deps / missing / non-dict / incomplete `log` all throw, with the `di.merge` error prefix); the `requireinit` guard rejecting every exported function before `init` has run; config application via row-count vs byte-size batching, defaults and overrides; `trackpartition`/`getpartsizes`/`clearpartsizes`/`syncpartsizes`; the `getpartchunks` batching and `partlimit` splitting logic; `version`/`getapimeta` shape; and end-to-end `mergebypart`, `mergebycol` and `mergehybrid` against real on-disk segments written to a scratch directory (cleaned up afterwards). --- ## Notes -- Extracted from TorQ `code/common/merge.q`. The only behavioural change on extraction is the removal of the `.sort.params` lookup (`getextrapartitiontype`): the parted column(s) are now a parameter (`extrapartitiontype`) supplied by the caller, keeping the module standalone. -- `init` must be called before any other function (it wires the logger). +- Extracted from TorQ `code/common/merge.q`. The one behavioural change on extraction is the removal of the `.sort.params`/`sort.csv` lookup (`getextrapartitiontype`): the parted column(s) are now a parameter (`extrapartitiontype`) supplied by the caller, keeping the module standalone. All error/info messages that referenced `sort.csv` in the legacy source have been reworded accordingly. +- `init` must be called before any other function - enforced by the `requireinit` guard on every other exported function, not just documented convention. +- The `VERSION`-file read pattern (fail loud on missing/unreadable/empty, `trim` against trailing whitespace) follows `di.servers`, not `di.eodtime` (which has no `VERSION` handling at all). - The merge functions operate on on-disk paths (`get`/`set`/`upsert` against file symbols); they do not manage the segment or destination directory lifecycle - that remains the caller's responsibility. diff --git a/di/merge/merge.q b/di/merge/merge.q index 75ddbb2a..4b55787f 100644 --- a/di/merge/merge.q +++ b/di/merge/merge.q @@ -7,71 +7,55 @@ / than held in memory and written once at end-of-day / the parted (`p#) column(s) for a table are supplied by the caller as extrapartitiontype; / the module never reads sort config itself, so it stays independent of di.sort -/ config and dependencies are passed to init in a single dictionary: config keys (see -/ merge.md) are optional with defaults; log is required and init errors immediately if it -/ is missing - see merge.md -/ module-local state convention: read config bare, mutate via .z.m, and access injected -/ dependencies via .z.m at every call site +/ config and the injected log dependency are passed to init in a single dictionary: config +/ keys (see merge.md) are optional with defaults; log is required and init errors +/ immediately if it is missing or does not provide info/error - see merge.md +/ module-local state convention: mutable state and injected deps are held under .z.m and +/ accessed via .z.m at every call site; nothing is read bare / ============================================================ / module state and defaults / ============================================================ -/ row count and byte-size estimate of each on-disk segment as it is written, keyed by -/ partition directory - drives the batch sizing and merge-method decisions below -partsizes:([ptdir:`symbol$()] rowcount:`long$(); bytes:`long$()); +/ schema template for the partition-size tracking table - the live copy is held in +/ .z.m.partsizes, seeded fresh by init and reset by clearpartsizes +partsizesschema:([ptdir:`symbol$()] rowcount:`long$(); bytes:`long$()); -/ configuration defaults - overridden by the config dictionary passed to init -mergebybytelimit:0b; / 0b = size batches / choose method by row count, 1b = by byte-size estimate -partlimit:1000; / maximum number of partitions merged together in a single batch +/ configuration defaults - overridden by the config keys in the dict passed to init +mergebybytelimitdefault:0b; / 0b = size batches / choose method by row count, 1b = by byte-size estimate +partlimitdefault:1000; / maximum number of partitions merged together in a single batch / ============================================================ -/ internal helpers +/ init guard / ============================================================ -/ normalise an injected logger to the binary {[c;m]} contract -normlog:{[logdict] - / detect a kx.log instance by its marker keys and wrap its monadic level functions into - / binary {[c;m]}, folding the context tag into the message; a plain {[c;m]} dict passes through - $[any `getlvl`sinks`fmts in key logdict; - `info`warn`error!( - {[fn;c;m] fn[string[c],": ",m]}[logdict`info;]; - {[fn;c;m] fn[string[c],": ",m]}[logdict`warn;]; - {[fn;c;m] fn[string[c],": ",m]}[logdict`error;]); - logdict] +initialised:{[] + / has init run? .z.m.loginfo has no load-time default - only init ever sets it - so this + / probe can't be fooled by a same-named constant, unlike partsizes/mergebybytelimit/partlimit + / which all start with real, valid-looking defaults whether init has run or not + :@[{.z.m.loginfo;1b};::;{[e] :0b}]; }; -/ validate and wire the injected dependencies from the deps dict -setdeps:{[deps] - / log is the only dependency - required, and must provide info/warn/error (a kx.log - / instance is auto-wrapped by normlog); validate before writing any module state - if[99h<>type deps; - '"di.merge: deps must be a dictionary of config and injected dependencies - see merge.md"]; - if[not `log in key deps; - '"di.merge: log dependency is required; pass `info`warn`error (or a kx.log logger) keyed on `log"]; - if[99h<>type deps`log; - '"di.merge: log must be a dict of info/warn/error functions (or a kx.log logger)"]; - lg:normlog deps`log; - if[not all `info`warn`error in key lg; - '"di.merge: log must provide info/warn/error; got: ",", " sv string key lg]; - .z.m.log:lg; +requireinit:{[ctx] + / every exported function except init/getapimeta refuses to run before init has wired the + / deps - without this a pre-init call would fail on an unset .z.m.loginfo with a confusing + / raw error instead of a clear one + if[not initialised[]; + '"di.merge: ",string[ctx],": init must be called before any other function"]; }; -/ apply recognised config overrides from the deps dict, defaulting where a key is absent -setconfig:{[deps] - cfg:$[99h=type deps;deps;()!()]; - .z.m.mergebybytelimit:$[`mergebybytelimit in key cfg;cfg`mergebybytelimit;0b]; - .z.m.partlimit:$[`partlimit in key cfg;cfg`partlimit;1000]; - }; +/ ============================================================ +/ internal helpers +/ ============================================================ / merge a single column from a segment into the destination partition, logging on failure mergeonecol:{[dest;segment;col] / filepaths to the destination column and the matching column in the segment destcol:` sv dest,col; destdata:get segcol:` sv segment,col; - .z.m.log[`info][`merge;"merging ",(string segcol)," to ",string destcol]; + .z.m.loginfo[`merge;"merging ",(string segcol)," to ",string destcol]; .[upsert;(destcol;destdata); - {[dc;e] .z.m.log[`error][`merge;"failed to save data to ",(string dc)," with error : ",e]}[destcol;]]; + {[dc;e] .z.m.logerr[`merge;"failed to save data to ",(string dc)," with error : ",e]}[destcol;]]; }; / ============================================================ @@ -81,15 +65,30 @@ mergeonecol:{[dest;segment;col] / accumulate the row count and byte-size estimate for a freshly-written segment trackpartition:{[ptdir;rowcount;bytes] / call each time data is written to a segment; keyed by partition directory + requireinit[`trackpartition]; .z.m.partsizes[ptdir]+:(rowcount;bytes); }; / drop all tracked segment sizes - call once end-of-day merging is complete -clearpartsizes:{ - .z.m.partsizes:0#partsizes; +clearpartsizes:{[] + requireinit[`clearpartsizes]; + .z.m.partsizes:0#partsizesschema; }; -getpartsizes:{partsizes}; / current tracked segment sizes (e.g. to sync to sort-worker processes) +/ current tracked segment sizes (e.g. to sync to sort-worker processes) +getpartsizes:{[] + requireinit[`getpartsizes]; + .z.m.partsizes + }; + +/ upsert a partsizes-shaped table received from a peer process into local tracked state +syncpartsizes:{[t] + / t: a table shaped like getpartsizes[] (ptdir/rowcount/bytes), typically received over IPC + / from a peer process's own getpartsizes[] call - upserts wholesale into local state, giving + / the receive side of the legacy raw-IPC partsizes fan-out a real function to go through + requireinit[`syncpartsizes]; + .z.m.partsizes:.z.m.partsizes upsert t; + }; / ============================================================ / public api - merging @@ -99,14 +98,16 @@ getpartsizes:{partsizes}; / current tracked segment sizes (e.g. to sync to sor getpartchunks:{[partdirs;mergelimit] / each batch stays within mergelimit (row count or byte estimate per mergebybytelimit) and / holds no more than partlimit partitions + requireinit[`getpartchunks]; / tracked sizes for just the partitions we are merging - t:select from partsizes where ptdir in partdirs; + t:select from .z.m.partsizes where ptdir in partdirs; / the measure we accumulate against the limit - r:$[mergebybytelimit;exec bytes from t;exec rowcount from t]; + r:$[.z.m.mergebybytelimit;exec bytes from t;exec rowcount from t]; / running-total scan resets to the current partition whenever adding it would breach the limit l:(where r={$[ztype chunks;chunks:(,/)chunks]; - .z.m.log[`info][`resort;"checking that the contents of this subpartition conform"]; + .z.m.loginfo[`resort;"checking that the contents of this subpartition conform"]; / can the p# attribute be applied as-is? if not, the data must be re-sorted by the parted column pattrtest:@[{@[x;y;`p#];0b}[chunks;];extrapartitiontype;{1b}]; if[pattrtest; - .z.m.log[`info][`resort;"re-sorting contents of subpartition"]; + .z.m.loginfo[`resort;"re-sorting contents of subpartition"]; chunks:xasc[extrapartitiontype;chunks]; - .z.m.log[`info][`resort;"the p attribute can now be applied"]; + .z.m.loginfo[`resort;"the p attribute can now be applied"]; ]; - .z.m.log[`info][`merge;"upserting ",(string count chunks)," rows to ",string dest]; + .z.m.loginfo[`merge;"upserting ",(string count chunks)," rows to ",string dest]; / append the merged rows to permanent storage, logging (not throwing) on failure + / e arrives already a string from the protected-apply mechanism - do not re-stringify it: string + / of an already-string value maps over each char individually, corrupting the message and making + / this handler itself throw, which defeats the log-not-throw contract this line exists to provide .[upsert;(dest;chunks); - {[e;d;p] .z.m.log[`error][`merge;"failed to merge to ",string[d]," from segments ",(", " sv string p)," Error is - ",string e]}[;dest;partchunks]]; + {[e;d;p] .z.m.logerr[`merge;"failed to merge to ",string[d]," from segments ",(", " sv string p)," Error is - ",e]}[;dest;partchunks]]; }; / merge one segment into the destination partition a column at a time mergebycol:{[tableinfo;dest;segment] / holds at most a single column in memory rather than the whole partition - .z.m.log[`info][`merge;"upserting columns from ",(string segment)," to ",string dest]; + requireinit[`mergebycol]; + .z.m.loginfo[`merge;"upserting columns from ",(string segment)," to ",string dest]; mergeonecol[dest;segment;] each cols tableinfo[1]; }; / merge the given partitions using whichever method fits each one mergehybrid:{[extrapartitiontype;tableinfo;dest;partdirs;mergelimit] / whole-partition for those within the limit, column-by-column for any single partition over it - overlimit:$[mergebybytelimit; - exec ptdir from partsizes where ptdir in partdirs,bytes>mergelimit; - exec ptdir from partsizes where ptdir in partdirs,rowcount>mergelimit]; + requireinit[`mergehybrid]; + overlimit:$[.z.m.mergebybytelimit; + exec ptdir from .z.m.partsizes where ptdir in partdirs,bytes>mergelimit; + exec ptdir from .z.m.partsizes where ptdir in partdirs,rowcount>mergelimit]; if[(count overlimit)<>count partdirs; partdirs:partdirs except overlimit; - .z.m.log[`info][`merge;"merging ",(", " sv string partdirs)," by whole partition"]; + .z.m.loginfo[`merge;"merging ",(", " sv string partdirs)," by whole partition"]; mergebypart[extrapartitiontype;` sv dest,`]'[getpartchunks[partdirs;mergelimit]]; ]; if[0<>count overlimit; - .z.m.log[`info][`merge;"merging ",(", " sv string overlimit)," column by column"]; + .z.m.loginfo[`merge;"merging ",(", " sv string overlimit)," column by column"]; mergebycol[tableinfo;dest]'[overlimit]; / column-by-column merge writes no .d file - create one if none exists yet if[()~key ` sv dest,`.d; - .z.m.log[`info][`merge;"creating file ",string ` sv dest,`.d]; + .z.m.loginfo[`merge;"creating file ",string ` sv dest,`.d]; (` sv dest,`.d) set cols tableinfo[1]; ]; ]; @@ -167,27 +174,34 @@ mergehybrid:{[extrapartitiontype;tableinfo;dest;partdirs;mergelimit] / error-log any parted column named for the table that is absent from it checkpartitiontype:{[tablename;extrapartitiontype] + requireinit[`checkpartitiontype]; $[count colsnotintab:extrapartitiontype where not extrapartitiontype in cols get tablename; - .z.m.log[`error][`checkpart;"parted columns ",(", " sv string colsnotintab)," are defined in sort.csv but not present in ",(string tablename)," table"]; - .z.m.log[`info][`checkpart;"all parted columns defined in sort.csv are present in ",(string tablename)," table"]]; + .z.m.logerr[`checkpart;"parted columns ",(", " sv string colsnotintab), + " not present in the parted columns supplied for ",(string tablename)," table"]; + .z.m.loginfo[`checkpart;"all parted columns supplied are present in ",(string tablename)," table"]]; }; / confirm every parted column has an enumerable type (h/i/j/s) so it can key a partition checkenumerabletype:{[tablename;extrapartitiontype] + requireinit[`checkenumerabletype]; $[all extrapartitiontype in exec c from meta[tablename] where t in "hijs"; - .z.m.log[`info][`checkenumerable;"all columns do have an enumerable type in ",(string tablename)," table"]; - .z.m.log[`error][`checkenumerable;"not all columns ",string[extrapartitiontype]," do have an enumerable type in ",(string tablename)," table"]]; + .z.m.loginfo[`checkenumerable;"all columns do have an enumerable type in ",(string tablename)," table"]; + .z.m.logerr[`checkenumerable;"not all columns ",string[extrapartitiontype], + " do have an enumerable type in ",(string tablename)," table"]]; }; / distinct combinations of the parted column values - one per partition directory getextrapartitions:{[tablename;extrapartitiontype] / functional form of: select distinct extrapartitiontype from tablename + requireinit[`getextrapartitions]; value each ?[tablename;();1b;extrapartitiontype!extrapartitiontype] }; / partition values grouped by the first character of the (single) parted column getfirstcharpartitions:{[tablename;extrapartitiontype] - raze each value (?[tablename;();();(distinct;first extrapartitiontype)]) group ?[tablename;();();({first each string x};(distinct;first extrapartitiontype))] + requireinit[`getfirstcharpartitions]; + raze each value (?[tablename;();();(distinct;first extrapartitiontype)]) + group ?[tablename;();();({first each string x};(distinct;first extrapartitiontype))] }; / ============================================================ @@ -195,15 +209,73 @@ getfirstcharpartitions:{[tablename;extrapartitiontype] / ============================================================ init:{[deps] - / initialise from a single dictionary holding config overrides and the injected log dependency - / config keys are optional and fall back to defaults: - / `mergebybytelimit - 0b (row count) or 1b (byte-size estimate) - default 0b - / `partlimit - max partitions merged together in a batch - default 1000 - / dependency is required: - / `log - a logger providing info/warn/error (a kx.log instance is auto-wrapped) - / example: - / merge.init[enlist[`log]!enlist kxlog] - setdeps deps; - setconfig deps; - .z.m.log[`info][`merge;"di.merge initialised"]; + / deps: dict - required `log (info, error only - merge.q never calls warn, confirmed + / function-by-function against the legacy source); + / optional `mergebybytelimit (default 0b), `partlimit (default 1000) + / example: merge.init[enlist[`log]!enlist logdep] + if[99h<>type deps; + '"di.merge: deps must be a dict with `log key"]; + if[not `log in key deps; + '"di.merge: log dependency is required; pass `info`error functions - see di.log"]; + if[99h<>type deps`log; + '"di.merge: log value must be a dict; pass `info`error functions"]; + if[not all `info`error in key deps`log; + '"di.merge: log dict must have `info`error keys; got: ",(", " sv string key deps`log)]; + .z.m.loginfo:(deps`log)`info; + .z.m.logerr:(deps`log)`error; + .z.m.mergebybytelimit:$[`mergebybytelimit in key deps;deps`mergebybytelimit;mergebybytelimitdefault]; + .z.m.partlimit:$[`partlimit in key deps;deps`partlimit;partlimitdefault]; + / preserve any partitions already tracked across a re-init (e.g. a live config reload) - partsizes + / is independent of the log/mergebybytelimit/partlimit deps a re-init is typically changing, and + / silently wiping tracked-but-unmerged segment sizes is a worse failure mode than leaving them be + priorpartsizes:@[{.z.m.partsizes};::;{[e] 0#partsizesschema}]; + .z.m.partsizes:priorpartsizes; + .z.m.loginfo[`merge;$[0 Date: Fri, 21 Aug 2026 10:01:51 +0100 Subject: [PATCH 3/4] Close review findings on di.merge: getfirstcharpartitions coverage, getpartchunks drop logging A senior review of the initial push flagged three gaps before this is PR-ready: getfirstcharpartitions had no positive-path test (only the pre-init rejection case), getpartchunks silently dropped untracked partitions with no trace, and merge.md's "two callers" note had gone half-stale now that checkenumerabletype is covered. Adds a positive-path test for getfirstcharpartitions, adds an info-level log line (and test) when getpartchunks drops an untracked partition without changing its filtering behaviour, and updates merge.md accordingly. --- di/merge/merge.md | 10 ++++++---- di/merge/merge.q | 6 ++++++ di/merge/test.csv | 13 +++++++++++++ 3 files changed, 25 insertions(+), 4 deletions(-) diff --git a/di/merge/merge.md b/di/merge/merge.md index 47c92a99..df7b1cef 100644 --- a/di/merge/merge.md +++ b/di/merge/merge.md @@ -132,7 +132,8 @@ Both legacy callers (`wdb.q`, `tickerlogreplay.q`) call `checkpartitiontype`, `g writedown modes, which `tickerlogreplay.q`'s simpler `partandmerge` replay mode has no equivalent of. A future consumer that only exercises the tickerlogreplay-style call pattern (as `di.wdb` will initially, most likely) should not assume the full API surface is exercised end-to-end by that -usage alone - `checkenumerabletype`/`getfirstcharpartitions` need their own coverage. +usage alone - `getfirstcharpartitions` needs its own coverage, which it now has (see `test.csv`); +`checkenumerabletype` is covered by the parted-column-checks block. --- @@ -146,10 +147,11 @@ usage alone - `checkenumerabletype`/`getfirstcharpartitions` need their own cove | rowcount | `long` | accumulated row count written to that segment | | bytes | `long` | accumulated byte-size estimate of that segment | -`getpartchunks[partdirs;mergelimit]` silently drops any requested `partdirs` entry that has not -been `trackpartition`'d - it filters against `getpartsizes[]`, so an untracked partition is simply +`getpartchunks[partdirs;mergelimit]` drops any requested `partdirs` entry that has not been +`trackpartition`'d - it filters against `getpartsizes[]`, so an untracked partition is simply absent from every batch rather than raising an error. Callers must ensure every partition they mean -to merge has been tracked first. +to merge has been tracked first. `getpartchunks` logs an info line naming how many requested +partitions had no tracked size whenever it drops any - the filtering behaviour itself is unchanged. --- diff --git a/di/merge/merge.q b/di/merge/merge.q index 4b55787f..3ed24a53 100644 --- a/di/merge/merge.q +++ b/di/merge/merge.q @@ -101,6 +101,12 @@ getpartchunks:{[partdirs;mergelimit] requireinit[`getpartchunks]; / tracked sizes for just the partitions we are merging t:select from .z.m.partsizes where ptdir in partdirs; + / a requested partdir with no tracked size is silently absent from t above - log it so the gap + / is visible rather than something a caller has to notice missing from the merge batch. distinct + / on both sides so a duplicate partdir in the request doesn't skew the count + if[0 Date: Fri, 21 Aug 2026 14:14:56 +0100 Subject: [PATCH 4/4] Fix di.merge smoke-test findings: attribute application, batch isolation, checkenumerabletype A full write-down-and-merge smoke test (real segments, real di.log, both of TorQ's partbyenum and partbyfirstchar write patterns) surfaced three real gaps beyond what the mock-logger k4unit suite could catch: checkenumerabletype built its error message with string[list] instead of ", " sv string list, producing a malformed nested value that crashed under a real logger the moment a non-enumerable parted column was checked - the one case the function exists to catch. Fixed to match checkpartitiontype's already-correct pattern. mergehybrid never actually left the destination `p#-attributed, even when every batch merged cleanly: upsert appends raw values onto an on-disk column without persisting an in-memory attribute, and no single batch/column write can guarantee the whole destination stays grouped once multiple batches (and mergebycol's un-resorted, multi-value segments) have all appended to it. Confirmed empirically both ways: a pure mergebypart merge left no attribute at all, and a partbyfirstchar-style merge through mergebycol left the parted column genuinely unsorted on disk. Fixed with a single, final resort-and-reattribute pass over the whole destination in mergehybrid, after every batch and column has merged - a deliberate, documented departure from the module's "keeps memory flat" goal for that one step, since a destination that's silently never truly parted is worse. mergebypart read every segment in a batch as one unprotected unit, so one missing/corrupt segment file crashed uncaught and took its healthy batch-mates down with it. Now reads each segment individually and protected - a bad segment is error-logged and dropped, the rest of the batch still merges. Adds regression coverage for all three (message-shape assertion, mergehybrid attribute/resort test spanning both merge paths, mergebypart batch-isolation test) and documents the attribute and batch-isolation behaviour in merge.md. --- di/merge/merge.md | 67 ++++++++++++++++++++++++++++++++++++++--------- di/merge/merge.q | 39 ++++++++++++++++++++++++--- di/merge/test.csv | 32 ++++++++++++++++++++++ 3 files changed, 122 insertions(+), 16 deletions(-) diff --git a/di/merge/merge.md b/di/merge/merge.md index df7b1cef..178b5f64 100644 --- a/di/merge/merge.md +++ b/di/merge/merge.md @@ -90,7 +90,7 @@ merge.getpartchunks[partdirs; 1000000] ``` ### `mergebypart[extrapartitiontype;dest;partchunks]` -Merge one batch of whole partition segments into destination partition `dest`, re-sorting by the parted column(s) `extrapartitiontype` first if the `p#` attribute cannot otherwise be applied. Typically iterated over the output of `getpartchunks`. +Merge one batch of whole partition segments into destination partition `dest`, re-sorting by the parted column(s) `extrapartitiontype` first if the `p#` attribute cannot otherwise be applied. Typically iterated over the output of `getpartchunks`. Each segment in `partchunks` is read individually and protected: a missing or corrupt segment file is error-logged and dropped, it does not take its batch-mates down with it - the rest of the batch still merges. If every segment in the batch fails to read, nothing is upserted and that is error-logged too. ```q merge.mergebypart[`sym; ` sv dest,`] each merge.getpartchunks[partdirs;lim] ``` @@ -102,7 +102,7 @@ merge.mergebycol[(`trade;schema); dest] each partdirs ``` ### `mergehybrid[extrapartitiontype;tableinfo;dest;partdirs;mergelimit]` -Merge `partdirs` using whichever method fits each: whole-partition for those within `mergelimit`, column-by-column for any single partition over it (creating the `.d` file if column-by-column merging produced none). +Merge `partdirs` using whichever method fits each: whole-partition for those within `mergelimit`, column-by-column for any single partition over it (creating the `.d` file if column-by-column merging produced none). If `extrapartitiontype` is non-empty, once both paths have finished it re-sorts and re-applies the `p#` attribute to `dest` as a whole - see "The parted attribute is applied once, on the whole destination" below for why that happens here rather than per-batch. ### `checkpartitiontype[tablename;extrapartitiontype]` Error-log any parted column supplied for the table that is absent from it. @@ -173,16 +173,54 @@ the log rather than something a future debugger has to discover by reading sourc **`mergebycol` deliberately keeps failing uncaught on a missing/corrupt segment column; `mergebypart` does not.** These are not equivalent failure modes, so making them match would not obviously be the -safer choice - it might be the wrong one. `mergebypart` fails per whole partition: if one partition's -upsert fails, `dest` simply doesn't get that partition's data yet, which is incomplete but still -internally consistent. `mergebycol` merges one column at a time into the *same* `dest`; if a swallowed -failure let it carry on past a bad column, `dest` would end up with some columns reflecting the new -data and others silently stale - a genuinely worse, silently-inconsistent partition, not just a -delayed merge. So `mergebycol`'s column read is intentionally left unprotected: a missing or -unreadable segment column throws straight out of `mergebycol`/`mergehybrid`'s column-by-column path, -by design, rather than risking a half-updated destination. (This also happens to match the legacy -TorQ source and Olly's draft, both of which have the same unprotected-read structure - but the reason -to keep it here is the partial-column-write risk above, not merely that it matches prior behaviour.) +safer choice - it might be the wrong one. `mergebypart` reads each segment in a batch individually and +protected: a missing/corrupt segment file is error-logged and dropped without disturbing its +batch-mates, which still merge. The remaining, surviving segments in the batch are then joined and +upserted as one unit - if *that* upsert itself fails (e.g. a schema mismatch), the whole batch's data +is lost together, which is incomplete but still internally consistent: `dest` simply doesn't get that +batch's data yet, nothing else in `dest` is disturbed. `mergebycol` merges one column at a time into +the *same* `dest`; if a swallowed failure let it carry on past a bad column, `dest` would end up with +some columns reflecting the new data and others silently stale - a genuinely worse, +silently-inconsistent partition, not just a delayed merge. So `mergebycol`'s column read is +intentionally left unprotected: a missing or unreadable segment column throws straight out of +`mergebycol`/`mergehybrid`'s column-by-column path, by design, rather than risking a half-updated +destination. (This also happens to match the legacy TorQ source and Olly's draft, both of which have +the same unprotected-read structure - but the reason to keep it here is the partial-column-write risk +above, not merely that it matches prior behaviour.) + +--- + +## The parted attribute is applied once, on the whole destination + +A full write-down-and-merge smoke test (real segments, real `di.log`, both of TorQ's `partbyenum` +and `partbyfirstchar` write patterns) found that `mergehybrid` never actually left `dest` with the +`p#` attribute set, even when every batch merged cleanly - `mergebypart`'s own resort logic decides +*whether* `p#` could apply and reorders rows accordingly, but applying the attribute to an in-memory +batch and then `upsert`-ing it to disk does not persist the attribute: `upsert` appends raw values +onto the on-disk column, it does not carry an in-memory attribute through to the file. Confirmed +empirically (`meta` on the merged destination showed no attribute on the parted column, in a +mergebypart-only scenario with nothing left to resort). + +Worse, `mergebycol` never even tries: for TorQ's `partbyfirstchar` write mode, a single segment can +hold *several* parted-column values in arrival order, not one. If that segment is large enough to +route through `mergebycol` instead of `mergebypart`, its rows are appended as-is - the destination +ends up genuinely unsorted and non-contiguous by the parted column. Confirmed empirically too: after +merging first-character-grouped segments through `mergehybrid`, the destination's parted column was +neither sorted nor grouped on disk. + +Trying to fix this per-batch (inside `mergebypart`, or per-column inside `mergebycol`) cannot give a +real guarantee either way: a batch that is already correctly grouped in isolation can still land next +to a *different* batch's values on disk, and `upsert` does not re-validate the combined result. +The only way to genuinely guarantee `` `p# `` on `dest` is to look at the whole thing at once. So +`mergehybrid` does that as a final step, after every whole-partition batch and every column-by-column +segment has been merged: it reads `dest` back, re-sorts it by `extrapartitiontype`, re-applies the +attribute, and writes the whole table back. This is a deliberate, necessary departure from this +module's "keeps memory flat" design goal (see the file header) for that one final step - the +alternative is a `dest` that is silently never truly parted, which is worse. Batch-by-batch merging +into `dest` still only ever holds one batch/column in memory at a time; only this last step reads +`dest` in full, once, per `mergehybrid` call. `mergebypart` called standalone in a caller-driven loop +(bypassing `mergehybrid` - see its usage example above) does **not** get this guarantee automatically; +route through `mergehybrid` if a properly parted `dest` matters. --- @@ -236,7 +274,9 @@ k4unit:use`di.k4unit k4unit.moduletest`di.merge ``` -The test suite injects a no-op binary mock logger (and a capturing logger for the log-path assertions). It covers: dependency validation (non-dict deps / missing / non-dict / incomplete `log` all throw, with the `di.merge` error prefix); the `requireinit` guard rejecting every exported function before `init` has run; config application via row-count vs byte-size batching, defaults and overrides; `trackpartition`/`getpartsizes`/`clearpartsizes`/`syncpartsizes`; the `getpartchunks` batching and `partlimit` splitting logic; `version`/`getapimeta` shape; and end-to-end `mergebypart`, `mergebycol` and `mergehybrid` against real on-disk segments written to a scratch directory (cleaned up afterwards). +The test suite injects a no-op binary mock logger (and a capturing logger for the log-path assertions). It covers: dependency validation (non-dict deps / missing / non-dict / incomplete `log` all throw, with the `di.merge` error prefix); the `requireinit` guard rejecting every exported function before `init` has run; config application via row-count vs byte-size batching, defaults and overrides; `trackpartition`/`getpartsizes`/`clearpartsizes`/`syncpartsizes`; the `getpartchunks` batching and `partlimit` splitting logic; `version`/`getapimeta` shape; end-to-end `mergebypart`, `mergebycol` and `mergehybrid` against real on-disk segments written to a scratch directory (cleaned up afterwards); `mergebypart` isolating one missing segment from healthy batch-mates in the same batch; and `mergehybrid` re-sorting and re-attributing the whole destination once both merge paths have run. + +Mock loggers only check that a message was logged at the right level - they do not process the message content the way a real logger does, so they cannot catch a malformed message (a list where a flat string was expected, for instance). Where a message's *shape* matters, not just that it fired, the test asserts on structure too (e.g. `10h=type` on the captured message) rather than relying on the mock alone. --- @@ -246,3 +286,4 @@ The test suite injects a no-op binary mock logger (and a capturing logger for th - `init` must be called before any other function - enforced by the `requireinit` guard on every other exported function, not just documented convention. - The `VERSION`-file read pattern (fail loud on missing/unreadable/empty, `trim` against trailing whitespace) follows `di.servers`, not `di.eodtime` (which has no `VERSION` handling at all). - The merge functions operate on on-disk paths (`get`/`set`/`upsert` against file symbols); they do not manage the segment or destination directory lifecycle - that remains the caller's responsibility. +- A real-logger smoke test (see "The parted attribute is applied once, on the whole destination" above) also caught `checkenumerabletype`'s error-message construction building a malformed list instead of a flat string for a multi-symbol `extrapartitiontype` - the k4unit mock loggers don't process message content, so this was invisible there. Fixed to match `checkpartitiontype`'s already-correct `", " sv string ...` pattern, with a regression test that checks the message's shape (`10h=type`), not just that it fired. diff --git a/di/merge/merge.q b/di/merge/merge.q index 3ed24a53..e43f7edd 100644 --- a/di/merge/merge.q +++ b/di/merge/merge.q @@ -123,16 +123,30 @@ mergebypart:{[extrapartitiontype;dest;partchunks] / re-sorts the batch by the parted column(s) first if the p# attribute cannot otherwise be applied requireinit[`mergebypart]; .z.m.loginfo[`merge;"reading partition/partitions ",", " sv string partchunks]; - chunks:get each partchunks; + / read each segment separately and protected, so one missing/corrupt segment file does not take + / its batch-mates down with it - a bare get each/join here would fail the whole batch together + reads:{[p] @[get;p;{[p;e] .z.m.logerr[`merge;"failed to read segment ",(string p), + ", not merged this batch - error is - ",e];(::)}[p;]]} each partchunks; + ok:98h=type each reads; + if[not all ok; + .z.m.logerr[`merge;"skipping ",string[sum not ok]," of ",string[count partchunks], + " segment(s) in this batch that failed to read"]]; + chunks:reads where ok; + if[0=count chunks; + .z.m.logerr[`merge;"no segments in this batch read successfully, nothing upserted to ",string dest]; + :(::)]; / a single segment reads back as a table; multiple read back as a list of tables to join if[98<>type chunks;chunks:(,/)chunks]; .z.m.loginfo[`resort;"checking that the contents of this subpartition conform"]; / can the p# attribute be applied as-is? if not, the data must be re-sorted by the parted column + / applying p# here would not survive the upsert below - upsert appends raw values onto the + / on-disk column and does not persist an in-memory attribute, so this only orders the rows + / within this batch; mergehybrid applies the real, persisted attribute once across the whole + / destination after every batch (and any mergebycol columns) have been upserted pattrtest:@[{@[x;y;`p#];0b}[chunks;];extrapartitiontype;{1b}]; if[pattrtest; .z.m.loginfo[`resort;"re-sorting contents of subpartition"]; chunks:xasc[extrapartitiontype;chunks]; - .z.m.loginfo[`resort;"the p attribute can now be applied"]; ]; .z.m.loginfo[`merge;"upserting ",(string count chunks)," rows to ",string dest]; / append the merged rows to permanent storage, logging (not throwing) on failure @@ -172,6 +186,25 @@ mergehybrid:{[extrapartitiontype;tableinfo;dest;partdirs;mergelimit] (` sv dest,`.d) set cols tableinfo[1]; ]; ]; + / mergebypart applies p# per batch and mergebycol applies none at all, so neither guarantees the + / full destination ends up grouped once multiple batches and/or both methods have all appended + / to it - re-sort and re-apply the attribute to the whole destination once, here, after every + / path above has finished, rather than leaving that guarantee split across per-batch calls + if[0