diff --git a/README.md b/README.md index 83692593ac..a752d2d726 100644 --- a/README.md +++ b/README.md @@ -49,7 +49,8 @@ Node.js and Rust — or delegates to a database you already have. server-side, streaming only what's visible. - Virtual servers that run Perspective's UI directly on external engines like - [DuckDB](https://duckdb.org/), [ClickHouse](https://clickhouse.com/) and + [DuckDB](https://duckdb.org/), [ClickHouse](https://clickhouse.com/), + [PostgreSQL](https://www.postgresql.org/) and [Polars](https://pola.rs/), translating view configurations into native queries — no ETL or data copy required. @@ -76,6 +77,7 @@ Node.js and Rust — or delegates to a database you already have. - [`perspective.handlers.tornado`](https://perspective-dev.github.io/python/perspective/handlers/tornado.html) - [`perspective.virtual_servers.clickhouse`](https://perspective-dev.github.io/python/perspective/virtual_servers/clickhouse.html) - [`perspective.virtual_servers.duckdb`](https://perspective-dev.github.io/python/perspective/virtual_servers/duckdb.html) + - [`perspective.virtual_servers.postgres`](https://perspective-dev.github.io/python/perspective/virtual_servers/postgres.html) - Rust API - [`perspective`](https://docs.rs/perspective/latest/perspective/) diff --git a/docs/md/FAQ.md b/docs/md/FAQ.md index adeeafffac..55c08702f8 100644 --- a/docs/md/FAQ.md +++ b/docs/md/FAQ.md @@ -157,7 +157,8 @@ const view = await table.view({ ``` Window Columns are supported by Perspective's built-in engine, by the DuckDB, -ClickHouse and Polars [Virtual Servers](./explanation/virtual_servers.md), and +ClickHouse, PostgreSQL and Polars +[Virtual Servers](./explanation/virtual_servers.md), and by the `` UI. They update incrementally as the `Table` updates. @@ -373,9 +374,9 @@ No. The WebSocket `Server` is not a security boundary. Every connected `Client` is treated as the author of the queries it submits, and is permitted to create and delete `Table`/`View` resources, author arbitrary [expression columns](./explanation/view/config/expressions.md), and — for -[Virtual Server](./explanation/virtual_servers.md) backends like DuckDB or -ClickHouse — author SQL fragments executed under the configured database -role. The bundled WebSocket adapters +[Virtual Server](./explanation/virtual_servers.md) backends like DuckDB, +ClickHouse or PostgreSQL — author SQL fragments executed under the configured +database role. The bundled WebSocket adapters (`tornado.py`/`aiohttp.py`/`starlette.py`/`WebSocketServer`) are reference integrations and do not authenticate, authorize, or enforce origin policy. diff --git a/docs/md/SUMMARY.md b/docs/md/SUMMARY.md index 3c3a57c90b..7a0f732fe3 100644 --- a/docs/md/SUMMARY.md +++ b/docs/md/SUMMARY.md @@ -66,6 +66,7 @@ - [DuckDB](./how_to/python/virtual_server/duckdb.md) - [ClickHouse](./how_to/python/virtual_server/clickhouse.md) - [Polars](./how_to/python/virtual_server/polars.md) + - [PostgreSQL](./how_to/python/virtual_server/postgres.md) - [Custom](./how_to/python/virtual_server/custom.md) # Rust diff --git a/docs/md/explanation/view/config/windows.md b/docs/md/explanation/view/config/windows.md index 597faf728a..b582a38481 100644 --- a/docs/md/explanation/view/config/windows.md +++ b/docs/md/explanation/view/config/windows.md @@ -176,7 +176,7 @@ first row has no predecessor to difference against: ## Support Window Columns are implemented by Perspective's built-in engine, by the -DuckDB, ClickHouse and Polars +DuckDB, ClickHouse, PostgreSQL and Polars [Virtual Servers](../../virtual_servers.md), and by the `` UI. Virtual Servers advertise support through their _features_ declaration, so the UI control is hidden for backends which do not diff --git a/docs/md/explanation/virtual_servers.md b/docs/md/explanation/virtual_servers.md index d39a773e6a..a888a75559 100644 --- a/docs/md/explanation/virtual_servers.md +++ b/docs/md/explanation/virtual_servers.md @@ -57,6 +57,8 @@ Perspective ships with virtual server implementations for: - **ClickHouse** — query a ClickHouse server from the browser ([JavaScript](../how_to/javascript/virtual_server/clickhouse.md)) or from Python ([Python](../how_to/python/virtual_server/clickhouse.md)). +- **PostgreSQL** — query a PostgreSQL server (16 or later) from Python + ([Python](../how_to/python/virtual_server/postgres.md)). ## Custom implementations diff --git a/docs/md/how_to/python/virtual_server.md b/docs/md/how_to/python/virtual_server.md index f7b4333d12..53010ad6ba 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. +- [**PostgreSQL**](./virtual_server/postgres.md) — query a PostgreSQL server + (16 or later) using the `psycopg` 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/custom.md b/docs/md/how_to/python/virtual_server/custom.md index 20dc1e071b..d24e273a98 100644 --- a/docs/md/how_to/python/virtual_server/custom.md +++ b/docs/md/how_to/python/virtual_server/custom.md @@ -115,6 +115,6 @@ app = tornado.web.Application([ ]) ``` -The built-in [DuckDB](./duckdb.md), [ClickHouse](./clickhouse.md) and -[Polars](./polars.md) implementations all follow exactly this shape and are -worth reading as complete references. +The built-in [DuckDB](./duckdb.md), [ClickHouse](./clickhouse.md), +[PostgreSQL](./postgres.md) and [Polars](./polars.md) implementations all +follow exactly this shape and are worth reading as complete references. diff --git a/docs/md/how_to/python/virtual_server/postgres.md b/docs/md/how_to/python/virtual_server/postgres.md new file mode 100644 index 0000000000..05b0e6445f --- /dev/null +++ b/docs/md/how_to/python/virtual_server/postgres.md @@ -0,0 +1,84 @@ +# PostgreSQL Virtual Server + +Perspective provides a built-in virtual server for +[PostgreSQL](https://www.postgresql.org/), allowing `` +clients to query a PostgreSQL server over WebSocket. + +Requires PostgreSQL 16 or later. + +## Installation + +```bash +pip install perspective-python "psycopg[binary]" +``` + +## Usage + +Create a server that exposes PostgreSQL tables to browser clients: + +```python +import tornado.web +import tornado.ioloop +from perspective.virtual_servers.postgres import PostgresVirtualServer +from perspective.handlers.tornado import PerspectiveTornadoHandler + +# Create virtual server backed by PostgreSQL. Each browser session opens its +# own connection with this DSN. +server = PostgresVirtualServer("postgresql://user@localhost:5432/mydb") + +# 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 (table names are schema-qualified): + +```javascript +const websocket = await perspective.websocket("ws://localhost:8080/websocket"); +const table = await websocket.open_table("public.my_table"); +document.getElementById("viewer").load(table); +``` + +The server is read-only with respect to your data: each viewer session +materializes its queries as connection-scoped `TEMPORARY VIEW`s, which +PostgreSQL drops automatically when the session disconnects. + +## Aggregates + +Aggregates are PostgreSQL's own functions, under their PostgreSQL names, and +each column type advertises only the aggregates PostgreSQL defines for it — +for example `bit_and`/`bit_or`/`bit_xor` on integers only, and +`bool_and`/`bool_or`/`every` (rather than `min`/`max`) on booleans. +`any_value` is the default for columns with no explicit aggregate, which is +why PostgreSQL 16 is required. + +## Window functions + +Window columns are PostgreSQL's own functions, under their PostgreSQL names — +the advertised name is emitted into the `OVER` clause verbatim. + +| | | +| -------------------- | ------------------------------------------------------------------- | +| Aggregating | `sum` `avg` `count` `min` `max` | +| Deviation / variance | `stddev_samp` `stddev_pop` `var_samp` `var_pop` | +| Navigation | `first_value` `last_value` `nth_value` `lag` `lead` | +| Ranking | `row_number` `rank` `dense_rank` `percent_rank` `cume_dist` `ntile` | +| Perspective's own | `diff` | + +`range` frames require a numeric order key in PostgreSQL, so they are +advertised for numeric column types only. + +## Limitations + +- **Split by** is not supported — PostgreSQL has no `PIVOT` statement. +- Natural-order (unsorted) window functions are not supported, since + PostgreSQL has no stable row identity; window columns require an explicit + order key. + +## Examples + +- [Python PostgreSQL example](https://github.com/perspective-dev/perspective/tree/master/examples/python-postgres-virtual) diff --git a/examples/python-postgres-virtual/index.html b/examples/python-postgres-virtual/index.html new file mode 100644 index 0000000000..7469d90359 --- /dev/null +++ b/examples/python-postgres-virtual/index.html @@ -0,0 +1,29 @@ + + + + + + + + + + + + diff --git a/examples/python-postgres-virtual/package.json b/examples/python-postgres-virtual/package.json new file mode 100644 index 0000000000..efc37093ed --- /dev/null +++ b/examples/python-postgres-virtual/package.json @@ -0,0 +1,21 @@ +{ + "name": "python-postgres-virtual", + "private": true, + "version": "5.2.0", + "description": "An example of streaming a PostgreSQL-backed `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-postgres-virtual/server.py b/examples/python-postgres-virtual/server.py new file mode 100644 index 0000000000..41b9f079be --- /dev/null +++ b/examples/python-postgres-virtual/server.py @@ -0,0 +1,111 @@ +# ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ +# ┃ ██████ ██████ ██████ █ █ █ █ █ █▄ ▀███ █ ┃ +# ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█ ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄ ▀█ █ ▀▀▀▀▀ ┃ +# ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄ █ ▄▄▄▄▄ ┃ +# ┃ █ ██████ █ ▀█▄ █ ██████ █ ███▌▐███ ███████▄ █ ┃ +# ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫ +# ┃ 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.postgres +import psycopg +import pyarrow.parquet as pq +import tornado.ioloop +import tornado.web + +from tornado.web import StaticFileHandler + +logging.basicConfig( + level=logging.DEBUG, +) + +logger = logging.getLogger(__name__) + +# Requires a running PostgreSQL >= 16 - e.g. +# `docker run --rm -p 5432:5432 -e POSTGRES_HOST_AUTH_METHOD=trust postgres:16` +# with `PSP_POSTGRES_DSN="postgresql://postgres@localhost:5432/postgres"`. +DSN = os.environ.get("PSP_POSTGRES_DSN", "postgresql:///postgres") + +INPUT_FILE = ( + Path(__file__).parent.resolve() + / "node_modules" + / "superstore-arrow" + / "superstore.parquet" +) + + +def arrow_type_to_postgres(arrow_type): + t = str(arrow_type) + if t in ("int8", "int16", "int32", "uint8", "uint16"): + return "INTEGER" + + if t.startswith("int") or t.startswith("uint"): + return "BIGINT" + + if t in ("float", "double", "halffloat"): + return "DOUBLE PRECISION" + + if t.startswith("timestamp"): + return "TIMESTAMP" + + if t.startswith("date"): + return "DATE" + + if t == "bool": + return "BOOLEAN" + + return "TEXT" + + +if __name__ == "__main__": + db = psycopg.connect(DSN, autocommit=True) + + # Load superstore parquet data into Postgres + arrow_table = pq.read_table(str(INPUT_FILE)) + db.execute('DROP TABLE IF EXISTS "superstore"') + names = arrow_table.schema.names + cols = ", ".join( + f'"{field.name}" {arrow_type_to_postgres(field.type)}' + for field in arrow_table.schema + ) + + db.execute(f'CREATE TABLE "superstore" ({cols})') + with db.cursor() as cur: + with cur.copy('COPY "superstore" FROM STDIN') as copy: + for row in arrow_table.to_pylist(): + copy.write_row(tuple(row[n] for n in names)) + + logger.info("Loaded superstore data into Postgres") + + virtual_server = perspective.virtual_servers.postgres.PostgresVirtualServer(DSN) + + 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..6500d9bfdf 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -442,6 +442,28 @@ importers: specifier: 'catalog:' version: 4.1.5 + examples/python-postgres-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-starlette: {} examples/python-tornado: diff --git a/rust/perspective-client/src/rust/virtual_server/generic_sql_model.rs b/rust/perspective-client/src/rust/virtual_server/generic_sql_model.rs index 553baa340e..e91d8464b0 100644 --- a/rust/perspective-client/src/rust/virtual_server/generic_sql_model.rs +++ b/rust/perspective-client/src/rust/virtual_server/generic_sql_model.rs @@ -76,8 +76,21 @@ pub type GenericSQLResult = Result; #[derive(Clone, Debug, Deserialize, Default)] pub struct GenericSQLVirtualServerModelArgs { create_entity: Option, + + /// Entity keyword for `view_delete`'s `DROP {drop_entity} IF EXISTS`. + /// Must agree with `create_entity` — Postgres rejects `DROP TABLE` on a + /// view (`"VIEW"`), while DuckDB's temp tables take the default + /// (`"TABLE"`). + drop_entity: Option, + grouping_fn: Option, + /// Expression for the dialect's natural row identity, used for unsorted + /// view order and natural-order window frames — `rowid` (DuckDB, the + /// default) or `ctid` (Postgres). Dialects with no such pseudo-column + /// (ClickHouse) should advertise `unordered` instead. + row_id_expr: Option, + /// Separator joining `split_by` values and the column name in pivoted /// view column names, e.g. `"CA|Sales"` for separator `"|"`. Perspective's /// column-path separator is `"|"`, so any other value produces views the @@ -236,9 +249,10 @@ impl GenericSQLVirtualServerModel { /// * `view_id` - The identifier of the view to delete. /// /// # Returns - /// SQL: `DROP TABLE IF EXISTS {view_id}` + /// SQL: `DROP {drop_entity} IF EXISTS {view_id}` pub fn view_delete(&self, view_id: &str) -> GenericSQLResult { - Ok(format!("DROP TABLE IF EXISTS {}", view_id)) + let entity = self.0.drop_entity.as_deref().unwrap_or("TABLE"); + Ok(format!("DROP {} IF EXISTS {}", entity, view_id)) } /// Returns the SQL query to create a view from a table with the given @@ -389,7 +403,7 @@ impl GenericSQLVirtualServerModel { let has_grouping_id = !config.group_by.is_empty() && config.group_rollup_mode != GroupRollupMode::Flat; let where_clause = if has_grouping_id { - " WHERE __GROUPING_ID__ = 0" + " WHERE \"__GROUPING_ID__\" = 0" } else { "" }; diff --git a/rust/perspective-client/src/rust/virtual_server/generic_sql_model/table_make_view.rs b/rust/perspective-client/src/rust/virtual_server/generic_sql_model/table_make_view.rs index f38823cd2d..daea179074 100644 --- a/rust/perspective-client/src/rust/virtual_server/generic_sql_model/table_make_view.rs +++ b/rust/perspective-client/src/rust/virtual_server/generic_sql_model/table_make_view.rs @@ -61,7 +61,12 @@ enum QueryOrientation { TotalPivoted, } -fn window_over_clause(w: &WindowSpec, frame: Option<&str>, order_expr: Option<&str>) -> String { +fn window_over_clause( + w: &WindowSpec, + frame: Option<&str>, + order_expr: Option<&str>, + row_id: &str, +) -> String { let mut parts: Vec = Vec::new(); if !w.partition_by.is_empty() { parts.push(format!( @@ -95,7 +100,7 @@ fn window_over_clause(w: &WindowSpec, frame: Option<&str>, order_expr: Option<&s } )) }, - None => parts.push("ORDER BY rowid ASC".to_string()), + None => parts.push(format!("ORDER BY {} ASC", row_id)), } if let Some(f) = frame { parts.push(f.to_string()); @@ -125,6 +130,7 @@ fn window_sql( w: &WindowSpec, resolve: &dyn Fn(&str) -> String, order_type: Option, + row_id: &str, ) -> Result { // `range` frame interval arithmetic is defined on the order key's // units - the natural (`rowid`) fallback is meaningless for it, so an @@ -170,7 +176,7 @@ fn window_sql( "{}({}) OVER ({})", op, src, - window_over_clause(w, Some(&frame), lin_order.as_deref()) + window_over_clause(w, Some(&frame), lin_order.as_deref(), row_id) )); } @@ -181,7 +187,7 @@ fn window_sql( return Ok(format!( "{}() OVER ({})", op, - window_over_clause(w, None, None) + window_over_clause(w, None, None, row_id) )); } @@ -191,7 +197,7 @@ fn window_sql( op, src, w.offset.unwrap_or(1), - window_over_clause(w, None, None) + window_over_clause(w, None, None, row_id) )), "nth_value" => { let frame = window_frame_sql(w.frame.as_ref()); @@ -199,13 +205,13 @@ fn window_sql( "nth_value({}, {}) OVER ({})", src, w.offset.unwrap_or(1), - window_over_clause(w, Some(&frame), lin_order.as_deref()) + window_over_clause(w, Some(&frame), lin_order.as_deref(), row_id) )) }, "ntile" => Ok(format!( "ntile({}) OVER ({})", w.offset.unwrap_or(1), - window_over_clause(w, None, None) + window_over_clause(w, None, None, row_id) )), // `diff` and `rate` are Perspective's, not any SQL dialect's - they // are synthesized here so a config authored against the engine keeps @@ -215,7 +221,7 @@ fn window_sql( src, src, w.offset.unwrap_or(1), - window_over_clause(w, None, None) + window_over_clause(w, None, None, row_id) )), "rate" => { let Some(order_by) = &w.order_by else { @@ -234,16 +240,16 @@ fn window_sql( } let frame = window_frame_sql(w.frame.as_ref()); - let over = window_over_clause(w, Some(&frame), lin_order.as_deref()); + let over = window_over_clause(w, Some(&frame), lin_order.as_deref(), row_id); // Δk in the denominator is measured on the same linear scale // the frame is defined on (a raw temporal key would not CAST - // to DOUBLE at all). + // to DOUBLE PRECISION at all). let okey = lin_order .clone() .unwrap_or_else(|| format!("\"{}\"", quote_ident(&order_by.0))); Ok(format!( - "(({} - first_value({}) OVER ({})) / NULLIF(CAST({} AS DOUBLE) - \ - CAST(first_value({}) OVER ({}) AS DOUBLE), 0))", + "(({} - first_value({}) OVER ({})) / NULLIF(CAST({} AS DOUBLE PRECISION) - \ + CAST(first_value({}) OVER ({}) AS DOUBLE PRECISION), 0))", src, src, over, okey, okey, over )) }, @@ -309,6 +315,7 @@ pub(crate) struct ViewQueryContext<'a> { like_escape_clause: Option<&'a str>, backslash_escaped_literals: bool, regex_fn: Option<&'a str>, + row_id_expr: &'a str, row_path_aliases: Vec, } @@ -331,6 +338,8 @@ impl<'a> ViewQueryContext<'a> { .unwrap_or_else(|| format!("\"{}\"", col)) }; + let row_id_expr = model.0.row_id_expr.as_deref().unwrap_or("rowid"); + // Window columns materialize in a wrapping sub-select, so every // downstream clause (filter, group_by, aggregate, sort) sees them as // plain columns in all four query orientations - mirroring the @@ -352,17 +361,19 @@ impl<'a> ViewQueryContext<'a> { selects.push(format!( "{} AS \"{}\"", - window_sql(w, &col_name_resolve, order_type)?, + window_sql(w, &col_name_resolve, order_type, row_id_expr)?, quote_ident(name) )); } - // `rowid` is a virtual column bound only on base tables - it - // does not survive `SELECT *` into the sub-select, so the - // natural row order is re-exported under an internal alias for - // the outer query's `ORDER BY` (see [`Self::natural_order_col`]). + // The natural row identity (`rowid`, `ctid`) is a system + // pseudo-column bound only on base tables - it does not survive + // `SELECT *` into the sub-select, so the natural row order is + // re-exported under an internal alias for the outer query's + // `ORDER BY` (see [`Self::natural_order_col`]). format!( - "(SELECT *, rowid AS __PSP_ROWID__, {} FROM {}) AS __PSP_WINDOW_SRC__", + "(SELECT *, {} AS \"__PSP_ROWID__\", {} FROM {}) AS __PSP_WINDOW_SRC__", + row_id_expr, selects.join(", "), table ) @@ -377,7 +388,7 @@ impl<'a> ViewQueryContext<'a> { .collect(); let row_path_aliases: Vec = (0..config.group_by.len()) - .map(|i| format!("__ROW_PATH_{}__", i)) + .map(|i| format!("\"__ROW_PATH_{}__\"", i)) .collect(); Ok(Self { @@ -389,6 +400,7 @@ impl<'a> ViewQueryContext<'a> { like_escape_clause: model.0.like_escape_clause.as_deref(), backslash_escaped_literals: model.0.backslash_escaped_literals.unwrap_or(false), regex_fn: model.0.regex_fn.as_deref(), + row_id_expr, row_path_aliases, }) } @@ -431,7 +443,7 @@ impl<'a> ViewQueryContext<'a> { let mut src_clauses = self.select_clauses(); src_clauses.extend(self.split_select_clauses()); src_clauses.push(format!( - "ROW_NUMBER() OVER (ORDER BY {}) as __ROW_NUM__", + "ROW_NUMBER() OVER (ORDER BY {}) as \"__ROW_NUM__\"", self.pivot_row_num_order() )); @@ -449,7 +461,7 @@ impl<'a> ViewQueryContext<'a> { .chain((1..=n).map(|k| (1u64 << k) - 1)) .map(|mask| { format!( - "SELECT *, {} AS __CGROUPING_ID__ FROM __PSP_PIVOT_BASE__", + "SELECT *, {} AS \"__CGROUPING_ID__\" FROM __PSP_PIVOT_BASE__", mask ) }) @@ -458,20 +470,20 @@ impl<'a> ViewQueryContext<'a> { format!( "WITH __PSP_PIVOT_BASE__ AS ({}), __PSP_PIVOT_SRC__ AS ({}) SELECT * \ - EXCLUDE (__ROW_NUM__) FROM {}", + EXCLUDE (\"__ROW_NUM__\") FROM {}", src, union, - self.pivot_join(&cols, &["__ROW_NUM__".to_string()]) + self.pivot_join(&cols, &["\"__ROW_NUM__\"".to_string()]) ) } else { let from = if cols.is_empty() { "__PSP_PIVOT_SRC__".to_string() } else { - self.pivot_join(&cols, &["__ROW_NUM__".to_string()]) + self.pivot_join(&cols, &["\"__ROW_NUM__\"".to_string()]) }; format!( - "WITH __PSP_PIVOT_SRC__ AS ({}) SELECT * EXCLUDE (__ROW_NUM__) FROM {}", + "WITH __PSP_PIVOT_SRC__ AS ({}) SELECT * EXCLUDE (\"__ROW_NUM__\") FROM {}", src, from ) } @@ -499,12 +511,12 @@ impl<'a> ViewQueryContext<'a> { let sort_source = self.sort_source_expr(sort_col); if self.is_flat_mode() { inner_clauses.push(format!( - "sum({}) OVER (PARTITION BY {}) AS __SORT_{}__", + "sum({}) OVER (PARTITION BY {}) AS \"__SORT_{}__\"", sort_source, groups_joined, sidx, )); } else { inner_clauses.push(format!( - "sum({}) OVER (PARTITION BY {}({}), {}) AS __SORT_{}__", + "sum({}) OVER (PARTITION BY {}({}), {}) AS \"__SORT_{}__\"", sort_source, self.grouping_fn, groups_joined, groups_joined, sidx, )); } @@ -533,11 +545,11 @@ impl<'a> ViewQueryContext<'a> { let mut row_id_cols = self.row_path_aliases.clone(); if !self.is_flat_mode() { - row_id_cols.push("__GROUPING_ID__".to_string()); + row_id_cols.push("\"__GROUPING_ID__\"".to_string()); } for (sidx, Sort(_, sort_dir)) in self.config.sort.iter().enumerate() { if *sort_dir != SortDir::None && !is_col_sort(sort_dir) { - row_id_cols.push(format!("__SORT_{}__", sidx)); + row_id_cols.push(format!("\"__SORT_{}__\"", sidx)); } } @@ -560,7 +572,7 @@ impl<'a> ViewQueryContext<'a> { QueryOrientation::TotalPivoted if self.is_split_rollup() => { let cols: Vec<&String> = self.config.columns.iter().flatten().collect(); let mut src_clauses = self.select_clauses(); - src_clauses.push("1 AS __TOTAL_KEY__".to_string()); + src_clauses.push("1 AS \"__TOTAL_KEY__\"".to_string()); src_clauses.extend(self.split_select_clauses()); src_clauses.push(self.cgrouping_id_clause()); let src = format!( @@ -574,11 +586,11 @@ impl<'a> ViewQueryContext<'a> { let from = if cols.is_empty() { "__PSP_PIVOT_SRC__".to_string() } else { - self.pivot_join(&cols, &["__TOTAL_KEY__".to_string()]) + self.pivot_join(&cols, &["\"__TOTAL_KEY__\"".to_string()]) }; format!( - "WITH __PSP_PIVOT_SRC__ AS ({}) SELECT * EXCLUDE (__TOTAL_KEY__) FROM {}", + "WITH __PSP_PIVOT_SRC__ AS ({}) SELECT * EXCLUDE (\"__TOTAL_KEY__\") FROM {}", src, from ) }, @@ -650,7 +662,7 @@ impl<'a> ViewQueryContext<'a> { let default_order = if self.config.split_by.is_empty() { self.natural_order_col() } else { - "__ROW_NUM__" + "\"__ROW_NUM__\"" }; query = format!("{} ORDER BY {}", query, default_order); @@ -797,7 +809,11 @@ impl<'a> ViewQueryContext<'a> { arms.push(format!("WHEN {} THEN {}", mask, name)); } - format!("CASE __CGROUPING_ID__ {} ELSE {} END", arms.join(" "), leaf) + format!( + "CASE \"__CGROUPING_ID__\" {} ELSE {} END", + arms.join(" "), + leaf + ) } fn split_prefix_expr(&self, kept: usize, sep: &str) -> String { @@ -814,7 +830,7 @@ impl<'a> ViewQueryContext<'a> { fn cgrouping_id_clause(&self) -> String { format!( - "{}({}) AS __CGROUPING_ID__", + "{}({}) AS \"__CGROUPING_ID__\"", self.grouping_fn, self.pivot_on_expr() ) @@ -947,14 +963,11 @@ impl<'a> ViewQueryContext<'a> { } /// The natural row-order column as visible to clauses that select `FROM - /// {from_expr}` - the base table's virtual `rowid` directly, or its - /// re-export when windows wrap the table in a sub-select (through which - /// `rowid` does not propagate). - fn natural_order_col(&self) -> &'static str { + fn natural_order_col(&self) -> &str { if self.config.windows.is_empty() { - "rowid" + self.row_id_expr } else { - "__PSP_ROWID__" + "\"__PSP_ROWID__\"" } } @@ -988,7 +1001,7 @@ impl<'a> ViewQueryContext<'a> { fn grouping_id_clause(&self) -> String { format!( - "{}({}) AS __GROUPING_ID__", + "{}({}) AS \"__GROUPING_ID__\"", self.grouping_fn, self.group_col_names.join(", ") ) @@ -999,7 +1012,7 @@ impl<'a> ViewQueryContext<'a> { .group_by .iter() .enumerate() - .map(|(i, col)| format!("{} as __ROW_PATH_{}__", self.col_name(col), i)) + .map(|(i, col)| format!("{} as \"__ROW_PATH_{}__\"", self.col_name(col), i)) .collect() } @@ -1023,10 +1036,10 @@ impl<'a> ViewQueryContext<'a> { let dir = sort_dir_to_string(sort_dir); if !self.config.split_by.is_empty() { if is_leaf { - clauses.push(format!("__SORT_{}__ {}", sidx, dir)); + clauses.push(format!("\"__SORT_{}__\" {}", sidx, dir)); } else { clauses.push(format!( - "first(__SORT_{}__) OVER __WINDOW_{}__ {}", + "first_value(\"__SORT_{}__\") OVER __WINDOW_{}__ {}", sidx, gidx, dir )); } @@ -1041,7 +1054,7 @@ impl<'a> ViewQueryContext<'a> { )); } else { clauses.push(format!( - "first({}({})) OVER __WINDOW_{}__ {}", + "first_value({}({})) OVER __WINDOW_{}__ {}", agg, self.col_name(sort_col), gidx, @@ -1059,7 +1072,7 @@ impl<'a> ViewQueryContext<'a> { if *sort_dir != SortDir::None && !is_col_sort(sort_dir) { let dir = sort_dir_to_string(sort_dir); if !self.config.split_by.is_empty() { - clauses.push(format!("__SORT_{}__ {}", sidx, dir)); + clauses.push(format!("\"__SORT_{}__\" {}", sidx, dir)); } else { let agg = self.get_aggregate(sort_col); clauses.push(format!("{}({}) {}", agg, self.col_name(sort_col), dir)); @@ -1072,9 +1085,9 @@ impl<'a> ViewQueryContext<'a> { if !self.config.split_by.is_empty() { let shift = self.config.group_by.len() - 1 - gidx; if shift > 0 { - clauses.push(format!("(__GROUPING_ID__ >> {}) DESC", shift)); + clauses.push(format!("(\"__GROUPING_ID__\" >> {}) DESC", shift)); } else { - clauses.push("__GROUPING_ID__ DESC".to_string()); + clauses.push("\"__GROUPING_ID__\" DESC".to_string()); } } else { let groups_up_to = self.config.group_by[..=gidx] @@ -1094,10 +1107,10 @@ impl<'a> ViewQueryContext<'a> { let dir = sort_dir_to_string(sort_dir); if !self.config.split_by.is_empty() { if is_leaf { - clauses.push(format!("__SORT_{}__ {}", sidx, dir)); + clauses.push(format!("\"__SORT_{}__\" {}", sidx, dir)); } else { clauses.push(format!( - "first(__SORT_{}__) OVER __WINDOW_{}__ {}", + "first_value(\"__SORT_{}__\") OVER __WINDOW_{}__ {}", sidx, gidx, dir )); } @@ -1107,7 +1120,7 @@ impl<'a> ViewQueryContext<'a> { clauses.push(format!("{}({}) {}", agg, self.col_name(sort_col), dir)); } else { clauses.push(format!( - "first({}({})) OVER __WINDOW_{}__ {}", + "first_value({}({})) OVER __WINDOW_{}__ {}", agg, self.col_name(sort_col), gidx, @@ -1158,9 +1171,9 @@ impl<'a> ViewQueryContext<'a> { } else if !self.config.split_by.is_empty() { let shift = self.config.group_by.len() - 1 - gidx; let grouping_expr = if shift > 0 { - format!("(__GROUPING_ID__ >> {})", shift) + format!("(\"__GROUPING_ID__\" >> {})", shift) } else { - "__GROUPING_ID__".to_string() + "\"__GROUPING_ID__\"".to_string() }; let order = self.row_path_aliases.join(", "); diff --git a/rust/perspective-client/src/rust/virtual_server/generic_sql_model/tests.rs b/rust/perspective-client/src/rust/virtual_server/generic_sql_model/tests.rs index a5257f616a..c98564e3bd 100644 --- a/rust/perspective-client/src/rust/virtual_server/generic_sql_model/tests.rs +++ b/rust/perspective-client/src/rust/virtual_server/generic_sql_model/tests.rs @@ -124,13 +124,13 @@ fn test_table_make_view_with_sort_group_by_and_split_by() { assert!(sql.contains("__SORT_0__"), "expected __SORT_0__: {}", sql); assert!( - sql.contains("__GROUPING_ID__, __SORT_0__"), + sql.contains("\"__GROUPING_ID__\", \"__SORT_0__\""), "expected __SORT_0__ in GROUP BY: {}", sql ); assert!( - sql.contains("__SORT_0__ ASC"), + sql.contains("\"__SORT_0__\" ASC"), "expected __SORT_0__ ASC in ORDER BY: {}", sql ); @@ -160,14 +160,14 @@ fn test_table_make_view_with_sort_multi_group_by_and_split_by() { .unwrap(); assert!( - sql.contains("PARTITION BY (__GROUPING_ID__ >> 1)"), + sql.contains("PARTITION BY (\"__GROUPING_ID__\" >> 1)"), "expected shifted __GROUPING_ID__ in WINDOW: {}", sql ); assert!( - sql.contains("first(__SORT_0__) OVER __WINDOW_0__"), - "expected first(__SORT_0__) OVER __WINDOW_0__: {}", + sql.contains("first_value(\"__SORT_0__\") OVER __WINDOW_0__"), + "expected first_value(\"__SORT_0__\") OVER __WINDOW_0__: {}", sql ); @@ -297,8 +297,8 @@ fn test_table_make_view_pivoted_with_sort() { sql ); assert!( - sql.ends_with("ORDER BY __ROW_NUM__)"), - "should end with ORDER BY __ROW_NUM__: {}", + sql.ends_with("ORDER BY \"__ROW_NUM__\")"), + "should end with ORDER BY \"__ROW_NUM__\": {}", sql ); } @@ -519,7 +519,7 @@ fn test_table_make_view_flat_group_by_with_split_by_and_sort() { sql ); assert!( - sql.contains("__SORT_0__ DESC"), + sql.contains("\"__SORT_0__\" DESC"), "expected __SORT_0__ DESC in ORDER BY: {}", sql ); @@ -700,13 +700,14 @@ fn test_table_make_view_grouped_pivoted_null_safe_join() { .unwrap(); assert!( - sql.contains("GROUP BY __ROW_PATH_0__, __GROUPING_ID__"), + sql.contains("GROUP BY \"__ROW_PATH_0__\", \"__GROUPING_ID__\""), "expected pivot GROUP BY on row keys: {}", sql ); assert!( sql.contains( - "__PSP_PIVOT_0__.__ROW_PATH_0__ IS NOT DISTINCT FROM __PSP_PIVOT_1__.__ROW_PATH_0__" + "__PSP_PIVOT_0__.\"__ROW_PATH_0__\" IS NOT DISTINCT FROM \ + __PSP_PIVOT_1__.\"__ROW_PATH_0__\"" ), "rollup rows have NULL row-path keys, join must be NULL-safe: {}", sql @@ -1030,7 +1031,7 @@ fn test_table_make_view_window_rate() { .unwrap(); assert!(sql.contains("first_value(\"price\") OVER")); - assert!(sql.contains("NULLIF(CAST(\"t\" AS DOUBLE)")); + assert!(sql.contains("NULLIF(CAST(\"t\" AS DOUBLE PRECISION)")); assert!(sql.contains("RANGE BETWEEN 10 PRECEDING AND CURRENT ROW")); } @@ -1121,3 +1122,127 @@ fn filter_sql(args: GenericSQLVirtualServerModelArgs, filter: serde_json::Value) .table_make_view("source_table", "dest_view", &config, &IndexMap::new()) .unwrap() } + +fn postgres_args() -> GenericSQLVirtualServerModelArgs { + GenericSQLVirtualServerModelArgs { + create_entity: Some("TEMPORARY VIEW".to_string()), + drop_entity: Some("VIEW".to_string()), + grouping_fn: Some("GROUPING".to_string()), + row_id_expr: Some("ctid".to_string()), + like_escape_clause: Some("\\".to_string()), + regex_fn: Some("regexp_like".to_string()), + ..Default::default() + } +} + +#[test] +fn test_view_delete_custom_drop_entity() { + let builder = GenericSQLVirtualServerModel::new(postgres_args()); + assert_eq!( + builder.view_delete("my_view").unwrap(), + "DROP VIEW IF EXISTS my_view" + ); +} + +#[test] +fn test_table_make_view_custom_create_entity() { + let builder = GenericSQLVirtualServerModel::new(postgres_args()); + let mut config = ViewConfig::default(); + config.columns = vec![Some("col1".to_string())]; + let sql = builder + .table_make_view("source_table", "dest_view", &config, &IndexMap::new()) + .unwrap(); + + assert!( + sql.starts_with("CREATE TEMPORARY VIEW dest_view AS"), + "expected TEMPORARY VIEW: {}", + sql + ); +} + +#[test] +fn test_table_make_view_flat_row_id_expr() { + let builder = GenericSQLVirtualServerModel::new(postgres_args()); + let mut config = ViewConfig::default(); + config.columns = vec![Some("col1".to_string())]; + let sql = builder + .table_make_view("source_table", "dest_view", &config, &IndexMap::new()) + .unwrap(); + + assert!( + sql.ends_with("ORDER BY ctid)"), + "flat default order should use the dialect row id: {}", + sql + ); + assert!(!sql.contains("rowid"), "no rowid for postgres: {}", sql); +} + +#[test] +fn test_table_make_view_window_natural_order_row_id_expr() { + let builder = GenericSQLVirtualServerModel::new(postgres_args()); + let mut config = ViewConfig::default(); + config.columns = vec![Some("t".to_string()), Some("cumsum".to_string())]; + let (name, mut spec) = window_spec("cumsum", "sum", Some(WindowFrame::Cumulative)); + spec.order_by = None; + config.windows = Windows(HashMap::from([(name, spec)])); + let sql = builder + .table_make_view("source_table", "dest_view", &config, &IndexMap::new()) + .unwrap(); + + assert!( + sql.contains("PARTITION BY \"sym\" ORDER BY ctid ASC"), + "natural window order should use the dialect row id: {}", + sql + ); +} + +#[test] +fn test_table_make_view_window_src_projects_row_id() { + // System pseudo-columns (`rowid`, `ctid`) do not survive `SELECT *` + // through the window sub-select, so the model projects them under + // `__PSP_ROWID__` and refers to the alias in the outer default order. + let builder = GenericSQLVirtualServerModel::new(GenericSQLVirtualServerModelArgs::default()); + let mut config = ViewConfig::default(); + config.columns = vec![Some("t".to_string()), Some("cumsum".to_string())]; + config.windows = Windows(HashMap::from([window_spec( + "cumsum", + "sum", + Some(WindowFrame::Cumulative), + )])); + let sql = builder + .table_make_view("source_table", "dest_view", &config, &IndexMap::new()) + .unwrap(); + + assert!( + sql.contains("SELECT *, rowid AS \"__PSP_ROWID__\","), + "window sub-select should project the row id: {}", + sql + ); + assert!( + sql.ends_with("ORDER BY \"__PSP_ROWID__\")"), + "outer default order should use the projected alias: {}", + sql + ); +} + +#[test] +fn test_table_make_view_grouping_fn_in_rollup_order() { + let builder = GenericSQLVirtualServerModel::new(postgres_args()); + let mut config = ViewConfig::default(); + config.columns = vec![Some("value".to_string())]; + config.group_by = vec!["category".to_string()]; + let sql = builder + .table_make_view("source_table", "dest_view", &config, &IndexMap::new()) + .unwrap(); + + assert!( + sql.contains("GROUPING(\"category\") AS \"__GROUPING_ID__\""), + "expected custom grouping fn: {}", + sql + ); + assert!( + !sql.contains("GROUPING_ID("), + "should not use the default grouping fn: {}", + sql + ); +} diff --git a/rust/perspective-python/perspective/tests/table/test_column_paths.py b/rust/perspective-python/perspective/tests/table/test_column_paths.py index eba64ae973..4ea63e7fa7 100644 --- a/rust/perspective-python/perspective/tests/table/test_column_paths.py +++ b/rust/perspective-python/perspective/tests/table/test_column_paths.py @@ -15,7 +15,7 @@ client = psp.Server().new_local_client() -class TestViewColumnPaths(object): +class TestViewColumnPaths: def test_column_paths(self, superstore): tbl = client.table(superstore) view = tbl.view() diff --git a/rust/perspective-python/perspective/tests/virtual_servers/test_postgres.py b/rust/perspective-python/perspective/tests/virtual_servers/test_postgres.py new file mode 100644 index 0000000000..9e78136bbc --- /dev/null +++ b/rust/perspective-python/perspective/tests/virtual_servers/test_postgres.py @@ -0,0 +1,963 @@ +# ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ +# ┃ ██████ ██████ ██████ █ █ █ █ █ █▄ ▀███ █ ┃ +# ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█ ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄ ▀█ █ ▀▀▀▀▀ ┃ +# ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄ █ ▄▄▄▄▄ ┃ +# ┃ █ ██████ █ ▀█▄ █ ██████ █ ███▌▐███ ███████▄ █ ┃ +# ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫ +# ┃ 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 os +import tempfile +import urllib.request +from datetime import datetime + +import pytest + +psycopg = pytest.importorskip("psycopg") +pq = pytest.importorskip("pyarrow.parquet") + +from perspective import Client +from perspective.virtual_servers.postgres import PostgresVirtualServer + +# Set `PSP_TEST_POSTGRES_DSN` to point these tests at a server; without a +# reachable PostgreSQL >= 16 they skip rather than fail. +DSN = os.environ.get("PSP_TEST_POSTGRES_DSN", "postgresql:///postgres") + +_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 + + +# Perspective-relevant Postgres types per superstore column - `BIGINT` and +# `DOUBLE PRECISION` both present as `float`, matching the DuckDB handler's +# schema for the same dataset. +_SUPERSTORE_TYPES = { + "Row ID": "INTEGER", + "Order ID": "TEXT", + "Order Date": "DATE", + "Ship Date": "DATE", + "Ship Mode": "TEXT", + "Customer ID": "TEXT", + "Customer Name": "TEXT", + "Segment": "TEXT", + "Country": "TEXT", + "City": "TEXT", + "State": "TEXT", + "Postal Code": "BIGINT", + "Region": "TEXT", + "Product ID": "TEXT", + "Category": "TEXT", + "Sub-Category": "TEXT", + "Product Name": "TEXT", + "Sales": "DOUBLE PRECISION", + "Quantity": "INTEGER", + "Discount": "DOUBLE PRECISION", + "Profit": "DOUBLE PRECISION", +} + + +def _load_superstore(conn): + arrow_table = pq.read_table(_get_superstore_parquet()) + names = arrow_table.schema.names + cols = ", ".join(f'"{name}" {_SUPERSTORE_TYPES[name]}' for name in names) + conn.execute(f"CREATE TABLE psp_test.superstore ({cols})") + with conn.cursor() as cur: + with cur.copy('COPY psp_test.superstore FROM STDIN') as copy: + for row in arrow_table.to_pylist(): + copy.write_row( + tuple( + v.date() if isinstance(v, datetime) else v + for v in (row[name] for name in names) + ) + ) + + +def _load_coerce_types(conn): + """A column of each Postgres type whose Perspective type is not its own. + + `mood` is the interesting one: enum OIDs are user-defined and unknown to + the OID map, so the column must fall back to `string` rather than fail. + """ + conn.execute("CREATE TYPE psp_test.mood AS ENUM ('happy', 'sad')") + conn.execute(""" + CREATE TABLE psp_test.coerce_types ( + "small" SMALLINT, + "big" BIGINT, + "float" REAL, + "decimal" NUMERIC(18, 3), + "time" TIME, + "timestamp" TIMESTAMP, + "timestamptz" TIMESTAMPTZ, + "date" DATE, + "enum" psp_test.mood, + "uuid" UUID, + "json" JSONB, + "string" TEXT + ) + """) + + conn.execute(""" + INSERT INTO psp_test.coerce_types VALUES + (-300, 9007199254740992, 1.5, 1.234, TIME '01:01:01', + TIMESTAMP '2023-01-01 00:00:00', + TIMESTAMPTZ '2023-01-01 00:00:00+00', DATE '2023-01-01', + 'happy', '00000000-0000-0000-0000-000000000001', + '{"a": 1}', 'a'), + (300, -9007199254740992, -1.5, -5.678, TIME '00:00:01', + TIMESTAMP '2023-01-02 00:00:00', + TIMESTAMPTZ '2023-01-02 00:00:00+00', DATE '2023-01-02', + 'sad', '00000000-0000-0000-0000-000000000002', + '{"b": 2}', 'b') + """) + + +@pytest.fixture(scope="module") +def pg_db(): + try: + conn = psycopg.connect(DSN, autocommit=True, connect_timeout=5) + except psycopg.OperationalError as e: + pytest.skip(f"no PostgreSQL server at '{DSN}': {e}") + + if conn.info.server_version < 160000: + pytest.skip("PostgreSQL >= 16 required (`any_value`)") + + conn.execute("DROP SCHEMA IF EXISTS psp_test CASCADE") + conn.execute("CREATE SCHEMA psp_test") + _load_superstore(conn) + _load_coerce_types(conn) + yield conn + conn.execute("DROP SCHEMA psp_test CASCADE") + conn.close() + + +@pytest.fixture +def client(pg_db): + server = PostgresVirtualServer(DSN) + + 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 + + +def approx(x): + # Float aggregation order (and so its rounding) is Postgres's own, + # not DuckDB's. + return pytest.approx(x, rel=1e-7) + + +class TestPostgresClient: + def test_get_hosted_table_names(self, client): + tables = client.get_hosted_table_names() + assert {"psp_test.superstore", "psp_test.coerce_types"} <= set(tables) + + +class TestPostgresTable: + def test_schema(self, client): + table = client.open_table("psp_test.superstore") + schema = table.schema() + assert schema == { + "Product Name": "string", + "Ship Date": "date", + "City": "string", + "Row ID": "integer", + "Customer Name": "string", + "Quantity": "integer", + "Discount": "float", + "Sub-Category": "string", + "Segment": "string", + "Category": "string", + "Order Date": "date", + "Order ID": "string", + "Sales": "float", + "State": "string", + "Postal Code": "float", + "Country": "string", + "Customer ID": "string", + "Ship Mode": "string", + "Region": "string", + "Profit": "float", + "Product ID": "string", + } + + def test_size(self, client): + table = client.open_table("psp_test.superstore") + size = table.size() + assert size == 9994 + + +class TestPostgresView: + def test_num_rows(self, client): + table = client.open_table("psp_test.superstore") + view = table.view(columns=["Sales", "Profit"]) + num_rows = view.num_rows() + assert num_rows == 9994 + view.delete() + + def test_num_columns(self, client): + table = client.open_table("psp_test.superstore") + view = table.view(columns=["Sales", "Profit", "State"]) + num_columns = view.num_columns() + assert num_columns == 3 + view.delete() + + def test_schema(self, client): + table = client.open_table("psp_test.superstore") + view = table.view(columns=["Sales", "Profit", "State"]) + schema = view.schema() + assert schema == { + "Sales": "float", + "Profit": "float", + "State": "string", + } + view.delete() + + def test_to_json(self, client): + table = client.open_table("psp_test.superstore") + view = table.view(columns=["Sales", "Quantity"]) + json = view.to_json(start_row=0, end_row=5) + assert json == [ + {"Sales": 261.96, "Quantity": 2}, + {"Sales": 731.94, "Quantity": 3}, + {"Sales": 14.62, "Quantity": 2}, + {"Sales": 957.5775, "Quantity": 5}, + {"Sales": 22.368, "Quantity": 2}, + ] + view.delete() + + def test_to_columns(self, client): + table = client.open_table("psp_test.superstore") + view = table.view(columns=["Sales", "Quantity"]) + columns = view.to_columns(start_row=0, end_row=5) + assert columns == { + "Sales": [261.96, 731.94, 14.62, 957.5775, 22.368], + "Quantity": [2, 3, 2, 5, 2], + } + view.delete() + + def test_column_paths(self, client): + table = client.open_table("psp_test.superstore") + view = table.view(columns=["Sales", "Profit", "State"]) + paths = view.column_paths() + assert paths == ["Sales", "Profit", "State"] + view.delete() + + +class TestPostgresGroupBy: + def test_single_group_by(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales"], + group_by=["Region"], + aggregates={"Sales": "sum"}, + ) + num_rows = view.num_rows() + assert num_rows == 5 + json = view.to_json() + assert json == [ + {"__ROW_PATH__": [], "Sales": approx(2297200.860299955)}, + {"__ROW_PATH__": ["Central"], "Sales": approx(501239.8908000005)}, + {"__ROW_PATH__": ["East"], "Sales": approx(678781.2399999979)}, + {"__ROW_PATH__": ["South"], "Sales": approx(391721.9050000003)}, + {"__ROW_PATH__": ["West"], "Sales": approx(725457.8245000006)}, + ] + view.delete() + + def test_multi_level_group_by(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales"], + group_by=["Region", "Category"], + aggregates={"Sales": "sum"}, + ) + json = view.to_json() + assert json == [ + {"__ROW_PATH__": [], "Sales": approx(2297200.860299955)}, + {"__ROW_PATH__": ["Central"], "Sales": approx(501239.8908000005)}, + { + "__ROW_PATH__": ["Central", "Furniture"], + "Sales": approx(163797.16380000004), + }, + { + "__ROW_PATH__": ["Central", "Office Supplies"], + "Sales": approx(167026.41500000027), + }, + { + "__ROW_PATH__": ["Central", "Technology"], + "Sales": approx(170416.3119999999), + }, + {"__ROW_PATH__": ["East"], "Sales": approx(678781.2399999979)}, + { + "__ROW_PATH__": ["East", "Furniture"], + "Sales": approx(208291.20400000009), + }, + { + "__ROW_PATH__": ["East", "Office Supplies"], + "Sales": approx(205516.0549999999), + }, + { + "__ROW_PATH__": ["East", "Technology"], + "Sales": approx(264973.9810000003), + }, + {"__ROW_PATH__": ["South"], "Sales": approx(391721.9050000003)}, + { + "__ROW_PATH__": ["South", "Furniture"], + "Sales": approx(117298.6840000001), + }, + { + "__ROW_PATH__": ["South", "Office Supplies"], + "Sales": approx(125651.31299999992), + }, + { + "__ROW_PATH__": ["South", "Technology"], + "Sales": approx(148771.9079999999), + }, + {"__ROW_PATH__": ["West"], "Sales": approx(725457.8245000006)}, + { + "__ROW_PATH__": ["West", "Furniture"], + "Sales": approx(252612.7435000003), + }, + { + "__ROW_PATH__": ["West", "Office Supplies"], + "Sales": approx(220853.24900000007), + }, + { + "__ROW_PATH__": ["West", "Technology"], + "Sales": approx(251991.83199999997), + }, + ] + view.delete() + + def test_group_by_with_count_aggregate(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales"], + group_by=["Region"], + aggregates={"Sales": "count"}, + ) + json = view.to_json() + assert json == [ + {"__ROW_PATH__": [], "Sales": 9994}, + {"__ROW_PATH__": ["Central"], "Sales": 2323}, + {"__ROW_PATH__": ["East"], "Sales": 2848}, + {"__ROW_PATH__": ["South"], "Sales": 1620}, + {"__ROW_PATH__": ["West"], "Sales": 3203}, + ] + view.delete() + + def test_group_by_with_avg_aggregate(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales"], + group_by=["Category"], + aggregates={"Sales": "avg"}, + ) + json = view.to_json() + assert json == [ + {"__ROW_PATH__": [], "Sales": approx(229.8580008304938)}, + {"__ROW_PATH__": ["Furniture"], "Sales": approx(349.83488698727007)}, + { + "__ROW_PATH__": ["Office Supplies"], + "Sales": approx(119.32410089611732), + }, + {"__ROW_PATH__": ["Technology"], "Sales": approx(452.70927612344155)}, + ] + view.delete() + + def test_group_by_with_min_aggregate(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Quantity"], + group_by=["Region"], + aggregates={"Quantity": "min"}, + ) + json = view.to_json() + assert json == [ + {"__ROW_PATH__": [], "Quantity": 1}, + {"__ROW_PATH__": ["Central"], "Quantity": 1}, + {"__ROW_PATH__": ["East"], "Quantity": 1}, + {"__ROW_PATH__": ["South"], "Quantity": 1}, + {"__ROW_PATH__": ["West"], "Quantity": 1}, + ] + view.delete() + + def test_group_by_with_max_aggregate(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Quantity"], + group_by=["Region"], + aggregates={"Quantity": "max"}, + ) + json = view.to_json() + assert json == [ + {"__ROW_PATH__": [], "Quantity": 14}, + {"__ROW_PATH__": ["Central"], "Quantity": 14}, + {"__ROW_PATH__": ["East"], "Quantity": 14}, + {"__ROW_PATH__": ["South"], "Quantity": 14}, + {"__ROW_PATH__": ["West"], "Quantity": 14}, + ] + view.delete() + + def test_group_by_with_stddev_aggregate(self, client): + # A Postgres-vocabulary aggregate with a `numeric` result, which + # arrives as `Decimal` and must normalize to `float`. + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Quantity"], + group_by=["Region"], + aggregates={"Quantity": "stddev_samp"}, + ) + json = view.to_json() + assert len(json) == 5 + assert all(isinstance(row["Quantity"], float) for row in json) + view.delete() + + +class TestPostgresFilter: + def test_filter_with_equals(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales", "Region"], + filter=[["Region", "==", "West"]], + ) + json = view.to_json(start_row=0, end_row=5) + assert json == [ + {"Sales": 14.62, "Region": "West"}, + {"Sales": 48.86, "Region": "West"}, + {"Sales": 7.28, "Region": "West"}, + {"Sales": 907.152, "Region": "West"}, + {"Sales": 18.504, "Region": "West"}, + ] + view.delete() + + def test_filter_with_not_equals(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales", "Region"], + filter=[["Region", "!=", "West"]], + ) + json = view.to_json(start_row=0, end_row=5) + assert json == [ + {"Sales": 261.96, "Region": "South"}, + {"Sales": 731.94, "Region": "South"}, + {"Sales": 957.5775, "Region": "South"}, + {"Sales": 22.368, "Region": "South"}, + {"Sales": 15.552, "Region": "South"}, + ] + view.delete() + + def test_filter_with_greater_than(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales", "Quantity"], + filter=[["Quantity", ">", 5]], + ) + json = view.to_json(start_row=0, end_row=5) + assert json == [ + {"Sales": 48.86, "Quantity": 7}, + {"Sales": 907.152, "Quantity": 6}, + {"Sales": 1706.184, "Quantity": 9}, + {"Sales": 665.88, "Quantity": 6}, + {"Sales": 19.46, "Quantity": 7}, + ] + view.delete() + + def test_filter_with_less_than(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales", "Quantity"], + filter=[["Quantity", "<", 3]], + ) + json = view.to_json(start_row=0, end_row=5) + assert json == [ + {"Sales": 261.96, "Quantity": 2}, + {"Sales": 14.62, "Quantity": 2}, + {"Sales": 22.368, "Quantity": 2}, + {"Sales": 55.5, "Quantity": 2}, + {"Sales": 8.56, "Quantity": 2}, + ] + view.delete() + + def test_filter_with_like(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales", "State"], + filter=[["State", "LIKE", "Cal%"]], + ) + json = view.to_json(start_row=0, end_row=5) + assert json == [ + {"Sales": 14.62, "State": "California"}, + {"Sales": 48.86, "State": "California"}, + {"Sales": 7.28, "State": "California"}, + {"Sales": 907.152, "State": "California"}, + {"Sales": 18.504, "State": "California"}, + ] + view.delete() + + def test_multiple_filters(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales", "Region", "Quantity"], + filter=[ + ["Region", "==", "West"], + ["Quantity", ">", 3], + ], + ) + json = view.to_json(start_row=0, end_row=5) + assert json == [ + {"Sales": 48.86, "Region": "West", "Quantity": 7}, + {"Sales": 7.28, "Region": "West", "Quantity": 4}, + {"Sales": 907.152, "Region": "West", "Quantity": 6}, + {"Sales": 114.9, "Region": "West", "Quantity": 5}, + {"Sales": 1706.184, "Region": "West", "Quantity": 9}, + ] + view.delete() + + def test_filter_with_group_by(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales"], + group_by=["Category"], + filter=[["Region", "==", "West"]], + aggregates={"Sales": "sum"}, + ) + num_rows = view.num_rows() + assert num_rows == 4 + json = view.to_json() + assert json == [ + {"__ROW_PATH__": [], "Sales": approx(725457.8245000006)}, + {"__ROW_PATH__": ["Furniture"], "Sales": approx(252612.7435000003)}, + { + "__ROW_PATH__": ["Office Supplies"], + "Sales": approx(220853.24900000007), + }, + {"__ROW_PATH__": ["Technology"], "Sales": approx(251991.83199999997)}, + ] + view.delete() + + +class TestPostgresSort: + def test_sort_ascending(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales", "Quantity"], + sort=[["Sales", "asc"]], + ) + json = view.to_json(start_row=0, end_row=5) + assert json == [ + {"Sales": 0.444, "Quantity": 1}, + {"Sales": 0.556, "Quantity": 1}, + {"Sales": 0.836, "Quantity": 1}, + {"Sales": 0.852, "Quantity": 1}, + {"Sales": 0.876, "Quantity": 1}, + ] + view.delete() + + def test_sort_descending(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales", "Quantity"], + sort=[["Sales", "desc"]], + ) + json = view.to_json(start_row=0, end_row=5) + assert json == [ + {"Sales": 22638.48, "Quantity": 6}, + {"Sales": 17499.95, "Quantity": 5}, + {"Sales": 13999.96, "Quantity": 4}, + {"Sales": 11199.968, "Quantity": 4}, + {"Sales": 10499.97, "Quantity": 3}, + ] + view.delete() + + def test_sort_with_group_by(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales"], + group_by=["Region"], + sort=[["Sales", "desc"]], + aggregates={"Sales": "sum"}, + ) + json = view.to_json() + assert json == [ + {"__ROW_PATH__": [], "Sales": approx(2297200.860299955)}, + {"__ROW_PATH__": ["West"], "Sales": approx(725457.8245000006)}, + {"__ROW_PATH__": ["East"], "Sales": approx(678781.2399999979)}, + {"__ROW_PATH__": ["Central"], "Sales": approx(501239.8908000005)}, + {"__ROW_PATH__": ["South"], "Sales": approx(391721.9050000003)}, + ] + view.delete() + + def test_sort_with_multi_level_group_by(self, client): + # Hierarchical rollup sort - exercises the `first_value(...) OVER + # __WINDOW_N__` emission, which must be that spelling for Postgres + # (DuckDB's `first` does not exist). + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales"], + group_by=["Region", "Category"], + sort=[["Sales", "desc"]], + aggregates={"Sales": "sum"}, + ) + json = view.to_json() + assert json[0]["__ROW_PATH__"] == [] + assert json[1]["__ROW_PATH__"] == ["West"] + assert json[2]["__ROW_PATH__"] == ["West", "Furniture"] + assert json[5]["__ROW_PATH__"] == ["East"] + view.delete() + + def test_multi_column_sort(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Region", "Sales", "Quantity"], + sort=[ + ["Region", "asc"], + ["Sales", "desc"], + ], + ) + json = view.to_json(start_row=0, end_row=5) + assert json == [ + {"Region": "Central", "Sales": 17499.95, "Quantity": 5}, + {"Region": "Central", "Sales": 9892.74, "Quantity": 13}, + {"Region": "Central", "Sales": 9449.95, "Quantity": 5}, + {"Region": "Central", "Sales": 8159.952, "Quantity": 8}, + {"Region": "Central", "Sales": 5443.96, "Quantity": 4}, + ] + view.delete() + + +class TestPostgresExpressions: + def test_simple_expression(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales", "doublesales"], + expressions={"doublesales": '"Sales" * 2'}, + ) + json = view.to_json(start_row=0, end_row=5) + assert json == [ + {"Sales": 261.96, "doublesales": 523.92}, + {"Sales": 731.94, "doublesales": 1463.88}, + {"Sales": 14.62, "doublesales": 29.24}, + {"Sales": 957.5775, "doublesales": 1915.155}, + {"Sales": 22.368, "doublesales": 44.736}, + ] + view.delete() + + def test_expression_with_multiple_columns(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales", "Profit", "margin"], + expressions={"margin": '"Profit" / "Sales"'}, + ) + json = view.to_json(start_row=0, end_row=5) + assert json == [ + {"Sales": 261.96, "Profit": 41.9136, "margin": approx(0.16)}, + {"Sales": 731.94, "Profit": 219.582, "margin": approx(0.3)}, + {"Sales": 14.62, "Profit": 6.8714, "margin": approx(0.47)}, + {"Sales": 957.5775, "Profit": -383.031, "margin": approx(-0.4)}, + {"Sales": 22.368, "Profit": 2.5164, "margin": approx(0.1125)}, + ] + view.delete() + + def test_expression_with_group_by(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["total"], + group_by=["Region"], + expressions={"total": '"Sales" + "Profit"'}, + aggregates={"total": "sum"}, + ) + json = view.to_json() + assert json == [ + {"__ROW_PATH__": [], "total": approx(2583597.882000014)}, + {"__ROW_PATH__": ["Central"], "total": approx(540946.2532999996)}, + {"__ROW_PATH__": ["East"], "total": approx(770304.0199999991)}, + {"__ROW_PATH__": ["South"], "total": approx(438471.33530000027)}, + {"__ROW_PATH__": ["West"], "total": approx(833876.2733999988)}, + ] + view.delete() + + +class TestPostgresViewport: + def test_start_row_and_end_row(self, client): + table = client.open_table("psp_test.superstore") + view = table.view(columns=["Sales", "Profit"]) + json = view.to_json(start_row=10, end_row=15) + assert json == [ + {"Sales": 1706.184, "Profit": 85.3092}, + {"Sales": 911.424, "Profit": 68.3568}, + {"Sales": 15.552, "Profit": 5.4432}, + {"Sales": 407.976, "Profit": 132.5922}, + {"Sales": 68.81, "Profit": -123.858}, + ] + view.delete() + + def test_start_col_and_end_col(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales", "Profit", "Quantity", "Discount"], + ) + json = view.to_json(start_row=0, end_row=5, start_col=1, end_col=3) + assert json == [ + {"Profit": 41.9136, "Quantity": 2}, + {"Profit": 219.582, "Quantity": 3}, + {"Profit": 6.8714, "Quantity": 2}, + {"Profit": -383.031, "Quantity": 5}, + {"Profit": 2.5164, "Quantity": 2}, + ] + view.delete() + + +class TestPostgresDataTypes: + def test_date_columns(self, client): + table = client.open_table("psp_test.superstore") + view = table.view(columns=["Order Date"]) + json = view.to_json(start_row=0, end_row=5) + assert json == [ + {"Order Date": 1478563200000}, + {"Order Date": 1478563200000}, + {"Order Date": 1465689600000}, + {"Order Date": 1444521600000}, + {"Order Date": 1444521600000}, + ] + view.delete() + + +class TestPostgresCombinedOperations: + def test_group_by_filter_sort(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales"], + group_by=["Category"], + filter=[["Region", "==", "West"]], + sort=[["Sales", "desc"]], + aggregates={"Sales": "sum"}, + ) + json = view.to_json() + assert json == [ + {"__ROW_PATH__": [], "Sales": approx(725457.8245000006)}, + {"__ROW_PATH__": ["Furniture"], "Sales": approx(252612.7435000003)}, + {"__ROW_PATH__": ["Technology"], "Sales": approx(251991.83199999997)}, + { + "__ROW_PATH__": ["Office Supplies"], + "Sales": approx(220853.24900000007), + }, + ] + view.delete() + + def test_expressions_group_by_sort(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["profitmargin"], + group_by=["Region"], + expressions={"profitmargin": '"Profit" / "Sales" * 100'}, + sort=[["profitmargin", "desc"]], + aggregates={"profitmargin": "avg"}, + ) + json = view.to_json() + assert json == [ + {"__ROW_PATH__": [], "profitmargin": approx(12.031392972104467)}, + {"__ROW_PATH__": ["West"], "profitmargin": approx(21.948661793784012)}, + {"__ROW_PATH__": ["East"], "profitmargin": approx(16.722695960406636)}, + {"__ROW_PATH__": ["South"], "profitmargin": approx(16.35190329218107)}, + { + "__ROW_PATH__": ["Central"], + "profitmargin": approx(-10.407293926323575), + }, + ] + view.delete() + + +class TestPostgresMinMax: + def test_min_max_integer(self, client): + table = client.open_table("psp_test.superstore") + view = table.view(columns=["Quantity"]) + min_val, max_val = view.get_min_max("Quantity") + assert min_val == 1 + assert max_val == 14 + view.delete() + + def test_min_max_float(self, client): + table = client.open_table("psp_test.superstore") + view = table.view(columns=["Sales"]) + min_val, max_val = view.get_min_max("Sales") + assert min_val == 0.444 + assert max_val == 22638.48 + view.delete() + + def test_min_max_string(self, client): + table = client.open_table("psp_test.superstore") + view = table.view(columns=["Category"]) + min_val, max_val = view.get_min_max("Category") + assert min_val == "Furniture" + assert max_val == "Technology" + view.delete() + + def test_min_max_with_group_by(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Sales"], + group_by=["Region"], + aggregates={"Sales": "sum"}, + ) + min_val, max_val = view.get_min_max("Sales") + assert min_val > 0 + assert max_val > 0 + assert max_val >= min_val + view.delete() + + def test_min_max_with_filter(self, client): + table = client.open_table("psp_test.superstore") + view = table.view( + columns=["Quantity"], + filter=[["Quantity", ">", 10]], + ) + min_val, max_val = view.get_min_max("Quantity") + assert min_val >= 11 + assert max_val == 14 + view.delete() + + +class TestPostgresCoerceTypes: + """The Postgres types whose Perspective type is not their own.""" + + def test_schema(self, client): + table = client.open_table("psp_test.coerce_types") + assert table.schema() == { + "small": "integer", + "big": "float", + "float": "float", + "decimal": "float", + "time": "datetime", + "timestamp": "datetime", + "timestamptz": "datetime", + "date": "date", + "enum": "string", + "uuid": "string", + "json": "string", + "string": "string", + } + + def test_numbers_flat(self, client): + table = client.open_table("psp_test.coerce_types") + view = table.view(columns=["small", "big", "float", "decimal"]) + assert view.to_json() == [ + { + "small": -300, + "big": 9007199254740992.0, + "float": 1.5, + "decimal": pytest.approx(1.234), + }, + { + "small": 300, + "big": -9007199254740992.0, + "float": -1.5, + "decimal": pytest.approx(-5.678), + }, + ] + view.delete() + + def test_temporal_flat(self, client): + table = client.open_table("psp_test.coerce_types") + view = table.view(columns=["time", "timestamp", "timestamptz", "date"]) + assert view.to_json() == [ + { + "time": 3661000, + "timestamp": 1672531200000, + "timestamptz": 1672531200000, + "date": 1672531200000, + }, + { + "time": 1000, + "timestamp": 1672617600000, + "timestamptz": 1672617600000, + "date": 1672617600000, + }, + ] + view.delete() + + def test_stringly_flat(self, client): + # Enums are unknown OIDs (string with a warning), uuid/jsonb + # stringify. + table = client.open_table("psp_test.coerce_types") + view = table.view(columns=["enum", "uuid", "json", "string"]) + assert view.to_json() == [ + { + "enum": "happy", + "uuid": "00000000-0000-0000-0000-000000000001", + "json": '{"a": 1}', + "string": "a", + }, + { + "enum": "sad", + "uuid": "00000000-0000-0000-0000-000000000002", + "json": '{"b": 2}', + "string": "b", + }, + ] + view.delete() + + def test_enum_group_by(self, client): + table = client.open_table("psp_test.coerce_types") + view = table.view( + group_by=["enum"], + columns=["small"], + aggregates={"small": "sum"}, + ) + assert view.to_json() == [ + {"__ROW_PATH__": [], "small": 0}, + {"__ROW_PATH__": ["happy"], "small": -300}, + {"__ROW_PATH__": ["sad"], "small": 300}, + ] + view.delete() + + def test_column_values_view(self, client): + # The filter dropdown's query shape - group by the column, select + # no columns at all. + table = client.open_table("psp_test.coerce_types") + view = table.view(group_by=["enum"], columns=[]) + csv = view.to_csv() + assert [line for line in csv.splitlines() if line] == [ + "__ROW_PATH_0__", + "null", + '"happy"', + '"sad"', + ] + view.delete() + + def test_filter_matching_nothing(self, client): + table = client.open_table("psp_test.coerce_types") + view = table.view( + columns=["small"], + filter=[["string", "==", "no such value"]], + ) + assert view.num_rows() == 0 + assert view.to_json() == [] + view.delete() diff --git a/rust/perspective-python/perspective/virtual_servers/postgres.py b/rust/perspective-python/perspective/virtual_servers/postgres.py new file mode 100644 index 0000000000..7fa62c376a --- /dev/null +++ b/rust/perspective-python/perspective/virtual_servers/postgres.py @@ -0,0 +1,434 @@ +# ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ +# ┃ ██████ ██████ ██████ █ █ █ █ █ █▄ ▀███ █ ┃ +# ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█ ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄ ▀█ █ ▀▀▀▀▀ ┃ +# ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄ █ ▄▄▄▄▄ ┃ +# ┃ █ ██████ █ ▀█▄ █ ██████ █ ███▌▐███ ███████▄ █ ┃ +# ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫ +# ┃ 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 json +import logging +from datetime import date, datetime, time, timezone +from decimal import Decimal + +import psycopg + +import perspective +from perspective.virtual_servers import VirtualServerHandler + +logger = logging.getLogger(__name__) + +# Requires PostgreSQL >= 16: `any_value` is both advertised below and the SQL +# builder's fallback aggregate for columns with no explicit aggregate. +# +# Aggregates, in Postgres's own vocabulary. Only single-argument aggregates +# with scalar results are expressible - the builder emits `{agg}({col})`, so +# e.g. two-argument `string_agg` cannot be advertised. +INT_AGGS = [ + "sum", + "avg", + "count", + "min", + "max", + "stddev", + "stddev_pop", + "stddev_samp", + "variance", + "var_pop", + "var_samp", + "bit_and", + "bit_or", + "bit_xor", + "any_value", +] + +# `bit_and`/`bit_or`/`bit_xor` are defined for integral types only. +FLOAT_AGGS = [a for a in INT_AGGS if not a.startswith("bit_")] + +STRING_AGGS = [ + "count", + "min", + "max", + "any_value", +] + +# Postgres has no `min(boolean)`/`max(boolean)`. +BOOL_AGGS = [ + "count", + "bool_and", + "bool_or", + "every", + "any_value", +] + +# Window functions, in Postgres's own vocabulary - the advertised name is the +# SQL function, emitted verbatim. Excluded relative to the DuckDB handler: +# `product` and `median` (no such functions in Postgres), `rate` (its +# synthesis casts the order key to DOUBLE PRECISION, which errors for +# timestamp keys). `range` frames are advertised only for numeric source +# types - Postgres `RANGE n PRECEDING` requires a numeric order key, and +# errors on timestamps without an interval literal. +FRAMES = ["rows", "range", "cumulative"] +FRAMES_ANY = ["rows", "cumulative"] + +WINDOW_AGGREGATES = [ + {"name": "sum", "frames": FRAMES, "result_type": "float"}, + {"name": "avg", "frames": FRAMES, "result_type": "float"}, + {"name": "count", "frames": FRAMES, "result_type": "float"}, + {"name": "min", "frames": FRAMES}, + {"name": "max", "frames": FRAMES}, + {"name": "stddev_samp", "frames": FRAMES, "result_type": "float"}, + {"name": "stddev_pop", "frames": FRAMES, "result_type": "float"}, + {"name": "var_samp", "frames": FRAMES, "result_type": "float"}, + {"name": "var_pop", "frames": FRAMES, "result_type": "float"}, + # Navigation. + {"name": "first_value", "frames": FRAMES}, + {"name": "last_value", "frames": FRAMES}, + {"name": "nth_value", "frames": FRAMES, "offset": True}, + {"name": "lag", "offset": True}, + {"name": "lead", "offset": True}, + # Ranking. These take no source column - the window's `order_by` is their + # input - but Perspective requires one, so the choice of source is + # immaterial for them. Ranks are `bigint`, which Perspective's 32-bit + # `integer` cannot hold, so they are `float`. + {"name": "row_number", "result_type": "float"}, + {"name": "rank", "result_type": "float"}, + {"name": "dense_rank", "result_type": "float"}, + {"name": "percent_rank", "result_type": "float"}, + {"name": "cume_dist", "result_type": "float"}, + # `ntile`'s argument is a bucket count rather than a row offset. + {"name": "ntile", "offset": True, "result_type": "float"}, + # Perspective's own, with no Postgres equivalent - the SQL translation + # synthesizes it from `lag`. + {"name": "diff", "offset": True, "result_type": "float"}, +] + +# Arithmetic is undefined for the non-numeric types; ordering and navigation +# are not. +WINDOW_AGGREGATES_ANY = [ + {"name": "count", "frames": FRAMES_ANY, "result_type": "float"}, + {"name": "min", "frames": FRAMES_ANY}, + {"name": "max", "frames": FRAMES_ANY}, + {"name": "first_value", "frames": FRAMES_ANY}, + {"name": "last_value", "frames": FRAMES_ANY}, + {"name": "nth_value", "frames": FRAMES_ANY, "offset": True}, + {"name": "lag", "offset": True}, + {"name": "lead", "offset": True}, + {"name": "row_number", "result_type": "float"}, + {"name": "rank", "result_type": "float"}, + {"name": "dense_rank", "result_type": "float"}, + {"name": "percent_rank", "result_type": "float"}, + {"name": "cume_dist", "result_type": "float"}, + {"name": "ntile", "offset": True, "result_type": "float"}, +] + +COMPARE_OPS = [ + "==", + "!=", + "IS DISTINCT FROM", + "IS NOT DISTINCT FROM", + ">=", + "<=", + ">", + "<", +] + +# `LIKE` only for strings - Postgres does not implicitly cast its operand. +STRING_FILTER_OPS = COMPARE_OPS + ["LIKE"] + + +class PostgresVirtualSession: + def __init__(self, callback, conninfo): + self.session = perspective.VirtualServer(PostgresVirtualServerHandler(conninfo)) + self.callback = callback + + def handle_request(self, msg): + self.callback(self.session.handle_request(msg)) + + +class PostgresVirtualServer: + def __init__(self, conninfo=""): + self.conninfo = conninfo + + def new_session(self, callback): + return PostgresVirtualSession(callback, self.conninfo) + + +class PostgresVirtualServerHandler(VirtualServerHandler): + """ + An implementation of a `perspective.VirtualServerHandler` for PostgreSQL + (16 or later, for `any_value`). + + Each handler owns one connection, and views are created as `TEMPORARY + VIEW`s: view names are connection-scoped so sessions cannot collide, every + viewport read re-plans the view's inner `ORDER BY` (a materialized + `CREATE TABLE AS` would depend on seq-scan order, which Postgres does not + guarantee), and all views drop automatically on disconnect. + """ + + def __init__(self, conninfo): + self.db = psycopg.connect(conninfo, autocommit=True) + self.sql_builder = perspective.GenericSQLVirtualServerModel( + { + "create_entity": "TEMPORARY VIEW", + "drop_entity": "VIEW", + "grouping_fn": "GROUPING", + "row_id_expr": "ctid", + "like_escape_clause": "\\", + "regex_fn": "regexp_like", + } + ) + + def get_features(self): + return { + "group_by": True, + # Postgres has no `PIVOT`; pivoting needs a two-pass + # filtered-aggregate strategy the SQL builder does not yet have. + "split_by": False, + "sort": True, + "expressions": True, + "group_rollup_mode": ["rollup", "flat", "total"], + # `ctid` is not a stable row identity, so natural-order windows + # are unsupported. + "unordered": True, + "filter_ops": { + "integer": COMPARE_OPS, + "float": COMPARE_OPS, + "string": STRING_FILTER_OPS, + "boolean": COMPARE_OPS, + "date": COMPARE_OPS, + "datetime": COMPARE_OPS, + }, + "aggregates": { + "integer": INT_AGGS, + "float": FLOAT_AGGS, + "string": STRING_AGGS, + "boolean": BOOL_AGGS, + "date": STRING_AGGS, + "datetime": STRING_AGGS, + }, + "window_aggregates": { + "integer": WINDOW_AGGREGATES, + "float": WINDOW_AGGREGATES, + "string": WINDOW_AGGREGATES_ANY, + "boolean": WINDOW_AGGREGATES_ANY, + "date": WINDOW_AGGREGATES_ANY, + "datetime": WINDOW_AGGREGATES_ANY, + }, + } + + def get_hosted_tables(self): + # `information_schema.tables` omits materialized views, so query the + # catalog directly. + query = """ + SELECT n.nspname, c.relname + FROM pg_catalog.pg_class c + JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace + WHERE c.relkind IN ('r', 'v', 'm', 'f', 'p') + AND n.nspname NOT IN ('pg_catalog', 'information_schema') + AND n.nspname NOT LIKE 'pg_toast%' + AND n.nspname NOT LIKE 'pg_temp%' + ORDER BY 1, 2 + """ + results = run_query(self.db, query) + return [f"{result[0]}.{result[1]}" for result in results] + + def _describe(self, entity_id): + """The column name/OID pairs of any relation or query result. + + `SELECT ... LIMIT 0` planning is the one schema mechanism that works + uniformly for tables, temporary views (which hide in `pg_temp`), and + expression validation - Postgres has no `DESCRIBE`. + """ + cur = run_query(self.db, f"SELECT * FROM {entity_id} LIMIT 0", cursor=True) + return [(d.name, d.type_code) for d in cur.description] + + def table_schema(self, table_name, config=None): + schema = {} + for col_name, oid in self._describe(table_name): + if not col_name.startswith("__"): + schema[col_name] = pg_oid_to_psp(oid, col_name) + + return schema + + def view_column_size(self, view_name, config): + n = len(self._describe(view_name)) + gs = len(config["group_by"]) + if gs > 0: + # Group views carry `__ROW_PATH_N__` columns, plus + # `__GROUPING_ID__` except in flat mode. + n -= gs + if config.get("group_rollup_mode", "rollup") != "flat": + n -= 1 + + return n + + def table_size(self, table_name): + query = self.sql_builder.table_size(table_name) + results = run_query(self.db, query) + return results[0][0] + + def table_make_view(self, table_name, view_name, config): + # Window order keys need column types for `range` frame emission. + schema = self.table_schema(table_name) if config.get("windows") else None + query = self.sql_builder.table_make_view(table_name, view_name, config, schema) + run_query(self.db, query, execute=True) + + def table_validate_expression(self, view_name, expression): + cur = run_query( + self.db, + f"SELECT {expression} FROM {view_name} LIMIT 0", + cursor=True, + ) + + return pg_oid_to_psp(cur.description[0].type_code, expression) + + def view_delete(self, view_name): + query = self.sql_builder.view_delete(view_name) + run_query(self.db, query, execute=True) + + def view_get_min_max(self, view_name, column_name, config): + query = self.sql_builder.view_get_min_max(view_name, column_name, config) + results = run_query(self.db, query) + row = results[0] + return (pg_to_py(row[0]), pg_to_py(row[1])) + + def view_get_data(self, view_name, config, schema, viewport, data): + group_by = config["group_by"] + is_flat = config.get("group_rollup_mode", "rollup") == "flat" + query = self.sql_builder.view_get_data(view_name, config, viewport, schema) + cur = run_query(self.db, query, cursor=True) + results = cur.fetchall() + for cidx, desc in enumerate(cur.description): + dtype = pg_oid_to_psp(desc.type_code, desc.name) + for ridx, row in enumerate(results): + # `__ROW_PATH_N__` cells are kept only while the row's + # grouping id marks that level as un-rolled-up; flat views + # have no `__GROUPING_ID__` column and keep every level. + if len(group_by) == 0: + grouping_id = None + elif is_flat: + grouping_id = 0 + else: + grouping_id = row[0] + + value = pg_to_py(row[cidx]) + if ( + dtype == "string" + and value is not None + and not isinstance(value, str) + ): + value = str(value) + + data.set_col(dtype, desc.name, ridx, value, grouping_id) + + +################################################################################ +# +# Postgres Utils + +# Standard `pg_type` OIDs, stable across every supported Postgres version. +_BOOL_OIDS = frozenset([16]) +_INT_OIDS = frozenset([21, 23]) # int2, int4 +_FLOAT_OIDS = frozenset( + [20, 26, 700, 701, 790, 1700] +) # int8, oid, float4/8, money, numeric +_DATE_OIDS = frozenset([1082]) +_DATETIME_OIDS = frozenset([1083, 1114, 1184, 1266]) # time, timestamp(tz), timetz +_STRING_OIDS = frozenset( + [ + 17, # bytea + 18, # char + 19, # name + 25, # text + 114, # json + 142, # xml + 1042, # bpchar + 1043, # varchar + 1186, # interval + 2950, # uuid + 3802, # jsonb + ] +) + + +def pg_oid_to_psp(oid, col_name=""): + """Convert a `pg_type` OID to a Perspective `ColumnType`. + + Must agree with the value normalization in `pg_to_py`, which decides the + Python value the same column's data arrives as - the two are halves of + one contract. + + `int8` and `numeric` go to `float` because Perspective's `integer` is + 32-bit; `time` goes to `datetime` because it renders as one. Everything + unrecognized - arrays, enums, ranges, user-defined types - renders as + text. Unknown is not fatal: raising here would take down the whole table + for one odd column. + """ + if oid in _BOOL_OIDS: + return "boolean" + + if oid in _INT_OIDS: + return "integer" + + if oid in _FLOAT_OIDS: + return "float" + + if oid in _DATE_OIDS: + return "date" + + if oid in _DATETIME_OIDS: + return "datetime" + + if oid not in _STRING_OIDS: + logger.warning(f"Unknown type OID '{oid}' for column '{col_name}'") + + return "string" + + +def pg_to_py(value): + """Normalize a psycopg result value for `PerspectiveColumn.set_col`. + + `datetime` first: it subclasses `date`, and the binding's date branch + would truncate the time-of-day. Naive timestamps are taken as UTC (the + engine's convention); `time` renders on the epoch date, matching how + narrower temporal types coerce in the Arrow path. + """ + if isinstance(value, Decimal): + return float(value) + + if isinstance(value, datetime): + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + + return int(value.timestamp() * 1000) + + if isinstance(value, time): + return pg_to_py(datetime.combine(date(1970, 1, 1), value)) + + if isinstance(value, (dict, list)): + return json.dumps(value) + + return value + + +def run_query(db, query, execute=False, cursor=False): + query = " ".join(query.split()) + start = datetime.now(datetime.UTC) + try: + cur = db.execute(query) + except psycopg.Error as e: + logger.error(e) + logger.error(f"{query}") + raise + else: + logger.debug(f"{datetime.now(datetime.UTC) - start} {query}") + if cursor: + return cur + elif not execute: + return cur.fetchall()