diff --git a/docs/md/how_to/python/virtual_server.md b/docs/md/how_to/python/virtual_server.md index f7b4333d12..a0deb637a5 100644 --- a/docs/md/how_to/python/virtual_server.md +++ b/docs/md/how_to/python/virtual_server.md @@ -15,6 +15,8 @@ Perspective ships with built-in virtual server implementations for: using the `clickhouse-connect` Python package. - [**Polars**](./virtual_server/polars.md) — query in-memory Polars DataFrames using the `polars` Python package. +- [**kdb+**](./virtual_server/kdb.md) — query a q process using the `pykx` + Python package. You can also [**implement your own**](./virtual_server/custom.md) virtual server to connect Perspective to any data source by subclassing `VirtualServerHandler`. diff --git a/docs/md/how_to/python/virtual_server/kdb.md b/docs/md/how_to/python/virtual_server/kdb.md new file mode 100644 index 0000000000..ebd3a2ba4f --- /dev/null +++ b/docs/md/how_to/python/virtual_server/kdb.md @@ -0,0 +1,207 @@ +# kdb+ Virtual Server + +Perspective provides a built-in virtual server for +[kdb+](https://kx.com/), allowing `` clients to query a q +process over WebSocket. + +## Installation + +```bash +pip install perspective-python pykx +``` + +## Usage + +Start a q process listening on a port: + +```bash +q -p 5001 +``` + +Create a server that exposes its tables to browser clients: + +```python +import pykx +import tornado.web +import tornado.ioloop +from perspective.virtual_servers.kdb import KdbVirtualServer +from perspective.handlers.tornado import PerspectiveTornadoHandler + +# Connect to q over IPC +conn = pykx.SyncQConnection(host="localhost", port=5001) + +# Create virtual server backed by kdb+ +server = KdbVirtualServer(conn) + +# Serve over WebSocket +app = tornado.web.Application([ + (r"/websocket", PerspectiveTornadoHandler, {"perspective_server": server}), +]) + +app.listen(8080) +tornado.ioloop.IOLoop.current().start() +``` + +Connect from the browser: + +```javascript +const websocket = await perspective.websocket("ws://localhost:8080/websocket"); +const table = await websocket.open_table("trades"); +document.getElementById("viewer").load(table); +``` + +Tables in q's root namespace are discoverable, as reported by `tables[]`. + +## Requirements of the q process + +Views are materialized as globals under a `.psp` namespace, so the connected +handle **must permit global assignment**. A read-only handle (`q -b`) or a +gateway that rejects writes will not work; point the virtual server at a +process you control, which may of course proxy a read-only store. + +Views are cleaned up when the UI closes them. If a view leaks — because a +client disconnected uncleanly, say — it remains as a global under `.psp` until +the process restarts. + +## Type mapping + +q's type system is richer than Perspective's six visual types. Columns whose q +type has no Perspective analogue are cast on the way out, so the schema +Perspective reports always matches the data it receives. + +| q type | Perspective | Notes | +| ------------------------------- | ----------- | ---------------------------------- | +| `boolean` | `boolean` | | +| `short`, `int`, `byte` | `integer` | | +| `long`, `real`, `float` | `float` | `long` is 64-bit; `integer` is 32 | +| `symbol`, `char`, string | `string` | | +| `guid` | `string` | cast with `string` | +| `date` | `date` | | +| `month` | `date` | cast with `"d"$` | +| `timestamp` | `datetime` | | +| `datetime` | `datetime` | cast with `"p"$` | +| `time`, `minute`, `second`, `timespan` | `string` | cast with `string` | + +### Nulls + +q has no validity bitmap — a null long *is* `0Nj`, the minimum 64-bit integer. +The handler translates these sentinels to Arrow nulls so they render as empty +cells rather than as `-9223372036854775808`. The empty symbol `` ` `` is +likewise q's symbol null and arrives in Perspective as null, not as `""`. + +Infinities (`0W`, `0w`) are genuine values in q and are passed through. + +## Supported features + +| Feature | Supported | Notes | +| ------------ | --------- | -------------------------------------------------- | +| Group By | ✔ | `rollup`, `flat` and `total` modes | +| Sort | ✔ | | +| Filter | ✔ | see below | +| Aggregates | ✔ | q's own — see below | +| Expressions | ✔ | written in **q**, not Perspective's expression language | +| Windows | ✔ | q's own — `mdev`, `mcount`, `xprev`, `ema` | +| Split By | ✘ | | + +This handler exposes **q's data model**, not a lowest common denominator +shared with the other virtual servers. Aggregate and filter names are q's, and +they mean what q means by them. If you know kdb+, the menus should read as +kdb+; if you are moving a saved layout from the DuckDB virtual server, expect +to re-pick aggregates. + +### Aggregates + +| q aggregate | Applies to | Notes | +| ----------- | ---------- | ----- | +| `sum` `avg` `min` `max` `count` `first` `last` | numeric | `sum` over a boolean column counts the trues | +| `count distinct` | any | q's `count distinct` | +| `med` | numeric | median | +| `dev` / `sdev` | numeric | **population** / **sample** standard deviation | +| `var` / `svar` | numeric | **population** / **sample** variance | +| `prd` | numeric | product | +| `any` / `all` | numeric, boolean | | +| `wavg` / `wsum` | numeric | **weighted** by a second column — `Quantity wavg Sales` | +| `cor` / `cov` | numeric | correlation / covariance against a second column | + +`dev` and `sdev` are offered separately because in q they are different +statistics, and the same goes for `var` and `svar`. Nothing here is renamed to +match another backend: there is no `stddev`, no `median`, no `product`. + +`wavg`, `wsum`, `cor` and `cov` take a second column, and appear in the +aggregate menu as a submenu of the columns they can pair with. + +### Filters + +Filter *operators* — `==`, `!=`, `<`, `>`, `<=`, `>=` — are Perspective's +spelling of q's `=`, `<>`, `<`, `>`, `<=`, `>=` and mean the same thing. The +*predicates* are q's: + +| Filter op | q | +| --------- | - | +| `like` | q's `like`, taking **q's** pattern language — `*` and `?`. `%` and `_` are literal characters, not wildcards | +| `in` / `not in` | q's `in` over a vector of values | + +Ordering comparisons (`<`, `>`, `<=`, `>=`) on string columns compare +lexicographically as symbols, which is q's ordering for `symbol` columns and +the intuitive one for the char-list types. + +> If you are used to the DuckDB virtual server, note that `like` patterns are +> **not** translated: `"Bos%"` matches a literal percent sign here. Write +> `"Bos*"`. + +`within` is absent despite being idiomatic q, because Perspective's filter UI +infers an operator's operand count from a fixed list of names and would render +a two-operand range filter as a single value box. + +### Expressions + +Expressions are q, passed through to the q process verbatim — there is no +translation from Perspective's ExprTK-style expression language, so what you +write in the expression editor is q: + +```q +Sales * 0.9 +``` + +```q +10 xbar Sales +``` + +```q +upper City +``` + +An expression may reference any column of the source table by name, provided +that name is a q identifier. Expressions are validated by q as you type: the +error q reports is the error the editor shows. + +An expression's type is resolved by q — you do not declare it — and it behaves +like any other column thereafter, so it can be grouped by, sorted, filtered +and aggregated. An expression whose alias matches a source column shadows it, +matching the SQL virtual servers. + +> **Expressions execute in your q process.** They are passed through +> unmodified, so an expression can call anything q can — including your own +> functions and `system`, which reaches the shell. This is the same trust model +> as the DuckDB virtual server, which inlines SQL fragments, but q's reach is +> wider. Only expose a kdb+ virtual server to clients you would grant query +> access to that process, and disable `expressions` in `get_features` if that +> is not true of your deployment. + +### Window functions + +Window columns are q's own moving and running primitives, under their q names. +A q developer already knows what `mdev` computes; it is not renamed to +`stddev`, which would both rename it and misdescribe it (`mdev` is a +*population* deviation, SQL's `STDDEV_SAMP` a sample one). + +| Window aggregate | q | +| ---------------- | - | +| `msum` `mavg` `mmin` `mmax` | the moving verbs; a cumulative frame takes the running form (`sums`, `avgs`, `mins`, `maxs`) | +| `mcount` | moving count of non-nulls | +| `mdev` | moving **population** deviation | +| `mvar` | its square — q has no moving-variance primitive | +| `first` | the earliest row in the frame | +| `xprev` / `xnext` | shift back / forward by an offset | +| `deltas` | difference from the previous row | +| `ema` | exponential moving average — **rejected by the SQL virtual servers**, which have no recursive `OVER` equivalent | diff --git a/examples/python-kdb-virtual/Dockerfile b/examples/python-kdb-virtual/Dockerfile new file mode 100644 index 0000000000..ee14404574 --- /dev/null +++ b/examples/python-kdb-virtual/Dockerfile @@ -0,0 +1,36 @@ +# ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ +# ┃ ██████ ██████ ██████ █ █ █ █ █ █▄ ▀███ █ ┃ +# ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█ ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄ ▀█ █ ▀▀▀▀▀ ┃ +# ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄ █ ▄▄▄▄▄ ┃ +# ┃ █ ██████ █ ▀█▄ █ ██████ █ ███▌▐███ ███████▄ █ ┃ +# ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫ +# ┃ Copyright (c) 2017, the Perspective Authors. ┃ +# ┃ ╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌ ┃ +# ┃ This file is part of the Perspective library, distributed under the terms ┃ +# ┃ of the [Apache License 2.0](https://www.apache.org/licenses/LICENSE-2.0). ┃ +# ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛ + +FROM --platform=linux/386 i386/debian:bookworm-slim + +RUN apt-get update \ + && apt-get install -y --no-install-recommends unzip \ + && rm -rf /var/lib/apt/lists/* + +# kdb+ is not redistributable, so it is not downloaded here. Fetch the 32-bit +# Linux archive from https://kx.com yourself and leave it beside this file: +# +# examples/python-kdb-virtual/l32.zip +# +# Note the 32-bit terms permit development and proof-of-concept use, but not +# commercial production use. +COPY l32.zip /tmp/l32.zip +RUN unzip -q /tmp/l32.zip -d /opt && rm /tmp/l32.zip + +ENV QHOME=/opt/q + +EXPOSE 5001 + +# `-p` listens on all interfaces, so the port maps out to the host. The +# example's server loads the superstore table over IPC on startup, so this +# process starts empty. +CMD ["/opt/q/l32/q", "-p", "5001"] diff --git a/examples/python-kdb-virtual/README.md b/examples/python-kdb-virtual/README.md new file mode 100644 index 0000000000..9af8386d24 --- /dev/null +++ b/examples/python-kdb-virtual/README.md @@ -0,0 +1,31 @@ +# kdb+ Virtual Server example + +Serves a `` backed by a **kdb+** process, via the +[kdb+ virtual server](../../docs/md/how_to/python/virtual_server/kdb.md). + +## Prerequisites + +A running q process listening on a port. Nothing needs to be in it — this +example loads the superstore dataset over IPC on startup. + +```bash +pip install pykx tornado +``` + +[PyKX](https://code.kx.com/pykx/) connects over IPC in its **unlicensed +mode**, so no kdb+ license is required by this process — only by the q you +connect to. + +## Running + +```bash +pnpm start +``` + +Then open . + +Point elsewhere with `PSP_KDB_HOST` / `PSP_KDB_PORT`: + +```bash +PSP_KDB_PORT=5010 pnpm start +``` diff --git a/examples/python-kdb-virtual/index.html b/examples/python-kdb-virtual/index.html new file mode 100644 index 0000000000..efd7c2dc39 --- /dev/null +++ b/examples/python-kdb-virtual/index.html @@ -0,0 +1,29 @@ + + + + + + + + + + + + diff --git a/examples/python-kdb-virtual/package.json b/examples/python-kdb-virtual/package.json new file mode 100644 index 0000000000..69ce517aaa --- /dev/null +++ b/examples/python-kdb-virtual/package.json @@ -0,0 +1,21 @@ +{ + "name": "python-kdb-virtual", + "private": true, + "version": "5.1.0", + "description": "An example of streaming a `perspective-python` server to the browser.", + "scripts": { + "start": "PYTHONPATH=../../python/perspective python3 server.py" + }, + "keywords": [], + "license": "Apache-2.0", + "dependencies": { + "@perspective-dev/client": "workspace:^", + "@perspective-dev/viewer": "workspace:^", + "@perspective-dev/viewer-charts": "workspace:^", + "@perspective-dev/viewer-datagrid": "workspace:^", + "superstore-arrow": "catalog:" + }, + "devDependencies": { + "npm-run-all": "catalog:" + } +} diff --git a/examples/python-kdb-virtual/server.py b/examples/python-kdb-virtual/server.py new file mode 100644 index 0000000000..719025ece9 --- /dev/null +++ b/examples/python-kdb-virtual/server.py @@ -0,0 +1,152 @@ +# ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ +# ┃ ██████ ██████ ██████ █ █ █ █ █ █▄ ▀███ █ ┃ +# ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█ ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄ ▀█ █ ▀▀▀▀▀ ┃ +# ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄ █ ▄▄▄▄▄ ┃ +# ┃ █ ██████ █ ▀█▄ █ ██████ █ ███▌▐███ ███████▄ █ ┃ +# ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫ +# ┃ Copyright (c) 2017, the Perspective Authors. ┃ +# ┃ ╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌ ┃ +# ┃ This file is part of the Perspective library, distributed under the terms ┃ +# ┃ of the [Apache License 2.0](https://www.apache.org/licenses/LICENSE-2.0). ┃ +# ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛ + +import logging +import os +from pathlib import Path + +import perspective +import perspective.handlers.tornado +import perspective.virtual_servers.kdb +import pyarrow as pa +import pyarrow.parquet as pq +import pykx +import tornado.ioloop +import tornado.web + +from tornado.web import StaticFileHandler + +logging.basicConfig( + level=logging.DEBUG, +) + +logger = logging.getLogger(__name__) + +INPUT_FILE = ( + Path(__file__).parent.resolve() + / "node_modules" + / "superstore-arrow" + / "superstore.parquet" +) + +KDB_HOST = os.environ.get("PSP_KDB_HOST", "localhost") +KDB_PORT = int(os.environ.get("PSP_KDB_PORT", "5001")) + +# Load the table with q's own CSV reader, `0:`. +# +# Every argument is a plain string, so nothing but char vectors crosses IPC. +# Handing PyKX a column dictionary of ten thousand mixed Python values +# instead makes q signal `nyi` while decoding what PyKX encoded, and a +# delimited text blob sidesteps that conversion layer completely — it is also +# how kdb+ ingests bulk data normally. +# +# The type template does the typing, so text columns arrive as *symbols*, +# which is how kdb+ stores low-cardinality text and which exercises the +# handler's symbol paths (`in` filters, `like`, row-path padding). +LOAD_TABLE = """ +{[name; types; names; csv] + (`$name) set flip (`$"\\t" vs names)!(types; "\\t") 0: csv } +""" + +# Arrow type -> q's `0:` type character. +Q_TYPE_CHARS = [ + (pa.types.is_boolean, "B"), + (pa.types.is_date, "D"), + (pa.types.is_timestamp, "P"), + (pa.types.is_floating, "F"), + (pa.types.is_int64, "J"), + (pa.types.is_integer, "I"), +] + + +def q_type_char(arrow_type): + """The `0:` template character for an Arrow type, defaulting to symbol.""" + for predicate, char in Q_TYPE_CHARS: + if predicate(arrow_type): + return char + + return "S" + + +def to_text(column): + """One Arrow column as the text `0:` will parse. + + Built by hand rather than with `pyarrow.csv`, which refuses to write + unquoted values containing a `"` — and superstore's product names are + full of them, as inches. Quoting instead is not an option: `0:` splits on + its delimiter and has no notion of RFC-4180, so it would read the quotes + as part of the value. A tab delimiter over data that contains no tabs + makes the split unambiguous, and a bare `"` passes through as what it is. + + A null becomes an empty field, which `0:` reads as that column's null. + """ + values = column.to_pylist() + if pa.types.is_boolean(column.type): + return ["" if x is None else ("1" if x else "0") for x in values] + + if pa.types.is_timestamp(column.type): + # `0:` wants `2016-11-08D00:00:00`, not the space `str()` gives. + return ["" if x is None else str(x).replace(" ", "D") for x in values] + + return ["" if x is None else str(x) for x in values] + + +if __name__ == "__main__": + try: + conn = pykx.SyncQConnection(host=KDB_HOST, port=KDB_PORT) + except BaseException: + logger.exception( + "Could not reach a q process at %s:%s - start one with `q -p %s`, or " + "point PSP_KDB_HOST / PSP_KDB_PORT elsewhere", + KDB_HOST, + KDB_PORT, + KDB_PORT, + ) + raise + + arrow_table = pq.read_table(str(INPUT_FILE)) + + columns = [to_text(arrow_table[name]) for name in arrow_table.column_names] + rows = "\n".join("\t".join(row) for row in zip(*columns)) + + conn( + LOAD_TABLE, + "data_source_one", + "".join(q_type_char(field.type) for field in arrow_table.schema), + "\t".join(arrow_table.column_names), + rows, + ) + + logger.info("Loaded superstore data into kdb+ at %s:%s", KDB_HOST, KDB_PORT) + + virtual_server = perspective.virtual_servers.kdb.KdbVirtualServer(conn) + app = tornado.web.Application( + [ + ( + r"/websocket", + perspective.handlers.tornado.PerspectiveTornadoHandler, + {"perspective_server": virtual_server}, + ), + (r"/node_modules/(.*)", StaticFileHandler, {"path": "../../node_modules/"}), + ( + r"/(.*)", + StaticFileHandler, + {"path": "./", "default_filename": "index.html"}, + ), + ], + websocket_max_message_size=100 * 1024 * 1024, + ) + + app.listen(3000) + logger.info("Listening on http://localhost:3000") + loop = tornado.ioloop.IOLoop.current() + loop.start() diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 8768f3ca39..a5c89afeed 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -420,6 +420,28 @@ importers: specifier: 'catalog:' version: 4.1.5 + examples/python-kdb-virtual: + dependencies: + '@perspective-dev/client': + specifier: workspace:^ + version: link:../../rust/perspective-js + '@perspective-dev/viewer': + specifier: workspace:^ + version: link:../../rust/perspective-viewer + '@perspective-dev/viewer-charts': + specifier: workspace:^ + version: link:../../packages/viewer-charts + '@perspective-dev/viewer-datagrid': + specifier: workspace:^ + version: link:../../packages/viewer-datagrid + superstore-arrow: + specifier: 'catalog:' + version: 3.2.0 + devDependencies: + npm-run-all: + specifier: 'catalog:' + version: 4.1.5 + examples/python-polars-virtual: dependencies: '@perspective-dev/client': diff --git a/rust/perspective-python/perspective/tests/virtual_servers/test_kdb.py b/rust/perspective-python/perspective/tests/virtual_servers/test_kdb.py new file mode 100644 index 0000000000..6d7bbe226a --- /dev/null +++ b/rust/perspective-python/perspective/tests/virtual_servers/test_kdb.py @@ -0,0 +1,1254 @@ +# ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ +# ┃ ██████ ██████ ██████ █ █ █ █ █ █▄ ▀███ █ ┃ +# ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█ ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄ ▀█ █ ▀▀▀▀▀ ┃ +# ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄ █ ▄▄▄▄▄ ┃ +# ┃ █ ██████ █ ▀█▄ █ ██████ █ ███▌▐███ ███████▄ █ ┃ +# ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫ +# ┃ Copyright (c) 2017, the Perspective Authors. ┃ +# ┃ ╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌ ┃ +# ┃ This file is part of the Perspective library, distributed under the terms ┃ +# ┃ of the [Apache License 2.0](https://www.apache.org/licenses/LICENSE-2.0). ┃ +# ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛ + +"""Tests for the kdb+ virtual server. + +`TestKdb*` are integration tests against a real q process. Start one with + + q -p 5001 + +and point `PSP_KDB_HOST` / `PSP_KDB_PORT` at it; they skip otherwise. +""" + +import os +import socket +import tempfile +import urllib.request + +import pytest + +from perspective.virtual_servers.kdb import ( + Columns, + marker_columns, + q_aggregate, + q_constraints, + q_column, + q_dict, + q_expression_types, + q_hosted_tables, + q_list, + q_string, + q_symbol, + q_table_make_view, + q_table_schema, + q_table_size, + q_view_delete, + q_view_min_max, + q_view_slice, + q_window, + q_window_body, + q_window_types, + q_windows, + sanitize, + sort_specs, + window_specs, +) + +# A stand-in for a q `meta`, covering the types whose Perspective mapping is +# non-obvious: `j` (64-bit, so `float`), `C` (a column of strings) and the +# cast-through types `g` / `z` / `m`. +SCHEMA = { + "Region": "s", + "City": "s", + "Sales": "f", + "Quantity": "j", + "Order Date": "d", + "Stamp": "p", + "Note": "C", + "Flag": "b", + "Id": "g", + "Month": "m", + "Local": "z", +} + + +def columns(schema=None, expressions=None, expression_types=None): + return Columns( + schema if schema is not None else SCHEMA, expressions, expression_types + ) + + +def make_view(config, schema=None, cols=None): + query, _ = q_table_make_view( + "trades", ".psp.v1", config, cols if cols is not None else columns(schema) + ) + return query + + +def columns_of(config, schema=None, cols=None): + _, view_columns = q_table_make_view( + "trades", ".psp.v1", config, cols if cols is not None else columns(schema) + ) + return view_columns + + +class TestKdbQueryLiterals: + def test_symbol_uses_identifier_form(self): + assert q_symbol("Sales") == "`Sales" + + def test_symbol_falls_back_for_non_identifiers(self): + # q identifiers must start with a letter and cannot contain a space, + # which is why the whole builder emits functional qSQL. + assert q_symbol("Product Name") == '`$"Product Name"' + assert q_symbol("Sub-Category") == '`$"Sub-Category"' + assert q_symbol("_leading") == '`$"_leading"' + assert q_symbol("__ROW_PATH_0__") == '`$"__ROW_PATH_0__"' + + def test_string_escapes(self): + assert q_string('a"b') == '"a\\"b"' + assert q_string("a\\b") == '"a\\\\b"' + assert q_string("a\nb") == '"a\\nb"' + + def test_symbol_escapes_injection(self): + # The payload has to stay inside the string literal. + assert q_symbol('x"; delete from `t; /') == '`$"x\\"; delete from `t; /"' + + def test_list_parenthesizes_enlist(self): + # Unparenthesized, q's right-to-left application would let `enlist` + # swallow the following operator. + assert q_list([]) == "()" + assert q_list(["`a"]) == "(enlist `a)" + assert q_list(["`a", "`b"]) == "(`a;`b)" + + def test_dict(self): + assert q_dict([], []) == "()!()" + assert q_dict(["`a"], ["1"]) == "(enlist `a)!(enlist 1)" + assert q_dict(["`a", "`b"], ["1", "2"]) == "(`a;`b)!(1;2)" + + def test_column_projections_cast_to_the_declared_type(self): + assert q_column("Sales", columns()) == "`Sales" + assert q_column("Id", columns()) == "(string;`Id)" + assert q_column("Local", columns()) == '($;"p";`Local)' + assert q_column("Month", columns()) == '($;"d";`Month)' + assert q_column("Note", columns()) == "`Note" + + def test_sanitize(self): + assert sanitize("view-1/2 3") == "view_1_2_3" + assert sanitize("a" * 64) == "a" * 32 + + def test_marker_columns(self): + assert marker_columns(0, False) == [] + assert marker_columns(2, False) == ["pspGid", "pspRp0", "pspRp1"] + # Flat mode has no rollup rows, so no discriminator. + assert marker_columns(2, True) == ["pspRp0", "pspRp1"] + + def test_sort_specs_drops_inactive_and_column_sorts(self): + specs = sort_specs( + { + "sort": [ + ["a", "desc"], + ["b", "none"], + ["c", "col asc"], + ["d", "asc abs"], + ] + } + ) + assert specs == [("a", "desc", False), ("d", "asc", True)] + + +class TestKdbQueryAggregates: + """The advertised aggregate names are q's own primitives, emitted as + written. There is no translation table mapping Perspective's or DuckDB's + vocabulary onto q's.""" + + def test_q_primitives_are_emitted_as_written(self): + for name in ["sum", "avg", "count", "min", "max", "first", "last", "prd"]: + assert q_aggregate(name, "Sales", columns()) == f"({name};`Sales)" + + def test_population_and_sample_are_distinct_choices(self): + # A kdb+ user picks between these deliberately; collapsing them onto + # one `stddev` would hide half of q's model. + assert q_aggregate("dev", "Sales", columns()) == "(dev;`Sales)" + assert q_aggregate("sdev", "Sales", columns()) == "(sdev;`Sales)" + assert q_aggregate("var", "Sales", columns()) == "(var;`Sales)" + assert q_aggregate("svar", "Sales", columns()) == "(svar;`Sales)" + + def test_median_is_med(self): + assert q_aggregate("med", "Sales", columns()) == "(med;`Sales)" + + def test_boolean_predicates(self): + assert q_aggregate("any", "Flag", columns()) == "(any;`Flag)" + assert q_aggregate("all", "Flag", columns()) == "(all;`Flag)" + + def test_count_distinct_is_the_one_compound_spelling(self): + assert q_aggregate("count distinct", "City", columns()) == ( + "(count;(distinct;`City))" + ) + + def test_weighted_aggregates_take_a_second_column(self): + # `Quantity wavg Sales` — the reason a lot of people run kdb+. + assert q_aggregate(["wavg", ["Quantity"]], "Sales", columns()) == ( + "(wavg;`Quantity;`Sales)" + ) + assert q_aggregate(["wsum", ["Quantity"]], "Sales", columns()) == ( + "(wsum;`Quantity;`Sales)" + ) + + def test_correlation_and_covariance(self): + assert q_aggregate(["cor", ["Quantity"]], "Sales", columns()) == ( + "(cor;`Quantity;`Sales)" + ) + assert q_aggregate(["cov", ["Quantity"]], "Sales", columns()) == ( + "(cov;`Quantity;`Sales)" + ) + + def test_multi_aggregate_resolves_an_expression_argument(self): + cols = columns(expressions={"E": "Sales*2"}, expression_types={"E": "f"}) + assert q_aggregate(["wavg", ["E"]], "Sales", cols) == ( + '(wavg;(parse "Sales*2");`Sales)' + ) + + def test_foreign_vocabulary_is_rejected(self): + # Names from the other backends are not q, and are not silently + # translated into it. + for name in ["stddev", "median", "product", "any_value", "distinct_count"]: + with pytest.raises(ValueError, match="Unknown aggregate"): + q_aggregate(name, "Sales", columns()) + + def test_the_advertised_set_is_an_allowlist(self): + # Emitting the name verbatim would otherwise be a way to name any q + # function at all. + with pytest.raises(ValueError, match="Unknown aggregate"): + q_aggregate("system", "Sales", columns()) + + def test_default_aggregate_is_q_native(self): + query = make_view({"columns": ["Sales", "City"], "group_rollup_mode": "total"}) + assert "(sum;`Sales)" in query + assert "(count;`City)" in query + + +class TestKdbQueryFilters: + def constraint(self, column, op, value, cols=None): + return q_constraints( + {"filter": [[column, op, value]]}, cols if cols is not None else columns() + ) + + def test_symbol_equality_enlists_the_constant(self): + # A bare symbol in a parse tree is a *column reference*; enlisting it + # is what makes q read it as a constant. + assert self.constraint("City", "==", "Boston") == ["(in;`City;enlist `Boston)"] + + def test_symbol_inequality(self): + assert self.constraint("City", "!=", "Boston") == [ + "(not;(in;`City;enlist `Boston))" + ] + + def test_like_is_qs_like_with_qs_wildcards(self): + # q's pattern language, not SQL's rewritten into it — `*` and `?` are + # the wildcards, and `%` / `_` are literal characters. + assert self.constraint("City", "like", "Bos*") == ['(like;`City;"Bos*")'] + assert self.constraint("City", "like", "B?s") == ['(like;`City;"B?s")'] + assert self.constraint("City", "like", "50%*") == ['(like;`City;"50%*")'] + + def test_in_takes_a_vector(self): + # A q vector is already a parse-tree constant, so no `enlist` dance. + assert self.constraint("City", "in", ["Boston", "Austin"]) == [ + "(in;`City;(`Boston;`Austin))" + ] + assert self.constraint("Sales", "in", [1, 2.5]) == ["(in;`Sales;(1.0;2.5))"] + + def test_in_with_one_value_still_enlists(self): + assert self.constraint("City", "in", ["Boston"]) == [ + "(in;`City;(enlist `Boston))" + ] + + def test_not_in(self): + assert self.constraint("City", "not in", ["Boston"]) == [ + "(not;(in;`City;(enlist `Boston)))" + ] + + def test_in_over_char_lists(self): + assert self.constraint("Note", "in", ["a", "b"]) == ['(in;`Note;("a";"b"))'] + + def test_in_requires_a_non_empty_list(self): + assert self.constraint("City", "in", []) == [] + assert self.constraint("City", "in", "Boston") == [] + + def test_numeric(self): + assert self.constraint("Sales", ">", 5) == ["(>;`Sales;5.0)"] + assert self.constraint("Quantity", "==", 3) == ["(=;`Quantity;3)"] + + def test_boolean(self): + assert self.constraint("Flag", "==", True) == ["(=;`Flag;1b)"] + + def test_date_is_epoch_arithmetic(self): + assert self.constraint("Order Date", "<", 86_400_000.0) == [ + '(<;`$"Order Date";(1970.01.01+1))' + ] + + def test_datetime_is_epoch_arithmetic(self): + assert self.constraint("Stamp", ">=", 1000.0) == [ + "(>=;`Stamp;(1970.01.01D00:00:00.000000000+1000000000))" + ] + + def test_char_list_equality_uses_in(self): + # `=` over a column of char lists compares character-wise. + assert self.constraint("Note", "==", "hello") == ['(in;`Note;enlist "hello")'] + + def test_char_list_ordering_casts_to_symbol(self): + assert self.constraint("Note", ">", "m") == ["(>;({[x] `$x};`Note);enlist `m)"] + + def test_filter_on_cast_column_uses_the_projection(self): + assert self.constraint("Id", "==", "abc") == ['(in;(string;`Id);enlist "abc")'] + + def test_value_injection_is_escaped(self): + assert self.constraint("City", "==", '"; delete') == [ + '(in;`City;enlist `$"\\"; delete")' + ] + + def test_null_and_array_operands_are_dropped(self): + assert q_constraints({"filter": [["City", "==", None]]}, columns()) == [] + assert q_constraints({"filter": [["City", "==", ["a", "b"]]]}, columns()) == [] + + +class TestKdbQueryFlat: + def test_flat_select(self): + assert make_view({"columns": ["City", "Sales"]}) == ( + ".psp.v1 set (`City;`Sales) xcols " + "?[`trades;();0b;(`City;`Sales)!(`City;`Sales)];" + ) + + def test_flat_select_defaults_to_every_column(self): + assert columns_of({"columns": []}, {"a": "j", "b": "s"}) == ["a", "b"] + + def test_flat_sort_drops_its_scratch_column(self): + query = make_view({"columns": ["Sales"], "sort": [["Sales", "desc"]]}) + assert query == ( + ".psp.v1 set (enlist `Sales) xcols " + "![(enlist `pspSrt0) xdesc " + "?[`trades;();0b;(`Sales;`pspSrt0)!(`Sales;`Sales)];" + "();0b;(enlist `pspSrt0)];" + ) + + def test_total_is_an_aggregate_with_no_by(self): + # `0b` with aggregate expressions collapses to a single row. + assert make_view( + { + "columns": ["Sales"], + "group_rollup_mode": "total", + "aggregates": {"Sales": "sum"}, + } + ) == ( + ".psp.v1 set (enlist `Sales) xcols " + "?[`trades;();0b;(enlist `Sales)!(enlist (sum;`Sales))];" + ) + + def test_aggregate_chain(self): + query = make_view( + { + "columns": ["City"], + "group_rollup_mode": "total", + "aggregates": {"City": "count distinct"}, + } + ) + assert "(count;(distinct;`City))" in query + + def test_unknown_aggregate_raises(self): + with pytest.raises(ValueError, match="Unknown aggregate"): + make_view( + { + "columns": ["Sales"], + "group_rollup_mode": "total", + "aggregates": {"Sales": "median_absolute_deviation"}, + } + ) + + def test_default_aggregate_is_type_aware(self): + query = make_view({"columns": ["Sales", "City"], "group_rollup_mode": "total"}) + assert "(sum;`Sales)" in query + assert "(count;`City)" in query + + +class TestKdbQueryGroupBy: + def test_single_level_rollup(self): + assert make_view( + { + "columns": ["Sales"], + "group_by": ["Region"], + "aggregates": {"Sales": "sum"}, + } + ) == ( + ".psp.v1 set (`pspGid;`pspRp0;`Sales) xcols " + "![(`pspOrd0;`pspRp0) xasc raze (" + "(`pspGid;`pspRp0;`pspOrd0;`Sales) xcols " + "{[t] t,'flip (`pspRp0;`pspGid;`pspOrd0)!" + "(count[t]#`;count[t]#1;count[t]#0)}" + "[?[`trades;();0b;(enlist `Sales)!(enlist (sum;`Sales))]];" + "(`pspGid;`pspRp0;`pspOrd0;`Sales) xcols " + "{[t] t,'flip (`pspGid;`pspOrd0)!(count[t]#0;count[t]#1)}" + "[0!?[`trades;();(enlist `pspRp0)!(enlist `Region);" + "(enlist `Sales)!(enlist (sum;`Sales))]]" + ");();0b;(enlist `pspOrd0)];" + ) + + def test_rollup_emits_one_level_per_depth(self): + query = make_view( + { + "columns": ["Sales"], + "group_by": ["Region", "City"], + "aggregates": {"Sales": "sum"}, + } + ) + # Three levels: total, Region, Region x City. + assert query.count("?[`trades;") == 3 + assert "(enlist `pspRp0)!(enlist `Region)" in query + assert "(`pspRp0;`pspRp1)!(`Region;`City)" in query + + def test_grouping_id_encodes_depth(self): + query = make_view( + { + "columns": ["Sales"], + "group_by": ["Region", "City"], + "aggregates": {"Sales": "sum"}, + } + ) + # `GROUPING_ID` is a bitmask of the columns aggregated away, so a + # level retaining `k` of `n` columns is `2 ** (n - k) - 1`. That is + # what `VirtualDataSlice` decodes each row's depth from. + assert "count[t]#3" in query # total + assert "count[t]#1" in query # Region + assert "count[t]#0" in query # leaf + + def test_row_path_padding_is_typed(self): + # Untyped nulls would not concatenate across levels. + assert "count[t]#`" in make_view({"columns": ["Sales"], "group_by": ["Region"]}) + assert "count[t]#0Nd" in make_view( + {"columns": ["Sales"], "group_by": ["Order Date"]} + ) + assert "count[t]#0Nj" in make_view( + {"columns": ["Sales"], "group_by": ["Quantity"]} + ) + + def test_row_path_padding_matches_the_projected_type(self): + # `Month` projects through a `"d"$` cast, so its pad must be a date + # null and not a month null. + query = make_view({"columns": ["Sales"], "group_by": ["Month"]}) + assert "count[t]#0Nd" in query + assert "0Nm" not in query + + def test_flat_mode_is_a_single_level(self): + assert make_view( + { + "columns": ["Sales"], + "group_by": ["Region"], + "group_rollup_mode": "flat", + "aggregates": {"Sales": "sum"}, + } + ) == ( + ".psp.v1 set (`pspRp0;`Sales) xcols (enlist `pspRp0) xasc " + "(`pspRp0;`Sales) xcols " + "0!?[`trades;();(enlist `pspRp0)!(enlist `Region);" + "(enlist `Sales)!(enlist (sum;`Sales))];" + ) + + def test_flat_mode_has_no_grouping_id(self): + query = make_view( + { + "columns": ["Sales"], + "group_by": ["Region"], + "group_rollup_mode": "flat", + } + ) + assert "pspGid" not in query + assert "pspOrd" not in query + + def test_view_columns_exclude_metadata(self): + assert columns_of({"columns": ["Sales"], "group_by": ["Region", "City"]}) == [ + "Sales" + ] + + +class TestKdbQueryGroupBySort: + def config(self, group_by, direction="desc"): + return { + "columns": ["Sales"], + "group_by": group_by, + "aggregates": {"Sales": "sum"}, + "sort": [["Sales", direction]], + } + + def test_single_level_sorts_on_its_own_aggregate(self): + query = make_view(self.config(["Region"])) + # Depth first, then the sort, then the row path — the tuple that puts + # the total row above its children. + assert ( + "(enlist `pspOrd0) xasc (enlist `pspSrt0) xdesc (enlist `pspRp0) xasc" + in query + ) + assert " lj " not in query + + def test_multi_level_orders_children_under_their_parent(self): + query = make_view(self.config(["Region", "City"])) + assert ( + "(enlist `pspOrd0) xasc (enlist `pspAnc0_0) xdesc " + "(`pspRp0;`pspOrd1) xasc (enlist `pspSrt0) xdesc " + "(enlist `pspRp1) xasc" in query + ) + + def test_multi_level_joins_the_ancestor_aggregate(self): + # A row sorts by its *parent's* aggregate before its own, so sibling + # subtrees stay contiguous. + query = make_view(self.config(["Region", "City"])) + assert ( + "lj (?[`trades;();(enlist `pspRp0)!(enlist `Region);" + "(enlist `pspAnc0_0)!(enlist (sum;`Sales))])" in query + ) + + def test_ancestor_join_is_skipped_without_sorts(self): + query = make_view( + { + "columns": ["Sales"], + "group_by": ["Region", "City"], + "aggregates": {"Sales": "sum"}, + } + ) + assert " lj " not in query + assert "(`pspOrd0;`pspRp0;`pspOrd1;`pspRp1) xasc" in query + + def test_scratch_columns_are_dropped(self): + query = make_view(self.config(["Region", "City"])) + assert query.endswith("();0b;(`pspSrt0;`pspOrd0;`pspOrd1;`pspAnc0_0)];") + + def test_abs_sort_wraps_the_aggregate(self): + query = make_view(self.config(["Region"], "desc abs")) + assert "(abs;(sum;`Sales))" in query + + def test_ascending(self): + query = make_view(self.config(["Region"], "asc")) + assert "(`pspOrd0;`pspSrt0;`pspRp0) xasc" in query + + +class TestKdbQueryExpressions: + """Expressions are q, passed through verbatim. `parse` is the bridge from + expression text to the parse tree functional qSQL takes.""" + + def cols(self, expressions, expression_types): + return columns(expressions=expressions, expression_types=expression_types) + + def test_expression_passes_q_through_parse(self): + cols = self.cols({"Net": "Sales*0.9"}, {"Net": "f"}) + assert q_column("Net", cols) == '(parse "Sales*0.9")' + + def test_expression_is_selectable(self): + cols = self.cols({"Net": "Sales*0.9"}, {"Net": "f"}) + query = make_view({"columns": ["Net"]}, cols=cols) + assert query == ( + ".psp.v1 set (enlist `Net) xcols " + '?[`trades;();0b;(enlist `Net)!(enlist (parse "Sales*0.9"))];' + ) + + def test_expression_text_is_escaped(self): + cols = self.cols({"E": 'x like "a\\"b"'}, {"E": "b"}) + assert q_column("E", cols) == '(parse "x like \\"a\\\\\\"b\\"")' + + def test_expression_shadows_a_source_column(self): + # Matching the SQL handlers, whose `col_name` resolves an alias before + # falling back to quoting it as an identifier. + cols = self.cols({"Sales": "2*Sales"}, {"Sales": "f"}) + assert q_column("Sales", cols) == '(parse "2*Sales")' + + def test_expression_is_groupable_and_pads_by_its_own_type(self): + cols = self.cols({"Bucket": "10 xbar Sales"}, {"Bucket": "f"}) + query = make_view({"columns": ["Quantity"], "group_by": ["Bucket"]}, cols=cols) + assert '(enlist `pspRp0)!(enlist (parse "10 xbar Sales"))' in query + # `f` is a float, so its row-path pad is a float null. + assert "count[t]#0n" in query + + def test_expression_is_filterable_by_its_resolved_type(self): + cols = self.cols({"Net": "Sales*0.9"}, {"Net": "f"}) + assert q_constraints({"filter": [["Net", ">", 5]]}, cols) == [ + '(>;(parse "Sales*0.9");5.0)' + ] + + def test_expression_of_string_type_filters_as_a_string(self): + cols = self.cols({"Upper": "upper City"}, {"Upper": "s"}) + assert q_constraints({"filter": [["Upper", "==", "BOSTON"]]}, cols) == [ + '(in;(parse "upper City");enlist `BOSTON)' + ] + + def test_expression_default_aggregate_follows_its_type(self): + cols = self.cols({"Net": "Sales*0.9"}, {"Net": "f"}) + query = make_view({"columns": ["Net"], "group_by": ["Region"]}, cols=cols) + assert '(sum;(parse "Sales*0.9"))' in query + + def test_untyped_expression_defaults_to_string(self): + # A missing type must not crash the builder; `string` is the safe + # default, as it is for an unknown source column. + cols = self.cols({"E": "x"}, {}) + assert q_constraints({"filter": [["E", "==", "a"]]}, cols) == [ + '(in;(parse "x");enlist `a)' + ] + + def test_expression_types_probe(self): + assert q_expression_types("trades", {"Net": "Sales*0.9"}) == ( + "{[m] m`t}[0!meta ?[(1 sublist get `trades);();0b;" + '(enlist `Net)!(enlist (parse "Sales*0.9"))]]' + ) + + def test_expression_types_probe_is_ordered(self): + query = q_expression_types("trades", {"A": "1+1", "B": "2+2"}) + assert '(`A;`B)!((parse "1+1");(parse "2+2"))' in query + + def test_names_excludes_expression_aliases(self): + # `columns()` defaulting must not re-select an alias as a source + # column — it is not one. + cols = self.cols({"Net": "Sales*0.9"}, {"Net": "f"}) + assert "Net" not in cols.names() + assert columns_of({"columns": []}, cols=cols) == [ + c for c in SCHEMA if c != "Net" + ] + + +class TestKdbQueryWindows: + """Windows are computed on the source before filtering and grouping, so a + window alias is an ordinary column everywhere downstream — mirroring the + SQL translation's `__PSP_WINDOW_SRC__` subquery.""" + + def window(self, **spec): + spec.setdefault("column", "Sales") + return {"W": spec} + + def cols(self, type_char="f"): + # The handler folds resolved window types into the source schema. + return columns({**SCHEMA, "W": type_char}) + + def body(self, **spec): + spec.setdefault("column", "Sales") + return q_window_body(spec) + + def test_running_and_moving_are_one_aggregate(self): + # q spells the running and moving forms of an aggregate differently; + # the frame chooses between them rather than the menu. + assert self.body(aggregate="msum") == "sums v" + assert self.body(aggregate="msum", rows=4) == "5 msum v" + assert self.body(aggregate="mavg") == "avgs v" + assert self.body(aggregate="mmin") == "mins v" + assert self.body(aggregate="mmax") == "maxs v" + + def test_rows_frame_is_one_wider_than_perspectives(self): + # Perspective frames `rows` *preceding* plus the current row; q counts + # the current row as one of its `n`. + assert self.body(aggregate="msum", rows=4) == "5 msum v" + assert self.body(aggregate="mavg", rows=0) == "1 mavg v" + + def test_mcount(self): + # `mcount` is q's moving count of non-nulls; the running case has no + # primitive and accumulates the same predicate. + assert self.body(aggregate="mcount", rows=2) == "3 mcount v" + assert self.body(aggregate="mcount") == "sums `long$not null v" + + def test_mdev_is_the_primitive_not_an_alias(self): + # The menu says `mdev`, so a q developer already knows it is a + # *population* deviation — no rename, no re-derivation. + assert self.body(aggregate="mdev", rows=4) == "5 mdev v" + assert self.body(aggregate="mvar", rows=4) == "{[d] d*d} 5 mdev v" + + def test_cumulative_deviation_is_derived_to_match_mdev(self): + # q has no running `mdev`, so the cumulative case is derived — as a + # population statistic, to agree with the framed case above. + assert self.body(aggregate="mvar") == ( + "{[s1;s2;n] (s2%n)-(s1%n) xexp 2}" + "[sums v;sums (v*v);sums (`long$not null v)]" + ) + assert self.body(aggregate="mdev").startswith("sqrt {[s1;s2;n]") + + def test_xprev_and_xnext(self): + assert self.body(aggregate="xprev", offset=3) == "3 xprev v" + # q has no `xnext` primitive; a negative shift is one. + assert self.body(aggregate="xnext", offset=2) == "-2 xprev v" + + def test_deltas(self): + # `deltas` is the 1-step difference; a wider offset spells out. + assert self.body(aggregate="deltas") == "deltas v" + assert self.body(aggregate="deltas", offset=3) == "v-3 xprev v" + + def test_offset_defaults_to_one(self): + assert self.body(aggregate="xprev") == "1 xprev v" + + def test_ema_is_native(self): + # The SQL translation rejects `ema` outright as recursive — q has it + # as a primitive, under its own name. + assert self.body(aggregate="ema", alpha=0.3) == "0.3 ema v" + + def test_ema_requires_alpha(self): + with pytest.raises(ValueError, match="requires an `alpha`"): + self.body(aggregate="ema") + + def test_range_frame_is_rejected(self): + with pytest.raises(ValueError, match="`range` frames are not supported"): + self.body(aggregate="msum", range=5.0) + + def test_first_is_supported(self): + # The SQL translation rejects `first`; q does it as plain indexing. + assert self.body(aggregate="first") == "first v" + assert self.body(aggregate="first", rows=4) == "v 0|(til count v)-4" + + def test_foreign_vocabulary_is_rejected(self): + # Perspective's and DuckDB's window names are not q's, and are not + # silently translated into it. + for name in ["sum", "stddev", "lag", "lead", "diff", "rate"]: + with pytest.raises(ValueError, match="Unknown window aggregate"): + self.body(aggregate=name) + + def test_window_is_a_lambda_applied_to_its_column(self): + cols = self.cols() + assert q_window(self.window(aggregate="msum")["W"], cols) == ( + "({[v] sums v};`Sales)" + ) + + def test_window_over_an_expression(self): + cols = columns(expressions={"E": "Sales*2"}, expression_types={"E": "f"}) + assert q_window({"column": "E", "aggregate": "msum"}, cols) == ( + '({[v] sums v};(parse "Sales*2"))' + ) + + def test_source_is_a_value_not_a_symbol(self): + # `![`trades;…]` would update the user's table *in place*. + query = make_view( + {"columns": ["W"], "windows": self.window(aggregate="msum")}, + cols=self.cols(), + ) + assert "![(get `trades);" in query + assert "![`trades;" not in query + + def test_unpartitioned_unordered_window(self): + assert make_view( + {"columns": ["W"], "windows": self.window(aggregate="msum")}, + cols=self.cols(), + ) == ( + ".psp.v1 set {[pspSrc] (enlist `W) xcols " + "?[pspSrc;();0b;(enlist `W)!(enlist `W)]}" + "[![(get `trades);();0b;(enlist `W)!(enlist ({[v] sums v};`Sales))]];" + ) + + def test_partition_by_uses_update_by(self): + # `update … by …` broadcasts within each partition and preserves row + # order; `select … by …` would not. + query = make_view( + { + "columns": ["W"], + "windows": self.window(aggregate="msum", partition_by=["Region"]), + }, + cols=self.cols(), + ) + assert "![(get `trades);();(enlist `Region)!(enlist `Region);" in query + + def test_order_by_sorts_then_restores_row_order(self): + # A window's `order_by` orders rows within the frame only — it must + # not reorder the view. + query = make_view( + { + "columns": ["W"], + "windows": self.window(aggregate="msum", order_by=["Stamp", "desc"]), + }, + cols=self.cols(), + ) + assert "(enlist `pspIdx)!(enlist `i)" in query + assert "(enlist `Stamp) xdesc" in query + assert "(enlist `pspIdx) xasc" in query + assert query.count("();0b;(enlist `pspIdx)]") == 1 + + def test_no_index_stamp_without_an_order_by(self): + query = make_view( + {"columns": ["W"], "windows": self.window(aggregate="msum")}, + cols=self.cols(), + ) + assert "pspIdx" not in query + + def test_windows_sharing_a_frame_compute_in_one_pass(self): + query = make_view( + { + "columns": ["A", "B"], + "windows": { + "A": {"column": "Sales", "aggregate": "msum"}, + "B": {"column": "Quantity", "aggregate": "mmax"}, + }, + }, + cols=columns({**SCHEMA, "A": "f", "B": "j"}), + ) + assert "(`A;`B)!(({[v] sums v};`Sales);({[v] maxs v};`Quantity))" in query + + def test_windows_are_emitted_in_alias_order(self): + # The map is unordered; sorting by alias keeps the q deterministic. + query = make_view( + { + "columns": ["A", "B"], + "windows": { + "B": {"column": "Quantity", "aggregate": "mmax"}, + "A": {"column": "Sales", "aggregate": "msum"}, + }, + }, + cols=columns({**SCHEMA, "A": "f", "B": "j"}), + ) + assert "(`A;`B)!(({[v] sums v};`Sales);({[v] maxs v};`Quantity))" in query + + def test_differing_partitions_get_their_own_pass(self): + query = make_view( + { + "columns": ["A", "B"], + "windows": { + "A": {"column": "Sales", "aggregate": "msum"}, + "B": { + "column": "Sales", + "aggregate": "msum", + "partition_by": ["Region"], + }, + }, + }, + cols=columns({**SCHEMA, "A": "f", "B": "f"}), + ) + assert "();0b;(enlist `A)!" in query + assert "();(enlist `Region)!(enlist `Region);(enlist `B)!" in query + + def test_windowed_source_is_bound_once_for_a_rollup(self): + # A rollup references its source once per level; inlining the windowed + # source would recompute the windows for each. + query = make_view( + { + "columns": ["W"], + "group_by": ["Region"], + "aggregates": {"W": "sum"}, + "windows": self.window(aggregate="msum"), + }, + cols=self.cols(), + ) + assert query.count("{[v] sums v}") == 1 + assert query.count("?[pspSrc;") == 2 # one select per rollup level + + def test_window_is_groupable_and_pads_by_its_resolved_type(self): + query = make_view( + { + "columns": ["Sales"], + "group_by": ["W"], + "windows": self.window(aggregate="msum"), + }, + cols=self.cols(), + ) + assert "(enlist `pspRp0)!(enlist `W)" in query + assert "count[t]#0n" in query + + def test_window_is_filterable_by_its_resolved_type(self): + assert q_constraints({"filter": [["W", ">", 100]]}, self.cols()) == [ + "(>;`W;100.0)" + ] + + def test_window_types_probe(self): + assert q_window_types( + "trades", [("W", {"column": "Sales", "aggregate": "msum"})], columns() + ) == ( + "{[m] m`t}[0!meta (enlist `W)#![(1 sublist get `trades);();0b;" + "(enlist `W)!(enlist ({[v] sums v};`Sales))]]" + ) + + def test_specs_are_sorted_by_alias(self): + specs = window_specs({"windows": {"b": {"column": "x"}, "a": {"column": "y"}}}) + assert [alias for alias, _ in specs] == ["a", "b"] + + def test_no_windows_leaves_the_source_alone(self): + assert q_windows("`trades", [], columns()) == "`trades" + + +class TestKdbQueryViews: + def test_hosted_tables(self): + assert q_hosted_tables() == "tables[]" + + def test_table_schema(self): + assert q_table_schema("trades") == "{[m] (m`c;m`t)}[0!meta `trades]" + assert q_table_schema("my table") == '{[m] (m`c;m`t)}[0!meta `$"my table"]' + + def test_table_size(self): + assert q_table_size("trades") == "count get `trades" + + def test_view_slice(self): + assert q_view_slice(".psp.v1", ["pspGid", "Product Name"], 10, 100) == ( + '(`pspGid;`$"Product Name")#(10;100) sublist .psp.v1' + ) + + def test_view_delete_is_protected(self): + # A view the UI has already lost is not an error. + assert q_view_delete("v1_abc") == ("@[{![`.psp;();0b;enlist x]};`v1_abc;()]") + + def test_min_max_numeric(self): + assert q_view_min_max(".psp.v1", "Sales", "f") == ( + "{[c] {[v] `float$v} each (min c;max c)}[flip[.psp.v1][`Sales]]" + ) + + def test_min_max_projects_temporals_to_epoch_millis(self): + # `Scalar` has no temporal variant, so a `datetime` would otherwise + # degrade to null crossing into Rust. + assert '(`float$"j"$v)*86400000' in q_view_min_max(".psp.v1", "d", "d") + assert '(`float$"j"$v)%1e6' in q_view_min_max(".psp.v1", "p", "p") + + def test_min_max_is_undefined_for_strings(self): + assert q_view_min_max(".psp.v1", "City", "s") is None + + +################################################################################ +# +# Integration — requires a running q process. + +KDB_HOST = os.environ.get("PSP_KDB_HOST", "localhost") +KDB_PORT = int(os.environ.get("PSP_KDB_PORT", "5001")) + +_SUPERSTORE_LOCAL = os.path.join( + os.path.dirname(__file__), + "..", + "..", + "..", + "node_modules", + "superstore-arrow", + "superstore.parquet", +) + +_SUPERSTORE_URL = ( + "https://cdn.jsdelivr.net/npm/superstore-arrow@3.2.0/superstore.parquet" +) + + +def _get_superstore_parquet(): + if os.path.exists(_SUPERSTORE_LOCAL): + return _SUPERSTORE_LOCAL + path = os.path.join(tempfile.gettempdir(), "superstore.parquet") + if not os.path.exists(path): + urllib.request.urlretrieve(_SUPERSTORE_URL, path) + return path + + +def _have_pykx(): + try: + import pykx # noqa: F401 + except ImportError: + return False + return True + + +def _q_is_reachable(): + try: + with socket.create_connection((KDB_HOST, KDB_PORT), timeout=1): + return True + except OSError: + return False + + +# Gated per-class rather than per-module: the golden tests above are the ones +# that cover the q translation in CI, and they must run everywhere. +requires_q = pytest.mark.skipif( + not _have_pykx() or not _q_is_reachable(), + reason=f"needs PyKX and a q process at {KDB_HOST}:{KDB_PORT} (`q -p {KDB_PORT}`)", +) + + +@pytest.fixture(scope="module") +def client(): + import pyarrow.parquet as pq + import pykx + + from perspective import Client + from perspective.virtual_servers.kdb import KdbVirtualServer + + db = pykx.SyncQConnection(host=KDB_HOST, port=KDB_PORT) + arrow_table = pq.read_table(_get_superstore_parquet()) + db( + "{[name;cols] (`$name) set flip (`$key cols)!value cols}", + "superstore", + {name: arrow_table[name].to_pylist() for name in arrow_table.column_names}, + ) + + server = KdbVirtualServer(db) + + def handle_request(msg): + session.handle_request(msg) + + def handle_response(msg): + c.handle_response(msg) + + session = server.new_session(handle_response) + c = Client(handle_request) + return c + + +@requires_q +class TestKdbClient: + def test_get_hosted_table_names(self, client): + assert "superstore" in client.get_hosted_table_names() + + +@requires_q +class TestKdbTable: + def test_schema(self, client): + schema = client.open_table("superstore").schema() + assert schema["Sales"] == "float" + assert schema["City"] == "string" + assert schema["Order Date"] == "date" + # `j` is 64-bit, so it maps to `float` rather than `integer`. + assert schema["Row ID"] in ("integer", "float") + + def test_size(self, client): + assert client.open_table("superstore").size() == 9994 + + +@requires_q +class TestKdbView: + def test_flat_columns(self, client): + table = client.open_table("superstore") + view = table.view(columns=["Sales"]) + assert view.to_columns(start_row=0, end_row=3)["Sales"] == pytest.approx( + [261.96, 731.94, 14.62] + ) + + def test_single_group_by(self, client): + table = client.open_table("superstore") + view = table.view( + columns=["Sales"], group_by=["Region"], aggregates={"Sales": "sum"} + ) + assert view.to_records() == [ + {"__ROW_PATH__": [], "Sales": pytest.approx(2297200.860299955)}, + {"__ROW_PATH__": ["Central"], "Sales": pytest.approx(501239.8908000005)}, + {"__ROW_PATH__": ["East"], "Sales": pytest.approx(678781.2399999979)}, + {"__ROW_PATH__": ["South"], "Sales": pytest.approx(391721.9050000003)}, + {"__ROW_PATH__": ["West"], "Sales": pytest.approx(725457.8245000006)}, + ] + + def test_multi_level_group_by_is_depth_first(self, client): + table = client.open_table("superstore") + view = table.view( + columns=["Sales"], + group_by=["Region", "Category"], + aggregates={"Sales": "sum"}, + ) + paths = [row["__ROW_PATH__"] for row in view.to_records()] + assert paths[:5] == [ + [], + ["Central"], + ["Central", "Furniture"], + ["Central", "Office Supplies"], + ["Central", "Technology"], + ] + + def test_group_by_sorted_keeps_subtrees_contiguous(self, client): + table = client.open_table("superstore") + view = table.view( + columns=["Sales"], + group_by=["Region", "Category"], + aggregates={"Sales": "sum"}, + sort=[["Sales", "desc"]], + ) + records = view.to_records() + assert records[0]["__ROW_PATH__"] == [] + # Regions descend by total, and each region's categories follow it. + regions = [ + r["__ROW_PATH__"][0] for r in records[1:] if len(r["__ROW_PATH__"]) == 1 + ] + assert regions == ["West", "East", "Central", "South"] + for index, row in enumerate(records): + if len(row["__ROW_PATH__"]) == 1: + children = records[index + 1 : index + 4] + assert all(len(c["__ROW_PATH__"]) == 2 for c in children) + + def test_filter(self, client): + table = client.open_table("superstore") + view = table.view( + columns=["Sales"], + group_by=["Region"], + aggregates={"Sales": "sum"}, + filter=[["Region", "==", "West"]], + ) + assert [r["__ROW_PATH__"] for r in view.to_records()] == [[], ["West"]] + + def test_expression_passthrough(self, client): + table = client.open_table("superstore") + view = table.view( + columns=["Net"], + expressions={"Net": "Sales*0.9"}, + ) + assert view.to_columns(start_row=0, end_row=2)["Net"] == pytest.approx( + [261.96 * 0.9, 731.94 * 0.9] + ) + + def test_expression_group_by(self, client): + table = client.open_table("superstore") + view = table.view( + columns=["Net"], + group_by=["Region"], + expressions={"Net": "Sales*0.9"}, + aggregates={"Net": "sum"}, + ) + records = view.to_records() + assert records[0]["__ROW_PATH__"] == [] + assert records[0]["Net"] == pytest.approx(2297200.860299955 * 0.9) + + def test_validate_expression_types(self, client): + table = client.open_table("superstore") + assert table.validate_expressions({"a": "Sales*2"})["expression_schema"] == { + "a": "float" + } + + def test_validate_expression_reports_errors(self, client): + table = client.open_table("superstore") + result = table.validate_expressions({"a": "this is not q ("}) + assert "a" in result["errors"] + + def test_weighted_average_aggregate(self, client): + # A q aggregate with no DuckDB counterpart in the menu. + table = client.open_table("superstore") + view = table.view( + columns=["Sales"], + group_by=["Region"], + aggregates={"Sales": ["wavg", ["Quantity"]]}, + ) + records = view.to_records() + assert records[0]["__ROW_PATH__"] == [] + # A weighted mean lies within the range of the values it weights. + assert 0 < records[0]["Sales"] < 10000 + + def test_population_and_sample_deviation_differ(self, client): + table = client.open_table("superstore") + + def deviation(name): + view = table.view( + columns=["Sales"], + group_by=["Region"], + aggregates={"Sales": name}, + ) + return view.to_records()[0]["Sales"] + + population, sample = deviation("dev"), deviation("sdev") + assert population != sample + assert population < sample + + def test_in_filter(self, client): + table = client.open_table("superstore") + view = table.view( + columns=["Sales"], + group_by=["Region"], + aggregates={"Sales": "sum"}, + filter=[["Region", "in", ["West", "East"]]], + ) + assert [r["__ROW_PATH__"] for r in view.to_records()] == [ + [], + ["East"], + ["West"], + ] + + def test_like_uses_q_wildcards(self, client): + table = client.open_table("superstore") + view = table.view( + columns=["Sales"], + group_by=["Region"], + aggregates={"Sales": "sum"}, + filter=[["Region", "like", "*est"]], + ) + assert [r["__ROW_PATH__"] for r in view.to_records()] == [[], ["West"]] + + def test_window_cumulative_sum(self, client): + table = client.open_table("superstore") + view = table.view( + columns=["Sales", "Cum"], + windows={"Cum": {"column": "Sales", "aggregate": "msum"}}, + ) + data = view.to_columns(start_row=0, end_row=3) + assert data["Cum"] == pytest.approx( + [261.96, 261.96 + 731.94, 261.96 + 731.94 + 14.62] + ) + + def test_window_rows_frame_includes_the_current_row(self, client): + # Perspective frames `rows` preceding *plus* the current row, so a + # `rows: 1` moving sum pairs each row with its predecessor. + table = client.open_table("superstore") + view = table.view( + columns=["Moving"], + windows={ + "Moving": {"column": "Sales", "aggregate": "msum", "rows": 1}, + }, + ) + data = view.to_columns(start_row=0, end_row=3) + assert data["Moving"] == pytest.approx( + [261.96, 261.96 + 731.94, 731.94 + 14.62] + ) + + def test_window_lag(self, client): + table = client.open_table("superstore") + view = table.view( + columns=["Prev"], + windows={"Prev": {"column": "Sales", "aggregate": "xprev"}}, + ) + data = view.to_columns(start_row=0, end_row=3) + assert data["Prev"][0] is None + assert data["Prev"][1:] == pytest.approx([261.96, 731.94]) + + def test_window_ema_has_no_sql_counterpart(self, client): + # `ema` is a q primitive; the SQL translation rejects it as recursive. + table = client.open_table("superstore") + view = table.view( + columns=["Ema"], + windows={ + "Ema": {"column": "Sales", "aggregate": "ema", "alpha": 0.5}, + }, + ) + data = view.to_columns(start_row=0, end_row=3) + first = 261.96 + second = 0.5 * 731.94 + 0.5 * first + third = 0.5 * 14.62 + 0.5 * second + assert data["Ema"] == pytest.approx([first, second, third]) + + def test_window_order_by_does_not_reorder_the_view(self, client): + # A window's `order_by` orders rows within the frame only. + table = client.open_table("superstore") + plain = table.view(columns=["Row ID"]).to_columns(start_row=0, end_row=5) + windowed = table.view( + columns=["Row ID"], + windows={ + "W": { + "column": "Sales", + "aggregate": "msum", + "order_by": ["Sales", "desc"], + } + }, + ).to_columns(start_row=0, end_row=5) + assert windowed["Row ID"] == plain["Row ID"] + + def test_window_partition_by(self, client): + table = client.open_table("superstore") + view = table.view( + columns=["Region", "Cum"], + windows={ + "Cum": { + "column": "Sales", + "aggregate": "msum", + "partition_by": ["Region"], + } + }, + ) + data = view.to_columns() + # Each region's running sum ends at that region's total. + totals = {} + for region, value in zip(data["Region"], data["Cum"]): + totals[region] = value + assert totals["West"] == pytest.approx(725457.8245000006) + + def test_window_is_groupable(self, client): + table = client.open_table("superstore") + view = table.view( + columns=["Cum"], + group_by=["Region"], + aggregates={"Cum": "max"}, + windows={"Cum": {"column": "Sales", "aggregate": "msum"}}, + ) + assert view.to_records()[0]["__ROW_PATH__"] == [] + + def test_nulls_are_not_sentinels(self, client): + # q nulls are in-band; without scrubbing an empty long renders as + # -9223372036854775808. + table = client.open_table("superstore") + view = table.view(columns=["Sales"], group_by=["Region"]) + for value in view.to_columns()["Sales"]: + assert value is None or abs(value) < 1e17 diff --git a/rust/perspective-python/perspective/virtual_servers/kdb.py b/rust/perspective-python/perspective/virtual_servers/kdb.py new file mode 100644 index 0000000000..2d228049ae --- /dev/null +++ b/rust/perspective-python/perspective/virtual_servers/kdb.py @@ -0,0 +1,1461 @@ +# ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ +# ┃ ██████ ██████ ██████ █ █ █ █ █ █▄ ▀███ █ ┃ +# ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█ ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄ ▀█ █ ▀▀▀▀▀ ┃ +# ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄ █ ▄▄▄▄▄ ┃ +# ┃ █ ██████ █ ▀█▄ █ ██████ █ ███▌▐███ ███████▄ █ ┃ +# ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫ +# ┃ Copyright (c) 2017, the Perspective Authors. ┃ +# ┃ ╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌ ┃ +# ┃ This file is part of the Perspective library, distributed under the terms ┃ +# ┃ of the [Apache License 2.0](https://www.apache.org/licenses/LICENSE-2.0). ┃ +# ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛ + +import io +import logging +import re +from datetime import date, datetime + +import pyarrow as pa +import pyarrow.compute as pc +from pyarrow import ipc + +import perspective +from perspective.virtual_servers import VirtualServerHandler + +logger = logging.getLogger(__name__) + +# The aggregates this handler advertises are q's own, by their q names — a +# kdb+ user picks `sdev` or `dev` because they are different statistics, and +# `wavg` because weighted aggregates are why they are running kdb+. Names are +# emitted into the query as written, so this doubles as the allowlist: a +# config naming anything else is rejected rather than passed through to q. +NUMBER_AGGS = [ + "sum", + "avg", + "count", + "count distinct", + "min", + "max", + "first", + "last", + "med", + "dev", + "sdev", + "var", + "svar", + "prd", + "any", + "all", +] + +# q aggregates over a second column, e.g. `Size wavg Price`. Advertised via +# `AggSpec::Multiple`, which the UI expands to one entry per column of the +# named type. +NUMBER_MULTI_AGGS = [ + "wavg", + "wsum", + "cor", + "cov", +] + +# `min`/`max` are defined for q temporals but not for symbols. +TEMPORAL_AGGS = [ + "count", + "count distinct", + "min", + "max", + "first", + "last", +] + +STRING_AGGS = [ + "count", + "count distinct", + "first", + "last", +] + +# `sum` over booleans counts the trues — a q idiom worth surfacing. +BOOLEAN_AGGS = [ + "count", + "count distinct", + "first", + "last", + "any", + "all", + "sum", +] + +AGGREGATES = set( + NUMBER_AGGS + NUMBER_MULTI_AGGS + TEMPORAL_AGGS + STRING_AGGS + BOOLEAN_AGGS +) + +# `AggSpec::Multiple` — the extra argument is a numeric column, which the UI +# expands into one menu entry per matching column. +NUMBER_MULTI_AGGS_SPEC = [[name, ["float"]] for name in NUMBER_MULTI_AGGS] + +# Aggregates whose q spelling is more than one primitive. Everything else is +# emitted as the q name itself, so this table stays at the exceptions. +AGGREGATE_CHAINS = { + "count distinct": ("count", "distinct"), +} + +# Comparison operators keep Perspective's spelling of q's `=` and `<>`: they +# are the UI's shared operator vocabulary — the value editor, the arity rules +# and the string autocomplete are all keyed off these exact strings — and they +# denote the same thing in both languages. The *predicates* below are where +# the data model shows: `like` is q's, with q's wildcards. +FILTER_OPS = [ + "==", + "!=", + ">=", + "<=", + ">", + "<", + "in", + "not in", +] + +STRING_FILTER_OPS = FILTER_OPS + ["like"] + +# Window aggregates are q's own primitives, by their q names. A q developer +# reads `mdev` and knows it is a moving *population* deviation; spelling it +# `stddev` would both rename it and misdescribe it. +# +# q's moving verbs take a frame; its running verbs are the cumulative case of +# the same aggregate, so `msum`/`sums` are one entry with two spellings (see +# `WINDOW_VERBS`) rather than two menu items. `range` frames are absent +# throughout: q has no range-framed primitive, which would need an as-of `wj`. +FRAMES = ["rows", "cumulative"] + +WINDOW_AGGREGATES = [ + {"name": "msum", "frames": FRAMES}, + {"name": "mavg", "frames": FRAMES, "result_type": "float"}, + {"name": "mcount", "frames": FRAMES, "result_type": "float"}, + {"name": "mmin", "frames": FRAMES}, + {"name": "mmax", "frames": FRAMES}, + {"name": "mdev", "frames": FRAMES, "result_type": "float"}, + {"name": "mvar", "frames": FRAMES, "result_type": "float"}, + {"name": "first", "frames": FRAMES}, + {"name": "xprev", "offset": True}, + {"name": "xnext", "offset": True}, + {"name": "deltas", "offset": True}, + {"name": "ema", "alpha": True, "result_type": "float"}, +] + +# `mmin`/`mmax` are defined for q temporals but not for symbols, and the +# arithmetic ones for neither. +WINDOW_AGGREGATES_TEMPORAL = [ + {"name": "mcount", "frames": FRAMES, "result_type": "float"}, + {"name": "mmin", "frames": FRAMES}, + {"name": "mmax", "frames": FRAMES}, + {"name": "first", "frames": FRAMES}, + {"name": "xprev", "offset": True}, + {"name": "xnext", "offset": True}, +] + +WINDOW_AGGREGATES_ANY = [ + {"name": "mcount", "frames": FRAMES, "result_type": "float"}, + {"name": "first", "frames": FRAMES}, + {"name": "xprev", "offset": True}, + {"name": "xnext", "offset": True}, +] + +# Each q window aggregate as its (running, moving) pair of primitives — the +# cumulative frame takes the first, a `rows` frame the second. +WINDOW_VERBS = { + "msum": ("sums", "msum"), + "mavg": ("avgs", "mavg"), + "mmin": ("mins", "mmin"), + "mmax": ("maxs", "mmax"), +} + +# The scratch column that restores a window's source ordering. A window's +# `order_by` orders rows within the frame only — it must not reorder the view. +WINDOW_INDEX = "pspIdx" + +# The lambda parameter a windowed source is bound to, so the levels of a +# rollup share one computation of it. +WINDOW_SOURCE = "pspSrc" + +# `meta`'s type character -> Perspective `ColumnType`. Upper case is the list +# variant of the same type; only `C` (a column of strings) is meaningful as a +# Perspective column. Types with no Perspective analogue are projected through +# a cast (see `TO_STRING` / `TO_DATE` / `TO_TIMESTAMP`) so the declared type +# and the type q actually yields cannot drift. +TYPES = { + "b": "boolean", + "x": "integer", + "h": "integer", + "i": "integer", + # `j` is 64-bit; like DuckDB's `BIGINT` it maps to `float` because + # Perspective's `integer` is 32-bit. + "j": "float", + "e": "float", + "f": "float", + "c": "string", + "C": "string", + "s": "string", + "g": "string", + "p": "datetime", + "z": "datetime", + "d": "date", + "m": "date", + "n": "string", + "u": "string", + "v": "string", + "t": "string", +} + +# Projections applied to source columns whose q type has no Perspective +# analogue, keeping `TYPES` honest about what lands in the Arrow payload. +TO_STRING = set("gnuvtc") +TO_DATE = {"m"} +TO_TIMESTAMP = {"z"} + +# `q` nulls are in-band sentinel values, not a validity bitmap. Rollup levels +# pad the row-path columns they don't fill, and the pad must be typed or the +# level tables won't concatenate. +NULLS = { + "b": "0b", + "x": "0x00", + "h": "0Nh", + "i": "0Ni", + "j": "0Nj", + "e": "0Ne", + "f": "0n", + "c": '" "', + "C": '""', + "s": "`", + "g": "0Ng", + "p": "0Np", + "z": "0Nz", + "d": "0Nd", + "m": "0Nm", + "n": "0Nn", + "u": "0Nu", + "v": "0Nv", + "t": "0Nt", +} + +# Views are materialized as globals under a dedicated namespace so a failed +# `view_delete` can never collide with user state. +NAMESPACE = ".psp" + +# Internal column names. They are q-legal identifiers (Perspective's +# `__ROW_PATH_0__` is not — q identifiers must start with a letter), and are +# renamed to the wire names in `view_get_data`. +GROUPING_ID = "pspGid" +ROW_PATH = "pspRp{}" +DEPTH = "pspOrd{}" +SORT = "pspSrt{}" +ANCESTOR = "pspAnc{}_{}" + +IDENTIFIER = re.compile(r"^[a-zA-Z][a-zA-Z0-9_]*$") + +STRING_ESCAPES = { + "\\": "\\\\", + '"': '\\"', + "\n": "\\n", + "\r": "\\r", + "\t": "\\t", +} + + +class KdbVirtualSession: + def __init__(self, callback, db): + self.session = perspective.VirtualServer(KdbVirtualServerHandler(db)) + self.callback = callback + + def handle_request(self, msg): + self.callback(self.session.handle_request(msg)) + + +class KdbVirtualServer: + def __init__(self, db): + self.db = db + + def new_session(self, callback): + return KdbVirtualSession(callback, self.db) + + +class KdbVirtualServerHandler(VirtualServerHandler): + """ + An implementation of a `perspective.VirtualServerHandler` for kdb+. + + `db` is a callable which evaluates a q expression and returns a PyKX + object, e.g. a `pykx.SyncQConnection`. The connected q process must accept + global assignment, as views are materialized as globals under `.psp`. + """ + + def __init__(self, db): + self.db = db + self.views = {} + self.schemas = {} + self.counter = 0 + + def get_features(self): + return { + "group_by": True, + # Phase 2 — the pivot idiom and its `|`-joined column naming are + # not implemented yet. + "split_by": False, + "sort": True, + # Expressions are q, passed through verbatim — there is no ExprTK + # translation, so the expression language here is q's. Note this + # executes client-authored q in the connected process (as the SQL + # handlers execute client-authored SQL); turn it off if that is + # not an acceptable trust boundary for a deployment. + "expressions": True, + "group_rollup_mode": ["rollup", "flat", "total"], + "filter_ops": { + "integer": FILTER_OPS, + "float": FILTER_OPS, + "boolean": FILTER_OPS, + "date": FILTER_OPS, + "datetime": FILTER_OPS, + "string": STRING_FILTER_OPS, + }, + "aggregates": { + "integer": NUMBER_AGGS + NUMBER_MULTI_AGGS_SPEC, + "float": NUMBER_AGGS + NUMBER_MULTI_AGGS_SPEC, + "string": STRING_AGGS, + "boolean": BOOLEAN_AGGS, + "date": TEMPORAL_AGGS, + "datetime": TEMPORAL_AGGS, + }, + "window_aggregates": { + "integer": WINDOW_AGGREGATES, + "float": WINDOW_AGGREGATES, + "date": WINDOW_AGGREGATES_TEMPORAL, + "datetime": WINDOW_AGGREGATES_TEMPORAL, + "string": WINDOW_AGGREGATES_ANY, + "boolean": WINDOW_AGGREGATES_ANY, + }, + } + + def get_hosted_tables(self): + return [to_str(x) for x in to_py(run_query(self.db, q_hosted_tables()))] + + def table_schema(self, table_name, config=None): + return { + name: TYPES[type_char] + for name, type_char in self.q_schema(table_name).items() + } + + def table_size(self, table_name): + return int(to_py(run_query(self.db, q_table_size(table_name)))) + + def view_schema(self, view_name, config=None): + view = self.views.get(view_name) + if view is None: + return self.table_schema(view_name) + return view["schema"] + + def view_size(self, view_name): + view = self.views.get(view_name) + if view is None: + return self.table_size(view_name) + return int(to_py(run_query(self.db, f"count {view['expression']}"))) + + def view_column_size(self, view_name, config=None): + view = self.views.get(view_name) + if view is None: + return len(self.table_schema(view_name)) + return len(view["schema"]) + + def table_validate_expression(self, table_name, expression): + """Type a q expression by asking q, which is also what rejects it — + an unparseable or ill-typed expression raises, and the UI renders the + q error against the offending expression.""" + return TYPES[self.expression_types(table_name, {"x": expression})["x"]] + + def table_make_view(self, table_name, view_name, config): + expressions = config.get("expressions") or {} + cols = Columns( + self.q_schema(table_name), + expressions, + self.expression_types(table_name, expressions), + ) + + # A window alias is an ordinary column of the windowed source by the + # time anything downstream sees it, so folding its type into the + # source schema is all that is needed to make it groupable, + # filterable and sortable. + windows = window_specs(config) + if windows: + cols = Columns( + {**cols.q_types, **self.window_types(table_name, windows, cols)}, + expressions, + cols.expression_types, + ) + + self.counter += 1 + q_name = f"v{self.counter}_{sanitize(view_name)}" + expression = f"{NAMESPACE}.{q_name}" + query, columns = q_table_make_view(table_name, expression, config, cols) + run_query(self.db, query) + + # The view's own `meta` types it, not the source table's: an aggregate + # changes a column's type (`count` over symbols is a long), and an + # expression column has no source column at all. + view_types = self.meta(expression) + self.views[view_name] = { + "q_name": q_name, + "table": table_name, + "expression": expression, + "columns": columns, + "schema": {c: TYPES[view_types[c]] for c in columns}, + "group_by": list(config.get("group_by") or []), + "flat": (config.get("group_rollup_mode") or "rollup") == "flat", + } + + def view_delete(self, view_name): + view = self.views.pop(view_name, None) + if view is not None: + run_query(self.db, q_view_delete(view["q_name"])) + + def view_get_min_max(self, view_name, column_name, config=None): + view = self.views.get(view_name) + if view is None: + return (None, None) + + # A view column is an aggregate of the source column of the same name, + # so the source table's `meta` is what types it — the view itself is a + # `.psp` global, not something `meta` can be asked about by view id. + type_char = self.q_schema(view["table"]).get(column_name) + if type_char is None: + return (None, None) + + query = q_view_min_max(view["expression"], column_name, type_char) + if query is None: + return (None, None) + + low, high = to_py(run_query(self.db, query)) + return (scrub_scalar(low), scrub_scalar(high)) + + def view_get_data(self, view_name, config, schema, viewport, data): + view = self.views.get(view_name) + if view is None: + return + + columns = [c for c in view["columns"] if c in schema] + start_col = viewport.get("start_col") or 0 + end_col = viewport.get("end_col") + columns = ( + columns[start_col:end_col] if end_col is not None else columns[start_col:] + ) + + start_row = viewport.get("start_row") or 0 + end_row = viewport.get("end_row") + if end_row is None: + end_row = self.view_size(view_name) + + length = end_row - start_row + if length <= 0: + return + + markers = marker_columns(len(view["group_by"]), view["flat"]) + query = q_view_slice(view["expression"], markers + columns, start_row, length) + + arrow_table = to_arrow(run_query(self.db, query)) + arrow_table = scrub_nulls(arrow_table) + arrow_table = rename_markers(arrow_table, len(view["group_by"])) + + buf = io.BytesIO() + with ipc.new_stream(buf, arrow_table.schema) as writer: + writer.write_table(arrow_table) + data.from_arrow_ipc(buf.getvalue()) + + ############################################################################ + # + # Internals + + def meta(self, table): + """`{column: q type char}` for any table expression.""" + names, type_chars = to_py(run_query(self.db, q_meta(table))) + return { + to_str(name): type_char + for name, type_char in zip(names, to_type_chars(type_chars)) + } + + def q_schema(self, table_name): + """The source table's q `meta`, memoized — every query builder needs + it to type its literals, casts and row-path padding.""" + if table_name not in self.schemas: + self.schemas[table_name] = { + name: type_char + for name, type_char in self.meta(q_symbol(table_name)).items() + if not name.startswith("__") + } + + return self.schemas[table_name] + + def expression_types(self, table_name, expressions): + """Type every expression alias in one round trip, by projecting them + over a one-row sample and reading the result's `meta`.""" + if not expressions: + return {} + + query = q_expression_types(table_name, expressions) + type_chars = to_type_chars(to_py(run_query(self.db, query))) + return dict(zip(expressions, type_chars)) + + def window_types(self, table_name, windows, cols): + """Type every window alias in one round trip. Window aggregates are + row-count independent in *type*, so a one-row sample suffices.""" + query = q_window_types(table_name, windows, cols) + type_chars = to_type_chars(to_py(run_query(self.db, query))) + return dict(zip([alias for alias, _ in windows], type_chars)) + + +################################################################################ +# +# q literals +# +# `q` has no bound parameters, so every literal is emitted through these — they +# are the injection boundary. Identifiers become symbols (never code), and +# strings are escaped. + + +def q_string(value): + """A q char-list literal.""" + out = "".join(STRING_ESCAPES.get(c, c) for c in str(value)) + return f'"{out}"' + + +def q_symbol(name): + """A q symbol literal, spelled `` `name `` when q's identifier grammar + allows it (Perspective column names like `"Product Name"` do not).""" + if IDENTIFIER.match(name): + return f"`{name}" + return f"`${q_string(name)}" + + +def q_list(items): + """A q list literal, handling the singleton `enlist` case. The parens are + load-bearing — q applies right-to-left, so an unparenthesized + `enlist x` would swallow whatever follows it.""" + if not items: + return "()" + if len(items) == 1: + return f"(enlist {items[0]})" + return "({})".format(";".join(items)) + + +def q_dict(names, values): + """A q dictionary literal, as taken by functional select's `by` and + aggregate arguments.""" + if not names: + return "()!()" + return f"{q_list(names)}!{q_list(values)}" + + +def q_number(value, type_char): + """A numeric literal typed to survive comparison against `type_char`.""" + if type_char in "ef": + return repr(float(value)) + return str(int(value)) + + +def q_temporal(value, type_char): + """A date or timestamp literal, emitted as epoch arithmetic so no + formatting or locale can come between Python and q. Perspective sends + temporal filter operands as epoch milliseconds.""" + ms = to_epoch_ms(value) + if ms is None: + return None + if type_char in ("d", "m"): + return f"(1970.01.01+{ms // 86_400_000})" + return f"(1970.01.01D00:00:00.000000000+{ms * 1_000_000})" + + +def to_epoch_ms(value): + """Coerce a Perspective filter operand to epoch milliseconds. `Scalar` is + a float for temporals, but ISO-8601 strings are accepted too.""" + if isinstance(value, bool): + return None + if isinstance(value, (int, float)): + return int(value) + if isinstance(value, datetime): + return int(value.timestamp() * 1000) + if isinstance(value, date): + return int(datetime(value.year, value.month, value.day).timestamp() * 1000) + if isinstance(value, str): + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + return int(parsed.timestamp() * 1000) + + return None + + +class Columns: + """What every builder needs to know about a name: its q type, and — if it + is an expression alias rather than a real column — the q it stands for. + + Expression aliases shadow source columns of the same name, matching the + SQL handlers, where `col_name` resolves an alias before quoting it as an + identifier. + """ + + def __init__(self, q_types, expressions=None, expression_types=None): + self.q_types = q_types + self.expressions = expressions or {} + self.expression_types = expression_types or {} + + def expression(self, column): + return self.expressions.get(column) + + def type_char(self, column): + if column in self.expressions: + return self.expression_types.get(column, "") + return self.q_types.get(column, "") + + def psp_type(self, column): + return TYPES.get(self.type_char(column), "string") + + def null(self, column): + return NULLS[projected_type(self.type_char(column) or "s")] + + def names(self): + return [c for c in self.q_types if c not in self.expressions] + + +def q_column(column, cols): + """The parse-tree expression for a column, applying the cast that makes + q's type match the one `TYPES` declares. + + An expression alias passes its q through verbatim via `parse`, which is + what turns expression *text* into the parse tree functional qSQL takes — + the same pass-through the SQL handlers do by inlining the fragment. + """ + expression = cols.expression(column) + if expression is not None: + return f"(parse {q_string(expression)})" + + type_char = cols.type_char(column) + symbol = q_symbol(column) + if type_char in TO_STRING: + return f"(string;{symbol})" + if type_char in TO_DATE: + return f'($;"d";{symbol})' + if type_char in TO_TIMESTAMP: + return f'($;"p";{symbol})' + return symbol + + +def projected_type(type_char): + """The q type a column has *after* `q_column`'s cast — the type a row-path + pad has to match for the rollup levels to concatenate.""" + if type_char in TO_STRING: + return "C" + if type_char in TO_DATE: + return "d" + if type_char in TO_TIMESTAMP: + return "p" + return type_char + + +def q_aggregate(aggregate, column, cols): + """An aggregate parse tree over a column. + + The advertised name *is* the q primitive, so it is emitted as written; + `AGGREGATE_CHAINS` covers only the spellings that are more than one + primitive. Validating against the advertised set keeps this from becoming + a way to name arbitrary q functions. + """ + arguments = [] + if isinstance(aggregate, (list, tuple)): + # `Aggregate::MultiAggregate` — `["wavg", ["Size"]]`. + arguments = list(aggregate[1] or []) if len(aggregate) > 1 else [] + aggregate = aggregate[0] if aggregate else "count" + + aggregate = str(aggregate) + if aggregate not in AGGREGATES: + msg = f"Unknown aggregate '{aggregate}'" + raise ValueError(msg) + + tree = q_column(column, cols) + if arguments: + # A binary q aggregate, whose left argument is the weight or the + # second series: `Size wavg Price`. + return f"({aggregate};{q_column(arguments[0], cols)};{tree})" + + for name in reversed(AGGREGATE_CHAINS.get(aggregate, (aggregate,))): + tree = f"({name};{tree})" + + return tree + + +################################################################################ +# +# q query builders +# +# Every builder is a pure `args -> q source` function so the emitted q can be +# asserted against goldens without a q process or a license. + + +def q_hosted_tables(): + return "tables[]" + + +def q_meta(table): + """`meta` as a `(names; types)` pair rather than as a table, so the result + has one unambiguous shape to unpack regardless of how PyKX orients a q + table.""" + return f"{{[m] (m`c;m`t)}}[0!meta {table}]" + + +def q_table_schema(table_name): + return q_meta(q_symbol(table_name)) + + +def q_expression_types(table_name, expressions): + """The q type char of each expression alias, in declaration order. + + Projecting over `1 sublist` keeps this cheap — expressions are row-wise, + so one row types them as well as the whole table does, and an empty table + still yields typed empty vectors. + """ + probe = q_select( + f"(1 sublist get {q_symbol(table_name)})", + [], + "0b", + q_dict( + [q_symbol(alias) for alias in expressions], + [f"(parse {q_string(text)})" for text in expressions.values()], + ), + ) + + return f"{{[m] m`t}}[0!meta {probe}]" + + +def q_table_size(table_name): + return f"count get {q_symbol(table_name)}" + + +def q_select(table, constraints, by, aggregates): + """Functional `?[t;c;b;a]`. Perspective's column names are not q + identifiers, so the text form of qSQL is unavailable to us.""" + return f"?[{table};{q_list(constraints)};{by};{aggregates}]" + + +def q_constraints(config, cols): + """`where` clauses for a view config's filters.""" + constraints = [] + for entry in config.get("filter") or []: + column, op = entry[0], entry[1] + value = entry[2] if len(entry) > 2 else None + constraint = q_constraint(column, op, value, cols) + if constraint is not None: + constraints.append(constraint) + + return constraints + + +def q_literal(value, type_char, psp_type): + """A filter operand as a q literal of the column's type.""" + if psp_type == "boolean": + return "1b" if value else "0b" + if psp_type in ("date", "datetime"): + return q_temporal(value, type_char) + if psp_type == "string": + # Symbols are what kdb+ string columns overwhelmingly are, and an + # unknown type is assumed to be one — the char-list handling would be + # actively wrong against a symbol column. + if type_char in ("s", ""): + return q_symbol(str(value)) + return q_string(value) + + return q_number(value, type_char) + + +def q_constraint(column, op, value, cols): + type_char = cols.type_char(column) + psp_type = TYPES.get(type_char, "string") + expression = q_column(column, cols) + + if op in ("in", "not in"): + # q's `in` takes a vector, which is also a parse-tree constant — no + # `enlist` dance needed once there is more than one value. + if not isinstance(value, (list, tuple)) or not value: + return None + + literals = [q_literal(v, type_char, psp_type) for v in value] + if any(literal is None for literal in literals): + return None + + match = f"(in;{expression};{q_list(literals)})" + return match if op == "in" else f"(not;{match})" + + if value is None or isinstance(value, (list, tuple)): + # A null operand would be `is null`, which is not advertised; a list + # operand only means `in`. + return None + + if psp_type == "string": + return q_string_constraint(expression, op, value, type_char) + + literal = q_literal(value, type_char, psp_type) + if literal is None: + return None + + operator = {"==": "=", "!=": "<>"}.get(op, op) + if operator not in ("=", "<>", "<", ">", "<=", ">="): + return None + + return f"({operator};{expression};{literal})" + + +def q_string_constraint(expression, op, value, type_char): + """String comparisons in q are not the scalar comparisons they look like: + `=` over a column of char lists compares character-wise, and a symbol + constant must be enlisted to distinguish it from a column reference. + """ + is_symbol = type_char in ("s", "") + if op == "like": + # q's `like`, taking q's pattern language — `*` and `?`, not SQL's + # `%` and `_`. The pattern is the user's, passed through as written. + return f"(like;{expression};{q_string(value)})" + + if op in ("==", "!="): + # `in` rather than `=`, which would compare char lists element-wise. + literal = q_literal(value, type_char, "string") + match = f"(in;{expression};enlist {literal})" + return match if op == "==" else f"(not;{match})" + + if op not in ("<", ">", "<=", ">="): + return None + + # Only symbols order lexicographically; char lists compare element-wise, + # so cast through a lambda — inside one, `` `$ `` is ordinary q and not a + # parse-tree column reference. + if not is_symbol: + expression = f"({{[x] `$x}};{expression})" + + return f"({op};{expression};enlist {q_symbol(str(value))})" + + +def q_view_delete(q_name): + """Deleting a view is best-effort — the UI recovers from a missing one, so + a protected evaluation is preferable to an error.""" + return f"@[{{![`{NAMESPACE};();0b;enlist x]}};{q_symbol(q_name)};()]" + + +def q_view_slice(expression, columns, start_row, length): + symbols = q_list([q_symbol(c) for c in columns]) + return f"{symbols}#({start_row};{length}) sublist {expression}" + + +def q_view_min_max(expression, column, type_char): + """`min`/`max` as a float pair. Perspective's `Scalar` has no temporal + variant, so temporals are projected to epoch milliseconds rather than + silently degrading to null on the way through `py_to_scalar`.""" + psp_type = TYPES.get(type_char, "string") + if psp_type == "string": + return None + + if psp_type == "date": + scale = '(`float$"j"$v)*86400000' + elif psp_type == "datetime": + scale = '(`float$"j"$v)%1e6' + else: + scale = "`float$v" + + # `flip` unwraps the table to its column dictionary, so the symbol lookup + # is unambiguously a column and not a row index. + column_expression = f"flip[{expression}][{q_symbol(column)}]" + return f"{{[c] {{[v] {scale}}} each (min c;max c)}}[{column_expression}]" + + +def marker_columns(group_by_len, is_flat): + """The metadata columns `VirtualDataSlice` reads back out of the payload. + Flat mode carries no `__GROUPING_ID__` — every row is a leaf.""" + if group_by_len == 0: + return [] + markers = [] if is_flat else [GROUPING_ID] + return markers + [ROW_PATH.format(i) for i in range(group_by_len)] + + +def sort_specs(config): + """The active row sorts. `col ...` directions sort split-by columns, which + this handler does not advertise.""" + specs = [] + for column, direction in config.get("sort") or []: + if direction == "none" or direction.startswith("col "): + continue + specs.append( + ( + column, + "desc" if direction.startswith("desc") else "asc", + direction.endswith("abs"), + ) + ) + + return specs + + +def q_sort(expression, keys): + """Apply an ordering tuple. q's sorts are stable and it evaluates + right-to-left, so emitting the keys in tuple order gives the leftmost key + the highest precedence.""" + if not keys: + return expression + + runs = [] + for name, direction in keys: + if runs and runs[-1][0] == direction: + runs[-1][1].append(name) + else: + runs.append((direction, [name])) + + for direction, names in reversed(runs): + verb = "xasc" if direction == "asc" else "xdesc" + symbols = q_list([q_symbol(n) for n in names]) + expression = f"{symbols} {verb} {expression}" + + return expression + + +def window_specs(config): + """A config's windows, sorted by alias so the emitted q is + deterministic — the map itself is unordered.""" + return sorted((config.get("windows") or {}).items()) + + +def window_frame(spec): + """The q window width, or `None` for a cumulative frame. + + Perspective frames `rows` *preceding* plus the current row, so a q window + — which counts the current row as one of its `n` — is one wider. + """ + if spec.get("range") is not None: + raise ValueError( + "window `range` frames are not supported by the kdb+ handler; q has no " + "range-framed primitive (an as-of `wj` would be required)" + ) + + rows = spec.get("rows") + return None if rows is None else int(rows) + 1 + + +def q_window_accumulate(operand, verbs, width): + """Apply a window's (cumulative, moving) verb pair to an operand.""" + cumulative, moving = verbs + if width is None: + return f"{cumulative} {operand}" + return f"{width} {moving} {operand}" + + +def q_window_body(spec): + """A window aggregate as q over the bound source vector `v`.""" + aggregate = spec.get("aggregate") + width = window_frame(spec) + offset = int(spec.get("offset") or 1) + + if aggregate in WINDOW_VERBS: + return q_window_accumulate("v", WINDOW_VERBS[aggregate], width) + + if aggregate == "mcount": + # `mcount` counts the non-nulls in each window; the running case has + # no primitive, so it accumulates the same predicate. + if width is not None: + return f"{width} mcount v" + return "sums `long$not null v" + + if aggregate in ("mdev", "mvar"): + # `mdev` is a population statistic, and `mvar` is its square — q has + # no moving-variance primitive, and no running form of either, so the + # cumulative case is derived to agree with `mdev`. + if width is not None: + deviation = f"{width} mdev v" + return deviation if aggregate == "mdev" else f"{{[d] d*d}} {deviation}" + + sums = "sums v" + squares = "sums (v*v)" + counts = "sums (`long$not null v)" + variance = f"{{[s1;s2;n] (s2%n)-(s1%n) xexp 2}}[{sums};{squares};{counts}]" + return variance if aggregate == "mvar" else f"sqrt {variance}" + + if aggregate == "first": + # The earliest row still inside the frame — index arithmetic, which is + # null-safe in a way that an `xprev`-and-patch would not be. + if width is not None: + return f"v 0|(til count v)-{width - 1}" + return "first v" + + if aggregate == "xprev": + return f"{offset} xprev v" + + if aggregate == "xnext": + # q has no `xnext` primitive; a negative shift is one. + return f"{-offset} xprev v" + + if aggregate == "deltas": + # `deltas` is the 1-step difference; the general offset spells out. + return "deltas v" if offset == 1 else f"v-{offset} xprev v" + + if aggregate == "ema": + alpha = spec.get("alpha") + if alpha is None: + raise ValueError("window `ema` requires an `alpha`") + return f"{float(alpha)!r} ema v" + + msg = f"Unknown window aggregate '{aggregate}'" + raise ValueError(msg) + + +def q_window(spec, cols): + """One window as a parse tree: a q lambda applied to its source column. + + A lambda rather than an inline parse tree because inside one the body is + ordinary q — no `enlist`-the-constant rules, and the source vector is + bound once however many times the aggregate needs it. + """ + return f"({{[v] {q_window_body(spec)}}};{q_column(spec['column'], cols)})" + + +def q_windows(table, windows, cols): + """Extend `table` with a column per window. + + Windows are computed on the source, before filtering and grouping, so a + window alias is an ordinary column everywhere downstream — mirroring the + SQL translation's `__PSP_WINDOW_SRC__` subquery. + + Windows sharing a partition and an ordering are computed in one pass, and + an ordered pass sorts the source, computes, then restores the original + row order: `order_by` orders rows within the frame, not the view. + """ + if not windows: + return table + + groups = {} + for alias, spec in windows: + order_by = spec.get("order_by") + key = ( + tuple(order_by) if order_by else None, + tuple(spec.get("partition_by") or []), + ) + groups.setdefault(key, []).append((alias, spec)) + + ordered = any(order_by for order_by, _ in groups) + if ordered: + table = q_update(table, [WINDOW_INDEX], ["`i"]) + + for (order_by, partition_by), specs in groups.items(): + if order_by: + column, direction = order_by + table = q_sort(table, [(column, direction)]) + + by = ( + q_dict( + [q_symbol(c) for c in partition_by], + [q_column(c, cols) for c in partition_by], + ) + if partition_by + else "0b" + ) + + # `update ... by ...` broadcasts within each partition and preserves + # row order, which `select ... by ...` would not. + table = q_update( + table, + [alias for alias, _ in specs], + [q_window(spec, cols) for _, spec in specs], + by, + ) + + if order_by: + table = q_sort(table, [(WINDOW_INDEX, "asc")]) + + if ordered: + table = q_drop(table, [WINDOW_INDEX]) + + return table + + +def q_window_types(table_name, windows, cols): + """The q type char of each window alias, in sorted-alias order. Typing is + row-count independent, so a one-row sample is enough.""" + source = q_windows(f"(1 sublist get {q_symbol(table_name)})", windows, cols) + aliases = q_list([q_symbol(alias) for alias, _ in windows]) + return f"{{[m] m`t}}[0!meta {aliases}#{source}]" + + +def q_update(table, names, values, by="0b"): + """Functional update, adding or replacing columns.""" + dictionary = q_dict([q_symbol(n) for n in names], values) + return f"![{table};();{by};{dictionary}]" + + +def q_set(target, expression, table_name, windows, cols): + """The statement materializing a view as a global. + + With windows, the view query is wrapped in a lambda taking the windowed + source, so however many times the query references it — once per rollup + level — the windows are computed once. + """ + if not windows: + return f"{target} set {expression};" + + source = q_windows(f"(get {q_symbol(table_name)})", windows, cols) + return f"{target} set {{[{WINDOW_SOURCE}] {expression}}}[{source}];" + + +def q_table_make_view(table_name, target, config, cols): + """Materialize a view as a global, returning `(statement, columns)`. + + A rollup is `n + 1` grouped selects — one per level — padded to a common + schema and concatenated, which is how a `GROUP BY ROLLUP` is spelled in a + language that has no `GROUPING SETS`. + """ + columns = [c for c in (config.get("columns") or []) if c] + group_by = list(config.get("group_by") or []) + mode = config.get("group_rollup_mode") or "rollup" + aggregates = config.get("aggregates") or {} + sorts = sort_specs(config) + constraints = q_constraints(config, cols) + windows = window_specs(config) + + # A rollup references its source once per level, so a windowed source is + # bound as a lambda argument rather than inlined — otherwise every level + # would recompute the windows. + table = WINDOW_SOURCE if windows else q_symbol(table_name) + + if not columns: + columns = [c for c in cols.names() if c not in group_by] + + is_total = mode == "total" + is_flat = mode == "flat" + grouped = bool(group_by) + levels = [] + + if not grouped: + # A flat or total select is a single level with no metadata columns. + if is_total: + selects = {c: q_aggregate_for(c, aggregates, cols) for c in columns} + else: + selects = {c: q_column(c, cols) for c in columns} + + sort_selects = { + SORT.format(j): q_sort_expression( + column, is_abs, aggregates, cols, aggregate=is_total + ) + for j, (column, _, is_abs) in enumerate(sorts) + } + selects.update(sort_selects) + expression = q_select( + table, + constraints, + "0b", + q_dict( + [q_symbol(c) for c in selects], + list(selects.values()), + ), + ) + + keys = ( + [] + if is_total + else [(SORT.format(j), d) for j, (_, d, _) in enumerate(sorts)] + ) + expression = q_sort(expression, keys) + expression = q_drop(expression, list(sort_selects)) + expression = q_reorder(expression, columns) + return (q_set(target, expression, table_name, windows, cols), columns) + + n = len(group_by) + depth_levels = [n] if is_flat else range(n + 1) + for k in depth_levels: + levels.append( + q_level( + table, + constraints, + aggregates, + cols, + group_by, + columns, + sorts, + k, + n, + is_flat, + ) + ) + + expression = f"raze {q_list(levels)}" if len(levels) > 1 else levels[0] + + # Sibling groups must sort as blocks, which means ordering a row by its + # ancestors' aggregates before its own — the ancestor's value lives in the + # level table one deeper than the ordering key, so join it back on. + if sorts: + for i in range(n - 1): + ancestors = q_ancestors( + table, constraints, aggregates, cols, group_by, sorts, i + ) + expression = f"({expression}) lj ({ancestors})" + + keys = [] + for i in range(n): + if not is_flat: + keys.append((DEPTH.format(i), "asc")) + for j, (_, direction, _) in enumerate(sorts): + name = SORT.format(j) if i == n - 1 else ANCESTOR.format(i, j) + keys.append((name, direction)) + keys.append((ROW_PATH.format(i), "asc")) + + expression = q_sort(expression, keys) + + helpers = [SORT.format(j) for j in range(len(sorts))] + if not is_flat: + helpers += [DEPTH.format(i) for i in range(n)] + if sorts: + helpers += [ + ANCESTOR.format(i, j) for i in range(n - 1) for j in range(len(sorts)) + ] + + expression = q_drop(expression, helpers) + expression = q_reorder(expression, marker_columns(n, is_flat) + columns) + return (q_set(target, expression, table_name, windows, cols), columns) + + +def q_level( + table, constraints, aggregates, cols, group_by, columns, sorts, k, n, is_flat +): + """One rollup level: grouped by the first `k` group-by columns, padded with + typed nulls for the levels it does not reach.""" + by = q_dict( + [q_symbol(ROW_PATH.format(i)) for i in range(k)], + [q_column(group_by[i], cols) for i in range(k)], + ) + + selects = {c: q_aggregate_for(c, aggregates, cols) for c in columns} + for j, (column, _, is_abs) in enumerate(sorts): + selects[SORT.format(j)] = q_sort_expression( + column, is_abs, aggregates, cols, aggregate=True + ) + + aggregate_dict = q_dict([q_symbol(c) for c in selects], list(selects.values())) + # A `by` of `0b` with aggregate expressions collapses to the single total + # row, which is exactly the `k == 0` level. + expression = q_select(table, constraints, "0b" if k == 0 else by, aggregate_dict) + if k > 0: + expression = f"0!{expression}" + + pad_names = [ROW_PATH.format(i) for i in range(k, n)] + pad_values = [cols.null(group_by[i]) for i in range(k, n)] + order = marker_columns(n, is_flat) + if not is_flat: + # A `GROUPING_ID` bitmask over `n` columns with the trailing `n - k` + # aggregated away — the encoding `VirtualDataSlice` decodes depth from. + pad_names.append(GROUPING_ID) + pad_values.append(str(2 ** (n - k) - 1)) + # How deep this row sits, clamped per level. Ordering by it ahead of + # each level's key is what interleaves a subtotal before its children. + pad_names += [DEPTH.format(i) for i in range(n)] + pad_values += [str(min(k, i + 1)) for i in range(n)] + order = order + [DEPTH.format(i) for i in range(n)] + + if pad_names: + expression = q_pad(expression, pad_names, pad_values) + + return q_reorder(expression, order + list(selects)) + + +def q_ancestors(table, constraints, aggregates, cols, group_by, sorts, i): + """A keyed table of the level-`i + 1` sort aggregates, for `lj` onto the + concatenated levels.""" + by = q_dict( + [q_symbol(ROW_PATH.format(x)) for x in range(i + 1)], + [q_column(group_by[x], cols) for x in range(i + 1)], + ) + + selects = { + ANCESTOR.format(i, j): q_sort_expression( + column, is_abs, aggregates, cols, aggregate=True + ) + for j, (column, _, is_abs) in enumerate(sorts) + } + + return q_select( + table, + constraints, + by, + q_dict([q_symbol(c) for c in selects], list(selects.values())), + ) + + +def q_aggregate_for(column, aggregates, cols): + aggregate = aggregates.get(column) + if aggregate is None: + psp_type = cols.psp_type(column) + aggregate = "sum" if psp_type in ("integer", "float") else "count" + + return q_aggregate(aggregate, column, cols) + + +def q_sort_expression(column, is_abs, aggregates, cols, aggregate): + expression = q_column(column, cols) + if aggregate: + expression = q_aggregate_for(column, aggregates, cols) + return f"(abs;{expression})" if is_abs else expression + + +def q_pad(expression, names, values): + """Append constant metadata columns. + + Deliberately *not* a functional update: inside a parse tree a symbol is a + column reference, so the symbol null a row-path pad needs would be read as + a column named `""`. Building the columns inside a lambda keeps them + literals, and `count[t]#` broadcasts them explicitly. + """ + columns = q_dict([q_symbol(n) for n in names], [f"count[t]#{v}" for v in values]) + return f"{{[t] t,'flip {columns}}}[{expression}]" + + +def q_drop(expression, names): + """Functional delete.""" + if not names: + return expression + return f"![{expression};();0b;{q_list([q_symbol(n) for n in names])}]" + + +def q_reorder(expression, columns): + """`xcols` over the full column list, pinning an order the levels can be + concatenated in.""" + if not columns: + return expression + return f"{q_list([q_symbol(c) for c in columns])} xcols {expression}" + + +def sanitize(view_name): + """Perspective's view names are opaque; q's globals are identifiers.""" + return re.sub(r"[^A-Za-z0-9]", "_", view_name)[:32] + + +################################################################################ +# +# kdb+ Utils + + +def run_query(db, query): + query = " ".join(query.split()) + try: + result = db(query) + except Exception as e: + logger.error(e) + logger.error(f"{query}") + raise e + else: + logger.debug(f"{query}") + return result + + +def to_py(result): + return result.py() if hasattr(result, "py") else result + + +def to_arrow(result): + return result.pa() if hasattr(result, "pa") else result + + +def to_str(value): + """PyKX yields symbols and chars as `bytes`.""" + if isinstance(value, bytes): + return value.decode("utf-8") + return str(value) + + +def to_type_chars(value): + """Normalize a q char vector to a list of single-character strings. PyKX + may hand one back as `bytes`, which iterates to the same chars once + decoded.""" + if isinstance(value, (bytes, str)): + return list(to_str(value)) + return [to_str(char) for char in value] + + +def scrub_scalar(value): + """q's null and infinity sentinels are in-band, so a `min`/`max` of an + empty or all-null column comes back as a sentinel rather than as null.""" + if value is None: + return None + value = float(value) + if value != value or abs(value) == float("inf"): + return None + if abs(value) >= 9.0e18: + return None + return value + + +def scrub_nulls(arrow_table): + """Replace q's in-band null sentinels with Arrow nulls. + + q has no validity bitmap — a null long *is* `INT64_MIN`. Perspective + treats a value as missing only when the Arrow validity bit says so, so + without this pass an empty cell renders as -9223372036854775808. + """ + sentinels = { + pa.int8(): -(2**7), + pa.int16(): -(2**15), + pa.int32(): -(2**31), + pa.int64(): -(2**63), + } + + columns = [scrub_column(column, sentinels) for column in arrow_table.columns] + schema = pa.schema( + [ + pa.field(name, column.type) + for name, column in zip(arrow_table.column_names, columns) + ] + ) + + return pa.Table.from_arrays(columns, schema=schema) + + +def scrub_column(column, sentinels): + dtype = column.type + + if pa.types.is_dictionary(dtype): + # Decode rather than re-encode: `VirtualDataSlice` dictionary-encodes + # `Utf8` itself, so the round trip would buy nothing. + column = column.cast(pa.string()) + dtype = column.type + + null = pa.scalar(None, type=dtype) + + if pa.types.is_floating(dtype): + return pc.if_else(pc.is_nan(column), null, column) + + if dtype in sentinels: + return pc.if_else(pc.equal(column, sentinels[dtype]), null, column) + + if pa.types.is_temporal(dtype): + # `0Nd` / `0Np` are the minimum value of the underlying integer. + width = 32 if pa.types.is_date32(dtype) or pa.types.is_time32(dtype) else 64 + underlying = pa.int32() if width == 32 else pa.int64() + mask = pc.equal(column.cast(underlying, safe=False), -(2 ** (width - 1))) + return pc.if_else(mask, null, column) + + if pa.types.is_string(dtype) or pa.types.is_large_string(dtype): + # The empty symbol is q's symbol null. + return pc.if_else(pc.equal(column, ""), null, column) + + return column + + +def rename_markers(arrow_table, group_by_len): + """Rename the q-legal metadata columns to the names `VirtualDataSlice` + looks for. q identifiers cannot start with an underscore, so the wire + names can only be applied here.""" + if group_by_len == 0: + return arrow_table + + renames = {GROUPING_ID: "__GROUPING_ID__"} + for i in range(group_by_len): + renames[ROW_PATH.format(i)] = f"__ROW_PATH_{i}__" + + return arrow_table.rename_columns( + [renames.get(name, name) for name in arrow_table.column_names] + )