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/init.q b/di/merge/init.q new file mode 100644 index 00000000..89aaff43 --- /dev/null +++ b/di/merge/init.q @@ -0,0 +1,21 @@ +/ load core functionality into the module +\l ::merge.q + +/ 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;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 new file mode 100644 index 00000000..178b5f64 --- /dev/null +++ b/di/merge/merge.md @@ -0,0 +1,289 @@ +# 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 +- 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`) +- 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 | 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 (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 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, 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 at least `info`/`error` | + +```q +merge:use`di.merge +logging:use`di.log +merge.init[logging.logdict] +/ or with config overrides: +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 + +### `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. + +### `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 +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`. 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] +``` + +### `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). 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. + +### `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. + +### `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 - `getfirstcharpartitions` needs its own coverage, which it now has (see `test.csv`); +`checkenumerabletype` is covered by the parted-column-checks block. + +--- + +## 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 | + +`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. `getpartchunks` logs an info line naming how many requested +partitions had no tracked size whenever it drops any - the filtering behaviour itself is unchanged. + +--- + +## 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` 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. + +--- + +## 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 +logging:use`di.log +merge:use`di.merge +merge.init[logging.logdict] + +/ 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); 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. + +--- + +## Notes + +- 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. +- 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 new file mode 100644 index 00000000..e43f7edd --- /dev/null +++ b/di/merge/merge.q @@ -0,0 +1,320 @@ +/ 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 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 +/ ============================================================ + +/ 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 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 + +/ ============================================================ +/ init guard +/ ============================================================ + +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}]; + }; + +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"]; + }; + +/ ============================================================ +/ 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.loginfo[`merge;"merging ",(string segcol)," to ",string destcol]; + .[upsert;(destcol;destdata); + {[dc;e] .z.m.logerr[`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 + requireinit[`trackpartition]; + .z.m.partsizes[ptdir]+:(rowcount;bytes); + }; + +/ drop all tracked segment sizes - call once end-of-day merging is complete +clearpartsizes:{[] + requireinit[`clearpartsizes]; + .z.m.partsizes:0#partsizesschema; + }; + +/ 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 +/ ============================================================ + +/ 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 + 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[0type 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[`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.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 + 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 + 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.loginfo[`merge;"merging ",(", " sv string partdirs)," by whole partition"]; + mergebypart[extrapartitiontype;` sv dest,`]'[getpartchunks[partdirs;mergelimit]]; + ]; + if[0<>count overlimit; + .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.loginfo[`merge;"creating file ",string ` sv dest,`.d]; + (` 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[0type 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