diff --git a/docs/md/explanation/derived_table.md b/docs/md/explanation/derived_table.md index d780dac399..3b01d83ce1 100644 --- a/docs/md/explanation/derived_table.md +++ b/docs/md/explanation/derived_table.md @@ -58,8 +58,8 @@ let totals = client.table(TableData::View(view), TableInitOptions::default()).aw An unpivoted `View` yields a `Table` with the `View`'s own columns, including `expressions` and window columns, and one row for every row passing its -`filter`. The source's `index` is inherited when the index column is among the -`View`'s `columns`, and its `limit` is inherited as well. +`filter`. The source's `index` is inherited when every index column is among +the `View`'s `columns`, and its `limit` is inherited as well. A pivoted `View` yields the table you see on screen. Each `group_by` column becomes a typed key column of the same name, and each data column takes the name @@ -346,7 +346,7 @@ changes several behaviors for code written against the old implementation: - A row updated out of the source `View`'s `filter` is removed, where the copy retained it with stale values. - Updates arrive in the same cycle as the source rather than eventually. -- A `View` which omits its source's index column is keyed by row identity, so - updates apply in place rather than appending. +- A `View` which omits any of its source's index columns is keyed by row + identity, so updates apply in place rather than appending. - A pivoted `View` yields named key columns and only the rows the `View` displays, where the copy appended its changed rows on every update. diff --git a/docs/md/explanation/table/options.md b/docs/md/explanation/table/options.md index f14259057f..2c7859f0cf 100644 --- a/docs/md/explanation/table/options.md +++ b/docs/md/explanation/table/options.md @@ -2,9 +2,10 @@
`limit` cannot be used in conjunction with `index`.
-Initializing a `Table` with an `index` tells Perspective to treat a column as -the primary key, allowing in-place updates of rows. Only a single column (of any -type) can be used as an `index`. Indexed `Table` instances allow: +Initializing a `Table` with an `index` tells Perspective to treat a column, or +a combination of columns, as the primary key, allowing in-place updates of +rows. A column of any type can be used as an `index`. Indexed `Table` instances +allow: - In-place _updates_ whenever a new row shares an `index` values with an existing row @@ -33,6 +34,52 @@ indexed_table = perspective.Table(data, index="a"); +### Composite index + +A list of column names keys each row by the combination of those columns' +values, in list order. A list of one column is the same as naming that column +directly, and an empty list is an error. + +
+ +JavaScript: + +```javascript +const table = await perspective.table(data, { index: ["name", "sector"] }); +await table.update([{ name: "AAPL", sector: "Tech", price: 1 }]); +await table.remove([["AAPL", "Tech"]]); +await table.get_index(); // ["name", "sector"] +``` + +
+
+ +Python + +```python +table = perspective.table(data, index=["name", "sector"]) +table.update([{"name": "AAPL", "sector": "Tech", "price": 1}]) +table.remove([["AAPL", "Tech"]]) +table.get_index() # ["name", "sector"] +``` + +
+ +A composite index differs from a single-column index in a few ways: + +- Every `update()` row must supply every index column, and an update which + omits one is rejected. A partial update may still omit any other column. +- `remove()` takes a list of keys, each a list of the index column values in + index order, or an Arrow with one column per index column. +- The `__INDEX__` column of an `update()`, and the `__ID__` and `__INDEX__` + columns of a `View`, carry each key as a list of the index column values. + `__INDEX__` is not supported in Arrow updates of a composite-indexed + `Table`; include the index columns instead. +- A `View` with no `sort` orders rows by the index columns in index order, + with each column ordered by its own type and `null` first. +- `vlookup("column", key1, key2, ...)` takes one key argument per index + column. + Initializing a `Table` with a `limit` sets the total number of rows the `Table` is allowed to have. When the `Table` is updated, and the resulting size of the `Table` would exceed its `limit`, rows that exceed `limit` overwrite the oldest diff --git a/docs/md/explanation/view/advanced.md b/docs/md/explanation/view/advanced.md index fe7ac770fa..a0287ae04e 100644 --- a/docs/md/explanation/view/advanced.md +++ b/docs/md/explanation/view/advanced.md @@ -236,8 +236,8 @@ tables across clients. Register a callback to be notified whenever rows are removed from the underlying `Table` by `remove()`, which requires an `index`. The callback receives the -`port_id` and the removed `index` column values as an Apache Arrow of a single -column named after the index. It fires once per update step, only for rows which +`port_id` and the removed `index` column values as an Apache Arrow with one +column per index column, named after that column. It fires once per update step, only for rows which existed before that step; `replace()` reports the keys it does not re-supply, and `clear()` reports every key: diff --git a/docs/md/explanation/view/config/expressions.md b/docs/md/explanation/view/config/expressions.md index d117b7cc0c..7617d8a61f 100644 --- a/docs/md/explanation/view/config/expressions.md +++ b/docs/md/explanation/view/config/expressions.md @@ -166,7 +166,7 @@ to the nearest multiple of 10. | `max(a, b, ...)` | Maximum of inputs | | `random()` | Random float between 0.0 and 1.0 | | `col(name)` | Look up a column by string name at runtime | -| `vlookup(col, key)` | Look up a value in another column by row key | +| `vlookup(col, key, ...)` | Look up a value in another column by row key, one key argument per `index` column | ## See also diff --git a/docs/md/how_to/python/table.md b/docs/md/how_to/python/table.md index c815dd7be3..c2b6e666c5 100644 --- a/docs/md/how_to/python/table.md +++ b/docs/md/how_to/python/table.md @@ -67,6 +67,14 @@ data.set_index("datetime") table = perspective.table(data, index="index") ``` +A `pandas.MultiIndex` arrives as one column per level, named after the level, +so a composite `index` keyed by those columns preserves it: + +```python +data = data.set_index(["name", "sector"]) +table = perspective.table(data, index=["name", "sector"]) +``` + ## Time Zone Handling When parsing `"datetime"` strings, times without an explicit timezone offset are diff --git a/packages/viewer-datagrid/src/ts/style_handlers/body.ts b/packages/viewer-datagrid/src/ts/style_handlers/body.ts index 18142e068c..7ddad3ff8c 100644 --- a/packages/viewer-datagrid/src/ts/style_handlers/body.ts +++ b/packages/viewer-datagrid/src/ts/style_handlers/body.ts @@ -103,6 +103,14 @@ interface StyleMemo { value: unknown; } +function same_key(a: unknown, b: unknown): boolean { + if (Array.isArray(a) && Array.isArray(b)) { + return a.length === b.length && a.every((v, i) => v === b[i]); + } + + return a === b; +} + function row_header_depth(row_header: unknown[] | undefined): number { if (!row_header) { return 0; @@ -302,7 +310,7 @@ export function applyBodyCellStyles( const id = model._ids[(metadata.y ?? 0) - (metadata.y0 ?? 0)]; let key_match = true; for (let i = 0; i < selectedId.length; i++) { - if (selectedId[i] !== id[i]) { + if (!same_key(selectedId[i], id[i])) { key_match = false; break; } diff --git a/rust/perspective-client/perspective.proto b/rust/perspective-client/perspective.proto index 1a27458b81..153f622192 100644 --- a/rust/perspective-client/perspective.proto +++ b/rust/perspective-client/perspective.proto @@ -298,7 +298,7 @@ message GetHostedTablesResp { message HostedTable { string entity_id = 1; - optional string index = 2; + repeated string index = 2; optional uint32 limit = 3; } @@ -382,6 +382,11 @@ message ViewGetConfigResp { //////////////////////////////////////////////////////////////////////////////// +// The columns of a composite `index`, in key order. +message IndexColumns { + repeated string columns = 1; +} + // `Client::table`. message MakeTableReq { MakeTableData data = 1; @@ -390,6 +395,7 @@ message MakeTableReq { oneof make_table_type { string make_index_table = 1; uint32 make_limit_table = 2; + IndexColumns make_composite_index_table = 6; }; // Back this Table's canonical data with the on-disk storage backend diff --git a/rust/perspective-client/src/rust/client.rs b/rust/perspective-client/src/rust/client.rs index fad1620d2c..ff40fdac67 100644 --- a/rust/perspective-client/src/rust/client.rs +++ b/rust/perspective-client/src/rust/client.rs @@ -29,7 +29,7 @@ use crate::proto::{ HostedTable, JoinType, MakeJoinTableReq, MakeTableReq, RemoveHostedTablesUpdateReq, Request, Response, ServerError, ServerSystemInfoReq, }; -use crate::table::{JoinOptions, Table, TableInitOptions, TableOptions, ViewBinding}; +use crate::table::{JoinOptions, Table, TableIndex, TableInitOptions, TableOptions, ViewBinding}; use crate::table_data::{TableData, UpdateData}; use crate::table_ref::TableRef; use crate::utils::*; @@ -576,6 +576,11 @@ impl Client { /// # Ok(()) } /// ``` pub async fn table(&self, input: TableData, options: TableInitOptions) -> ClientResult { + let mut options = options; + if let Some(index) = options.index.take() { + options.index = Some(index.normalize()?); + } + let entity_id = match options.name.clone() { Some(x) => x.to_owned(), None => randid(), @@ -588,7 +593,11 @@ impl Client { if let (None, Some(index)) = (&options.index, &source_index) { let config = view.get_config().await?; let is_flat = config.group_by.is_empty() && config.split_by.is_empty(); - let has_index = config.columns.iter().flatten().any(|x| x == index); + let has_index = index + .columns() + .iter() + .all(|column| config.columns.iter().flatten().any(|x| x == column)); + if is_flat && has_index { options.index = Some(index.clone()); } @@ -713,7 +722,7 @@ impl Client { let client = self.clone(); match self.oneshot(&msg).await? { ClientResp::MakeJoinTableResp(_) => Ok(Table::new(entity_id, client, TableOptions { - index: Some(on.to_owned()), + index: Some(TableIndex::Column(on.to_owned())), limit: None, page_to_disk: None, list_flatten: None, @@ -760,7 +769,7 @@ impl Client { // TODO fix this - name is repeated 2x if let Some(info) = infos.into_iter().find(|i| i.entity_id == entity_id) { let options = TableOptions { - index: info.index, + index: TableIndex::from_columns(info.index), limit: info.limit, page_to_disk: None, list_flatten: None, diff --git a/rust/perspective-client/src/rust/lib.rs b/rust/perspective-client/src/rust/lib.rs index fa59b57144..92d2f98e8c 100644 --- a/rust/perspective-client/src/rust/lib.rs +++ b/rust/perspective-client/src/rust/lib.rs @@ -56,7 +56,7 @@ pub use crate::proto::JoinType; pub use crate::session::{ProxySession, Session}; pub use crate::table::{ DeleteOptions, DescribeError, DescribeVerdict, Description, ExprValidationResult, JoinOptions, - Table, TableInitOptions, TableReadFormat, UpdateOptions, + Table, TableIndex, TableInitOptions, TableReadFormat, UpdateOptions, }; pub use crate::table_data::{TableData, UpdateData}; pub use crate::table_ref::TableRef; @@ -76,7 +76,7 @@ impl From<&str> for HostedTable { fn from(entity_id: &str) -> Self { HostedTable { entity_id: entity_id.to_string(), - index: None, + index: Vec::new(), limit: None, } } diff --git a/rust/perspective-client/src/rust/table.rs b/rust/perspective-client/src/rust/table.rs index 2344dee64d..5008645649 100644 --- a/rust/perspective-client/src/rust/table.rs +++ b/rust/perspective-client/src/rust/table.rs @@ -67,6 +67,66 @@ impl TableReadFormat { } } +/// The column, or columns in key order, a [`Table`] is keyed by. +/// +/// When serialized, a single column is a `string` and a composite key is an +/// array of `string`. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, TS)] +#[serde(untagged)] +pub enum TableIndex { + Column(String), + Columns(Vec), +} + +impl TableIndex { + /// The index columns in key order. + pub fn columns(&self) -> &[String] { + match self { + TableIndex::Column(column) => std::slice::from_ref(column), + TableIndex::Columns(columns) => columns, + } + } + + /// Fold a single-column list into [`TableIndex::Column`] and reject an + /// empty one. + pub fn normalize(self) -> ClientResult { + match self { + TableIndex::Columns(columns) if columns.is_empty() => Err(ClientError::BadTableOptions), + TableIndex::Columns(mut columns) if columns.len() == 1 => { + Ok(TableIndex::Column(columns.remove(0))) + }, + index => Ok(index), + } + } + + /// The normalized index for a list of columns, `None` when empty. + pub fn from_columns(mut columns: Vec) -> Option { + match columns.len() { + 0 => None, + 1 => Some(TableIndex::Column(columns.remove(0))), + _ => Some(TableIndex::Columns(columns)), + } + } +} + +impl From<&str> for TableIndex { + fn from(value: &str) -> Self { + TableIndex::Column(value.to_string()) + } +} + +impl From for TableIndex { + fn from(value: String) -> Self { + TableIndex::Column(value) + } +} + +impl From> for TableIndex { + fn from(value: Vec) -> Self { + TableIndex::Columns(value) + } +} + /// Options which impact the behavior of [`Client::table`], as well as /// subsequent calls to [`Table::update`]. #[derive(Clone, Debug, Default, Serialize, Deserialize, TS)] @@ -79,13 +139,14 @@ pub struct TableInitOptions { #[ts(optional)] pub format: Option, - /// This [`Table`] should use the column named by the `index` parameter as - /// the `index`, which causes [`Table::update`] and [`Client::table`] input - /// to either insert or update existing rows based on `index` column - /// value equality. + /// This [`Table`] should use the column, or columns, named by the `index` + /// parameter as the `index`, which causes [`Table::update`] and + /// [`Client::table`] input to either insert or update existing rows based + /// on `index` column value equality. A list of columns keys rows by the + /// combination of their values, in list order. #[serde(default)] #[ts(optional)] - pub index: Option, + pub index: Option, /// This [`Table`] should be limited to `limit` rows, after which the /// _earliest_ rows will be overwritten (where _earliest_ is defined as @@ -150,8 +211,15 @@ impl TryFrom for MakeTableOptions { .. } => Err(ClientError::BadTableOptions)?, TableOptions { - index: Some(index), .. + index: Some(TableIndex::Column(index)), + .. } => Some(MakeTableType::MakeIndexTable(index)), + TableOptions { + index: Some(TableIndex::Columns(columns)), + .. + } => Some(MakeTableType::MakeCompositeIndexTable(IndexColumns { + columns, + })), TableOptions { limit: Some(limit), .. } => Some(MakeTableType::MakeLimitTable(limit)), @@ -163,7 +231,7 @@ impl TryFrom for MakeTableOptions { #[derive(Clone, Debug)] pub(crate) struct TableOptions { - pub index: Option, + pub index: Option, pub limit: Option, pub page_to_disk: Option, pub list_flatten: Option, @@ -428,7 +496,7 @@ impl Table { /// # async fn run() -> Result<(), Box> { /// # let client: Client = todo!(); /// let options = TableInitOptions { - /// index: Some("x".to_string()), + /// index: Some("x".into()), /// ..TableInitOptions::default() /// }; /// let data = TableData::Update(UpdateData::Csv("x,y\n1,2\n3,4".into())); @@ -436,8 +504,8 @@ impl Table { /// let index = table.get_index(); /// # Ok(()) } /// ``` - pub fn get_index(&self) -> Option { - self.options.index.as_ref().map(|index| index.to_owned()) + pub fn get_index(&self) -> Option { + self.options.index.clone() } /// Returns the user-specified row limit for this table. diff --git a/rust/perspective-js/src/rust/client.rs b/rust/perspective-js/src/rust/client.rs index 1e065e3522..e2d269d3cb 100644 --- a/rust/perspective-js/src/rust/client.rs +++ b/rust/perspective-js/src/rust/client.rs @@ -358,9 +358,10 @@ impl Client { /// - `options` - Optional configuration which provides one of: /// - `limit` - The max number of rows the resulting [`Table`] can /// store. - /// - `index` - The column name to use as an _index_ column. If this - /// `Table` is being instantiated by _data_, this column name must be - /// present in the data. + /// - `index` - The column name, or an array of column names, to use as + /// an _index_. If this `Table` is being instantiated by _data_, these + /// column names must be present in the data. An array keys rows by + /// the combination of the columns' values. /// - `name` - The name of the table. This will be generated if it is /// not provided. /// - `format` - The explicit format of the input data, can be one of @@ -398,6 +399,12 @@ impl Client { /// ```javascript /// const table = await client.table(data, { index: "Row ID" }); /// ``` + /// + /// Create a table with a composite `index`: + /// + /// ```javascript + /// const table = await client.table(data, { index: ["name", "sector"] }); + /// ``` #[wasm_bindgen] pub async fn table( &self, diff --git a/rust/perspective-js/src/rust/table.rs b/rust/perspective-js/src/rust/table.rs index 77575d4813..3c4fa58951 100644 --- a/rust/perspective-js/src/rust/table.rs +++ b/rust/perspective-js/src/rust/table.rs @@ -81,11 +81,16 @@ impl Table { /// /// ```javascript /// const table = await client.table("x,y\n1,2\n3,4", { index: "x" }); - /// const index = table.get_index(); // "x" + /// const index = await table.get_index(); // "x" + /// const table2 = await client.table(data, { index: ["x", "y"] }); + /// const index2 = await table2.get_index(); // ["x", "y"] /// ``` - #[wasm_bindgen] - pub async fn get_index(&self) -> Option { - self.0.get_index() + #[wasm_bindgen(unchecked_return_type = "string | string[] | undefined")] + pub async fn get_index(&self) -> ApiResult { + match self.0.get_index() { + None => Ok(JsValue::UNDEFINED), + Some(index) => Ok(JsValue::from_serde_ext(&index)?), + } } /// Get a copy of the [`Client`] this [`Table`] came from. @@ -250,12 +255,14 @@ impl Table { /// # Arguments /// /// - `indices` - A list of `index` column values for rows that should be - /// removed. + /// removed. For a composite `index`, each entry is an array of the index + /// column values in key order. /// /// # JavaScript Examples /// /// ```javascript /// await table.remove([1, 2, 3]); + /// await composite_table.remove([["AAPL", "Tech"], ["MSFT", "Tech"]]); /// ``` #[wasm_bindgen] pub async fn remove(&self, value: &JsValue, options: Option) -> ApiResult<()> { diff --git a/rust/perspective-js/src/rust/virtual_server.rs b/rust/perspective-js/src/rust/virtual_server.rs index df36bfb0ea..dd6dc6c4d3 100644 --- a/rust/perspective-js/src/rust/virtual_server.rs +++ b/rust/perspective-js/src/rust/virtual_server.rs @@ -87,7 +87,7 @@ const TS_VIRTUAL_SERVER_HANDLER: &'static str = r#" */ export interface VirtualHostedTable { name: string; - index?: string; + index?: string | string[]; limit?: number; } @@ -326,7 +326,7 @@ impl VirtualServerHandler for JsServerHandler { if let Some(s) = item.as_string() { tables.push(HostedTable { entity_id: s, - index: None, + index: Vec::new(), limit: None, }); } else if item.is_object() { @@ -335,7 +335,19 @@ impl VirtualServerHandler for JsServerHandler { .ok_or_else(|| JsError(JsValue::from_str("name must be a string")))?; let index = Reflect::get(&item, &JsValue::from_str("index")) .ok() - .and_then(|v| v.as_string()); + .map(|v| { + if let Some(s) = v.as_string() { + vec![s] + } else if Array::is_array(&v) { + Array::from(&v) + .iter() + .filter_map(|x| x.as_string()) + .collect() + } else { + Vec::new() + } + }) + .unwrap_or_default(); let limit = Reflect::get(&item, &JsValue::from_str("limit")) .ok() .and_then(|v| v.as_f64().map(|x| x as u32)); diff --git a/rust/perspective-js/test/js/composite_index.spec.js b/rust/perspective-js/test/js/composite_index.spec.js new file mode 100644 index 0000000000..76949ba3c1 --- /dev/null +++ b/rust/perspective-js/test/js/composite_index.spec.js @@ -0,0 +1,609 @@ +// ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ +// ┃ ██████ ██████ ██████ █ █ █ █ █ █▄ ▀███ █ ┃ +// ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█ ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄ ▀█ █ ▀▀▀▀▀ ┃ +// ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄ █ ▄▄▄▄▄ ┃ +// ┃ █ ██████ █ ▀█▄ █ ██████ █ ███▌▐███ ███████▄ █ ┃ +// ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫ +// ┃ 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 { test, expect } from "@perspective-dev/test"; +import perspective from "./perspective_client"; + +const data = [ + { name: "AAPL", sector: "Tech", price: 1 }, + { name: "MSFT", sector: "Tech", price: 2 }, + { name: "AAPL", sector: "Hardware", price: 3 }, +]; + +const ordered = [ + { name: "AAPL", sector: "Hardware", price: 3 }, + { name: "AAPL", sector: "Tech", price: 1 }, + { name: "MSFT", sector: "Tech", price: 2 }, +]; + +const index = ["name", "sector"]; + +async function fixture(perspective, input = data, options = {}) { + const table = await perspective.table(input, { index, ...options }); + const view = await table.view(); + const cleanup = async () => { + await view.delete(); + await table.delete(); + }; + + return { table, view, cleanup }; +} + +async function failure(promise) { + let message = ""; + try { + await promise; + } catch (error) { + message = error.message; + } + + return message; +} + +async function read_keys(perspective, indices) { + const table = await perspective.table(indices); + const view = await table.view(); + const result = await view.to_json(); + await view.delete(); + await table.delete(); + return result; +} + +async function subscribe(view) { + const calls = []; + let resolve; + const id = await view.on_remove((removed) => { + calls.push(removed); + if (resolve) { + resolve(removed); + } + }); + + const next = () => + new Promise((x) => { + resolve = x; + }); + + return { calls, next, id }; +} + +((perspective) => { + test.describe("Composite index", function () { + test.describe("construction", function () { + test("from rows", async function () { + const { table, view, cleanup } = await fixture(perspective); + expect(await table.get_index()).toEqual(index); + expect(await table.size()).toEqual(3); + expect(await view.to_json()).toEqual(ordered); + await cleanup(); + }); + + test("from columns", async function () { + const columns = { + name: ["AAPL", "MSFT", "AAPL"], + sector: ["Tech", "Tech", "Hardware"], + price: [1, 2, 3], + }; + + const { table, view, cleanup } = await fixture( + perspective, + columns, + ); + + expect(await table.get_index()).toEqual(index); + expect(await view.to_json()).toEqual(ordered); + await cleanup(); + }); + + test("from csv", async function () { + const csv = + "name,sector,price\nAAPL,Tech,1\nMSFT,Tech,2\nAAPL,Hardware,3"; + const { table, view, cleanup } = await fixture( + perspective, + csv, + ); + expect(await table.get_index()).toEqual(index); + expect(await view.to_json()).toEqual(ordered); + await cleanup(); + }); + + test("from arrow", async function () { + const source = await perspective.table(data); + const source_view = await source.view(); + const arrow = await source_view.to_arrow(); + await source_view.delete(); + await source.delete(); + const { table, view, cleanup } = await fixture( + perspective, + arrow, + ); + + expect(await table.get_index()).toEqual(index); + expect(await view.to_json()).toEqual(ordered); + await cleanup(); + }); + + test("from schema", async function () { + const schema = { + name: "string", + sector: "string", + price: "integer", + }; + + const { table, view, cleanup } = await fixture( + perspective, + schema, + ); + + expect(await table.get_index()).toEqual(index); + await table.update(data); + expect(await view.to_json()).toEqual(ordered); + await cleanup(); + }); + + test("dedupes a repeated key on load, last wins", async function () { + const repeated = [ + { name: "AAPL", sector: "Tech", price: 1 }, + { name: "AAPL", sector: "Tech", price: 9 }, + { name: "MSFT", sector: "Tech", price: 2 }, + ]; + + const { table, view, cleanup } = await fixture( + perspective, + repeated, + ); + + expect(await table.size()).toEqual(2); + expect(await view.to_json()).toEqual([ + { name: "AAPL", sector: "Tech", price: 9 }, + { name: "MSFT", sector: "Tech", price: 2 }, + ]); + + await cleanup(); + }); + + test("a single-column list reports the column name", async function () { + const table = await perspective.table(data, { + index: ["price"], + }); + + expect(await table.get_index()).toEqual("price"); + await table.delete(); + }); + + test("rejects an empty list", async function () { + const message = await failure( + perspective.table(data, { index: [] }), + ); + + expect(message).not.toEqual(""); + }); + + test("rejects `index` with `limit`", async function () { + const message = await failure( + perspective.table(data, { index, limit: 2 }), + ); + + expect(message).not.toEqual(""); + }); + + test("rejects an unknown column", async function () { + const message = await failure( + perspective.table(data, { index: ["name", "nope"] }), + ); + + expect(message).toContain("nope"); + }); + + test("rejects a repeated column", async function () { + const message = await failure( + perspective.table(data, { index: ["name", "name"] }), + ); + + expect(message).toContain("repeated"); + }); + }); + + test.describe("updates", function () { + test("updates in place by matching tuple", async function () { + const { table, view, cleanup } = await fixture(perspective); + await table.update([ + { name: "AAPL", sector: "Tech", price: 10 }, + ]); + + expect(await table.size()).toEqual(3); + expect(await view.to_json()).toEqual([ + { name: "AAPL", sector: "Hardware", price: 3 }, + { name: "AAPL", sector: "Tech", price: 10 }, + { name: "MSFT", sector: "Tech", price: 2 }, + ]); + + await cleanup(); + }); + + test("appends a new tuple", async function () { + const { table, view, cleanup } = await fixture(perspective); + await table.update([ + { name: "MSFT", sector: "Hardware", price: 4 }, + ]); + + expect(await table.size()).toEqual(4); + expect(await view.to_json()).toEqual([ + { name: "AAPL", sector: "Hardware", price: 3 }, + { name: "AAPL", sector: "Tech", price: 1 }, + { name: "MSFT", sector: "Hardware", price: 4 }, + { name: "MSFT", sector: "Tech", price: 2 }, + ]); + + await cleanup(); + }); + + test("applies a partial update with all key columns", async function () { + const wide = data.map((row) => ({ ...row, qty: 7 })); + const { table, view, cleanup } = await fixture( + perspective, + wide, + ); + + await table.update([{ name: "MSFT", sector: "Tech", qty: 8 }]); + expect(await view.to_json()).toEqual([ + { name: "AAPL", sector: "Hardware", price: 3, qty: 7 }, + { name: "AAPL", sector: "Tech", price: 1, qty: 7 }, + { name: "MSFT", sector: "Tech", price: 2, qty: 8 }, + ]); + + await cleanup(); + }); + + test("rejects an update which omits a key column", async function () { + const { table, cleanup } = await fixture(perspective); + const message = await failure( + table.update([{ name: "MSFT", price: 5 }]), + ); + + expect(message).toContain("missing index column `sector`"); + await cleanup(); + }); + + test("rejects a column update which omits a key column", async function () { + const { table, cleanup } = await fixture(perspective); + const message = await failure( + table.update({ sector: ["Tech"], price: [5] }), + ); + + expect(message).toContain("missing index column `name`"); + await cleanup(); + }); + + test("updates by `__INDEX__` array", async function () { + const { table, view, cleanup } = await fixture(perspective); + await table.update([ + { __INDEX__: ["AAPL", "Tech"], price: 100 }, + ]); + + expect(await view.to_json()).toEqual([ + { name: "AAPL", sector: "Hardware", price: 3 }, + { name: "AAPL", sector: "Tech", price: 100 }, + { name: "MSFT", sector: "Tech", price: 2 }, + ]); + + await cleanup(); + }); + + test("rejects a scalar `__INDEX__`", async function () { + const { table, cleanup } = await fixture(perspective); + const message = await failure( + table.update([{ __INDEX__: "AAPL", price: 100 }]), + ); + + expect(message).toContain("__INDEX__"); + await cleanup(); + }); + + test("rejects `__INDEX__` in an Arrow update", async function () { + const { table, cleanup } = await fixture(perspective); + const keys = await perspective.table({ + __INDEX__: ["x"], + price: [1], + }); + + const keys_view = await keys.view(); + const arrow = await keys_view.to_arrow(); + const message = await failure(table.update(arrow)); + expect(message).toContain("__INDEX__"); + await keys_view.delete(); + await keys.delete(); + await cleanup(); + }); + }); + + test.describe("removes", function () { + test("removes by a list of tuples", async function () { + const { table, view, cleanup } = await fixture(perspective); + await table.remove([["AAPL", "Tech"]]); + expect(await table.size()).toEqual(2); + expect(await view.to_json()).toEqual([ + { name: "AAPL", sector: "Hardware", price: 3 }, + { name: "MSFT", sector: "Tech", price: 2 }, + ]); + + await cleanup(); + }); + + test("removes by key columns", async function () { + const { table, view, cleanup } = await fixture(perspective); + await table.remove({ name: ["AAPL"], sector: ["Hardware"] }); + expect(await view.to_json()).toEqual([ + { name: "AAPL", sector: "Tech", price: 1 }, + { name: "MSFT", sector: "Tech", price: 2 }, + ]); + + await cleanup(); + }); + + test("removes by Arrow", async function () { + const { table, view, cleanup } = await fixture(perspective); + const keys = await perspective.table({ + name: ["MSFT"], + sector: ["Tech"], + }); + + const keys_view = await keys.view(); + await table.remove(await keys_view.to_arrow()); + expect(await view.to_json()).toEqual([ + { name: "AAPL", sector: "Hardware", price: 3 }, + { name: "AAPL", sector: "Tech", price: 1 }, + ]); + + await keys_view.delete(); + await keys.delete(); + await cleanup(); + }); + + test("rejects a tuple of the wrong arity", async function () { + const { table, cleanup } = await fixture(perspective); + const message = await failure(table.remove([["AAPL"]])); + expect(message).toContain("2 values"); + await cleanup(); + }); + + test("`on_remove` delivers one Arrow column per key column", async function () { + const { table, view, cleanup } = await fixture(perspective); + const sub = await subscribe(view); + const promise = sub.next(); + await table.remove([ + ["AAPL", "Tech"], + ["MSFT", "Tech"], + ]); + + const removed = await promise; + expect(await read_keys(perspective, removed.indices)).toEqual([ + { name: "AAPL", sector: "Tech" }, + { name: "MSFT", sector: "Tech" }, + ]); + + await cleanup(); + }); + + test("`replace` and `clear` report dropped keys", async function () { + const { table, view, cleanup } = await fixture(perspective); + const sub = await subscribe(view); + let promise = sub.next(); + await table.replace([ + { name: "AAPL", sector: "Tech", price: 5 }, + ]); + + let removed = await promise; + expect(await read_keys(perspective, removed.indices)).toEqual([ + { name: "MSFT", sector: "Tech" }, + { name: "AAPL", sector: "Hardware" }, + ]); + + promise = sub.next(); + await table.clear(); + removed = await promise; + expect(await read_keys(perspective, removed.indices)).toEqual([ + { name: "AAPL", sector: "Tech" }, + ]); + + await cleanup(); + }); + }); + + test.describe("keys", function () { + test("`__ID__` on a flat view is the tuple", async function () { + const { view, cleanup } = await fixture(perspective); + expect(await view.to_json({ id: true })).toEqual([ + { + __ID__: [["AAPL", "Hardware"]], + name: "AAPL", + sector: "Hardware", + price: 3, + }, + { + __ID__: [["AAPL", "Tech"]], + name: "AAPL", + sector: "Tech", + price: 1, + }, + { + __ID__: [["MSFT", "Tech"]], + name: "MSFT", + sector: "Tech", + price: 2, + }, + ]); + + await cleanup(); + }); + + test("`__INDEX__` on a flat view is the tuple", async function () { + const { view, cleanup } = await fixture(perspective); + expect(await view.to_columns({ index: true })).toEqual({ + __INDEX__: [ + [["AAPL", "Hardware"]], + [["AAPL", "Tech"]], + [["MSFT", "Tech"]], + ], + name: ["AAPL", "AAPL", "MSFT"], + sector: ["Hardware", "Tech", "Tech"], + price: [3, 1, 2], + }); + + await cleanup(); + }); + + test("`__INDEX__` on a grouped view lists each tuple", async function () { + const { table, cleanup } = await fixture(perspective); + const view = await table.view({ + group_by: ["sector"], + columns: ["price"], + }); + + const result = await view.to_json({ index: true }); + expect(result[2].__ROW_PATH__).toEqual(["Tech"]); + expect(result[2].__INDEX__).toEqual([ + ["MSFT", "Tech"], + ["AAPL", "Tech"], + ]); + + await view.delete(); + await cleanup(); + }); + }); + + test.describe("ordering", function () { + test("orders by the key columns with numbers compared numerically", async function () { + const rows = [ + { a: "x", n: 10 }, + { a: "x", n: 9 }, + { a: "w", n: 1 }, + ]; + + const table = await perspective.table(rows, { + index: ["a", "n"], + }); + + const view = await table.view(); + expect(await view.to_json()).toEqual([ + { a: "w", n: 1 }, + { a: "x", n: 9 }, + { a: "x", n: 10 }, + ]); + + await view.delete(); + await table.delete(); + }); + + test("orders a null key first", async function () { + const rows = [ + { a: "x", n: 2 }, + { a: "x", n: null }, + { a: "x", n: 1 }, + ]; + + const table = await perspective.table(rows, { + index: ["a", "n"], + }); + + const view = await table.view(); + expect(await view.to_json()).toEqual([ + { a: "x", n: null }, + { a: "x", n: 1 }, + { a: "x", n: 2 }, + ]); + + await view.delete(); + await table.delete(); + }); + + test("`sort` wins over key order", async function () { + const { table, cleanup } = await fixture(perspective); + const view = await table.view({ sort: [["price", "desc"]] }); + expect(await view.to_json()).toEqual([ + { name: "AAPL", sector: "Hardware", price: 3 }, + { name: "MSFT", sector: "Tech", price: 2 }, + { name: "AAPL", sector: "Tech", price: 1 }, + ]); + + await view.delete(); + await cleanup(); + }); + }); + + test.describe("derived tables", function () { + test("inherits the index when every key column is present", async function () { + const { table, cleanup } = await fixture(perspective); + const view = await table.view({ + columns: ["name", "sector", "price"], + }); + + const derived = await perspective.table(view); + expect(await derived.get_index()).toEqual(index); + await table.remove([["MSFT", "Tech"]]); + expect(await derived.size()).toEqual(2); + await derived.delete(); + await view.delete(); + await cleanup(); + }); + + test("has no index when a key column is missing", async function () { + const { table, cleanup } = await fixture(perspective); + const view = await table.view({ columns: ["name", "price"] }); + const derived = await perspective.table(view); + expect(await derived.get_index()).toBeUndefined(); + const derived_view = await derived.view(); + expect(await derived_view.to_json({ id: true })).toEqual([ + { __ID__: [["AAPL", "Hardware"]], name: "AAPL", price: 3 }, + { __ID__: [["AAPL", "Tech"]], name: "AAPL", price: 1 }, + { __ID__: [["MSFT", "Tech"]], name: "MSFT", price: 2 }, + ]); + + await derived_view.delete(); + await derived.delete(); + await view.delete(); + await cleanup(); + }); + }); + + test("`vlookup` takes one key per index column", async function () { + const { table, cleanup } = await fixture(perspective); + const view = await table.view({ + columns: ["name", "sector", "looked_up"], + expressions: { + looked_up: "vlookup('price', 'MSFT', 'Tech')", + }, + }); + + expect(await view.to_columns()).toEqual({ + name: ["AAPL", "AAPL", "MSFT"], + sector: ["Hardware", "Tech", "Tech"], + looked_up: [2, 2, 2], + }); + + await view.delete(); + await cleanup(); + }); + + test("`list_flatten` names a repeated key column", async function () { + const rows = [ + { name: "AAPL", sector: ["Tech", "Hardware"], price: 1 }, + ]; + + const message = await failure( + perspective.table(rows, { index, list_flatten: "zip" }), + ); + + expect(message).toContain("indexed on `name`"); + }); + }); +})(perspective); diff --git a/rust/perspective-js/test/js/remote.spec.js b/rust/perspective-js/test/js/remote.spec.js index 45c8441220..1a6a0b48b1 100644 --- a/rust/perspective-js/test/js/remote.spec.js +++ b/rust/perspective-js/test/js/remote.spec.js @@ -45,6 +45,21 @@ test.describe("WebSocketManager", function () { await table.delete(); }); + test("reports a composite index through `open_table`", async () => { + const data = [{ x: 1, y: "a" }]; + const table = await perspective.table(data, { + name: "composite", + index: ["x", "y"], + }); + + const client = await perspective.websocket(`ws://localhost:${port}`); + const client_table = await client.open_table("composite"); + expect(await client_table.get_index()).toEqual(["x", "y"]); + await client.terminate(); + await new Promise((x) => setTimeout(x, 10)); + await table.delete(); + }); + test("Throws an exception when the server is detached", async () => { const data = [{ x: 1 }]; const _table = await perspective.table(data, { name: "test2" }); diff --git a/rust/perspective-python/perspective/tests/table/test_on_remove.py b/rust/perspective-python/perspective/tests/table/test_on_remove.py index d56be765ef..962817860b 100644 --- a/rust/perspective-python/perspective/tests/table/test_on_remove.py +++ b/rust/perspective-python/perspective/tests/table/test_on_remove.py @@ -44,6 +44,23 @@ def test_on_remove_string_index(self): assert removed_indices(calls[0]) == [{"y": "b"}, {"y": "c"}] view.remove_remove(cid) + def test_on_remove_composite_index(self): + tbl = Table( + [ + {"name": "AAPL", "sector": "Tech", "price": 1}, + {"name": "MSFT", "sector": "Tech", "price": 2}, + ], + index=["name", "sector"], + ) + view = tbl.view() + calls = [] + cid = view.on_remove(lambda port_id, indices: calls.append(indices)) + tbl.remove([["MSFT", "Tech"]]) + assert len(calls) == 1 + assert removed_indices(calls[0]) == [{"name": "MSFT", "sector": "Tech"}] + assert tbl.size() == 1 + view.remove_remove(cid) + def test_on_remove_ignores_unknown_and_update(self): tbl = Table(data, index="x") view = tbl.view() diff --git a/rust/perspective-python/perspective/tests/table/test_table.py b/rust/perspective-python/perspective/tests/table/test_table.py index 51e86dd5a5..1edb245512 100644 --- a/rust/perspective-python/perspective/tests/table/test_table.py +++ b/rust/perspective-python/perspective/tests/table/test_table.py @@ -14,6 +14,7 @@ import perspective from pytest import mark, raises import pytest +import pandas as pd import perspective as psp client = psp.Server().new_local_client() @@ -523,6 +524,60 @@ def test_table_get_index_none(self): tbl = Table({"a": ["", "a", None, "b"], "b": [4, 3, 2, 1]}) assert tbl.get_index() is None + # composite index + + def test_table_composite_index(self): + data = [ + {"name": "AAPL", "sector": "Tech", "price": 1}, + {"name": "MSFT", "sector": "Tech", "price": 2}, + {"name": "AAPL", "sector": "Hardware", "price": 3}, + ] + tbl = Table(data, index=["name", "sector"]) + assert tbl.get_index() == ["name", "sector"] + assert tbl.size() == 3 + assert tbl.view().to_records() == [ + {"name": "AAPL", "sector": "Hardware", "price": 3}, + {"name": "AAPL", "sector": "Tech", "price": 1}, + {"name": "MSFT", "sector": "Tech", "price": 2}, + ] + + def test_table_composite_index_dedupes(self): + data = [ + {"name": "AAPL", "sector": "Tech", "price": 1}, + {"name": "AAPL", "sector": "Tech", "price": 9}, + ] + tbl = Table(data, index=["name", "sector"]) + assert tbl.size() == 1 + assert tbl.view().to_records() == [{"name": "AAPL", "sector": "Tech", "price": 9}] + + def test_table_composite_index_single_column_list(self): + tbl = Table({"a": [1, 2], "b": [3, 4]}, index=["a"]) + assert tbl.get_index() == "a" + + def test_table_composite_index_empty_raises(self): + with raises(Exception): + Table({"a": [1, 2], "b": [3, 4]}, index=[]) + + def test_table_composite_index_unknown_column_raises(self): + with raises(Exception): + Table({"a": [1, 2], "b": [3, 4]}, index=["a", "c"]) + + def test_table_composite_index_ids(self): + tbl = Table({"a": ["x", "x"], "b": [2, 1]}, index=["a", "b"]) + assert tbl.view().to_records(id=True) == [ + {"__ID__": [["x", 1]], "a": "x", "b": 1}, + {"__ID__": [["x", 2]], "a": "x", "b": 2}, + ] + + def test_table_composite_index_from_pandas_multiindex(self): + df = pd.DataFrame( + {"name": ["AAPL", "MSFT"], "sector": ["Tech", "Tech"], "price": [1, 2]} + ).set_index(["name", "sector"]) + tbl = Table(df, index=["name", "sector"]) + assert tbl.get_index() == ["name", "sector"] + tbl.update([{"name": "AAPL", "sector": "Tech", "price": 5}]) + assert tbl.view().to_columns()["price"] == [5, 2] + # limit def test_table_limit(self): diff --git a/rust/perspective-python/perspective/tests/table/test_update.py b/rust/perspective-python/perspective/tests/table/test_update.py index 6b093869fc..a11907c671 100644 --- a/rust/perspective-python/perspective/tests/table/test_update.py +++ b/rust/perspective-python/perspective/tests/table/test_update.py @@ -14,7 +14,7 @@ import pyarrow as pa import pandas as pd from datetime import date, datetime -from pytest import mark +from pytest import mark, raises import perspective as psp client = psp.Server().new_local_client() @@ -22,6 +22,39 @@ class TestUpdate(object): + def test_update_composite_index_in_place(self): + tbl = Table( + [ + {"name": "AAPL", "sector": "Tech", "price": 1}, + {"name": "MSFT", "sector": "Tech", "price": 2}, + ], + index=["name", "sector"], + ) + tbl.update([{"name": "AAPL", "sector": "Tech", "price": 10}]) + tbl.update({"name": ["MSFT"], "sector": ["Hardware"], "price": [4]}) + assert tbl.size() == 3 + assert tbl.view().to_records() == [ + {"name": "AAPL", "sector": "Tech", "price": 10}, + {"name": "MSFT", "sector": "Hardware", "price": 4}, + {"name": "MSFT", "sector": "Tech", "price": 2}, + ] + + def test_update_composite_index_by_index_array(self): + tbl = Table( + [{"name": "AAPL", "sector": "Tech", "price": 1}], + index=["name", "sector"], + ) + tbl.update([{"__INDEX__": ["AAPL", "Tech"], "price": 7}]) + assert tbl.view().to_records() == [{"name": "AAPL", "sector": "Tech", "price": 7}] + + def test_update_composite_index_missing_column_raises(self): + tbl = Table( + [{"name": "AAPL", "sector": "Tech", "price": 1}], + index=["name", "sector"], + ) + with raises(Exception, match="missing index column `sector`"): + tbl.update([{"name": "AAPL", "price": 7}]) + def test_update_with_missing_or_null_values(self): tbl = Table({"a": "string", "b": "string"}, index="a") tbl.update([{"a": "1", "b": "2"}, {"a": "3", "b": "4"}]) diff --git a/rust/perspective-python/perspective/tests/virtual_servers/test_hosted_tables.py b/rust/perspective-python/perspective/tests/virtual_servers/test_hosted_tables.py new file mode 100644 index 0000000000..f8c60273b2 --- /dev/null +++ b/rust/perspective-python/perspective/tests/virtual_servers/test_hosted_tables.py @@ -0,0 +1,58 @@ +# ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ +# ┃ ██████ ██████ ██████ █ █ █ █ █ █▄ ▀███ █ ┃ +# ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█ ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄ ▀█ █ ▀▀▀▀▀ ┃ +# ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄ █ ▄▄▄▄▄ ┃ +# ┃ █ ██████ █ ▀█▄ █ ██████ █ ███▌▐███ ███████▄ █ ┃ +# ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫ +# ┃ 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 perspective +from perspective import Client +from perspective.virtual_servers import VirtualServerHandler + + +class HostedTablesHandler(VirtualServerHandler): + def get_hosted_tables(self): + return [ + "plain", + {"name": "single", "index": "a"}, + {"name": "composite", "index": ["a", "b"]}, + ] + + def table_schema(self, table_name): + return {"a": "integer", "b": "string"} + + def table_size(self, table_name): + return 0 + + def table_make_view(self, table_name, view_name, config): + pass + + def view_delete(self, view_name): + pass + + def view_get_data(self, view_name, config, schema, viewport, data): + pass + + +def make_client(handler): + session = perspective.VirtualServer(handler) + + def handle_request(msg): + client.handle_response(session.handle_request(msg)) + + client = Client(handle_request) + return client + + +class TestHostedTables(object): + def test_hosted_table_index_shapes(self): + client = make_client(HostedTablesHandler()) + assert client.open_table("plain").get_index() is None + assert client.open_table("single").get_index() == "a" + assert client.open_table("composite").get_index() == ["a", "b"] diff --git a/rust/perspective-python/perspective/tests/widget/test_widget.py b/rust/perspective-python/perspective/tests/widget/test_widget.py index 818bac2b52..93222bbe7f 100644 --- a/rust/perspective-python/perspective/tests/widget/test_widget.py +++ b/rust/perspective-python/perspective/tests/widget/test_widget.py @@ -73,6 +73,12 @@ def test_widget_schema_with_index(self): widget = PerspectiveWidget({"a": "integer"}, index="a") assert widget.table.get_index() == "a" + def test_widget_schema_with_composite_index(self): + widget = PerspectiveWidget({"a": "integer", "b": "string"}, index=["a", "b"]) + assert widget.table.get_index() == ["a", "b"] + load_msg = widget._make_load_message() + assert load_msg.to_columns()["data"]["options"] == {"index": ["a", "b"]} + def test_widget_schema_with_limit(self): widget = PerspectiveWidget({"a": "integer"}, limit=5) assert widget.table.get_limit() == 5 diff --git a/rust/perspective-python/perspective/widget/__init__.py b/rust/perspective-python/perspective/widget/__init__.py index 51159b4fd4..3a0545c1c4 100644 --- a/rust/perspective-python/perspective/widget/__init__.py +++ b/rust/perspective-python/perspective/widget/__init__.py @@ -152,8 +152,8 @@ def __init__( # Keyword Arguments - - `index` (`str`): A column name to be used as the primary key. - Ignored if `server` is True. + - `index` (`str`|`list[str]`): A column name, or a list of column + names, to be used as the primary key. Ignored if `server` is True. - `binding_mode` (`str`): "client-server" or "server" - `limit` (`int`): A upper limit on the number of rows in the Table. Cannot be set at the same time as `index`, ignored if `server` diff --git a/rust/perspective-python/src/client/client_async.rs b/rust/perspective-python/src/client/client_async.rs index c3436dd24d..4f409f8201 100644 --- a/rust/perspective-python/src/client/client_async.rs +++ b/rust/perspective-python/src/client/client_async.rs @@ -21,12 +21,12 @@ use perspective_client::config::{ColumnType, ViewConfigUpdate}; use perspective_client::proto::ListFlatten; use perspective_client::{ Client, ColumnWindow, DeleteOptions, DescribeVerdict, OnRemoveData, OnUpdateData, OnUpdateMode, - OnUpdateOptions, Table, TableData, TableInitOptions, TableReadFormat, TableRef, UpdateData, - UpdateOptions, View, ViewWindow, assert_table_api, assert_view_api, asyncfn, + OnUpdateOptions, Table, TableData, TableIndex, TableInitOptions, TableReadFormat, TableRef, + UpdateData, UpdateOptions, View, ViewWindow, assert_table_api, assert_view_api, asyncfn, }; use pyo3::exceptions::PyValueError; use pyo3::prelude::*; -use pyo3::types::{PyAny, PyBytes, PyDict, PyString}; +use pyo3::types::{PyAny, PyBytes, PyDict, PyList, PyString}; use pythonize::depythonize; use super::pandas::arrow_to_pandas; @@ -209,7 +209,7 @@ impl AsyncClient { &self, input: Py, limit: Option, - index: Option>, + index: Option>, name: Option>, format: Option>, page_to_disk: Option, @@ -233,7 +233,7 @@ impl AsyncClient { match (limit, index) { (None, None) => {}, (None, Some(index)) => { - options.index = Some(index.extract::(py)?); + options.index = Some(parse_index(index.bind(py))?); }, (Some(limit), None) => options.limit = Some(limit), (Some(_), Some(_)) => { @@ -426,10 +426,35 @@ pub struct AsyncTable { assert_table_api!(AsyncTable); +pub(crate) fn parse_index(index: &Bound<'_, PyAny>) -> PyResult { + if let Ok(column) = index.extract::() { + return Ok(TableIndex::Column(column)); + } + + let columns = index + .extract::>() + .map_err(|_| PyValueError::new_err("`index` must be a str or a list of str"))?; + + Ok(TableIndex::Columns(columns)) +} + +pub(crate) fn index_to_py( + py: Python<'_>, + index: Option, +) -> PyResult>> { + Ok(match index { + None => None, + Some(TableIndex::Column(column)) => Some(PyString::new(py, &column).into_any().unbind()), + Some(TableIndex::Columns(columns)) => Some(PyList::new(py, columns)?.into_any().unbind()), + }) +} + #[pymethods] impl AsyncTable { - pub fn get_index(&self) -> Option { - self.table.get_index() + /// Returns the index column name, or the list of index column names for + /// a composite index. + pub fn get_index(&self, py: Python<'_>) -> PyResult>> { + index_to_py(py, self.table.get_index()) } /// Get a copy of the [`Client`] this [`Table`] came from. diff --git a/rust/perspective-python/src/client/client_sync.rs b/rust/perspective-python/src/client/client_sync.rs index c770583ed3..a17d6b4a83 100644 --- a/rust/perspective-python/src/client/client_sync.rs +++ b/rust/perspective-python/src/client/client_sync.rs @@ -148,9 +148,9 @@ impl Client { /// - `options` - Optional configuration which provides one of: /// - `limit` - The max number of rows the resulting [`Table`] can /// store. - /// - `index` - The column name to use as an _index_ column. If this - /// `Table` is being instantiated by _data_, this column name must be - /// present in the data. + /// - `index` - The column name, or list of column names, to use as an + /// _index_. If this `Table` is being instantiated by _data_, this + /// column name must be present in the data. /// - `name` - The name of the table. This will be generated if it is /// not provided. /// - `format` - The explicit format of the input data, can be one of @@ -174,7 +174,7 @@ impl Client { py: Python<'_>, input: Py, limit: Option, - index: Option>, + index: Option>, name: Option>, format: Option>, page_to_disk: Option, @@ -328,16 +328,17 @@ impl Table { )) } - /// Returns the name of the index column for the table. + /// Returns the name of the index column for the table, or the list of + /// index column names for a composite index. /// /// # Python Examples /// /// ```python /// table = perspective.table("x,y\n1,2\n3,4", index="x"); - /// index = client.get_index() + /// index = table.get_index() /// ``` - pub fn get_index(&self) -> Option { - self.0.get_index() + pub fn get_index(&self, py: Python<'_>) -> PyResult>> { + self.0.get_index(py) } /// Get a copy of the [`Client`] this [`Table`] came from. diff --git a/rust/perspective-python/src/server/virtual_server_sync.rs b/rust/perspective-python/src/server/virtual_server_sync.rs index 8cf7b6316e..3ff75f0892 100644 --- a/rust/perspective-python/src/server/virtual_server_sync.rs +++ b/rust/perspective-python/src/server/virtual_server_sync.rs @@ -105,13 +105,22 @@ impl VirtualServerHandler for PyServerHandler { Ok::<_, PyErr>(if x.is_instance_of::() { HostedTable { entity_id: x.to_string(), - index: None, + index: Vec::new(), limit: None, } } else { HostedTable { entity_id: x.get_item("name")?.to_string(), - index: x.get_item("index").ok().and_then(|x| x.extract().ok()), + index: x + .get_item("index") + .ok() + .and_then(|x| { + x.extract::() + .map(|s| vec![s]) + .or_else(|_| x.extract::>()) + .ok() + }) + .unwrap_or_default(), limit: x.get_item("limit").ok().and_then(|x| x.extract().ok()), } }) diff --git a/rust/perspective-server/cpp/perspective/CMakeLists.txt b/rust/perspective-server/cpp/perspective/CMakeLists.txt index 4a361d0757..3ea679149d 100644 --- a/rust/perspective-server/cpp/perspective/CMakeLists.txt +++ b/rust/perspective-server/cpp/perspective/CMakeLists.txt @@ -517,6 +517,7 @@ set(SOURCE_FILES ${PSP_CPP_SRC}/src/cpp/none.cpp ${PSP_CPP_SRC}/src/cpp/path.cpp ${PSP_CPP_SRC}/src/cpp/pivot.cpp + ${PSP_CPP_SRC}/src/cpp/pkey_codec.cpp ${PSP_CPP_SRC}/src/cpp/pool.cpp ${PSP_CPP_SRC}/src/cpp/port.cpp ${PSP_CPP_SRC}/src/cpp/process_state.cpp diff --git a/rust/perspective-server/cpp/perspective/src/cpp/arrow_loader.cpp b/rust/perspective-server/cpp/perspective/src/cpp/arrow_loader.cpp index 6914c07ad3..612a82e2d3 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/arrow_loader.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/arrow_loader.cpp @@ -257,12 +257,20 @@ void ArrowLoader::fill_table( t_data_table& tbl, const t_schema& input_schema, - const std::string& index, + const std::vector& index, std::uint32_t offset, bool is_update ) { bool implicit_index = false; const auto& arrow_fields = fields(); + if (index.size() > 1 + && std::find(m_names.begin(), m_names.end(), "__INDEX__") + != m_names.end()) { + PSP_COMPLAIN_AND_ABORT( + "`__INDEX__` is not supported for a Table with a composite " + "index; include the index columns instead.\n" + ); + } parallel_for(int(m_names.size()), [&](int cidx) { auto name = m_names[cidx]; @@ -304,16 +312,16 @@ ArrowLoader::fill_table( key_col->set_nth(ridx, (ridx + offset)); okey_col->set_nth(ridx, (ridx + offset)); } - } else { - if (!input_schema.has_column(index)) { + } else if (index.size() == 1) { + if (!input_schema.has_column(index[0])) { std::stringstream ss; - ss << "Specified index `" << index + ss << "Specified index `" << index[0] << "` is invalid as it does not appear in the Table.\n"; PSP_COMPLAIN_AND_ABORT(ss.str()); } - tbl.clone_column(index, "psp_pkey"); - tbl.clone_column(index, "psp_okey"); + tbl.clone_column(index[0], "psp_pkey"); + tbl.clone_column(index[0], "psp_okey"); } } } @@ -1129,7 +1137,7 @@ ArrowLoader::row_count() const { } std::optional -ArrowLoader::repeated_index(const std::string& index) const { +ArrowLoader::repeated_index(const std::vector& index) const { if (!m_expanded) { return std::nullopt; } @@ -1140,8 +1148,10 @@ ArrowLoader::repeated_index(const std::string& index) const { && m_normalized->per_element[std::distance(m_names.begin(), it)]; }; - if (!index.empty() && !is_per_element(index)) { - return index; + for (const auto& name : index) { + if (!is_per_element(name)) { + return name; + } } const std::string implicit{"__INDEX__"}; diff --git a/rust/perspective-server/cpp/perspective/src/cpp/arrow_writer.cpp b/rust/perspective-server/cpp/perspective/src/cpp/arrow_writer.cpp index 987ae0c61a..9fa93c0d63 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/arrow_writer.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/arrow_writer.cpp @@ -79,16 +79,18 @@ get_scalar(t_tscalar& t) { // } -std::shared_ptr -column_to_arrow_ipc( - const t_column& col, const std::string& name, t_uindex nrows +static void +column_to_arrow( + const t_column& col, + const std::string& name, + t_uindex nrows, + std::shared_ptr& field, + std::shared_ptr& array ) { t_get_data_extents extents{ 0, static_cast(nrows), 0, 1 }; auto get = [&col](t_uindex ridx) { return col.get_scalar(ridx); }; - std::shared_ptr field; - std::shared_ptr array; switch (col.get_dtype()) { case DTYPE_INT8: { field = arrow::field(name, arrow::int8()); @@ -178,11 +180,13 @@ column_to_arrow_ipc( PSP_COMPLAIN_AND_ABORT(ss.str()); } } +} - auto schema = arrow::schema({field}); - auto batch = arrow::RecordBatch::Make( - schema, static_cast(nrows), {array} - ); +static std::shared_ptr +batch_to_ipc( + const std::shared_ptr& schema, + const std::shared_ptr& batch +) { arrow::Result> allocated = arrow::AllocateResizableBuffer(0); if (!allocated.ok()) { @@ -204,4 +208,41 @@ column_to_arrow_ipc( return std::make_shared(buffer->ToString()); } +std::shared_ptr +column_to_arrow_ipc( + const t_column& col, const std::string& name, t_uindex nrows +) { + std::shared_ptr field; + std::shared_ptr array; + column_to_arrow(col, name, nrows, field, array); + auto schema = arrow::schema({field}); + auto batch = arrow::RecordBatch::Make( + schema, static_cast(nrows), {array} + ); + return batch_to_ipc(schema, batch); +} + +std::shared_ptr +table_to_arrow_ipc(const t_data_table& table) { + const auto nrows = table.size(); + const auto& names = table.get_schema().columns(); + std::vector> fields; + std::vector> arrays; + fields.reserve(names.size()); + arrays.reserve(names.size()); + for (const auto& name : names) { + std::shared_ptr field; + std::shared_ptr array; + column_to_arrow(*table.get_const_column(name), name, nrows, field, array); + fields.push_back(field); + arrays.push_back(array); + } + + auto schema = arrow::schema(fields); + auto batch = arrow::RecordBatch::Make( + schema, static_cast(nrows), arrays + ); + return batch_to_ipc(schema, batch); +} + } // namespace perspective::apachearrow \ No newline at end of file diff --git a/rust/perspective-server/cpp/perspective/src/cpp/computed_expression.cpp b/rust/perspective-server/cpp/perspective/src/cpp/computed_expression.cpp index c92e6495cc..ed2710f427 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/computed_expression.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/computed_expression.cpp @@ -849,7 +849,7 @@ t_computed_function_store::t_computed_function_store( computed_function::col(vocab, is_type_validator, source_table, row_idx) ), m_vlookup_fn(computed_function::vlookup( - vocab, is_type_validator, source_table, row_idx + vocab, is_type_validator, pkey_map, source_table, row_idx )) {} void diff --git a/rust/perspective-server/cpp/perspective/src/cpp/computed_function.cpp b/rust/perspective-server/cpp/perspective/src/cpp/computed_function.cpp index c5a38ed825..5634a05fd3 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/computed_function.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/computed_function.cpp @@ -11,6 +11,7 @@ // ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛ #include +#include #include #include #include @@ -2290,12 +2291,14 @@ col::operator()(t_parameter_list parameters) { vlookup::vlookup( t_expression_vocab& expression_vocab, bool is_type_validator, + const t_pkey_mapping& pkey_map, const std::shared_ptr& source_table, t_uindex& row_idx ) : - exprtk::igeneric_function("TT"), + exprtk::igeneric_function("TT*"), m_expression_vocab(expression_vocab), m_is_type_validator(is_type_validator), + m_pkey_map(pkey_map), m_source_table(source_table), m_row_idx(row_idx) {} vlookup::~vlookup() = default; @@ -2311,19 +2314,50 @@ vlookup::operator()(t_parameter_list parameters) { column_name.set(column_gt_view()); t_dtype column_name_dtype = column_name.get_dtype(); + if (column_name_dtype != DTYPE_STR) { + rval.m_status = STATUS_CLEAR; + return rval; + } - t_generic_type& index_gt = parameters[1]; - t_scalar_view index_gt_view(index_gt); + auto pkey_col = m_source_table->get_const_column("psp_pkey"); + const auto& codec = m_source_table->get_pkey_codec(); t_tscalar index; + std::string encoded; + if (codec) { + if (parameters.size() != 1 + codec->arity()) { + rval.m_status = STATUS_CLEAR; + return rval; + } - index.set(index_gt_view()); + std::vector key; + key.reserve(codec->arity()); + for (std::size_t i = 0; i < codec->arity(); ++i) { + t_scalar_view key_view(parameters[i + 1]); + t_tscalar part; + part.set(key_view()); + const bool want_str = codec->dtypes()[i] == DTYPE_STR; + if (part.is_valid() && want_str != (part.get_dtype() == DTYPE_STR)) { + rval.m_status = STATUS_CLEAR; + return rval; + } - auto pkey_col = m_source_table->get_const_column("psp_pkey"); + key.push_back(part); + } - if (column_name_dtype != DTYPE_STR - || index.get_dtype() != pkey_col->get_dtype()) { - rval.m_status = STATUS_CLEAR; - return rval; + codec->encode(key, encoded); + index.set(encoded.c_str()); + } else { + if (parameters.size() != 2) { + rval.m_status = STATUS_CLEAR; + return rval; + } + + t_scalar_view index_gt_view(parameters[1]); + index.set(index_gt_view()); + if (index.get_dtype() != pkey_col->get_dtype()) { + rval.m_status = STATUS_CLEAR; + return rval; + } } if (!column_name.is_valid()) { @@ -2343,6 +2377,16 @@ vlookup::operator()(t_parameter_list parameters) { return rval; } + if (codec) { + const auto it = m_pkey_map.find(index); + if (it != m_pkey_map.end()) { + rval.set(col->get_scalar(it->second)); + } + + rval.m_type = col->get_dtype(); + return rval; + } + auto idx = index.to_uint64(); if (idx < col->size()) { auto res = col->get_scalar(idx); diff --git a/rust/perspective-server/cpp/perspective/src/cpp/derived_source.cpp b/rust/perspective-server/cpp/perspective/src/cpp/derived_source.cpp index d2d85c5de0..8aaf361b3c 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/derived_source.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/derived_source.cpp @@ -206,9 +206,14 @@ namespace { } } // namespace -std::string +std::vector t_derived_source::index() const { - return ""; + return {}; +} + +std::shared_ptr +t_derived_source::pkey_codec() const { + return nullptr; } std::uint32_t @@ -1154,17 +1159,25 @@ class t_flat_derived_source : public t_derived_source { ); } - std::string + std::vector index() const override { - const std::string& index = m_parent->get_index(); + const auto& index = m_parent->get_index(); const auto columns = m_view->get_view_config()->get_columns(); - if (std::find(columns.begin(), columns.end(), index) == columns.end()) { - return ""; + for (const auto& name : index) { + if (std::find(columns.begin(), columns.end(), name) + == columns.end()) { + return {}; + } } return index; } + std::shared_ptr + pkey_codec() const override { + return m_parent->get_pkey_codec(); + } + std::uint32_t limit() const override { return m_parent->get_limit(); diff --git a/rust/perspective-server/cpp/perspective/src/cpp/join_engine.cpp b/rust/perspective-server/cpp/perspective/src/cpp/join_engine.cpp index 86b94ffca6..2bdfc98559 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/join_engine.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/join_engine.cpp @@ -299,7 +299,7 @@ JoinEngine::make_join_table( } t_schema merged_schema(merged_columns, merged_types); - auto join_table = Table::from_schema("", merged_schema); + auto join_table = Table::from_schema({}, merged_schema); return {join_table, ""}; } diff --git a/rust/perspective-server/cpp/perspective/src/cpp/json_loader.cpp b/rust/perspective-server/cpp/perspective/src/cpp/json_loader.cpp index 7b95436c9a..aca6f8eb01 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/json_loader.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/json_loader.cpp @@ -700,12 +700,13 @@ void JsonLoader::init( std::string_view data, t_json_format format, - const std::string& index, + const std::vector& index, const t_schema* existing, t_list_flatten mode ) { m_format = format; m_mode = mode; + m_index = index; if (format == JSON_FORMAT_NDJSON) { m_stream = rapidjson::StringStream(data.data()); m_document.ParseStream(m_stream); @@ -749,6 +750,15 @@ JsonLoader::init( m_names = existing->columns(); m_types = existing->types(); m_is_implicit = index.empty(); + if (index.size() > 1) { + m_codec.emplace(t_pkey_codec::from_schema(*existing, index)); + t_schema scratch_schema(index, m_codec->dtypes()); + m_index_scratch = std::make_unique(scratch_schema); + m_index_scratch->init(); + m_index_scratch->extend(1); + m_index_seen.assign(index.size(), false); + } + return; } @@ -777,15 +787,17 @@ JsonLoader::init( } std::optional -JsonLoader::repeated_index(const std::string& index) const { +JsonLoader::repeated_index(const std::vector& index) const { if (!m_expands) { return std::nullopt; } // An index this payload does not carry cannot be established as // per-element, so it is treated as repeated. - if (!index.empty() && m_per_element.count(index) == 0) { - return index; + for (const auto& name : index) { + if (m_per_element.count(name) == 0) { + return name; + } } const std::string implicit{"__INDEX__"}; @@ -802,7 +814,7 @@ JsonLoader::repeated_index(const std::string& index) const { std::uint32_t JsonLoader::fill_table( t_data_table& tbl, - const std::string& index, + const std::vector& index, std::uint32_t offset, bool is_update ) { @@ -830,7 +842,7 @@ JsonLoader::fill_table( static void infer_record( const rapidjson::Value& record, - const std::string& index, + const std::vector& index, std::set& seen, std::set& known, std::vector& names, @@ -863,7 +875,7 @@ infer_record( for (const auto& col : record.GetObj()) { path.assign(col.name.GetString()); for_each_leaf(path, col.value, mode, [&](const auto& name, const auto& leaf, bool) { - if (name == index) { + if (std::find(index.begin(), index.end(), name) != index.end()) { is_implicit = false; } @@ -930,7 +942,7 @@ JsonLoader::resolve_column( } void -JsonLoader::infer_rows(const std::string& index) { +JsonLoader::infer_rows(const std::vector& index) { std::set seen; std::set known; std::set literal; @@ -962,7 +974,7 @@ JsonLoader::infer_rows(const std::string& index) { } void -JsonLoader::infer_ndjson(const std::string& index) { +JsonLoader::infer_ndjson(const std::vector& index) { // Only the first record is available without consuming the stream; later // records grow the schema during the fill instead. std::set seen; @@ -988,7 +1000,7 @@ JsonLoader::infer_ndjson(const std::string& index) { } void -JsonLoader::infer_cols(const std::string& index) { +JsonLoader::infer_cols(const std::vector& index) { // https://github.com/Tencent/rapidjson/issues/1994 for (const auto& it : m_document.GetObj()) { if (!it.value.IsArray()) { @@ -999,7 +1011,8 @@ JsonLoader::infer_cols(const std::string& index) { PSP_COMPLAIN_AND_ABORT("Can't create table from empty columns") } - if (it.name.GetString() == index) { + if (std::find(index.begin(), index.end(), it.name.GetString()) + != index.end()) { m_is_implicit = false; } @@ -1094,10 +1107,146 @@ fill_cell( fill_column_json(tbl.get_column(name), ii, cell, is_update); } +bool +JsonLoader::is_single_index(std::string_view name) const { + return m_index.size() == 1 && m_index[0] == name; +} + +void +JsonLoader::begin_row() { + m_row_has_pkey = false; + std::fill(m_index_seen.begin(), m_index_seen.end(), false); +} + +void +JsonLoader::note_column(std::string_view name) { + if (!m_codec) { + return; + } + + for (std::size_t i = 0; i < m_index.size(); ++i) { + if (m_index[i] == name) { + m_index_seen[i] = true; + return; + } + } +} + +void +JsonLoader::fill_pkey_cell( + t_data_table& tbl, + const std::shared_ptr& pkey_col, + t_uindex ii, + const rapidjson::Value& cell, + bool is_update +) { + if (!m_codec) { + fill_cell(tbl, pkey_col, "psp_pkey", ii, cell, is_update); + return; + } + + if (!cell.IsArray() || cell.Size() != m_index.size()) { + std::stringstream ss; + ss << "`__INDEX__` for a composite index must be an array of " + << m_index.size() << " values at index " << ii << std::endl; + PSP_COMPLAIN_AND_ABORT(ss.str()); + } + + std::vector key; + key.reserve(m_index.size()); + for (std::size_t i = 0; i < m_index.size(); ++i) { + auto col = m_index_scratch->get_column(m_index[i]); + auto promote = fill_column_json(col, 0, cell[i], true); + if (promote) { + std::stringstream ss; + ss << "Cannot append value of type " << dtype_to_str(*promote) + << " to column \"" << m_index[i] << "\" of type " + << dtype_to_str(col->get_dtype()) << " at index " << ii + << std::endl; + PSP_COMPLAIN_AND_ABORT(ss.str()); + } + + key.push_back(col->get_scalar(0)); + } + + std::string buf; + m_codec->encode(key, buf); + pkey_col->set_nth(ii, buf.c_str()); + m_row_has_pkey = true; +} + +void +JsonLoader::finish_row( + t_data_table& tbl, const std::shared_ptr& pkey_col, t_uindex ii +) { + if (!m_codec || m_row_has_pkey) { + return; + } + + for (std::size_t i = 0; i < m_index.size(); ++i) { + if (!m_index_seen[i]) { + std::stringstream ss; + ss << "Update is missing index column `" << m_index[i] << "`\n"; + PSP_COMPLAIN_AND_ABORT(ss.str()); + } + } + + if (m_index_cols.empty()) { + for (const auto& name : m_index) { + m_index_cols.push_back(tbl.get_column(name)); + } + } + + std::vector cols; + cols.reserve(m_index_cols.size()); + for (const auto& col : m_index_cols) { + cols.push_back(col.get()); + } + + std::string buf; + m_codec->encode(cols, ii, buf); + pkey_col->set_nth(ii, buf.c_str()); +} + +void +JsonLoader::finish_cols( + t_data_table& tbl, + const std::shared_ptr& pkey_col, + t_uindex nrows, + bool has_index_member, + const std::vector& col_names +) { + if (!m_codec || has_index_member) { + return; + } + + for (const auto& name : m_index) { + if (std::find(col_names.begin(), col_names.end(), name) + == col_names.end()) { + std::stringstream ss; + ss << "Update is missing index column `" << name << "`\n"; + PSP_COMPLAIN_AND_ABORT(ss.str()); + } + } + + std::vector> owned; + std::vector cols; + for (const auto& name : m_index) { + owned.push_back(tbl.get_column(name)); + cols.push_back(owned.back().get()); + } + + std::string buf; + for (t_uindex ridx = 0; ridx < nrows; ++ridx) { + m_codec->encode(cols, ridx, buf); + pkey_col->set_nth(ridx, buf.c_str()); + } +} + std::uint32_t JsonLoader::fill_rows( t_data_table& tbl, - const std::string& index, + const std::vector& index, std::uint32_t offset, bool is_update ) { @@ -1112,6 +1261,7 @@ JsonLoader::fill_rows( t_uindex ii = 0; for (const auto& row : m_document.GetArray()) { bool nested = false; + begin_row(); if (m_is_implicit && is_update) { psp_pkey_col->set_nth(ii, (ii + offset)); } @@ -1124,10 +1274,7 @@ JsonLoader::fill_rows( const auto name = key_name(it.name); if (is_update && name == "__INDEX__") { - fill_cell( - tbl, psp_pkey_col, "psp_pkey", ii, it.value, is_update - ); - + fill_pkey_cell(tbl, psp_pkey_col, ii, it.value, is_update); continue; } @@ -1137,7 +1284,8 @@ JsonLoader::fill_rows( } fill_cell(tbl, col, name, ii, it.value, is_update); - if (!m_is_implicit && index == name) { + note_column(name); + if (is_single_index(name)) { fill_column_json(psp_pkey_col, ii, it.value, is_update); if (psp_okey_col) { fill_column_json(psp_okey_col, ii, it.value, is_update); @@ -1151,6 +1299,7 @@ JsonLoader::fill_rows( psp_okey_col->set_nth(ii, ii); } + finish_row(tbl, psp_pkey_col, ii); ii++; continue; } @@ -1168,6 +1317,7 @@ JsonLoader::fill_rows( } for (t_uindex k = 0; k < width; ++k) { + begin_row(); if (m_is_implicit && is_update) { psp_pkey_col->set_nth(ii, (ii + offset)); } @@ -1180,10 +1330,7 @@ JsonLoader::fill_rows( const auto sub = child_slot(k, child_width, stride, m_mode); if (is_update && std::string_view{it.name.GetString()} == "__INDEX__") { - fill_cell( - tbl, psp_pkey_col, "psp_pkey", ii, it.value, is_update - ); - + fill_pkey_cell(tbl, psp_pkey_col, ii, it.value, is_update); continue; } @@ -1201,7 +1348,8 @@ JsonLoader::fill_rows( } fill_cell(tbl, col, name, ii, v, is_update); - if (!m_is_implicit && index == name) { + note_column(name); + if (is_single_index(name)) { fill_column_json(psp_pkey_col, ii, v, is_update); if (psp_okey_col) { fill_column_json(psp_okey_col, ii, v, is_update); @@ -1216,6 +1364,7 @@ JsonLoader::fill_rows( psp_okey_col->set_nth(ii, ii); } + finish_row(tbl, psp_pkey_col, ii); ii++; } } @@ -1226,7 +1375,7 @@ JsonLoader::fill_rows( std::uint32_t JsonLoader::fill_cols( t_data_table& tbl, - const std::string& index, + const std::vector& index, std::uint32_t offset, bool is_update ) { @@ -1260,6 +1409,8 @@ JsonLoader::fill_cols( const auto psp_okey_col = is_update ? nullptr : tbl.get_column("psp_okey"); const bool implicit_pkey_from_offset = m_is_implicit && is_update && !m_document.GetObj().HasMember("__INDEX__"); + const bool has_index_member = + is_update && m_document.GetObj().HasMember("__INDEX__"); tbl.extend(nrows); bool expands = false; @@ -1279,7 +1430,7 @@ JsonLoader::fill_cols( } if (is_pkey_column[c]) { - fill_cell(tbl, psp_pkey_col, "psp_pkey", r, cell, is_update); + fill_pkey_cell(tbl, psp_pkey_col, r, cell, is_update); continue; } @@ -1289,7 +1440,7 @@ JsonLoader::fill_cols( } fill_cell(tbl, col, col_names[c], r, cell, is_update); - if (!m_is_implicit && index == col_names[c]) { + if (is_single_index(col_names[c])) { fill_column_json(psp_pkey_col, r, cell, is_update); if (psp_okey_col) { fill_column_json(psp_okey_col, r, cell, is_update); @@ -1306,6 +1457,7 @@ JsonLoader::fill_cols( } } + finish_cols(tbl, psp_pkey_col, nrows, has_index_member, col_names); return nrows; } @@ -1370,10 +1522,7 @@ JsonLoader::fill_cols( for (t_uindex k = 0; k < row_width[r]; ++k) { const auto ii = row_start[r] + k; if (is_pkey_column[c]) { - fill_cell( - tbl, psp_pkey_col, "psp_pkey", ii, cell, is_update - ); - + fill_pkey_cell(tbl, psp_pkey_col, ii, cell, is_update); continue; } @@ -1403,7 +1552,8 @@ JsonLoader::fill_cols( } fill_cell(tbl, col, name, ii, v, is_update); - if (!m_is_implicit && index == name) { + note_column(name); + if (is_single_index(name)) { fill_column_json(psp_pkey_col, ii, v, is_update); if (psp_okey_col) { fill_column_json( @@ -1424,13 +1574,14 @@ JsonLoader::fill_cols( } } + finish_cols(tbl, psp_pkey_col, size, has_index_member, col_names); return size; } std::uint32_t JsonLoader::fill_ndjson( t_data_table& tbl, - const std::string& index, + const std::vector& index, std::uint32_t offset, bool is_update ) { @@ -1451,6 +1602,7 @@ JsonLoader::fill_ndjson( if (!nested) { tbl.extend(ii + 1); + begin_row(); if (m_is_implicit && is_update) { psp_pkey_col->set_nth(ii, (ii + offset)); } @@ -1458,10 +1610,7 @@ JsonLoader::fill_ndjson( for (const auto& it : m_document.GetObj()) { const auto name = key_name(it.name); if (is_update && name == "__INDEX__") { - fill_cell( - tbl, psp_pkey_col, "psp_pkey", ii, it.value, is_update - ); - + fill_pkey_cell(tbl, psp_pkey_col, ii, it.value, is_update); continue; } @@ -1471,7 +1620,8 @@ JsonLoader::fill_ndjson( } fill_cell(tbl, col, name, ii, it.value, is_update); - if (!m_is_implicit && index == name) { + note_column(name); + if (is_single_index(name)) { fill_column_json(psp_pkey_col, ii, it.value, is_update); if (psp_okey_col) { fill_column_json(psp_okey_col, ii, it.value, is_update); @@ -1484,6 +1634,7 @@ JsonLoader::fill_ndjson( psp_okey_col->set_nth(ii, ii); } + finish_row(tbl, psp_pkey_col, ii); ii++; m_document.ParseStream(m_stream); if (m_document.HasParseError()) { @@ -1503,6 +1654,7 @@ JsonLoader::fill_ndjson( } for (t_uindex k = 0; k < width; ++k) { + begin_row(); if (m_is_implicit && is_update) { psp_pkey_col->set_nth(ii, (ii + offset)); } @@ -1515,10 +1667,7 @@ JsonLoader::fill_ndjson( const auto sub = child_slot(k, child_width, stride, m_mode); if (is_update && std::string_view{it.name.GetString()} == "__INDEX__") { - fill_cell( - tbl, psp_pkey_col, "psp_pkey", ii, it.value, is_update - ); - + fill_pkey_cell(tbl, psp_pkey_col, ii, it.value, is_update); continue; } @@ -1536,7 +1685,8 @@ JsonLoader::fill_ndjson( } fill_cell(tbl, col, name, ii, v, is_update); - if (!m_is_implicit && index == name) { + note_column(name); + if (is_single_index(name)) { fill_column_json(psp_pkey_col, ii, v, is_update); if (psp_okey_col) { fill_column_json(psp_okey_col, ii, v, is_update); @@ -1551,6 +1701,7 @@ JsonLoader::fill_ndjson( psp_okey_col->set_nth(ii, ii); } + finish_row(tbl, psp_pkey_col, ii); ii++; } diff --git a/rust/perspective-server/cpp/perspective/src/cpp/pkey_codec.cpp b/rust/perspective-server/cpp/perspective/src/cpp/pkey_codec.cpp new file mode 100644 index 0000000000..ac2932223c --- /dev/null +++ b/rust/perspective-server/cpp/perspective/src/cpp/pkey_codec.cpp @@ -0,0 +1,516 @@ +// ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ +// ┃ ██████ ██████ ██████ █ █ █ █ █ █▄ ▀███ █ ┃ +// ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█ ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄ ▀█ █ ▀▀▀▀▀ ┃ +// ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄ █ ▄▄▄▄▄ ┃ +// ┃ █ ██████ █ ▀█▄ █ ██████ █ ███▌▐███ ███████▄ █ ┃ +// ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫ +// ┃ 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). ┃ +// ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛ + + +#include +#include +#include +#include +#include +#include +#include + +namespace perspective { + +namespace { + constexpr unsigned char NULL_MARK = 0x00; + constexpr unsigned char VALUE_MARK = 0x01; + constexpr unsigned char ESCAPE = 0x01; + + template + void + append_be(std::string& raw, T v) { + for (int shift = (int)(sizeof(T) * 8) - 8; shift >= 0; shift -= 8) { + raw.push_back(static_cast((v >> shift) & 0xFF)); + } + } + + template + T + read_be(const unsigned char*& p) { + T v = 0; + for (std::size_t i = 0; i < sizeof(T); ++i) { + v = (v << 8) | static_cast(*p++); + } + + return v; + } + + template + U + flip_sign(S v) { + U u; + std::memcpy(&u, &v, sizeof(U)); + return u ^ (static_cast(1) << (sizeof(U) * 8 - 1)); + } + + template + S + unflip_sign(U u) { + u ^= (static_cast(1) << (sizeof(U) * 8 - 1)); + S s; + std::memcpy(&s, &u, sizeof(S)); + return s; + } + + std::uint64_t + float_to_ordered(double v) { + if (std::isnan(v)) { + v = std::numeric_limits::quiet_NaN(); + } else if (v == 0.0) { + v = 0.0; + } + + std::uint64_t bits; + std::memcpy(&bits, &v, sizeof(bits)); + if ((bits >> 63) != 0) { + return ~bits; + } + + return bits ^ (static_cast(1) << 63); + } + + double + ordered_to_float(std::uint64_t bits) { + if ((bits >> 63) != 0) { + bits ^= (static_cast(1) << 63); + } else { + bits = ~bits; + } + + double v; + std::memcpy(&v, &bits, sizeof(v)); + return v; + } + + std::uint32_t + float32_to_ordered(float v) { + if (std::isnan(v)) { + v = std::numeric_limits::quiet_NaN(); + } else if (v == 0.0F) { + v = 0.0F; + } + + std::uint32_t bits; + std::memcpy(&bits, &v, sizeof(bits)); + if ((bits >> 31) != 0) { + return ~bits; + } + + return bits ^ (static_cast(1) << 31); + } + + float + ordered_to_float32(std::uint32_t bits) { + if ((bits >> 31) != 0) { + bits ^= (static_cast(1) << 31); + } else { + bits = ~bits; + } + + float v; + std::memcpy(&v, &bits, sizeof(v)); + return v; + } + + void + escape_nul_free(const std::string& raw, std::string& out) { + out.clear(); + out.reserve(raw.size() + 8); + for (const char c : raw) { + const auto b = static_cast(c); + if (b == 0x00) { + out.push_back(static_cast(ESCAPE)); + out.push_back(static_cast(0x01)); + } else if (b == ESCAPE) { + out.push_back(static_cast(ESCAPE)); + out.push_back(static_cast(0x02)); + } else { + out.push_back(c); + } + } + } + + std::string + unescape_nul_free(const char* bytes) { + std::string raw; + for (const char* p = bytes; *p != '\0'; ++p) { + const auto b = static_cast(*p); + if (b == ESCAPE) { + ++p; + raw.push_back(static_cast(*p == 0x01 ? 0x00 : ESCAPE)); + } else { + raw.push_back(*p); + } + } + + return raw; + } +} // namespace + +t_pkey_codec::t_pkey_codec(std::vector dtypes) : + m_dtypes(std::move(dtypes)) { + for (const auto dtype : m_dtypes) { + if (!is_supported(dtype)) { + std::stringstream ss; + ss << "Unsupported type `" << get_dtype_descr(dtype) + << "` for `index`.\n"; + PSP_COMPLAIN_AND_ABORT(ss.str()); + } + } +} + +t_pkey_codec +t_pkey_codec::from_schema( + const t_schema& schema, const std::vector& index +) { + std::vector dtypes; + dtypes.reserve(index.size()); + for (const auto& name : index) { + if (!schema.has_column(name)) { + std::stringstream ss; + ss << "Specified index `" << name + << "` does not appear in the Table.\n"; + PSP_COMPLAIN_AND_ABORT(ss.str()); + } + + dtypes.push_back(schema.get_dtype(name)); + } + + return t_pkey_codec(std::move(dtypes)); +} + +bool +t_pkey_codec::is_supported(t_dtype dtype) { + switch (dtype) { + case DTYPE_INT64: + case DTYPE_INT32: + case DTYPE_INT16: + case DTYPE_INT8: + case DTYPE_UINT64: + case DTYPE_UINT32: + case DTYPE_UINT16: + case DTYPE_UINT8: + case DTYPE_FLOAT64: + case DTYPE_FLOAT32: + case DTYPE_BOOL: + case DTYPE_DATE: + case DTYPE_TIME: + case DTYPE_STR: + return true; + default: + return false; + } +} + +const std::vector& +t_pkey_codec::dtypes() const { + return m_dtypes; +} + +std::size_t +t_pkey_codec::arity() const { + return m_dtypes.size(); +} + +t_tscalar +t_pkey_codec::coerce(const t_tscalar& value, t_dtype dtype) const { + if (!value.is_valid()) { + return mknull(dtype); + } + + if (value.get_dtype() == dtype) { + return value; + } + + t_tscalar rval; + rval.clear(); + switch (dtype) { + case DTYPE_INT64: + rval.set(static_cast(value.to_int64())); + break; + case DTYPE_INT32: + rval.set(static_cast(value.to_int64())); + break; + case DTYPE_INT16: + rval.set(static_cast(value.to_int64())); + break; + case DTYPE_INT8: + rval.set(static_cast(value.to_int64())); + break; + case DTYPE_UINT64: + rval.set(static_cast(value.to_int64())); + break; + case DTYPE_UINT32: + rval.set(static_cast(value.to_int64())); + break; + case DTYPE_UINT16: + rval.set(static_cast(value.to_int64())); + break; + case DTYPE_UINT8: + rval.set(static_cast(value.to_int64())); + break; + case DTYPE_FLOAT64: + rval.set(value.to_double()); + break; + case DTYPE_FLOAT32: + rval.set(static_cast(value.to_double())); + break; + case DTYPE_BOOL: + rval.set(value.as_bool()); + break; + case DTYPE_DATE: + rval.set(t_date(static_cast(value.to_int64()))); + break; + case DTYPE_TIME: + rval.set(t_time(value.to_int64())); + break; + case DTYPE_STR: { + std::stringstream ss; + ss << "Cannot use a value of type `" + << get_dtype_descr(value.get_dtype()) + << "` as a string index column.\n"; + PSP_COMPLAIN_AND_ABORT(ss.str()); + } break; + default: + PSP_COMPLAIN_AND_ABORT("Unsupported type for `index`."); + } + + return rval; +} + +void +t_pkey_codec::encode_field( + const t_tscalar& value, t_dtype dtype, std::string& raw +) const { + if (!value.is_valid()) { + raw.push_back(static_cast(NULL_MARK)); + return; + } + + raw.push_back(static_cast(VALUE_MARK)); + const t_tscalar v = coerce(value, dtype); + switch (dtype) { + case DTYPE_INT64: + append_be(raw, flip_sign(v.m_data.m_int64)); + break; + case DTYPE_INT32: + append_be(raw, flip_sign(v.m_data.m_int32)); + break; + case DTYPE_INT16: + append_be(raw, flip_sign(v.m_data.m_int16)); + break; + case DTYPE_INT8: + append_be(raw, flip_sign(v.m_data.m_int8)); + break; + case DTYPE_UINT64: + append_be(raw, v.m_data.m_uint64); + break; + case DTYPE_UINT32: + case DTYPE_DATE: + append_be(raw, v.m_data.m_uint32); + break; + case DTYPE_UINT16: + append_be(raw, v.m_data.m_uint16); + break; + case DTYPE_UINT8: + append_be(raw, v.m_data.m_uint8); + break; + case DTYPE_TIME: + append_be(raw, flip_sign(v.m_data.m_int64)); + break; + case DTYPE_FLOAT64: + append_be(raw, float_to_ordered(v.m_data.m_float64)); + break; + case DTYPE_FLOAT32: + append_be(raw, float32_to_ordered(v.m_data.m_float32)); + break; + case DTYPE_BOOL: + raw.push_back(static_cast(v.m_data.m_bool ? 1 : 0)); + break; + case DTYPE_STR: { + const char* s = v.get_char_ptr(); + if (s != nullptr) { + raw.append(s); + } + + raw.push_back('\0'); + } break; + default: + PSP_COMPLAIN_AND_ABORT("Unsupported type for `index`."); + } +} + +void +t_pkey_codec::encode( + const std::vector& cols, t_uindex ridx, std::string& out +) const { + std::string raw; + for (std::size_t i = 0; i < m_dtypes.size(); ++i) { + encode_field(cols[i]->get_scalar(ridx), m_dtypes[i], raw); + } + + escape_nul_free(raw, out); +} + +void +t_pkey_codec::encode(const std::vector& key, std::string& out) + const { + if (key.size() != m_dtypes.size()) { + std::stringstream ss; + ss << "Composite index expects " << m_dtypes.size() + << " values but received " << key.size() << ".\n"; + PSP_COMPLAIN_AND_ABORT(ss.str()); + } + + std::string raw; + for (std::size_t i = 0; i < m_dtypes.size(); ++i) { + encode_field(key[i], m_dtypes[i], raw); + } + + escape_nul_free(raw, out); +} + +std::vector +t_pkey_codec::decode(const char* bytes, std::vector& storage) + const { + std::vector rval; + rval.reserve(m_dtypes.size()); + storage.assign(m_dtypes.size(), std::string()); + if (bytes == nullptr) { + for (const auto dtype : m_dtypes) { + rval.push_back(mknull(dtype)); + } + + return rval; + } + + const std::string raw = unescape_nul_free(bytes); + const auto* p = reinterpret_cast(raw.data()); + const auto* end = p + raw.size(); + for (std::size_t i = 0; i < m_dtypes.size(); ++i) { + const auto dtype = m_dtypes[i]; + if (p >= end || *p == NULL_MARK) { + if (p < end) { + ++p; + } + + rval.push_back(mknull(dtype)); + continue; + } + + ++p; + t_tscalar v; + v.clear(); + switch (dtype) { + case DTYPE_INT64: + v.set(unflip_sign(read_be(p))); + break; + case DTYPE_INT32: + v.set(unflip_sign(read_be(p))); + break; + case DTYPE_INT16: + v.set(unflip_sign(read_be(p))); + break; + case DTYPE_INT8: + v.set(unflip_sign(read_be(p))); + break; + case DTYPE_UINT64: + v.set(read_be(p)); + break; + case DTYPE_UINT32: + v.set(read_be(p)); + break; + case DTYPE_DATE: + v.set(t_date(read_be(p))); + break; + case DTYPE_UINT16: + v.set(read_be(p)); + break; + case DTYPE_UINT8: + v.set(read_be(p)); + break; + case DTYPE_TIME: + v.set(t_time(unflip_sign(read_be(p) + ))); + break; + case DTYPE_FLOAT64: + v.set(ordered_to_float(read_be(p))); + break; + case DTYPE_FLOAT32: + v.set(ordered_to_float32(read_be(p))); + break; + case DTYPE_BOOL: + v.set(*p++ != 0); + break; + case DTYPE_STR: { + const auto* start = reinterpret_cast(p); + const auto len = std::strlen(start); + storage[i].assign(start, len); + p += len + 1; + v.set(storage[i].c_str()); + } break; + default: + PSP_COMPLAIN_AND_ABORT("Unsupported type for `index`."); + } + + rval.push_back(v); + } + + return rval; +} + +void +fill_pkey_from_index( + t_data_table& tbl, + const std::vector& index, + const t_pkey_codec& codec, + bool is_update, + bool fill_okey +) { + std::vector> owned; + std::vector cols; + for (const auto& name : index) { + auto col = tbl.get_column_safe(name); + if (!col) { + std::stringstream ss; + if (is_update) { + ss << "Update is missing index column `" << name << "`\n"; + } else { + ss << "Specified index `" << name + << "` does not exist in dataset.\n"; + } + + PSP_COMPLAIN_AND_ABORT(ss.str()); + } + + owned.push_back(col); + cols.push_back(col.get()); + } + + auto pkey = tbl.add_column_sptr("psp_pkey", DTYPE_STR, true); + std::shared_ptr okey; + if (fill_okey) { + okey = tbl.add_column_sptr("psp_okey", DTYPE_STR, true); + } + + std::string key; + for (t_uindex ridx = 0; ridx < tbl.size(); ++ridx) { + codec.encode(cols, ridx, key); + pkey->set_nth(ridx, key.c_str()); + if (okey) { + okey->set_nth(ridx, key.c_str()); + } + } +} + +} // namespace perspective diff --git a/rust/perspective-server/cpp/perspective/src/cpp/server.cpp b/rust/perspective-server/cpp/perspective/src/cpp/server.cpp index 26d01625c6..d315d97e58 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/server.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/server.cpp @@ -2499,8 +2499,8 @@ ProtoServer::_handle_request(std::uint32_t client_id, Request&& req) { v->set_entity_id(name); const auto tbl = m_resources.get_table(name); - if (!tbl->get_index().empty()) { - v->set_index(tbl->get_index()); + for (const auto& name : tbl->get_index()) { + v->add_index(name); } if (tbl->get_limit() != std::numeric_limits::max()) { @@ -2538,7 +2538,7 @@ ProtoServer::_handle_request(std::uint32_t client_id, Request&& req) { break; } - std::string index; + std::vector index; std::uint32_t limit = std::numeric_limits::max(); bool has_limit = false; std::string make_error; @@ -2550,7 +2550,14 @@ ProtoServer::_handle_request(std::uint32_t client_id, Request&& req) { break; } case proto::MakeTableReq_MakeTableOptions::kMakeIndexTable: { - index = r.options().make_index_table(); + index.push_back(r.options().make_index_table()); + break; + } + case proto::MakeTableReq_MakeTableOptions:: + kMakeCompositeIndexTable: { + const auto& columns = + r.options().make_composite_index_table().columns(); + index.assign(columns.begin(), columns.end()); break; } case proto::MakeTableReq_MakeTableOptions:: @@ -3943,6 +3950,44 @@ ProtoServer::_poll() { return resp_envs; } +static std::shared_ptr +removed_keys_to_arrow( + const std::shared_ptr
& table, + const std::shared_ptr& removed +) { + const auto& index = table->get_index(); + const auto pkey = removed->get_const_column("psp_pkey"); + if (index.size() == 1) { + return apachearrow::column_to_arrow_ipc( + *pkey, index[0], removed->size() + ); + } + + const auto& codec = *table->get_pkey_codec(); + t_schema schema(index, codec.dtypes()); + t_data_table keys(schema); + keys.init(); + keys.extend(removed->size()); + std::vector> cols; + cols.reserve(index.size()); + for (const auto& name : index) { + cols.push_back(keys.get_column(name)); + } + + std::vector storage; + for (t_uindex ridx = 0; ridx < removed->size(); ++ridx) { + const auto key = pkey->get_scalar(ridx); + const auto values = codec.decode( + key.is_valid() ? key.get_char_ptr() : nullptr, storage + ); + for (std::size_t i = 0; i < values.size(); ++i) { + cols[i]->set_scalar(ridx, values[i]); + } + } + + return apachearrow::table_to_arrow_ipc(keys); +} + void ProtoServer::_notify_table( std::shared_ptr
& table, @@ -3966,11 +4011,7 @@ ProtoServer::_notify_table( if (has_removes) { auto remove_subs = m_resources.get_view_on_remove_sub(view_id); if (!remove_subs.empty() && !removed_indices) { - removed_indices = apachearrow::column_to_arrow_ipc( - *removed->get_const_column("psp_pkey"), - table->get_index(), - removed->size() - ); + removed_indices = removed_keys_to_arrow(table, removed); } for (auto& subscription : remove_subs) { @@ -4072,7 +4113,11 @@ ProtoServer::make_derived_table( } table = Table::make_derived( - schema, source->pkey_dtype(), source->index(), source->limit() + schema, + source->pkey_dtype(), + source->index(), + source->limit(), + source->pkey_codec() ); #ifdef PSP_PARALLEL_FOR diff --git a/rust/perspective-server/cpp/perspective/src/cpp/table.cpp b/rust/perspective-server/cpp/perspective/src/cpp/table.cpp index 67a7b1401f..0773dfbdf6 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/table.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/table.cpp @@ -17,6 +17,7 @@ #include "perspective/data_table.h" #include "perspective/raw_types.h" #include "perspective/json_loader.h" +#include "perspective/pkey_codec.h" #include "perspective/schema.h" #include "rapidjson/document.h" #include @@ -24,6 +25,7 @@ #include #include #include +#include #include #include #include @@ -40,7 +42,7 @@ Table::Table( std::vector column_names, std::vector data_types, std::uint32_t limit, - std::string index, + std::vector index, t_backing_store backing_store, apachearrow::t_list_flatten list_flatten ) : @@ -57,6 +59,124 @@ Table::Table( m_list_flatten(list_flatten) { validate_columns(m_column_names); + if (is_composite_index()) { + t_schema schema(m_column_names, m_data_types); + m_pkey_codec = std::make_shared( + t_pkey_codec::from_schema(schema, m_index) + ); + } +} + +static t_schema +index_schema( + const t_schema& output_schema, const std::vector& index +) { + std::vector types; + types.reserve(index.size()); + for (const auto& name : index) { + types.push_back(output_schema.get_dtype(name)); + } + + return {index, types}; +} + +static void +fill_index_cell( + const std::shared_ptr& col, + const std::string& name, + t_uindex ii, + const rapidjson::Value& cell +) { + auto promote = json::fill_column_json(col, ii, cell, true); + if (promote) { + std::stringstream ss; + ss << "Cannot append value of type " << dtype_to_str(*promote) + << " to column \"" << name << "\" of type " + << dtype_to_str(col->get_dtype()) << " at index " << ii + << std::endl; + PSP_COMPLAIN_AND_ABORT(ss.str()); + } +} + +static void +fill_composite_remove_table( + t_data_table& data_table, + const std::vector& index, + const rapidjson::Document& document +) { + std::vector> cols; + cols.reserve(index.size()); + for (const auto& name : index) { + cols.push_back(data_table.get_column(name)); + } + + if (document.IsArray()) { + data_table.extend(document.Size()); + t_uindex ii = 0; + for (const auto& row : document.GetArray()) { + if (!row.IsArray() || row.Size() != index.size()) { + std::stringstream ss; + ss << "Composite index expects an array of " << index.size() + << " values at index " << ii << std::endl; + PSP_COMPLAIN_AND_ABORT(ss.str()); + } + + for (std::size_t i = 0; i < index.size(); ++i) { + fill_index_cell(cols[i], index[i], ii, row[i]); + } + + ii++; + } + + return; + } + + if (!document.IsObject()) { + PSP_COMPLAIN_AND_ABORT("Cannot remove fish!\n") + } + + t_uindex nrows = 0; + for (std::size_t i = 0; i < index.size(); ++i) { + const auto& name = index[i]; + if (!document.HasMember(name.c_str()) + || !document[name.c_str()].IsArray()) { + std::stringstream ss; + ss << "Remove is missing index column `" << name << "`\n"; + PSP_COMPLAIN_AND_ABORT(ss.str()); + } + + const auto& cells = document[name.c_str()]; + if (i == 0) { + nrows = cells.Size(); + data_table.extend(nrows); + } else if (cells.Size() != nrows) { + std::stringstream ss; + ss << "Index column `" << name << "` has " << cells.Size() + << " values but expected " << nrows << std::endl; + PSP_COMPLAIN_AND_ABORT(ss.str()); + } + + t_uindex ii = 0; + for (const auto& cell : cells.GetArray()) { + fill_index_cell(cols[i], name, ii, cell); + ii++; + } + } +} + +static void +require_index_columns( + const std::vector& names, + const std::vector& index, + const char* verb +) { + for (const auto& name : index) { + if (std::find(names.begin(), names.end(), name) == names.end()) { + std::stringstream ss; + ss << verb << " is missing index column `" << name << "`\n"; + PSP_COMPLAIN_AND_ABORT(ss.str()); + } + } } void @@ -214,6 +334,7 @@ Table::make_gnode(const t_schema& in_schema) { auto gnode = std::make_shared(in_schema, out_schema, m_backing_store); gnode->init(); + gnode->get_table_sptr()->set_pkey_codec(m_pkey_codec); return gnode; } @@ -287,12 +408,35 @@ Table::get_data_types() const { return m_data_types; } -const std::string& +const std::vector& Table::get_index() const { PSP_VERBOSE_ASSERT(m_init, "touching uninited object"); return m_index; } +bool +Table::is_composite_index() const { + return m_index.size() > 1; +} + +t_dtype +Table::pkey_dtype(const t_schema& schema) const { + if (m_index.empty()) { + return DTYPE_INT32; + } + + if (m_index.size() == 1) { + return schema.get_dtype(m_index[0]); + } + + return DTYPE_STR; +} + +const std::shared_ptr& +Table::get_pkey_codec() const { + return m_pkey_codec; +} + std::uint32_t Table::get_offset() const { PSP_VERBOSE_ASSERT(m_init, "touching uninited object"); @@ -384,7 +528,15 @@ Table::update_csv(const std::string_view& data, std::uint32_t port_id) { t_data_table data_table(get_schema()); data_table.init(); data_table.extend(row_count); + if (is_composite_index()) { + require_index_columns(arrow_loader.names(), m_index, "Update"); + } + arrow_loader.fill_table(data_table, get_schema(), m_index, m_offset, true); + if (is_composite_index()) { + fill_pkey_from_index(data_table, m_index, *m_pkey_codec, true, true); + } + process_op_column(data_table, t_op::OP_INSERT); calculate_offset(row_count); m_pool->send(get_gnode()->get_id(), port_id, data_table); @@ -392,7 +544,7 @@ Table::update_csv(const std::string_view& data, std::uint32_t port_id) { std::shared_ptr
Table::from_csv( - const std::string& index, + const std::vector& index, std::string&& data, std::uint32_t limit, t_backing_store backing_store, @@ -435,6 +587,16 @@ Table::from_csv( loader.fill_table(*data_table, input_schema, index, 0, false); } + if (index.size() > 1) { + fill_pkey_from_index( + *data_table, + index, + t_pkey_codec::from_schema(data_table->get_schema(), index), + false, + true + ); + } + auto pool = std::make_shared(); pool->init(); auto tbl = std::make_shared
( @@ -477,9 +639,19 @@ Table::remove_rows(const std::string_view& data) { } const t_schema& output_schema = get_gnode()->get_output_schema(); + if (is_composite_index()) { + t_data_table data_table(index_schema(output_schema, m_index)); + data_table.init(); + fill_composite_remove_table(data_table, m_index, document); + fill_pkey_from_index(data_table, m_index, *m_pkey_codec, true, true); + process_op_column(data_table, OP_DELETE); + m_pool->send(get_gnode()->get_id(), 0, data_table); + return; + } - std::vector column_names{m_index}; - std::vector data_types{output_schema.get_dtype(m_index)}; + const std::string& index = m_index[0]; + std::vector column_names{index}; + std::vector data_types{output_schema.get_dtype(index)}; t_schema schema(column_names, data_types); @@ -488,33 +660,20 @@ Table::remove_rows(const std::string_view& data) { data_table.init(); data_table.extend(document.Size()); - data_table.add_column("psp_pkey", schema.get_dtype(m_index), true); + data_table.add_column("psp_pkey", schema.get_dtype(index), true); const auto& psp_pkey_col = data_table.get_column("psp_pkey"); // 3.) Fill table t_uindex ii = 0; - auto col = data_table.get_column(m_index); + auto col = data_table.get_column(index); for (const auto& cell : document.GetArray()) { - auto promote = json::fill_column_json(col, ii, cell, true); - if (promote) { - std::stringstream ss; - ss << "Cannot append value of type " << dtype_to_str(*promote) - << " to column \"" << m_index << "\" of type " << dtype_to_str(col->get_dtype()) - << " at index " << ii - << std::endl; - PSP_COMPLAIN_AND_ABORT(ss.str()); - } - - // if (!is_implicit && m_index == col_name) { + fill_index_cell(col, index, ii, cell); json::fill_column_json(psp_pkey_col, ii, cell, true); - // } - ii++; } data_table.clone_column("psp_pkey", "psp_okey"); - // calculate_offset(data_table.size()); process_op_column(data_table, OP_DELETE); m_pool->send(get_gnode()->get_id(), 0, data_table); } @@ -524,18 +683,29 @@ Table::remove_cols(const std::string_view& data) { // 1.) Infer schema rapidjson::Document document; document.Parse(data.data()); - if (!document.IsArray()) { - PSP_COMPLAIN_AND_ABORT("Cannot remove fish!\n") - } - if (m_index.empty()) { PSP_COMPLAIN_AND_ABORT("Cannot remove from unindexed Table\n") } const t_schema& output_schema = get_gnode()->get_output_schema(); + if (is_composite_index()) { + t_data_table data_table(index_schema(output_schema, m_index)); + data_table.init(); + fill_composite_remove_table(data_table, m_index, document); + fill_pkey_from_index(data_table, m_index, *m_pkey_codec, true, true); + calculate_offset(data_table.size()); + process_op_column(data_table, OP_DELETE); + m_pool->send(get_gnode()->get_id(), 0, data_table); + return; + } - std::vector column_names{m_index}; - std::vector data_types{output_schema.get_dtype(m_index)}; + if (!document.IsArray()) { + PSP_COMPLAIN_AND_ABORT("Cannot remove fish!\n") + } + + const std::string& index = m_index[0]; + std::vector column_names{index}; + std::vector data_types{output_schema.get_dtype(index)}; t_schema schema(column_names, data_types); @@ -544,31 +714,19 @@ Table::remove_cols(const std::string_view& data) { data_table.init(); data_table.extend(document.Size()); - data_table.add_column("psp_pkey", schema.get_dtype(m_index), true); - data_table.add_column("psp_okey", schema.get_dtype(m_index), true); + data_table.add_column("psp_pkey", schema.get_dtype(index), true); + data_table.add_column("psp_okey", schema.get_dtype(index), true); const auto& psp_pkey_col = data_table.get_column("psp_pkey"); const auto& psp_okey_col = data_table.get_column("psp_okey"); // 3.) Fill table t_uindex ii = 0; - auto col = data_table.get_column(m_index); + auto col = data_table.get_column(index); for (const auto& cell : document.GetArray()) { - auto promote = json::fill_column_json(col, ii, cell, true); - if (promote) { - std::stringstream ss; - ss << "Cannot append value of type " << dtype_to_str(*promote) - << " to column \"" << m_index << "\" of type " << dtype_to_str(col->get_dtype()) - << " at index " << ii - << std::endl; - PSP_COMPLAIN_AND_ABORT(ss.str()); - } - - // if (!is_implicit && m_index == col_name) { + fill_index_cell(col, index, ii, cell); json::fill_column_json(psp_pkey_col, ii, cell, true); json::fill_column_json(psp_okey_col, ii, cell, true); - // } - ii++; } @@ -592,19 +750,25 @@ Table::remove_arrow(const std::string_view& data) { ); const auto names = arrow_loader.names(); - if (std::find(names.begin(), names.end(), m_index) == names.end()) { - std::stringstream ss; - ss << "Cannot remove: Arrow is missing index column `" << m_index - << "`\n"; - PSP_COMPLAIN_AND_ABORT(ss.str()); + for (const auto& name : m_index) { + if (std::find(names.begin(), names.end(), name) == names.end()) { + std::stringstream ss; + ss << "Cannot remove: Arrow is missing index column `" << name + << "`\n"; + PSP_COMPLAIN_AND_ABORT(ss.str()); + } } const t_schema& output_schema = get_gnode()->get_output_schema(); - t_schema schema({m_index}, {output_schema.get_dtype(m_index)}); + t_schema schema = index_schema(output_schema, m_index); t_data_table data_table(schema); data_table.init(); data_table.extend(arrow_loader.row_count()); arrow_loader.fill_table(data_table, schema, m_index, m_offset, true); + if (is_composite_index()) { + fill_pkey_from_index(data_table, m_index, *m_pkey_codec, true, true); + } + process_op_column(data_table, OP_DELETE); m_pool->send(get_gnode()->get_id(), 0, data_table); } @@ -612,7 +776,7 @@ Table::remove_arrow(const std::string_view& data) { std::shared_ptr
Table::from_json_loader( json::JsonLoader& loader, - const std::string& index, + const std::vector& index, std::string&& data, std::uint32_t limit, t_backing_store backing_store, @@ -628,13 +792,24 @@ Table::from_json_loader( t_schema schema(loader.names(), loader.types()); auto data_table = std::make_unique(schema); data_table->init(); - const auto pkey_dtype = - loader.is_implicit() ? DTYPE_INT32 : schema.get_dtype(index); + t_dtype pkey_dtype = DTYPE_INT32; + if (!loader.is_implicit()) { + pkey_dtype = index.size() == 1 ? schema.get_dtype(index[0]) : DTYPE_STR; + } data_table->add_column("psp_pkey", pkey_dtype, true); data_table->add_column("psp_okey", pkey_dtype, true); const auto nrows = loader.fill_table(*data_table, index, 0, false); + if (!loader.is_implicit() && index.size() > 1) { + fill_pkey_from_index( + *data_table, + index, + t_pkey_codec::from_schema(data_table->get_schema(), index), + false, + true + ); + } // `names`/`types` may have grown during the fill -- an ndjson record can // introduce a column -- so the Table's column list is read back from the @@ -687,11 +862,7 @@ Table::update_json( t_data_table data_table(table_schema); data_table.init(); - data_table.add_column( - "psp_pkey", - m_index.empty() ? DTYPE_INT32 : table_schema.get_dtype(m_index), - true - ); + data_table.add_column("psp_pkey", pkey_dtype(table_schema), true); const auto size = loader.fill_table(data_table, m_index, m_offset, true); data_table.clone_column("psp_pkey", "psp_okey"); @@ -708,7 +879,7 @@ Table::update_cols(const std::string_view& data, std::uint32_t port_id) { std::shared_ptr
Table::from_cols( - const std::string& index, + const std::vector& index, std::string&& data, std::uint32_t limit, t_backing_store backing_store, @@ -736,7 +907,7 @@ Table::update_rows(const std::string_view& data, std::uint32_t port_id) { std::shared_ptr
Table::from_rows( - const std::string& index, + const std::vector& index, std::string&& data, std::uint32_t limit, t_backing_store backing_store, @@ -758,7 +929,7 @@ Table::update_ndjson(const std::string_view& data, std::uint32_t port_id) { std::shared_ptr
Table::from_ndjson( - const std::string& index, + const std::vector& index, std::string&& data, std::uint32_t limit, t_backing_store backing_store, @@ -775,7 +946,7 @@ Table::from_ndjson( std::shared_ptr
Table::from_schema( - const std::string& index, + const std::vector& index, const t_schema& schema, std::uint32_t limit, t_backing_store backing_store, @@ -787,20 +958,24 @@ Table::from_schema( t_data_table data_table(schema); data_table.init(); - // TODO check for implicit index; - if (index.empty()) { - data_table.add_column("psp_pkey", DTYPE_INT32, true); - data_table.add_column("psp_okey", DTYPE_INT32, true); - } else { - if (!schema.has_column(index)) { + for (const auto& name : index) { + if (!schema.has_column(name)) { std::stringstream ss; - ss << "Specified index `" << index + ss << "Specified index `" << name << "` does not appear in the Table." << '\n'; PSP_COMPLAIN_AND_ABORT(ss.str()); } + } - data_table.clone_column(index, "psp_pkey"); - data_table.clone_column(index, "psp_okey"); + if (index.empty()) { + data_table.add_column("psp_pkey", DTYPE_INT32, true); + data_table.add_column("psp_okey", DTYPE_INT32, true); + } else if (index.size() == 1) { + data_table.clone_column(index[0], "psp_pkey"); + data_table.clone_column(index[0], "psp_okey"); + } else { + data_table.add_column("psp_pkey", DTYPE_STR, true); + data_table.add_column("psp_okey", DTYPE_STR, true); } auto tbl = std::make_shared
( @@ -822,8 +997,9 @@ std::shared_ptr
Table::make_derived( const t_schema& schema, t_dtype pkey_dtype, - const std::string& index, - std::uint32_t limit + const std::vector& index, + std::uint32_t limit, + std::shared_ptr pkey_codec ) { auto pool = std::make_shared(); pool->init(); @@ -839,6 +1015,10 @@ Table::make_derived( ); tbl->m_derived = true; + if (!tbl->m_pkey_codec) { + tbl->m_pkey_codec = std::move(pkey_codec); + } + tbl->set_gnode(tbl->make_gnode(data_table.get_schema())); pool->register_gnode(tbl->m_gnode.get()); tbl->m_gnode->remove_input_port(0); @@ -874,18 +1054,35 @@ Table::update_arrow(const std::string_view& data, std::uint32_t port_id) { auto input_schema = this->get_schema(); auto arrow_names = arrow_loader.names(); - if (std::find(arrow_names.begin(), arrow_names.end(), "__INDEX__") - != arrow_names.end()) { + const bool has_index_column = + std::find(arrow_names.begin(), arrow_names.end(), "__INDEX__") + != arrow_names.end(); + + if (has_index_column) { + if (is_composite_index()) { + PSP_COMPLAIN_AND_ABORT( + "`__INDEX__` is not supported for Arrow updates of a Table " + "with a composite index; include the index columns instead.\n" + ); + } + if (m_index.empty()) { input_schema.add_column("__INDEX__", DTYPE_INT32); } else { input_schema.add_column( - "__INDEX__", input_schema.get_dtype(m_index) + "__INDEX__", input_schema.get_dtype(m_index[0]) ); } } + if (is_composite_index()) { + require_index_columns(arrow_names, m_index, "Update"); + } + arrow_loader.fill_table(data_table, input_schema, m_index, m_offset, true); + if (is_composite_index()) { + fill_pkey_from_index(data_table, m_index, *m_pkey_codec, true, true); + } process_op_column(data_table, t_op::OP_INSERT); calculate_offset(row_count); @@ -894,7 +1091,7 @@ Table::update_arrow(const std::string_view& data, std::uint32_t port_id) { std::shared_ptr
Table::from_arrow( - const std::string& index, + const std::vector& index, std::string&& data, std::uint32_t limit, t_backing_store backing_store, @@ -948,6 +1145,16 @@ Table::from_arrow( loader.fill_table(*data_table, input_schema, index, 0, false); } + if (index.size() > 1) { + fill_pkey_from_index( + *data_table, + index, + t_pkey_codec::from_schema(data_table->get_schema(), index), + false, + true + ); + } + // Make Table auto pool = std::make_shared(); pool->init(); @@ -966,7 +1173,7 @@ Table::make_table( const std::vector& column_names, const std::vector& data_types, std::uint32_t limit, - const std::string& index, + const std::vector& index, const std::string_view& data, t_backing_store backing_store, apachearrow::t_list_flatten list_flatten @@ -1003,14 +1210,18 @@ Table::make_table( void Table::validate_columns(const std::vector& column_names) { - if (!m_index.empty()) { - // Check if index is valid after getting column names - bool explicit_index = - std::find(column_names.begin(), column_names.end(), m_index) - != column_names.end(); - if (!explicit_index) { + std::set seen; + for (const auto& name : m_index) { + if (std::find(column_names.begin(), column_names.end(), name) + == column_names.end()) { + PSP_COMPLAIN_AND_ABORT( + "Specified index `" + name + "` does not exist in dataset." + ); + } + + if (!seen.insert(name).second) { PSP_COMPLAIN_AND_ABORT( - "Specified index `" + m_index + "` does not exist in dataset." + "Specified index column `" + name + "` is repeated." ); } } diff --git a/rust/perspective-server/cpp/perspective/src/cpp/view.cpp b/rust/perspective-server/cpp/perspective/src/cpp/view.cpp index 1c5eefc8f5..ca87ae1d0f 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/view.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/view.cpp @@ -55,6 +55,7 @@ View::View( m_name(std::move(name)), m_separator(std::move(separator)), m_view_config(std::move(std::move(view_config))) { + m_pkey_codec = m_table->get_pkey_codec(); m_row_pivots = m_view_config->get_row_pivots(); m_column_pivots = m_view_config->get_column_pivots(); m_aggregates = m_view_config->get_aggspecs(); @@ -1857,7 +1858,7 @@ View::write_row_path( // Question: Why are the row paths reversed? for (auto entry = row_path.size(); entry > 0; entry--) { const t_tscalar& scalar = row_path[entry - 1]; - write_scalar(scalar, is_formatted, writer); + write_id_entry(scalar, is_formatted, writer); } writer.EndArray(); @@ -1905,6 +1906,43 @@ View::write_column( writer.EndArray(); } +template +void +View::write_pkey( + const t_tscalar& scalar, + bool is_formatted, + rapidjson::Writer& writer +) const { + if (!m_pkey_codec || !scalar.is_valid() + || scalar.get_dtype() != DTYPE_STR) { + write_scalar(scalar, is_formatted, writer); + return; + } + + std::vector storage; + const auto values = m_pkey_codec->decode(scalar.get_char_ptr(), storage); + writer.StartArray(); + for (const auto& value : values) { + write_scalar(value, is_formatted, writer); + } + + writer.EndArray(); +} + +template +void +View::write_id_entry( + const t_tscalar& scalar, + bool is_formatted, + rapidjson::Writer& writer +) const { + if (m_column_only) { + write_pkey(scalar, is_formatted, writer); + } else { + write_scalar(scalar, is_formatted, writer); + } +} + template void View::write_index_column( @@ -1925,7 +1963,7 @@ View::write_index_column( for (auto i = keys.size(); i > 0; --i) { auto scalar = keys[i - 1]; - write_scalar(scalar, is_formatted, writer); + write_pkey(scalar, is_formatted, writer); } writer.EndArray(); @@ -1993,7 +2031,7 @@ View::to_rows( writer.Key("__ID__"); writer.StartArray(); if (!keys.empty()) { - write_scalar(keys[0], is_formatted, writer); + write_pkey(keys[0], is_formatted, writer); } writer.EndArray(); } @@ -2004,7 +2042,7 @@ View::to_rows( writer.StartArray(); for (auto i = keys.size(); i > 0; --i) { auto scalar = keys[i - 1]; - write_scalar(scalar, is_formatted, writer); + write_pkey(scalar, is_formatted, writer); } writer.EndArray(); @@ -2076,7 +2114,7 @@ View::to_rows( const auto row_path = get_row_path(r); for (auto entry = row_path.size(); entry > 0; entry--) { const t_tscalar& scalar = row_path[entry - 1]; - write_scalar(scalar, is_formatted, writer); + write_id_entry(scalar, is_formatted, writer); } writer.EndArray(); @@ -2086,7 +2124,7 @@ View::to_rows( writer.StartArray(); for (auto entry = row_path.size(); entry > 0; entry--) { const t_tscalar& scalar = row_path[entry - 1]; - write_scalar(scalar, is_formatted, writer); + write_id_entry(scalar, is_formatted, writer); } writer.EndArray(); @@ -2098,7 +2136,7 @@ View::to_rows( writer.StartArray(); for (auto i = keys.size(); i > 0; --i) { auto scalar = keys[i - 1]; - write_scalar(scalar, is_formatted, writer); + write_pkey(scalar, is_formatted, writer); } writer.EndArray(); @@ -2170,7 +2208,7 @@ View::to_rows( writer.StartArray(); for (auto entry = row_path.size(); entry > 0; entry--) { const t_tscalar& scalar = row_path[entry - 1]; - write_scalar(scalar, is_formatted, writer); + write_id_entry(scalar, is_formatted, writer); } writer.EndArray(); @@ -2181,7 +2219,7 @@ View::to_rows( writer.StartArray(); for (auto entry = row_path.size(); entry > 0; entry--) { const t_tscalar& scalar = row_path[entry - 1]; - write_scalar(scalar, is_formatted, writer); + write_id_entry(scalar, is_formatted, writer); } writer.EndArray(); @@ -2193,7 +2231,7 @@ View::to_rows( writer.StartArray(); for (auto i = keys.size(); i > 0; --i) { auto scalar = keys[i - 1]; - write_scalar(scalar, is_formatted, writer); + write_pkey(scalar, is_formatted, writer); } writer.EndArray(); @@ -2331,7 +2369,7 @@ View::to_ndjson( writer.Key("__ID__"); writer.StartArray(); if (!keys.empty()) { - write_scalar(keys[0], is_formatted, writer); + write_pkey(keys[0], is_formatted, writer); } writer.EndArray(); } @@ -2342,7 +2380,7 @@ View::to_ndjson( writer.StartArray(); for (auto i = keys.size(); i > 0; --i) { auto scalar = keys[i - 1]; - write_scalar(scalar, is_formatted, writer); + write_pkey(scalar, is_formatted, writer); } writer.EndArray(); @@ -2417,7 +2455,7 @@ View::to_ndjson( const auto row_path = get_row_path(r); for (auto entry = row_path.size(); entry > 0; entry--) { const t_tscalar& scalar = row_path[entry - 1]; - write_scalar(scalar, is_formatted, writer); + write_id_entry(scalar, is_formatted, writer); } writer.EndArray(); @@ -2427,7 +2465,7 @@ View::to_ndjson( writer.StartArray(); for (auto entry = row_path.size(); entry > 0; entry--) { const t_tscalar& scalar = row_path[entry - 1]; - write_scalar(scalar, is_formatted, writer); + write_id_entry(scalar, is_formatted, writer); } writer.EndArray(); @@ -2439,7 +2477,7 @@ View::to_ndjson( writer.StartArray(); for (auto i = keys.size(); i > 0; --i) { auto scalar = keys[i - 1]; - write_scalar(scalar, is_formatted, writer); + write_pkey(scalar, is_formatted, writer); } writer.EndArray(); @@ -2515,7 +2553,7 @@ View::to_ndjson( writer.StartArray(); for (auto entry = row_path.size(); entry > 0; entry--) { const t_tscalar& scalar = row_path[entry - 1]; - write_scalar(scalar, is_formatted, writer); + write_id_entry(scalar, is_formatted, writer); } writer.EndArray(); @@ -2526,7 +2564,7 @@ View::to_ndjson( writer.StartArray(); for (auto entry = row_path.size(); entry > 0; entry--) { const t_tscalar& scalar = row_path[entry - 1]; - write_scalar(scalar, is_formatted, writer); + write_id_entry(scalar, is_formatted, writer); } writer.EndArray(); @@ -2538,7 +2576,7 @@ View::to_ndjson( writer.StartArray(); for (auto i = keys.size(); i > 0; --i) { auto scalar = keys[i - 1]; - write_scalar(scalar, is_formatted, writer); + write_pkey(scalar, is_formatted, writer); } writer.EndArray(); @@ -2620,7 +2658,7 @@ View::to_columns( const auto keys = m_ctx->get_pkeys(vec); writer.StartArray(); if (!keys.empty()) { - write_scalar(keys[0], is_formatted, writer); + write_pkey(keys[0], is_formatted, writer); } writer.EndArray(); } @@ -2667,7 +2705,7 @@ View::to_columns( const auto row_path = m_ctx->get_row_path(r); for (auto entry = row_path.size(); entry > 0; entry--) { const t_tscalar& scalar = row_path[entry - 1]; - write_scalar(scalar, is_formatted, writer); + write_id_entry(scalar, is_formatted, writer); } writer.EndArray(); @@ -2741,7 +2779,7 @@ View::to_columns( const auto row_path = m_ctx->get_row_path(r); for (auto entry = row_path.size(); entry > 0; entry--) { const t_tscalar& scalar = row_path[entry - 1]; - write_scalar(scalar, is_formatted, writer); + write_id_entry(scalar, is_formatted, writer); } writer.EndArray(); diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/arrow_loader.h b/rust/perspective-server/cpp/perspective/src/include/perspective/arrow_loader.h index 0a0e2c6187..5845e17665 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/arrow_loader.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/arrow_loader.h @@ -61,10 +61,10 @@ namespace apachearrow { * the identical value on every row of the group and would silently * collide. * - * @param index - the explicit index column, or empty for none. + * @param index - the explicit index columns, or empty for none. */ - std::optional repeated_index(const std::string& index - ) const; + std::optional + repeated_index(const std::vector& index) const; /** * @brief Initialize the arrow loader with a CSV. @@ -92,7 +92,7 @@ namespace apachearrow { void fill_table( t_data_table& tbl, const t_schema& input_schema, - const std::string& index, + const std::vector& index, std::uint32_t offset, bool is_update ); diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/arrow_writer.h b/rust/perspective-server/cpp/perspective/src/include/perspective/arrow_writer.h index 568f36e594..59448a480c 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/arrow_writer.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/arrow_writer.h @@ -66,6 +66,11 @@ namespace apachearrow { const t_column& col, const std::string& name, t_uindex nrows ); + /** + * @brief Serialize every column of `table` as one Arrow IPC stream. + */ + std::shared_ptr table_to_arrow_ipc(const t_data_table& table); + /** * @brief Build an `arrow::Array` from a column typed as `DTYPE_BOOL.` * diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/computed_function.h b/rust/perspective-server/cpp/perspective/src/include/perspective/computed_function.h index f4768dbf52..f4e5e3b179 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/computed_function.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/computed_function.h @@ -232,6 +232,7 @@ namespace computed_function { vlookup( t_expression_vocab& expression_vocab, bool is_type_validator, + const t_pkey_mapping& pkey_map, const std::shared_ptr& source_table, t_uindex& row_idx ); @@ -241,6 +242,7 @@ namespace computed_function { private: t_expression_vocab& m_expression_vocab; bool m_is_type_validator; + const t_pkey_mapping& m_pkey_map; const std::shared_ptr& m_source_table; t_uindex& m_row_idx; }; diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/data_table.h b/rust/perspective-server/cpp/perspective/src/include/perspective/data_table.h index f200767240..7b4afab865 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/data_table.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/data_table.h @@ -40,6 +40,7 @@ struct t_flatten_record { }; class t_data_table; +class t_pkey_codec; class PERSPECTIVE_EXPORT t_tabular {}; @@ -133,6 +134,16 @@ class PERSPECTIVE_EXPORT t_data_table { std::shared_ptr flatten() const; bool is_pkey_table() const; + + void + set_pkey_codec(std::shared_ptr codec) { + m_pkey_codec = std::move(codec); + } + + const std::shared_ptr& + get_pkey_codec() const { + return m_pkey_codec; + } bool is_same_shape(t_data_table& tbl) const; void pprint() const; @@ -270,6 +281,7 @@ class PERSPECTIVE_EXPORT t_data_table { t_backing_store m_backing_store; bool m_init; std::vector> m_columns; + std::shared_ptr m_pkey_codec; }; PERSPECTIVE_EXPORT bool diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/derived_source.h b/rust/perspective-server/cpp/perspective/src/include/perspective/derived_source.h index 1e94407f8c..91f4896df5 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/derived_source.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/derived_source.h @@ -12,6 +12,7 @@ #pragma once #include +#include #include #include #include @@ -67,7 +68,12 @@ class PERSPECTIVE_EXPORT t_derived_source { virtual t_dtype pkey_dtype() const = 0; - virtual std::string index() const; + virtual std::vector index() const; + + /** + * @brief The parent's composite key codec, or null. + */ + virtual std::shared_ptr pkey_codec() const; virtual std::uint32_t limit() const; diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/json_loader.h b/rust/perspective-server/cpp/perspective/src/include/perspective/json_loader.h index 434baff81f..6719a560ef 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/json_loader.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/json_loader.h @@ -21,6 +21,7 @@ #include #include #include +#include #include #include #include @@ -60,7 +61,7 @@ namespace json { void init( std::string_view data, t_json_format format, - const std::string& index, + const std::vector& index, const t_schema* existing, t_list_flatten mode ); @@ -72,14 +73,14 @@ namespace json { bool empty() const; - std::optional repeated_index(const std::string& index - ) const; + std::optional + repeated_index(const std::vector& index) const; void release(); std::uint32_t fill_table( t_data_table& tbl, - const std::string& index, + const std::vector& index, std::uint32_t offset, bool is_update ); @@ -92,16 +93,42 @@ namespace json { bool is_update ); - void infer_rows(const std::string& index); - void infer_cols(const std::string& index); - void infer_ndjson(const std::string& index); + void infer_rows(const std::vector& index); + void infer_cols(const std::vector& index); + void infer_ndjson(const std::vector& index); - std::uint32_t - fill_rows(t_data_table&, const std::string&, std::uint32_t, bool); - std::uint32_t - fill_cols(t_data_table&, const std::string&, std::uint32_t, bool); - std::uint32_t - fill_ndjson(t_data_table&, const std::string&, std::uint32_t, bool); + std::uint32_t fill_rows( + t_data_table&, const std::vector&, std::uint32_t, bool + ); + std::uint32_t fill_cols( + t_data_table&, const std::vector&, std::uint32_t, bool + ); + std::uint32_t fill_ndjson( + t_data_table&, const std::vector&, std::uint32_t, bool + ); + + bool is_single_index(std::string_view name) const; + void begin_row(); + void note_column(std::string_view name); + void fill_pkey_cell( + t_data_table& tbl, + const std::shared_ptr& pkey_col, + t_uindex ii, + const rapidjson::Value& cell, + bool is_update + ); + void finish_row( + t_data_table& tbl, + const std::shared_ptr& pkey_col, + t_uindex ii + ); + void finish_cols( + t_data_table& tbl, + const std::shared_ptr& pkey_col, + t_uindex nrows, + bool has_index_member, + const std::vector& col_names + ); rapidjson::Document m_document; rapidjson::StringStream m_stream{nullptr}; @@ -115,6 +142,13 @@ namespace json { bool m_expands{false}; bool m_is_implicit{true}; bool m_empty{false}; + + std::vector m_index; + std::optional m_codec; + std::unique_ptr m_index_scratch; + std::vector> m_index_cols; + std::vector m_index_seen; + bool m_row_has_pkey{false}; }; } // namespace json diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/pkey_codec.h b/rust/perspective-server/cpp/perspective/src/include/perspective/pkey_codec.h new file mode 100644 index 0000000000..34d32e0844 --- /dev/null +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/pkey_codec.h @@ -0,0 +1,80 @@ +// ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ +// ┃ ██████ ██████ ██████ █ █ █ █ █ █▄ ▀███ █ ┃ +// ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█ ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄ ▀█ █ ▀▀▀▀▀ ┃ +// ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄ █ ▄▄▄▄▄ ┃ +// ┃ █ ██████ █ ▀█▄ █ ██████ █ ███▌▐███ ███████▄ █ ┃ +// ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫ +// ┃ 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). ┃ +// ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛ + + +#pragma once +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace perspective { + +/** + * @brief Encodes a composite primary key into one order-preserving, + * NUL-free string so a multi-column `index` can live in a `DTYPE_STR` + * `psp_pkey` column. + */ +class PERSPECTIVE_EXPORT t_pkey_codec { +public: + explicit t_pkey_codec(std::vector dtypes); + + static t_pkey_codec from_schema( + const t_schema& schema, const std::vector& index + ); + + static bool is_supported(t_dtype dtype); + + const std::vector& dtypes() const; + + std::size_t arity() const; + + void encode( + const std::vector& cols, + t_uindex ridx, + std::string& out + ) const; + + void encode(const std::vector& key, std::string& out) const; + + std::vector + decode(const char* bytes, std::vector& storage) const; + + t_tscalar coerce(const t_tscalar& value, t_dtype dtype) const; + +private: + void encode_field(const t_tscalar& value, t_dtype dtype, std::string& raw) + const; + + std::vector m_dtypes; +}; + +/** + * @brief Encode every row's composite key from `index` into `psp_pkey`, and + * `psp_okey` when requested. + */ +PERSPECTIVE_EXPORT void fill_pkey_from_index( + t_data_table& tbl, + const std::vector& index, + const t_pkey_codec& codec, + bool is_update, + bool fill_okey +); + +} // namespace perspective diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/table.h b/rust/perspective-server/cpp/perspective/src/include/perspective/table.h index 9f6cf1516c..989bbcad03 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/table.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/table.h @@ -21,6 +21,7 @@ #include #include #include +#include namespace perspective { @@ -46,8 +47,9 @@ class PERSPECTIVE_EXPORT Table { * @param data_types * @param limit - an upper bound on the number of rows in the Table * (optional). - * @param index - a string column name to be used as a primary key. If not - * explicitly set, a primary key will be generated. + * @param index - the column names forming the primary key, which is + * generated when empty and encoded into one `DTYPE_STR` key when there + * is more than one. * @param op */ Table( @@ -55,7 +57,7 @@ class PERSPECTIVE_EXPORT Table { std::vector column_names, std::vector data_types, std::uint32_t limit, - std::string index, + std::vector index, t_backing_store backing_store = BACKING_STORE_MEMORY, apachearrow::t_list_flatten list_flatten = apachearrow::LIST_FLATTEN_ZIP @@ -211,9 +213,24 @@ class PERSPECTIVE_EXPORT Table { const std::vector& get_data_types() const; std::uint32_t get_offset() const; std::uint32_t get_limit() const; - const std::string& get_index() const; + const std::vector& get_index() const; t_backing_store get_backing_store() const; + /** + * @brief Whether this `Table` is keyed by more than one column. + */ + bool is_composite_index() const; + + /** + * @brief The dtype of `psp_pkey` for this `Table`'s index given `schema`. + */ + t_dtype pkey_dtype(const t_schema& schema) const; + + /** + * @brief The codec for a composite key, or null for other indexes. + */ + const std::shared_ptr& get_pkey_codec() const; + // Setters void set_column_names(const std::vector& column_names); void set_data_types(const std::vector& data_types); @@ -230,7 +247,7 @@ class PERSPECTIVE_EXPORT Table { // void update_cols(const std::string_view& data) const; static std::shared_ptr
from_csv( - const std::string& index, + const std::vector& index, std::string&& data, std::uint32_t limit = std::numeric_limits::max(), t_backing_store backing_store = BACKING_STORE_MEMORY, @@ -239,7 +256,7 @@ class PERSPECTIVE_EXPORT Table { ); static std::shared_ptr
from_cols( - const std::string& index, + const std::vector& index, std::string&& data, std::uint32_t limit = std::numeric_limits::max(), t_backing_store backing_store = BACKING_STORE_MEMORY, @@ -248,7 +265,7 @@ class PERSPECTIVE_EXPORT Table { ); static std::shared_ptr
from_rows( - const std::string& index, + const std::vector& index, std::string&& data, std::uint32_t limit = std::numeric_limits::max(), t_backing_store backing_store = BACKING_STORE_MEMORY, @@ -257,7 +274,7 @@ class PERSPECTIVE_EXPORT Table { ); static std::shared_ptr
from_ndjson( - const std::string& index, + const std::vector& index, std::string&& data, std::uint32_t limit = std::numeric_limits::max(), t_backing_store backing_store = BACKING_STORE_MEMORY, @@ -266,7 +283,7 @@ class PERSPECTIVE_EXPORT Table { ); static std::shared_ptr
from_schema( - const std::string& index, + const std::vector& index, const t_schema& schema, std::uint32_t limit = std::numeric_limits::max(), t_backing_store backing_store = BACKING_STORE_MEMORY, @@ -280,14 +297,15 @@ class PERSPECTIVE_EXPORT Table { static std::shared_ptr
make_derived( const t_schema& schema, t_dtype pkey_dtype, - const std::string& index, - std::uint32_t limit + const std::vector& index, + std::uint32_t limit, + std::shared_ptr pkey_codec ); bool is_derived() const; static std::shared_ptr
from_arrow( - const std::string& index, + const std::vector& index, std::string&& data, std::uint32_t limit = std::numeric_limits::max(), t_backing_store backing_store = BACKING_STORE_MEMORY, @@ -299,7 +317,7 @@ class PERSPECTIVE_EXPORT Table { const std::vector& column_names, const std::vector& data_types, std::uint32_t limit, - const std::string& index, + const std::vector& index, const std::string_view& data, t_backing_store backing_store = BACKING_STORE_MEMORY, apachearrow::t_list_flatten list_flatten = @@ -313,7 +331,7 @@ class PERSPECTIVE_EXPORT Table { */ static std::shared_ptr
from_json_loader( json::JsonLoader& loader, - const std::string& index, + const std::vector& index, std::string&& data, std::uint32_t limit, t_backing_store backing_store, @@ -371,11 +389,10 @@ class PERSPECTIVE_EXPORT Table { const t_uindex m_limit; /** - * @brief The name of a column that should be used as the Table's primary - * key. - * + * @brief The columns that form the Table's primary key. */ - const std::string m_index; + const std::vector m_index; + std::shared_ptr m_pkey_codec; bool m_gnode_set; bool m_derived = false; const t_backing_store m_backing_store; diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/view.h b/rust/perspective-server/cpp/perspective/src/include/perspective/view.h index 6a901252b5..01e13d5ba8 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/view.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/view.h @@ -190,6 +190,25 @@ class PERSPECTIVE_EXPORT View { rapidjson::Writer& writer ) const; + /** + * @brief Write a primary key, decoding a composite key into an array. + */ + void write_pkey( + const t_tscalar& scalar, + bool is_formatted, + rapidjson::Writer& writer + ) const; + + /** + * @brief Write one `__ID__` row path entry, which is a primary key on a + * column-only view. + */ + void write_id_entry( + const t_tscalar& scalar, + bool is_formatted, + rapidjson::Writer& writer + ) const; + /** * @brief Returns shared pointer to a t_data_slice object, which * contains the underlying slice of data as well as the metadata @@ -435,6 +454,7 @@ class PERSPECTIVE_EXPORT View { std::shared_ptr
m_table; std::shared_ptr m_ctx; + std::shared_ptr m_pkey_codec; std::string m_name; std::string m_separator;