diff --git a/docs/md/explanation/architecture/client_server.md b/docs/md/explanation/architecture/client_server.md index be30ff4dd3..df9629ad0f 100644 --- a/docs/md/explanation/architecture/client_server.md +++ b/docs/md/explanation/architecture/client_server.md @@ -44,7 +44,10 @@ loop.start() Perspective's websocket client interfaces with the Python server, then _replicates_ the server-side Table. When the server-side `Table` has an `index`, the replica inherits it, and both `update()` and `remove()` on the server are -mirrored in the browser. +mirrored in the browser, including rows which an `update()` moves outside the +`View`'s `filter`. A replica of a `View` with `group_by` is indexed on the +`group_by` columns instead, so groups which gain, change or lose all of their +rows on the server are mirrored too. ```javascript const websocket = await perspective.websocket("ws://localhost:8080"); diff --git a/docs/md/explanation/view/advanced.md b/docs/md/explanation/view/advanced.md index 7a023b4cd8..fd7e5d64ef 100644 --- a/docs/md/explanation/view/advanced.md +++ b/docs/md/explanation/view/advanced.md @@ -233,24 +233,35 @@ view.remove_update(on_update) When `mode` is set to `"row"`, the callback receives a delta of only the rows that changed (as Apache Arrow), which is useful for efficiently synchronizing -tables across clients. The callback also receives `reset`, which is `true` when -the update step began with `clear()` or `replace()`; the delta is then relative -to an empty `Table`, so a replica should apply it with `replace()` rather than -`update()`. +tables across clients. For a `View` with `group_by`, each delta row also carries +the `group_by` columns, named and typed as `to_arrow()` emits them and `null` +below the row's depth, so the delta can be applied with `update()` to a replica +indexed on those columns. The callback also receives `reset`, which is `true` +when the update step began with `clear()` or `replace()`; the delta is then +relative to an empty `Table`, so a replica should apply it with `replace()` +rather than `update()`. In Python, a callback that does not declare `reset`, +positionally or as a keyword, is called without it. ## Remove Callbacks -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 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: +Register a callback to be notified whenever rows leave this `View`. The callback +receives the `port_id`, the rows' keys as an Apache Arrow, and a `cause` naming +the rule they left under. For a `View` without `group_by`, the keys are the +`index` column values, one column per index column named after it, and it +requires an `index`: `cause` is `"key"` when the row was removed from the +`Table` (`remove()`; `replace()` reports the keys it does not re-supply and +`clear()` every key) and `"filter"` when an `update()` moved the row outside the +View's `filter`. For a `View` with `group_by`, `cause` is `"group"` and the keys +are one column per `group_by` column named after it, `null` below the group's +depth, for each group which no longer has any rows. In every case a replica +indexed on those columns can pass the Arrow to `remove()` unchanged. It fires +once per cause per update step, `"key"` first, only for rows which existed +before that step:
```javascript -const callback = await view.on_remove(({ indices, port_id }) => { +const callback = await view.on_remove(({ indices, port_id, cause }) => { replica.remove(indices); }); @@ -262,13 +273,16 @@ await view.remove_remove(callback);
```python -def on_remove(port_id, indices): +def on_remove(port_id, indices, cause): replica.remove(indices) callback = view.on_remove(on_remove) view.remove_remove(callback) ``` +A callback that does not declare `cause`, positionally or as a keyword, is +called without it. +
## Constructing a `Table` from a `View` diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 5d263dfc4b..be161d2016 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1032,75 +1032,18 @@ importers: tools/bench: dependencies: - perspective-0-10-0: - specifier: npm:@finos/perspective@0.10.0 - version: '@finos/perspective@0.10.0' perspective-0-9-0: specifier: npm:@finos/perspective@0.9.0 version: '@finos/perspective@0.9.0' perspective-1-0-0: specifier: npm:@finos/perspective@1.0.0 version: '@finos/perspective@1.0.0' - perspective-1-1-0: - specifier: npm:@finos/perspective@1.1.0 - version: '@finos/perspective@1.1.0' - perspective-2-1-0: - specifier: npm:@finos/perspective@2.1.0 - version: '@finos/perspective@2.1.0' - perspective-2-10-0: - specifier: npm:@finos/perspective@2.10.0 - version: '@finos/perspective@2.10.0' perspective-2-2-0: specifier: npm:@finos/perspective@2.2.0 version: '@finos/perspective@2.2.0' - perspective-2-3-0: - specifier: npm:@finos/perspective@2.3.0 - version: '@finos/perspective@2.3.0' - perspective-2-4-0: - specifier: npm:@finos/perspective@2.4.0 - version: '@finos/perspective@2.4.0' - perspective-2-5-0: - specifier: npm:@finos/perspective@2.5.0 - version: '@finos/perspective@2.5.0' - perspective-2-6-0: - specifier: npm:@finos/perspective@2.6.0 - version: '@finos/perspective@2.6.0' - perspective-2-7-0: - specifier: npm:@finos/perspective@2.7.0 - version: '@finos/perspective@2.7.0' - perspective-2-8-0: - specifier: npm:@finos/perspective@2.8.0 - version: '@finos/perspective@2.8.0' - perspective-2-9-0: - specifier: npm:@finos/perspective@2.9.0 - version: '@finos/perspective@2.9.0' perspective-3-0-0: specifier: npm:@finos/perspective@3.0.0 version: '@finos/perspective@3.0.0' - perspective-3-1-0: - specifier: npm:@finos/perspective@3.1.0 - version: '@finos/perspective@3.1.0' - perspective-3-2-0: - specifier: npm:@finos/perspective@3.2.0 - version: '@finos/perspective@3.2.0' - perspective-3-3-0: - specifier: npm:@finos/perspective@3.3.0 - version: '@finos/perspective@3.3.0' - perspective-3-4-0: - specifier: npm:@finos/perspective@3.4.0 - version: '@finos/perspective@3.4.0' - perspective-3-5-0: - specifier: npm:@finos/perspective@3.5.0 - version: '@finos/perspective@3.5.0' - perspective-3-6-0: - specifier: npm:@finos/perspective@3.6.0 - version: '@finos/perspective@3.6.0' - perspective-3-7-0: - specifier: npm:@finos/perspective@3.7.0 - version: '@finos/perspective@3.7.0' - perspective-3-8-0: - specifier: npm:@finos/perspective@3.8.0 - version: '@finos/perspective@3.8.0(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1)' perspective-4-0-0: specifier: npm:@perspective-dev/client@4.0.0 version: '@perspective-dev/client@4.0.0(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1)' @@ -1137,6 +1080,15 @@ importers: perspective-5-5-0: specifier: npm:@perspective-dev/client@5.5.0 version: '@perspective-dev/client@5.5.0(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1)' + perspective-5-6-0: + specifier: npm:@perspective-dev/client@5.6.0 + version: '@perspective-dev/client@5.6.0(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1)' + perspective-5-7-0: + specifier: npm:@perspective-dev/client@5.7.0 + version: '@perspective-dev/client@5.7.0(@clickhouse/client-web@1.17.0)(@duckdb/duckdb-wasm@1.32.0)' + perspective-5-8-0: + specifier: npm:@perspective-dev/client@5.8.0 + version: '@perspective-dev/client@5.8.0(@clickhouse/client-web@1.17.0)(@duckdb/duckdb-wasm@1.32.0)' devDependencies: '@perspective-dev/client': specifier: workspace:^ @@ -1451,11 +1403,6 @@ packages: resolution: {integrity: sha512-43/qtrDUokr7LJqoF2c3+RInu/t4zfrpYdoSDfYyhg52rwLV6TnOvdG4fXm7IkSB3wErkcmJS9iEhjVtOSEjjA==} engines: {node: ^18.18.0 || ^20.9.0 || >=21.1.0} - '@finos/perspective@0.10.0': - resolution: {integrity: sha512-PmsUiaOjChJaivDxGLNH8vAw8lG98OLlPjKCvci3iAFAEclGKMhSkHP/qYDU+NUZSPqsvk2Rf66TFsK8aiy/jg==} - engines: {node: '>=12'} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - '@finos/perspective@0.9.0': resolution: {integrity: sha512-CkQ8thiCcSLpIr08aMYaYaMcHgk0Cj2PS1TPpz0hbwlR27UZeItgqrR4LNb3Fl4O6lZ3DaBQXIlqV05sP9yvGg==} engines: {node: '>=12'} @@ -1466,97 +1413,15 @@ packages: engines: {node: '>=12'} deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - '@finos/perspective@1.1.0': - resolution: {integrity: sha512-UHqkIKXmJ+q++hhVrVe9uFuNXifeyImHinUOi1Vyw172GpWZE1w8b53zRRk9P2U3WtYm7x7Jrhp1ILqkZ5F+aw==} - engines: {node: '>=12'} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@2.1.0': - resolution: {integrity: sha512-NgDJtM7BbPEry574qQuSkzKEDdhSBSJecFeMyTOOlfzQ8cldR9xMg7/Lo5/NQWWi5cKuyzTWjTphOn6bP2M6YA==} - engines: {node: '>=14'} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@2.10.0': - resolution: {integrity: sha512-8g6hsKhpS2gR7GYQWksNDm8kpymHZiSbJkC4HAp1dI95RArksr4U9CUAOnALgw8JI9B5CYI6b1SWQxbzCsTQ6A==} - engines: {node: '>=14'} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - '@finos/perspective@2.2.0': resolution: {integrity: sha512-OPSFrwnaUyDnAL4jQb4kICMDBO6tqZ1vGGgd9xgTssB0VToIXObPMzlNAFtuWDL7qwiyxupqMYkjXMhoRGY7pA==} engines: {node: '>=14'} deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - '@finos/perspective@2.3.0': - resolution: {integrity: sha512-SnsdYdYHXKarG/mFIUcZmUiTKo9yccnbd+r0a9BRUcY3p9VuDPdp8fnPanoXLKi1xUkIz7RJSSTVpbPL1T+jWg==} - engines: {node: '>=14'} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@2.4.0': - resolution: {integrity: sha512-JPYGAZq0WmcYUKXnyqdgZW+UPvgSAiWHmYx6I+ZVbjz1TIuuPFZ5Yw1n/VumUrjNGn/XNl7WnUUovh31Tez7tQ==} - engines: {node: '>=14'} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@2.5.0': - resolution: {integrity: sha512-4YUs0oLvvGG4HQzqLJQCkRic+FW4h623etmcjBSuqcC3tLT+OMFxNZ8Z3NjS6uvdFu6vPTO7fAA0RnTdZKhXKg==} - engines: {node: '>=14'} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@2.6.0': - resolution: {integrity: sha512-3YIBv4h1wQl8FBuabWLIV6H5aS1ufDC4/xjsrp3bIL5y5dgYwV5Uqr3gFj6cpmbWn0jNaFb6pEVm3Scn2QyDNw==} - engines: {node: '>=14'} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@2.7.0': - resolution: {integrity: sha512-/9UHAnH1iW8ANnfplDLGMS60dx3R8UWRoL0dKTx9A3jNbeP2l+5wd0twsal00FqR/wgUomd3+tbbkzOIMsNclA==} - engines: {node: '>=14'} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@2.8.0': - resolution: {integrity: sha512-D2LSwwFqs4mcZjwBBCNn7JdidK+Th29ZA6rdA1LMKn4pvedok0YzKii0beKTzovhVeKthlcVROKQpFY3DEjptw==} - engines: {node: '>=14'} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@2.9.0': - resolution: {integrity: sha512-O40EzleYRpmX2k8xwbcJfWlgi4QDJ2cUJlmEA0wG1LLVmSOQD7vFeQK9H2v6xQBwn6+/andpDLglrzDYCEBOHg==} - engines: {node: '>=14'} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - '@finos/perspective@3.0.0': resolution: {integrity: sha512-myWGlfxN97vV3qvv34QAxCH389Oo2SH0EGN4eyrGdZSb+UlFjv9HR1TYM18GRhdM1ASg7jFwnIvGDcw7oM+qNA==} deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - '@finos/perspective@3.1.0': - resolution: {integrity: sha512-nyO4NWU1+C4C0SOYDdHM7p01uPl9eeJUYMG+YNy+LNOg5xGSpNZ4CI4iT9+1N7mSKLl9mrly8OAoRQXEHoDKTA==} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@3.2.0': - resolution: {integrity: sha512-shV80NUE44V6AX2jhEAD4XS4uhMyv5A22SJ9kbop/ZKAC3Z9rbrmJffQqymZCgM2457fkjy/GxIcwcDur7GgFQ==} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@3.3.0': - resolution: {integrity: sha512-GYzXPqHnhrWQz930f9OejVvsBs5I6jjJLvbK/xfUK2AUijMhzAUsPI70uEKUFwsy5VMA6sxWPl/UQB9W9IBGDw==} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@3.4.0': - resolution: {integrity: sha512-7FlOdnotYkJOc5/+yerpcSv9VH+1e2fs53ZiOAm0/id2cke9RySbm30NpHICGq13SQW5J3ezByZElx3V72A8mQ==} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@3.5.0': - resolution: {integrity: sha512-J4Q6aGq541iFeNeCrnW6tuq1ELfg0hhILH8qYoI9oJUSx1O/kwackunHOkxcPxW6zeaT0vUfHZaJjS7u1dYTuA==} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@3.6.0': - resolution: {integrity: sha512-MOXifjSsPNO39Bc0LiXp1OvV6ujNes4EoFxJ58edzCutbeSWf0yknXyEBmwR0PB/QWNHx1Czl3cnnJHXgRPn3Q==} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@3.7.0': - resolution: {integrity: sha512-UCtIGE6fuT7A7ljS9R5EmDfN7PZd5DwMg3iAikB/WPVE7dOtulEbvbsnkUMKa78zL8aN5hAGR4h4hxWMlv7fnw==} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - - '@finos/perspective@3.8.0': - resolution: {integrity: sha512-9pot9YJq1RDdIRJlffp97ktbd0Sxk74mnoSOcf6U/luWceyEnfceQYHwJqUPR1IMNb9H1ZC6NQTbU4WRUT3hAQ==} - deprecated: This package is no longer maintained. Please upgrade to @perspective-dev/client - '@fontsource/roboto-mono@4.5.10': resolution: {integrity: sha512-KrJdmkqz6DszT2wV/bbhXef4r0hV3B0vw2mAqei8A2kRnvq+gcJLmmIeQ94vu9VEXrUQzos5M9lH1TAAXpRphw==} @@ -2018,6 +1883,33 @@ packages: '@perspective-dev/client@5.5.0': resolution: {integrity: sha512-c6nN671nxKphyL2U9cjzTNnZm6+mwnTECXsCtaL+HWixPBeE5KMd9LTXsigiXMinC7gYrvpGZsvaJlRecv2jKQ==} + '@perspective-dev/client@5.6.0': + resolution: {integrity: sha512-mDeORPXWbfZZ3QCMdETgw668DZM1wTD/5nU9nOIdM3cWF5yuXRhgaHV4XnVbYtTEWSDkhvEXiZtHesCx5Xx9hA==} + + '@perspective-dev/client@5.7.0': + resolution: {integrity: sha512-gpIT9uxKv0XgorPTwMzRJ1nWQQwB7u8InFmc9W8xCqKLJYcoRuH5gZCJYNf69jkFJmad6RxzFeKh2rJiE2qB/w==} + engines: {node: '>=22'} + peerDependencies: + '@clickhouse/client-web': ^1.12.0 + '@duckdb/duckdb-wasm': ^1.30.0 + peerDependenciesMeta: + '@clickhouse/client-web': + optional: true + '@duckdb/duckdb-wasm': + optional: true + + '@perspective-dev/client@5.8.0': + resolution: {integrity: sha512-gnpyzG5QuUjhAowKuSdho/ysKB3juUyDKzbRSG2nITTN8qwWCvscSTQeFBaIwL4/4GyzZAxK5vtnjgXkJ98zHQ==} + engines: {node: '>=22'} + peerDependencies: + '@clickhouse/client-web': ^1.12.0 + '@duckdb/duckdb-wasm': ^1.30.0 + peerDependenciesMeta: + '@clickhouse/client-web': + optional: true + '@duckdb/duckdb-wasm': + optional: true + '@perspective-dev/server@4.0.0': resolution: {integrity: sha512-S25Mr+cgRomU9aDiplXUj2gImuYZFXUUfytLdJC+f9iipFqX3MqFUCPI4jEzb1xTUnP3Fzg9xzs8p1SymHOF9g==} @@ -2054,6 +1946,9 @@ packages: '@perspective-dev/server@5.5.0': resolution: {integrity: sha512-4wVCiabrrCpe7kvZu5Kx8BNJEjTGj2An7nJJiqZ9CvBZ8bObke6rqFNgTj3o1oGDmai+eTNFpg5LrFH4dZlPEw==} + '@perspective-dev/server@5.8.0': + resolution: {integrity: sha512-kF7UgSKA1HLxYyfSsFwnKHZDjyNPlovixWjhz7bvOTqbDzphVY1EnZAGZUtJ7IB8wjty9lk0kMSp1AvtpfFyTw==} + '@playwright/experimental-ct-core@1.62.0': resolution: {integrity: sha512-jmd7zmlTHrDf4coLZKxIQSmc8Dqr9NuNnadSQx1xeBlkoCUYO9eiiL5yIQ71LKHm/xcipRpe6HSIFEjoxXsAvQ==} engines: {node: '>=20'} @@ -3598,6 +3493,7 @@ packages: image-size@0.5.5: resolution: {integrity: sha512-6TDAlDPZxUFCv+fuOkIoXT/V/f3Qbq8e37p+YOiYrUv3v9cc3/6x78VdfPgFVaB9dZYeLUfKgHRebpkm/oP2VQ==} engines: {node: '>=0.10.0'} + deprecated: Please migrate to image-size@2.x hasBin: true import-cwd@3.0.0: @@ -5522,21 +5418,6 @@ snapshots: '@eslint/core': 0.17.0 levn: 0.4.1 - '@finos/perspective@0.10.0': - dependencies: - '@babel/runtime': 7.28.4 - core-js: 3.46.0 - d3-array: 1.2.4 - detectie: 1.0.0 - flatbuffers: 1.12.0 - lodash: 4.17.21 - text-encoding-utf-8: 1.0.2 - tslib: 1.14.1 - ws: 6.2.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - '@finos/perspective@0.9.0': dependencies: '@babel/runtime': 7.28.4 @@ -5567,30 +5448,6 @@ snapshots: - bufferutil - utf-8-validate - '@finos/perspective@1.1.0': - dependencies: - fflate: 0.7.4 - ws: 6.2.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@2.1.0': - dependencies: - fflate: 0.7.4 - ws: 6.2.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@2.10.0': - dependencies: - stoppable: 1.1.0 - ws: 6.2.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - '@finos/perspective@2.2.0': dependencies: fflate: 0.7.4 @@ -5599,63 +5456,6 @@ snapshots: - bufferutil - utf-8-validate - '@finos/perspective@2.3.0': - dependencies: - fflate: 0.7.4 - ws: 6.2.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@2.4.0': - dependencies: - fflate: 0.7.4 - ws: 6.2.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@2.5.0': - dependencies: - fflate: 0.7.4 - ws: 6.2.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@2.6.0': - dependencies: - fflate: 0.7.4 - stoppable: 1.1.0 - ws: 6.2.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@2.7.0': - dependencies: - stoppable: 1.1.0 - ws: 6.2.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@2.8.0': - dependencies: - stoppable: 1.1.0 - ws: 6.2.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@2.9.0': - dependencies: - stoppable: 1.1.0 - ws: 6.2.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - '@finos/perspective@3.0.0': dependencies: stoppable: 1.1.0 @@ -5664,73 +5464,6 @@ snapshots: - bufferutil - utf-8-validate - '@finos/perspective@3.1.0': - dependencies: - stoppable: 1.1.0 - ws: 8.18.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@3.2.0': - dependencies: - stoppable: 1.1.0 - ws: 8.18.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@3.3.0': - dependencies: - stoppable: 1.1.0 - ws: 8.18.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@3.4.0': - dependencies: - stoppable: 1.1.0 - ws: 8.18.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@3.5.0': - dependencies: - stoppable: 1.1.0 - ws: 8.18.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@3.6.0': - dependencies: - stoppable: 1.1.0 - ws: 8.18.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@3.7.0': - dependencies: - stoppable: 1.1.0 - ws: 8.18.3 - transitivePeerDependencies: - - bufferutil - - utf-8-validate - - '@finos/perspective@3.8.0(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1)': - dependencies: - pro_self_extracting_wasm: 0.0.9(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1) - stoppable: 1.1.0 - ws: 8.18.3 - transitivePeerDependencies: - - bufferutil - - debug - - supports-color - - utf-8-validate - '@fontsource/roboto-mono@4.5.10': {} '@fortawesome/fontawesome-free@5.15.4': {} @@ -6686,6 +6419,42 @@ snapshots: - supports-color - utf-8-validate + '@perspective-dev/client@5.6.0(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1)': + dependencies: + '@perspective-dev/server': 5.8.0 + pro_self_extracting_wasm: 0.0.9(debug@4.4.3(supports-color@8.1.1))(supports-color@8.1.1) + stoppable: 1.1.0 + ws: 8.18.3 + transitivePeerDependencies: + - bufferutil + - debug + - supports-color + - utf-8-validate + + '@perspective-dev/client@5.7.0(@clickhouse/client-web@1.17.0)(@duckdb/duckdb-wasm@1.32.0)': + dependencies: + '@perspective-dev/server': 5.8.0 + stoppable: 1.1.0 + ws: 8.18.3 + optionalDependencies: + '@clickhouse/client-web': 1.17.0 + '@duckdb/duckdb-wasm': 1.32.0 + transitivePeerDependencies: + - bufferutil + - utf-8-validate + + '@perspective-dev/client@5.8.0(@clickhouse/client-web@1.17.0)(@duckdb/duckdb-wasm@1.32.0)': + dependencies: + '@perspective-dev/server': 5.8.0 + stoppable: 1.1.0 + ws: 8.18.3 + optionalDependencies: + '@clickhouse/client-web': 1.17.0 + '@duckdb/duckdb-wasm': 1.32.0 + transitivePeerDependencies: + - bufferutil + - utf-8-validate + '@perspective-dev/server@4.0.0': {} '@perspective-dev/server@4.1.0': {} @@ -6710,6 +6479,8 @@ snapshots: '@perspective-dev/server@5.5.0': {} + '@perspective-dev/server@5.8.0': {} + '@playwright/experimental-ct-core@1.62.0(@types/node@24.9.1)(jiti@1.21.7)(less@4.4.2)(terser@5.44.0)(tsx@4.20.6)(yaml@2.8.1)': dependencies: playwright: 1.62.0 diff --git a/rust/perspective-client/build.rs b/rust/perspective-client/build.rs index 10f71eece4..7ce6d57fbf 100644 --- a/rust/perspective-client/build.rs +++ b/rust/perspective-client/build.rs @@ -61,8 +61,6 @@ fn prost_build() -> Result<()> { "#[ts(type = \"Uint8Array | undefined\")]", ) .field_attribute("ViewOnUpdateResp.delta", "#[serde(with = \"serde_bytes\")]") - .type_attribute("ViewOnRemoveResp", "#[derive(ts_rs::TS)]") - .field_attribute("ViewOnRemoveResp.indices", "#[ts(type = \"Uint8Array\")]") .field_attribute( "ViewOnRemoveResp.indices", "#[serde(with = \"serde_bytes\")]", diff --git a/rust/perspective-client/perspective.proto b/rust/perspective-client/perspective.proto index 27fb29c6db..16c112b997 100644 --- a/rust/perspective-client/perspective.proto +++ b/rust/perspective-client/perspective.proto @@ -477,8 +477,14 @@ message TableRemoveDeleteResp {} // `View::on_remove` message ViewOnRemoveReq {} message ViewOnRemoveResp { + enum Cause { + KEY = 0; + GROUP = 1; + FILTER = 2; + } optional bytes indices = 1; uint32 port_id = 2; + Cause cause = 3; } // `View::remove_remove` diff --git a/rust/perspective-client/src/rust/client.rs b/rust/perspective-client/src/rust/client.rs index 41adae37dc..8fb37aa135 100644 --- a/rust/perspective-client/src/rust/client.rs +++ b/rust/perspective-client/src/rust/client.rs @@ -590,8 +590,11 @@ impl Client { let request_options: TableOptions = options.clone().into(); let mut options = options; let source_index = view.source.as_ref().and_then(|x| x.options.index.clone()); + let config = view.get_config().await?; + let is_pivoted = !config.group_by.is_empty(); + let keyed_by_group = + options.index.is_none() && is_pivoted && config.split_by.is_empty(); 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 = index .columns() @@ -619,6 +622,11 @@ impl Client { return Err(ClientError::BadTableOptions); } + if keyed_by_group { + options.index = Some(config.group_by.clone().into()); + options.limit = None; + } + let window = ViewWindow::default(); let arrow = view.to_arrow(window).await?; let mut table = self @@ -641,7 +649,11 @@ impl Client { }; let update_token = view.on_update(callback, options).await?; - let remove_token = if source_index.is_some() && source_index == table.get_index() { + let replica_index = table.get_index(); + let mirrors_removes = + replica_index.is_some() && (is_pivoted || source_index == replica_index); + + let remove_token = if mirrors_removes { let table_ = table.clone(); let callback = asyncfn!(table_, async move |removed: OnRemoveData| { if let Some(indices) = removed.indices.as_ref().filter(|x| !x.is_empty()) { diff --git a/rust/perspective-client/src/rust/lib.rs b/rust/perspective-client/src/rust/lib.rs index 92d2f98e8c..955fbf3323 100644 --- a/rust/perspective-client/src/rust/lib.rs +++ b/rust/perspective-client/src/rust/lib.rs @@ -61,7 +61,8 @@ pub use crate::table::{ pub use crate::table_data::{TableData, UpdateData}; pub use crate::table_ref::TableRef; pub use crate::view::{ - ColumnWindow, OnRemoveData, OnUpdateData, OnUpdateMode, OnUpdateOptions, View, ViewWindow, + ColumnWindow, OnRemoveData, OnUpdateData, OnUpdateMode, OnUpdateOptions, RemoveCause, View, + ViewWindow, }; pub type ClientError = utils::ClientError; diff --git a/rust/perspective-client/src/rust/view.rs b/rust/perspective-client/src/rust/view.rs index 7edf5a7d5e..f2bb5e7e07 100644 --- a/rust/perspective-client/src/rust/view.rs +++ b/rust/perspective-client/src/rust/view.rs @@ -159,16 +159,55 @@ impl Deref for OnUpdateData { } } -/// Removed index values and port ID corresponding to a remove batch, provided -/// to the callback argument to [`View::on_remove`]. -#[derive(TS)] -pub struct OnRemoveData(crate::proto::ViewOnRemoveResp); +/// The rule under which the rows reported by [`View::on_remove`] left the +/// [`View`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, TS)] +#[serde(rename_all = "lowercase")] +pub enum RemoveCause { + Key, + Group, + Filter, +} -impl Deref for OnRemoveData { - type Target = crate::proto::ViewOnRemoveResp; +impl RemoveCause { + /// This cause as its lowercase name. + pub fn as_str(&self) -> &'static str { + match self { + RemoveCause::Key => "key", + RemoveCause::Group => "group", + RemoveCause::Filter => "filter", + } + } +} - fn deref(&self) -> &Self::Target { - &self.0 +impl From for RemoveCause { + fn from(cause: view_on_remove_resp::Cause) -> Self { + match cause { + view_on_remove_resp::Cause::Key => RemoveCause::Key, + view_on_remove_resp::Cause::Group => RemoveCause::Group, + view_on_remove_resp::Cause::Filter => RemoveCause::Filter, + } + } +} + +/// The keys of rows which left a [`View`], the port ID and the [`RemoveCause`] +/// they left under, provided to the callback argument to [`View::on_remove`]. +#[derive(Clone, Debug, Serialize, TS)] +pub struct OnRemoveData { + pub port_id: u32, + #[ts(type = "Uint8Array | undefined")] + #[serde(with = "serde_bytes")] + pub indices: Option>, + pub cause: RemoveCause, +} + +impl From for OnRemoveData { + fn from(resp: ViewOnRemoveResp) -> Self { + OnRemoveData { + port_id: resp.port_id, + cause: resp.cause().into(), + indices: resp.indices, + } } } /// The [`View`] struct is Perspective's query and serialization interface. It @@ -638,13 +677,19 @@ impl View { } } - /// Register a callback which is invoked whenever rows are removed from - /// this [`View`]'s [`Table`] by [`Table::remove`], with the removed `index` - /// column values as an Apache Arrow of one column named after the index. + /// Register a callback which is invoked whenever rows leave this [`View`], + /// with their keys as an Apache Arrow of one column per key column named + /// after it and the [`RemoveCause`] they left under. /// - /// [`Table::replace`] reports the keys it does not re-supply and - /// [`Table::clear`] reports every key. It never fires for a - /// [`Table`] without an `index`. + /// For a [`View`] without `group_by` the keys are the `index` columns of + /// the [`Table`], which must have one: [`RemoveCause::Key`] when + /// [`Table::remove`] removed them ([`Table::replace`] reports the keys it + /// does not re-supply and [`Table::clear`] every key) and + /// [`RemoveCause::Filter`] when [`Table::update`] moved them outside the + /// [`View`]'s `filter`. For a [`View`] with `group_by` the keys are the + /// `group_by` columns of each group which no longer has any rows, + /// [`RemoveCause::Group`], `null` below the group's depth. It fires once + /// per cause per update step. /// /// # Examples /// @@ -668,7 +713,7 @@ impl View { async move { match resp.client_resp { Some(ClientResp::ViewOnRemoveResp(resp)) => { - on_remove(OnRemoveData(resp)).await; + on_remove(OnRemoveData::from(resp)).await; Ok(()) }, resp => Err(resp.into()), diff --git a/rust/perspective-client/src/rust/virtual_server/server.rs b/rust/perspective-client/src/rust/virtual_server/server.rs index b961bb2879..23133fcf24 100644 --- a/rust/perspective-client/src/rust/virtual_server/server.rs +++ b/rust/perspective-client/src/rust/virtual_server/server.rs @@ -651,7 +651,8 @@ impl VirtualServer { ViewOnRemoveReq(_) => { respond!(msg, ViewOnRemoveResp { indices: None, - port_id: 0 + port_id: 0, + cause: 0 }) }, ViewRemoveOnRemoveReq(_) => { diff --git a/rust/perspective-js/src/rust/lib.rs b/rust/perspective-js/src/rust/lib.rs index cb649f5a44..3cc3c3207f 100644 --- a/rust/perspective-js/src/rust/lib.rs +++ b/rust/perspective-js/src/rust/lib.rs @@ -53,7 +53,7 @@ export type * from "../ts-rs/ColumnWindow.d.ts"; export type * from "../ts-rs/TableInitOptions.d.ts"; export type * from "../ts-rs/ViewConfigUpdate.d.ts"; export type * from "../ts-rs/ViewOnUpdateResp.d.ts"; -export type * from "../ts-rs/ViewOnRemoveResp.d.ts"; +export type * from "../ts-rs/RemoveCause.d.ts"; export type * from "../ts-rs/OnRemoveData.d.ts"; export type * from "../ts-rs/OnUpdateData.d.ts"; export type * from "../ts-rs/OnUpdateOptions.d.ts"; diff --git a/rust/perspective-js/src/rust/view.rs b/rust/perspective-js/src/rust/view.rs index 7572690eaa..ff46f134c3 100644 --- a/rust/perspective-js/src/rust/view.rs +++ b/rust/perspective-js/src/rust/view.rs @@ -364,19 +364,24 @@ impl View { Ok(self.0.remove_update(callback_id).await?) } - /// Register a callback which is invoked whenever rows are removed from - /// this [`View`]'s [`Table`] by [`Table::remove`], with an object - /// containing `port_id` and `indices`, the removed `index` column - /// values as an Arrow of one column named after the index. + /// Register a callback which is invoked whenever rows leave this + /// [`View`], with an object containing `port_id`, `indices` (their keys + /// as an Arrow of one column per key column named after it) and `cause`, + /// one of `"key"`, `"filter"` or `"group"`. /// - /// [`Table::replace`] reports the keys it does not re-supply and - /// [`Table::clear`] reports every key. `on_remove` never fires for a - /// [`Table`] without an `index`. + /// For a [`View`] without `group_by` the keys are the `index` columns of + /// the [`Table`], which must have one: `"key"` when [`Table::remove`] + /// removed them ([`Table::replace`] reports the keys it does not + /// re-supply and [`Table::clear`] every key) and `"filter"` when + /// [`Table::update`] moved them outside the [`View`]'s `filter`. For a + /// [`View`] with `group_by` the keys are the `group_by` columns of each + /// group which no longer has any rows, `"group"`, `null` below the + /// group's depth. It fires once per cause per update step. /// /// # JavaScript Examples /// /// ```javascript - /// const id = await view.on_remove(({ indices, port_id }) => { + /// const id = await view.on_remove(({ indices, port_id, cause }) => { /// replica.remove(indices); /// }); /// ``` @@ -387,7 +392,7 @@ impl View { on_remove_js: Function, ) -> ApiFuture { let poll_loop = LocalPollLoop::new(move |args: OnRemoveData| { - let js_obj = JsValue::from_serde_ext(&*args)?; + let js_obj = JsValue::from_serde_ext(&args)?; on_remove_js.call1(&JsValue::UNDEFINED, &js_obj) }); diff --git a/rust/perspective-js/src/ts/public_types.ts b/rust/perspective-js/src/ts/public_types.ts index b66b4e842c..efd5ec6f4f 100644 --- a/rust/perspective-js/src/ts/public_types.ts +++ b/rust/perspective-js/src/ts/public_types.ts @@ -44,6 +44,7 @@ export type * from "../../dist/ts-rs/JoinType.d.ts"; export type * from "../../dist/ts-rs/OnRemoveData.d.ts"; export type * from "../../dist/ts-rs/OnUpdateData.d.ts"; export type * from "../../dist/ts-rs/OnUpdateOptions.d.ts"; +export type * from "../../dist/ts-rs/RemoveCause.d.ts"; export type * from "../../dist/ts-rs/Scalar.d.ts"; export type * from "../../dist/ts-rs/SortDir.d.ts"; export type * from "../../dist/ts-rs/SystemInfo.d.ts"; @@ -52,7 +53,6 @@ export type * from "../../dist/ts-rs/TypedArrayWindow.d.ts"; export type * from "../../dist/ts-rs/UpdateOptions.d.ts"; export type * from "../../dist/ts-rs/ViewConfig.d.ts"; export type * from "../../dist/ts-rs/ViewConfigUpdate.d.ts"; -export type * from "../../dist/ts-rs/ViewOnRemoveResp.d.ts"; export type * from "../../dist/ts-rs/ViewOnUpdateResp.d.ts"; export type * from "../../dist/ts-rs/ViewWindow.d.ts"; export type * from "../../dist/ts-rs/WindowAggSpec.d.ts"; diff --git a/rust/perspective-js/test/js/aggregates.spec.js b/rust/perspective-js/test/js/aggregates.spec.js index b5cabacb25..f4ed02db34 100644 --- a/rust/perspective-js/test/js/aggregates.spec.js +++ b/rust/perspective-js/test/js/aggregates.spec.js @@ -2430,6 +2430,191 @@ const std = (nums) => { view.delete(); table.delete(); }); + + const ORDER_EXPECTED = { + avg: 1.5, + min: 1, + max: 2, + low: 1, + high: 2, + q1: 1, + median: 1.5, + q3: 2, + "high minus low": 1, + }; + + async function grouped_order_aggregate(table, column, agg) { + const view = await table.view({ + group_by: ["g"], + columns: [column], + aggregates: { [column]: agg }, + }); + + const out = (await view.to_columns())[column]; + await view.delete(); + return out; + } + + test("order aggregates skip nulls in the first update", async function () { + const table = await perspective.table({ g: "string", x: "float" }); + await table.update({ g: ["a", "a", "a"], x: [1, null, 2] }); + const results = {}; + for (const agg of Object.keys(ORDER_EXPECTED)) { + results[agg] = await grouped_order_aggregate(table, "x", agg); + } + + const expected = Object.fromEntries( + Object.entries(ORDER_EXPECTED).map(([k, v]) => [k, [v, v]]), + ); + + expect(results).toEqual(expected); + await table.delete(); + }); + + test("order aggregates skip nulls appended in a later update", async function () { + const table = await perspective.table({ g: "string", x: "float" }); + await table.update({ g: ["a"], x: [1] }); + await table.update({ g: ["a", "a"], x: [null, 2] }); + const results = {}; + for (const agg of Object.keys(ORDER_EXPECTED)) { + results[agg] = await grouped_order_aggregate(table, "x", agg); + } + + const expected = Object.fromEntries( + Object.entries(ORDER_EXPECTED).map(([k, v]) => [k, [v, v]]), + ); + + expect(results).toEqual(expected); + await table.delete(); + }); + + test("order aggregates skip omitted keys", async function () { + const table = await perspective.table({ g: "string", x: "float" }); + await table.update([ + { g: "a", x: 1 }, + { g: "a" }, + { g: "a", x: 2 }, + ]); + const results = {}; + for (const agg of Object.keys(ORDER_EXPECTED)) { + results[agg] = await grouped_order_aggregate(table, "x", agg); + } + + const expected = Object.fromEntries( + Object.entries(ORDER_EXPECTED).map(([k, v]) => [k, [v, v]]), + ); + + expect(results).toEqual(expected); + await table.delete(); + }); + + test("high and low skip nulls with negative values", async function () { + const table = await perspective.table({ g: "string", x: "float" }); + await table.update({ g: ["a", "a", "a"], x: [-1, null, -2] }); + const results = {}; + for (const agg of ["high", "low", "max", "min", "high minus low"]) { + results[agg] = await grouped_order_aggregate(table, "x", agg); + } + + expect(results).toEqual({ + high: [-1, -1], + low: [-2, -2], + max: [-1, -1], + min: [-2, -2], + "high minus low": [1, 1], + }); + + await table.delete(); + }); + + test("order aggregates skip nulls for integer, string and date columns", async function () { + const table = await perspective.table({ + g: "string", + i: "integer", + s: "string", + d: "date", + }); + + await table.update({ + g: ["a", "a"], + i: [7, null], + s: ["x", null], + d: [new Date(2020, 0, 5), null], + }); + + const flat = await table.view({ columns: ["d"] }); + const date = (await flat.to_columns()).d[0]; + await flat.delete(); + + const results = {}; + for (const agg of ["max", "min", "median"]) { + for (const column of ["i", "s", "d"]) { + results[`${column} ${agg}`] = await grouped_order_aggregate( + table, + column, + agg, + ); + } + } + + expect(date).not.toBeNull(); + expect(results).toEqual({ + "i max": [7, 7], + "s max": ["x", "x"], + "d max": [date, date], + "i min": [7, 7], + "s min": ["x", "x"], + "d min": [date, date], + "i median": [7, 7], + "s median": ["x", "x"], + "d median": [date, date], + }); + + await table.delete(); + }); + + test("order aggregates report null for an all-null group", async function () { + const table = await perspective.table({ g: "string", x: "float" }); + await table.update({ g: ["a", "a"], x: [null, null] }); + const results = {}; + for (const agg of Object.keys(ORDER_EXPECTED)) { + results[agg] = await grouped_order_aggregate(table, "x", agg); + } + + const expected = Object.fromEntries( + Object.keys(ORDER_EXPECTED).map((k) => [k, [null, null]]), + ); + + expect(results).toEqual(expected); + await table.delete(); + }); + + test("high and low keep their mark when a row is updated to null", async function () { + const table = await perspective.table( + { g: "string", x: "float", k: "integer" }, + { index: "k" }, + ); + + await table.update({ g: ["a", "a"], x: [3, 5], k: [1, 2] }); + const high = await table.view({ + group_by: ["g"], + columns: ["x"], + aggregates: { x: "high" }, + }); + + const low = await table.view({ + group_by: ["g"], + columns: ["x"], + aggregates: { x: "low" }, + }); + + await table.update({ k: [2], x: [null] }); + expect((await high.to_columns()).x).toEqual([5, 5]); + expect((await low.to_columns()).x).toEqual([3, 3]); + await high.delete(); + await low.delete(); + await table.delete(); + }); }); test.describe("Invalid aggregates", function () { diff --git a/rust/perspective-js/test/js/constructors.spec.js b/rust/perspective-js/test/js/constructors.spec.js index 7388036918..ce543aad59 100644 --- a/rust/perspective-js/test/js/constructors.spec.js +++ b/rust/perspective-js/test/js/constructors.spec.js @@ -1503,6 +1503,73 @@ function validate_typed_array(typed_array, column_data) { table.delete(); }); + test("Preserves nulls before an integer column is promoted to float", async function () { + const table = await perspective.table([ + { x: 1 }, + { x: null }, + { x: 2.5 }, + ]); + + expect(await table.schema()).toEqual({ x: "float" }); + const view = await table.view(); + expect((await view.to_columns()).x).toEqual([1, null, 2.5]); + await view.delete(); + await table.delete(); + }); + + test("Preserves nulls before a column-oriented integer column is promoted to float", async function () { + const leading = await perspective.table({ x: [null, 1, 2.5] }); + expect(await leading.schema()).toEqual({ x: "float" }); + const leading_view = await leading.view(); + expect((await leading_view.to_columns()).x).toEqual([null, 1, 2.5]); + await leading_view.delete(); + await leading.delete(); + + const interior = await perspective.table({ x: [1, 2, null, 2.5] }); + expect(await interior.schema()).toEqual({ x: "float" }); + const interior_view = await interior.view(); + expect((await interior_view.to_columns()).x).toEqual([ + 1, + 2, + null, + 2.5, + ]); + await interior_view.delete(); + await interior.delete(); + }); + + test("Preserves nulls before an integer column is promoted to string", async function () { + const table = await perspective.table([ + { x: 1 }, + { x: null }, + { x: "a" }, + ]); + + expect(await table.schema()).toEqual({ x: "string" }); + const view = await table.view(); + expect((await view.to_columns()).x).toEqual(["1", null, "a"]); + await view.delete(); + await table.delete(); + }); + + test("Nulls before a float promotion do not count toward avg", async function () { + const table = await perspective.table([ + { g: "a", x: 1 }, + { g: "a", x: null }, + { g: "a", x: 2.5 }, + ]); + + const view = await table.view({ + group_by: ["g"], + columns: ["x"], + aggregates: { x: "avg" }, + }); + + expect((await view.to_columns()).x).toEqual([1.75, 1.75]); + await view.delete(); + await table.delete(); + }); + test("has correct size", async function () { var table = await perspective.table(data); let result = await table.size(); diff --git a/rust/perspective-js/test/js/delta.spec.js b/rust/perspective-js/test/js/delta.spec.js index 898375f47a..a2cee7fd67 100644 --- a/rust/perspective-js/test/js/delta.spec.js +++ b/rust/perspective-js/test/js/delta.spec.js @@ -39,6 +39,15 @@ let partial_change_nonseq = [ { x: 4, y: "string2", z: true }, ]; +async function arrow_rows(perspective, view) { + const table = await perspective.table(await view.to_arrow()); + const flat = await table.view(); + const rows = await flat.to_json(); + await flat.delete(); + await table.delete(); + return rows; +} + async function match_delta( perspective, delta, @@ -721,8 +730,18 @@ function it_old_behavior(name, capture) { view.on_update( async function (updated) { const expected = [ - { x: 1, y: 1, z: 1 }, - { x: 2, y: 1, z: 1 }, + { + y: "string1", + x: 1, + "y (distinct count)": 1, + z: 1, + }, + { + y: "string2", + x: 2, + "y (distinct count)": 1, + z: 1, + }, ]; await match_delta(perspective, updated.delta, expected); view.delete(); @@ -795,7 +814,10 @@ function it_old_behavior(name, capture) { }); view.on_update( async function (updated) { - const expected = [{ x: 1 }, { x: 2 }]; + const expected = [ + { y: "string1", x: 1 }, + { y: "string2", x: 2 }, + ]; await match_delta( perspective, updated.delta, @@ -844,9 +866,19 @@ function it_old_behavior(name, capture) { view.on_update( async function (updated) { const expected = [ - { x: 13, y: 6, z: 3 }, - { x: 1, y: 1, z: 1 }, - { x: 2, y: 1, z: 1 }, + { y: null, x: 13, "y (distinct count)": 6, z: 3 }, + { + y: "string1", + x: 1, + "y (distinct count)": 1, + z: 1, + }, + { + y: "string2", + x: 2, + "y (distinct count)": 1, + z: 1, + }, ]; await match_delta(perspective, updated.delta, expected); view.delete(); @@ -872,7 +904,11 @@ function it_old_behavior(name, capture) { }); view.on_update( async function (updated) { - const expected = [{ z: 3 }, { z: 1 }, { z: 1 }]; + const expected = [ + { y: null, z: 3 }, + { y: "string1", z: 1 }, + { y: "string2", z: 1 }, + ]; await match_delta( perspective, updated.delta, @@ -897,8 +933,8 @@ function it_old_behavior(name, capture) { view.on_update( async function (updated) { const expected = [ - { x: 10, y: 3, z: 2 }, - { x: 5, y: 1, z: 2 }, + { y: null, x: 10, "y (distinct count)": 3, z: 2 }, + { y: null, x: 5, "y (distinct count)": 1, z: 2 }, ]; await match_delta(perspective, updated.delta, expected); view.delete(); @@ -928,8 +964,18 @@ function it_old_behavior(name, capture) { async function (updated) { // aggregates are sorted, in this case by string comparator - "string1" and "string2" are at the end const expected = [ - { x: 1, y: 1, z: 1 }, - { x: 4, y: 1, z: 1 }, + { + y: "string1", + x: 1, + "y (distinct count)": 1, + z: 1, + }, + { + y: "string2", + x: 4, + "y (distinct count)": 1, + z: 1, + }, ]; await match_delta( perspective, @@ -961,7 +1007,11 @@ function it_old_behavior(name, capture) { view.on_update( async function (updated) { - const expected = [{ x: 13 }, { x: 2 }, { x: 1 }]; + const expected = [ + { y: null, x: 13 }, + { y: "string2", x: 2 }, + { y: "string1", x: 1 }, + ]; await match_delta( perspective, updated.delta, @@ -990,11 +1040,10 @@ function it_old_behavior(name, capture) { }); view.on_update( async function (updated) { - const json = await view.to_json(); - json.map((d) => { - delete d["__ROW_PATH__"]; - }); - const expected = json.slice(0, 3); + const json = await arrow_rows(perspective, view); + const expected = json.filter((d) => + [null, "string1", "string2"].includes(d.y), + ); await match_delta( perspective, updated.delta, @@ -1021,11 +1070,10 @@ function it_old_behavior(name, capture) { }); view.on_update( async function (updated) { - const json = await view.to_json(); - json.map((d) => { - delete d["__ROW_PATH__"]; - }); - const expected = json.slice(0, 3); + const json = await arrow_rows(perspective, view); + const expected = json.filter((d) => + [null, "string1", "string2"].includes(d.y), + ); await match_delta( perspective, updated.delta, @@ -1053,14 +1101,14 @@ function it_old_behavior(name, capture) { }); view.on_update( async function (updated) { - const expected = await view.to_json(); + const expected = await arrow_rows( + perspective, + view, + ); await match_delta( perspective, updated.delta, - expected.slice(0, 3).map((x) => { - delete x.__ROW_PATH__; - return x; - }), + expected.slice(0, 3), ); view.delete(); table.delete(); @@ -1117,10 +1165,7 @@ function it_old_behavior(name, capture) { }); view.on_update( async function (updated) { - const json = await view.to_json(); - json.map((d) => { - delete d["__ROW_PATH__"]; - }); + const json = await arrow_rows(perspective, view); const expected = json.slice(0, 3); await match_delta( perspective, @@ -1148,10 +1193,7 @@ function it_old_behavior(name, capture) { }); view.on_update( async function (updated) { - const json = await view.to_json(); - json.map((d) => { - delete d["__ROW_PATH__"]; - }); + const json = await arrow_rows(perspective, view); const expected = json.slice(0, 3); await match_delta( perspective, @@ -1178,10 +1220,7 @@ function it_old_behavior(name, capture) { }); view.on_update( async function (updated) { - const json = await view.to_json(); - json.map((d) => { - delete d["__ROW_PATH__"]; - }); + const json = await arrow_rows(perspective, view); const expected = [json[0], json[3], json[4]]; await match_delta( perspective, @@ -1209,10 +1248,7 @@ function it_old_behavior(name, capture) { }); view.on_update( async function (updated) { - const json = await view.to_json(); - json.map((d) => { - delete d["__ROW_PATH__"]; - }); + const json = await arrow_rows(perspective, view); const expected = [json[0], json[3], json[4]]; await match_delta( perspective, @@ -1262,11 +1298,10 @@ function it_old_behavior(name, capture) { }); view.on_update( async function (updated) { - const json = await view.to_json(); - json.map((d) => { - delete d["__ROW_PATH__"]; - }); - const expected = json.slice(0, 3); + const json = await arrow_rows(perspective, view); + const expected = json.filter((d) => + [null, "string1", "string2"].includes(d.y), + ); await match_delta(perspective, updated.delta, expected); view.delete(); table.delete(); @@ -1287,14 +1322,9 @@ function it_old_behavior(name, capture) { }); view.on_update( async function (updated) { - const expected = (await view.to_json()).filter( - (d) => - d["__ROW_PATH__"].length === 0 || - d["__ROW_PATH__"][0] !== "c", - ); - expected.map((d) => { - delete d["__ROW_PATH__"]; - }); + const expected = ( + await arrow_rows(perspective, view) + ).filter((d) => d.y !== "c"); await match_delta(perspective, updated.delta, expected); view.delete(); table.delete(); @@ -1321,14 +1351,9 @@ function it_old_behavior(name, capture) { }); view.on_update( async function (updated) { - const expected = (await view.to_json()).filter( - (d) => - d["__ROW_PATH__"].length === 0 || - d["__ROW_PATH__"][0] !== "c", - ); - expected.map((d) => { - delete d["__ROW_PATH__"]; - }); + const expected = ( + await arrow_rows(perspective, view) + ).filter((d) => d.y !== "c"); await match_delta( perspective, updated.delta, @@ -1363,10 +1388,7 @@ function it_old_behavior(name, capture) { view.on_update( async function (updated) { // aggregates are sorted, in this case by string comparator - "string1" and "string2" are at the end - const json = await view.to_json(); - json.map((d) => { - delete d["__ROW_PATH__"]; - }); + const json = await arrow_rows(perspective, view); const expected = [json[3], json[4]]; await match_delta( perspective, diff --git a/rust/perspective-js/test/js/derived/lifecycle.spec.ts b/rust/perspective-js/test/js/derived/lifecycle.spec.ts index eb5c430d71..e86a63e582 100644 --- a/rust/perspective-js/test/js/derived/lifecycle.spec.ts +++ b/rust/perspective-js/test/js/derived/lifecycle.spec.ts @@ -139,4 +139,19 @@ test.describe("Derived table lifecycle", function () { await source.delete(); }); } + + test("group_by: a dissolved group leaves the derived table", async function () { + const config = PARENTS.group_by; + const source = await make_source(); + const view = await source.view(config); + const derived = await perspective.table(view); + const ids = rows(24) + .filter((row) => row.g === "a") + .map((row) => row.id); + await source.remove(ids); + await expect_tree_parity(source, config, derived, view); + await derived.delete(); + await view.delete(); + await source.delete(); + }); }); diff --git a/rust/perspective-js/test/js/expressions/deltas.spec.js b/rust/perspective-js/test/js/expressions/deltas.spec.js index 1340d33687..671af8e5b1 100644 --- a/rust/perspective-js/test/js/expressions/deltas.spec.js +++ b/rust/perspective-js/test/js/expressions/deltas.spec.js @@ -224,9 +224,24 @@ function it_old_behavior(name, capture) { view.on_update( async function (updated) { const expected = [ - { x: 14, y: 6, 'lower("y")': 6 }, - { x: 1, y: 1, 'lower("y")': 1 }, - { x: 3, y: 1, 'lower("y")': 1 }, + { + 'lower("y")': null, + x: 14, + y: 6, + 'lower("y") (count)': 6, + }, + { + 'lower("y")': "hello", + x: 1, + y: 1, + 'lower("y") (count)': 1, + }, + { + 'lower("y")': "world", + x: 3, + y: 1, + 'lower("y") (count)': 1, + }, ]; await match_delta( perspective, @@ -260,7 +275,11 @@ function it_old_behavior(name, capture) { view.on_update( async function (updated) { - const expected = [{ x: 14 }, { x: 3 }, { x: 1 }]; + const expected = [ + { 'lower("y")': null, x: 14 }, + { 'lower("y")': "world", x: 3 }, + { 'lower("y")': "hello", x: 1 }, + ]; await match_delta( perspective, updated.delta, @@ -299,62 +318,65 @@ function it_old_behavior(name, capture) { async function (updated) { const expected = [ { - 'A|lower("y")': "a", + 'lower("y")': null, + 'A|lower("y") (last)': "a", "A|x": 1, "A|y": 1, - 'B|lower("y")': "b", + 'B|lower("y") (last)': "b", "B|x": 2, "B|y": 1, - 'C|lower("y")': "c", + 'C|lower("y") (last)': "c", "C|x": 3, "C|y": 1, - 'D|lower("y")': "d", + 'D|lower("y") (last)': "d", "D|x": 4, "D|y": 1, - 'HELLO|lower("y")': "hello", + 'HELLO|lower("y") (last)': "hello", "HELLO|x": 1, "HELLO|y": 1, - 'WORLD|lower("y")': "world", + 'WORLD|lower("y") (last)': "world", "WORLD|x": 3, "WORLD|y": 1, }, { - 'A|lower("y")': null, + 'lower("y")': "hello", + 'A|lower("y") (last)': null, "A|x": null, "A|y": null, - 'B|lower("y")': null, + 'B|lower("y") (last)': null, "B|x": null, "B|y": null, - 'C|lower("y")': null, + 'C|lower("y") (last)': null, "C|x": null, "C|y": null, - 'D|lower("y")': null, + 'D|lower("y") (last)': null, "D|x": null, "D|y": null, - 'HELLO|lower("y")': "hello", + 'HELLO|lower("y") (last)': "hello", "HELLO|x": 1, "HELLO|y": 1, - 'WORLD|lower("y")': null, + 'WORLD|lower("y") (last)': null, "WORLD|x": null, "WORLD|y": null, }, { - 'A|lower("y")': null, + 'lower("y")': "world", + 'A|lower("y") (last)': null, "A|x": null, "A|y": null, - 'B|lower("y")': null, + 'B|lower("y") (last)': null, "B|x": null, "B|y": null, - 'C|lower("y")': null, + 'C|lower("y") (last)': null, "C|x": null, "C|y": null, - 'D|lower("y")': null, + 'D|lower("y") (last)': null, "D|x": null, "D|y": null, - 'HELLO|lower("y")': null, + 'HELLO|lower("y") (last)': null, "HELLO|x": null, "HELLO|y": null, - 'WORLD|lower("y")': "world", + 'WORLD|lower("y") (last)': "world", "WORLD|x": 3, "WORLD|y": 1, }, diff --git a/rust/perspective-js/test/js/multi_server.spec.js b/rust/perspective-js/test/js/multi_server.spec.js index a8ad73b914..7555baca15 100644 --- a/rust/perspective-js/test/js/multi_server.spec.js +++ b/rust/perspective-js/test/js/multi_server.spec.js @@ -347,5 +347,222 @@ import perspective from "./perspective_client"; expect(json0[1]).toEqual(json0[0]); expect(json1).toEqual([[], []]); }); + + test("Dissolved groups transfer to a remote Table", async function ({ + page, + }) { + await page.goto("/rust/perspective-js/test/html/test.html"); + const result = await page.evaluate(async () => { + let perspective = await import( + "http://localhost:6598/node_modules/@perspective-dev/client/dist/cdn/perspective.js" + ); + + perspective.init_client( + fetch( + "http://localhost:6598/node_modules/@perspective-dev/client/dist/wasm/perspective-js.wasm", + ), + ); + + const worker0 = await perspective.worker(); + const worker1 = await perspective.worker(); + const sorted = (rows) => + rows.map((row) => JSON.stringify(row)).sort(); + + const rows_of = async (view) => { + const table = await worker1.table(await view.to_arrow()); + const flat = await table.view(); + const rows = await flat.to_json(); + await flat.delete(); + await table.delete(); + return rows; + }; + + const sync = async (view1, check, action) => { + const { promise, resolve } = Promise.withResolvers(); + const token = await view1.on_update(async () => { + if (await check()) { + resolve(); + } + }); + + await action(); + await Promise.race([ + promise, + new Promise((resolve) => setTimeout(resolve, 2000)), + ]); + + await view1.remove_update(token); + }; + + const table0 = await worker0.table("x,y\n1,a\n2,b\n3,a", { + index: "x", + }); + + const view0 = await table0.view({ + group_by: ["y"], + columns: ["x"], + }); + + const table1 = await worker1.table(view0); + const view1 = await table1.view(); + const same = async () => + JSON.stringify(sorted(await rows_of(view0))) === + JSON.stringify(sorted(await view1.to_json())); + + await sync(view1, same, () => table0.remove([2])); + return { + index: [await table1.get_index()].flat(), + same: await same(), + rows: await view1.to_json(), + }; + }); + + expect(result.index).toEqual(["y"]); + expect(result.same).toEqual(true); + expect(result.rows.some((row) => row.y === "b")).toEqual(false); + }); + + test("Dissolved groups of a limited remote Table transfer", async function ({ + page, + }) { + await page.goto("/rust/perspective-js/test/html/test.html"); + const result = await page.evaluate(async () => { + let perspective = await import( + "http://localhost:6598/node_modules/@perspective-dev/client/dist/cdn/perspective.js" + ); + + perspective.init_client( + fetch( + "http://localhost:6598/node_modules/@perspective-dev/client/dist/wasm/perspective-js.wasm", + ), + ); + + const worker0 = await perspective.worker(); + const worker1 = await perspective.worker(); + const sorted = (rows) => + rows.map((row) => JSON.stringify(row)).sort(); + + const rows_of = async (view) => { + const table = await worker1.table(await view.to_arrow()); + const flat = await table.view(); + const rows = await flat.to_json(); + await flat.delete(); + await table.delete(); + return rows; + }; + + const sync = async (view1, check, action) => { + const { promise, resolve } = Promise.withResolvers(); + const token = await view1.on_update(async () => { + if (await check()) { + resolve(); + } + }); + + await action(); + await Promise.race([ + promise, + new Promise((resolve) => setTimeout(resolve, 2000)), + ]); + + await view1.remove_update(token); + }; + + const table0 = await worker0.table("x,y\n1,a\n2,b\n3,c", { + limit: 3, + }); + + const view0 = await table0.view({ + group_by: ["y"], + columns: ["x"], + }); + + const table1 = await worker1.table(view0); + const view1 = await table1.view(); + const same = async () => + JSON.stringify(sorted(await rows_of(view0))) === + JSON.stringify(sorted(await view1.to_json())); + + await sync(view1, same, () => + table0.update([{ x: 4, y: "d" }]), + ); + return { + limit: await table1.get_limit(), + same: await same(), + rows: await view1.to_json(), + }; + }); + + expect(result.limit).toBeFalsy(); + expect(result.same).toEqual(true); + expect(result.rows.some((row) => row.y === "a")).toEqual(false); + }); + + test("Rows leaving a remote View's filter transfer", async function ({ + page, + }) { + await page.goto("/rust/perspective-js/test/html/test.html"); + const result = await page.evaluate(async () => { + let perspective = await import( + "http://localhost:6598/node_modules/@perspective-dev/client/dist/cdn/perspective.js" + ); + + perspective.init_client( + fetch( + "http://localhost:6598/node_modules/@perspective-dev/client/dist/wasm/perspective-js.wasm", + ), + ); + + const worker0 = await perspective.worker(); + const worker1 = await perspective.worker(); + const sorted = (rows) => + rows.map((row) => JSON.stringify(row)).sort(); + + const rows_of = async (view) => { + const table = await worker1.table(await view.to_arrow()); + const flat = await table.view(); + const rows = await flat.to_json(); + await flat.delete(); + await table.delete(); + return rows; + }; + + const sync = async (view1, check, action) => { + const { promise, resolve } = Promise.withResolvers(); + const token = await view1.on_update(async () => { + if (await check()) { + resolve(); + } + }); + + await action(); + await Promise.race([ + promise, + new Promise((resolve) => setTimeout(resolve, 2000)), + ]); + + await view1.remove_update(token); + }; + + const table0 = await worker0.table("x,y\n1,a\n2,b\n3,a", { + index: "x", + }); + + const view0 = await table0.view({ filter: [["y", "==", "a"]] }); + const table1 = await worker1.table(view0); + const view1 = await table1.view(); + const same = async () => + JSON.stringify(await view0.to_json()) === + JSON.stringify(await view1.to_json()); + + await sync(view1, same, () => + table0.update([{ x: 1, y: "b" }]), + ); + return [await view0.to_json(), await view1.to_json()]; + }); + + expect(result[0]).toEqual([{ x: 3, y: "a" }]); + expect(result[1]).toEqual(result[0]); + }); }); })(perspective); diff --git a/rust/perspective-js/test/js/on_remove.spec.js b/rust/perspective-js/test/js/on_remove.spec.js index 3b968fb179..0a096247e7 100644 --- a/rust/perspective-js/test/js/on_remove.spec.js +++ b/rust/perspective-js/test/js/on_remove.spec.js @@ -66,6 +66,7 @@ async function fixture(perspective, index) { await table.remove([2]); const removed = await promise; expect(removed.port_id).toEqual(0); + expect(removed.cause).toEqual("key"); expect(await removed_indices(perspective, removed.indices)).toEqual( [{ x: 2 }], ); @@ -247,4 +248,210 @@ async function fixture(perspective, index) { await cleanup(); }); }); + + test.describe("View.on_remove pivoted", function () { + async function pivoted(perspective, config, options = { index: "x" }) { + const table = await perspective.table(data, options); + const view = await table.view(config); + const cleanup = async () => { + await view.delete(); + await table.delete(); + }; + + return { table, view, cleanup }; + } + + const by_x = (a, b) => (a.x ?? -Infinity) - (b.x ?? -Infinity); + + test("fires when a `remove` dissolves a group", async function () { + const { table, view, cleanup } = await pivoted(perspective, { + group_by: ["y"], + }); + const sub = await subscribe(view); + const promise = sub.next(); + await table.remove([2]); + const removed = await promise; + expect(removed.cause).toEqual("group"); + expect(await removed_indices(perspective, removed.indices)).toEqual( + [{ y: "b" }], + ); + await view.remove_remove(sub.id); + await cleanup(); + }); + + test("fires when an `update` re-keys the last member of a group", async function () { + const { table, view, cleanup } = await pivoted(perspective, { + group_by: ["y"], + }); + const sub = await subscribe(view); + const promise = sub.next(); + await table.update([{ x: 2, y: "a", z: 2.5 }]); + const removed = await promise; + expect(removed.cause).toEqual("group"); + expect(await removed_indices(perspective, removed.indices)).toEqual( + [{ y: "b" }], + ); + await view.remove_remove(sub.id); + await cleanup(); + }); + + test("does not fire when a group only changes aggregates", async function () { + const { table, view, cleanup } = await pivoted(perspective, { + group_by: ["y"], + }); + const sub = await subscribe(view); + await table.update([{ x: 1, y: "a", z: 9.5 }]); + await table.size(); + expect(sub.calls.length).toEqual(0); + const promise = sub.next(); + await table.remove([3]); + await promise; + expect(sub.calls.length).toEqual(1); + expect( + await removed_indices(perspective, sub.calls[0].indices), + ).toEqual([{ y: "c" }]); + await view.remove_remove(sub.id); + await cleanup(); + }); + + test("reports every level of a dropped subtree", async function () { + const { table, view, cleanup } = await pivoted(perspective, { + group_by: ["y", "x"], + }); + const sub = await subscribe(view); + const promise = sub.next(); + await table.remove([2]); + const removed = await promise; + const rows = await removed_indices(perspective, removed.indices); + expect(rows.sort(by_x)).toEqual([ + { y: "b", x: null }, + { y: "b", x: 2 }, + ]); + await view.remove_remove(sub.id); + await cleanup(); + }); + + test("`group_rollup_mode: 'flat'` reports leaves only", async function () { + const { table, view, cleanup } = await pivoted(perspective, { + group_by: ["y", "x"], + group_rollup_mode: "flat", + }); + const sub = await subscribe(view); + const promise = sub.next(); + await table.remove([2]); + const removed = await promise; + expect(await removed_indices(perspective, removed.indices)).toEqual( + [{ y: "b", x: 2 }], + ); + await view.remove_remove(sub.id); + await cleanup(); + }); + + test("`clear` reports nothing", async function () { + const { table, view, cleanup } = await pivoted(perspective, { + group_by: ["y"], + }); + const sub = await subscribe(view); + await table.clear(); + await table.update(data); + await table.size(); + expect(sub.calls.length).toEqual(0); + expect(await view.num_rows()).toEqual(4); + await view.remove_remove(sub.id); + await cleanup(); + }); + + test("fires when a `limit` eviction dissolves a group", async function () { + const { table, view, cleanup } = await pivoted( + perspective, + { group_by: ["y"] }, + { limit: 3 }, + ); + const sub = await subscribe(view); + const promise = sub.next(); + await table.update([{ x: 4, y: "d", z: 4.5 }]); + const removed = await promise; + expect(removed.cause).toEqual("group"); + expect(await removed_indices(perspective, removed.indices)).toEqual( + [{ y: "a" }], + ); + await view.remove_remove(sub.id); + await cleanup(); + }); + }); + + test.describe("View.on_remove filter", function () { + async function filtered(perspective) { + const table = await perspective.table(data, { index: "x" }); + const view = await table.view({ filter: [["y", "==", "a"]] }); + const cleanup = async () => { + await view.delete(); + await table.delete(); + }; + + return { table, view, cleanup }; + } + + test("fires when an `update` moves a row outside the filter", async function () { + const { table, view, cleanup } = await filtered(perspective); + const sub = await subscribe(view); + const promise = sub.next(); + await table.update([{ x: 1, y: "b", z: 1.5 }]); + const removed = await promise; + expect(removed.cause).toEqual("filter"); + expect(await removed_indices(perspective, removed.indices)).toEqual( + [{ x: 1 }], + ); + expect(await view.to_json()).toEqual([]); + await view.remove_remove(sub.id); + await cleanup(); + }); + + test("does not fire when a row re-enters the filter", async function () { + const { table, view, cleanup } = await filtered(perspective); + const sub = await subscribe(view); + const promise = sub.next(); + await table.update([{ x: 1, y: "b", z: 1.5 }]); + await promise; + await table.update([{ x: 1, y: "a", z: 1.5 }]); + await table.size(); + expect(sub.calls.length).toEqual(1); + expect(await view.to_json()).toEqual([{ x: 1, y: "a", z: 1.5 }]); + await view.remove_remove(sub.id); + await cleanup(); + }); + + test("does not fire for an `update` inside the filter", async function () { + const { table, view, cleanup } = await filtered(perspective); + const sub = await subscribe(view); + await table.update([{ x: 1, y: "a", z: 9.5 }]); + await table.size(); + expect(sub.calls.length).toEqual(0); + await view.remove_remove(sub.id); + await cleanup(); + }); + + test("`key` and `filter` removals arrive as separate messages", async function () { + const { table, view, cleanup } = await filtered(perspective); + const sub = await subscribe(view); + const first = sub.next(); + await table.remove([2]); + await first; + const second = sub.next(); + await table.update([{ x: 1, y: "b", z: 1.5 }]); + await second; + expect(sub.calls.map((removed) => removed.cause)).toEqual([ + "key", + "filter", + ]); + expect( + await removed_indices(perspective, sub.calls[0].indices), + ).toEqual([{ x: 2 }]); + expect( + await removed_indices(perspective, sub.calls[1].indices), + ).toEqual([{ x: 1 }]); + await view.remove_remove(sub.id); + await cleanup(); + }); + }); })(perspective); diff --git a/rust/perspective-js/test/js/split_rollup_mode.spec.js b/rust/perspective-js/test/js/split_rollup_mode.spec.js index 5904a60df7..14812152b9 100644 --- a/rust/perspective-js/test/js/split_rollup_mode.spec.js +++ b/rust/perspective-js/test/js/split_rollup_mode.spec.js @@ -325,7 +325,7 @@ const data = { // The delta batch carries one column per emitted column // path, total and subtotal groups included. expect(Object.keys(delta).sort()).toStrictEqual( - ["w", "false|w", "true|w"].sort(), + ["y", "w", "false|w", "true|w"].sort(), ); view.delete(); table.delete(); 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 962817860b..2d5107cc9a 100644 --- a/rust/perspective-python/perspective/tests/table/test_on_remove.py +++ b/rust/perspective-python/perspective/tests/table/test_on_remove.py @@ -155,3 +155,116 @@ def test_table_from_pivoted_view_stays_unindexed(self): view = tbl.view(group_by=["y"]) tbl2 = Table(view) assert tbl2.get_index() is None + + def test_on_remove_cause_is_key(self): + tbl = Table(data, index="x") + view = tbl.view() + calls = [] + cid = view.on_remove( + lambda port_id, indices, cause: calls.append((cause, indices)) + ) + tbl.remove([2]) + assert [cause for cause, _ in calls] == ["key"] + assert removed_indices(calls[0][1]) == [{"x": 2}] + view.remove_remove(cid) + + def test_on_remove_keyword_only_cause(self): + tbl = Table(data, index="x") + view = tbl.view() + calls = [] + + def cb(port_id, indices, *, cause): + calls.append(cause) + + cid = view.on_remove(cb) + tbl.remove([2]) + assert calls == ["key"] + view.remove_remove(cid) + + def test_on_remove_group_by_reports_dissolved_group(self): + tbl = Table(data, index="x") + view = tbl.view(group_by=["y"]) + calls = [] + cid = view.on_remove( + lambda port_id, indices, cause: calls.append((cause, indices)) + ) + tbl.remove([2]) + assert [cause for cause, _ in calls] == ["group"] + assert removed_indices(calls[0][1]) == [{"y": "b"}] + view.remove_remove(cid) + + def test_on_remove_group_by_rekeying_update(self): + tbl = Table(data, index="x") + view = tbl.view(group_by=["y"]) + calls = [] + cid = view.on_remove( + lambda port_id, indices, cause: calls.append((cause, indices)) + ) + tbl.update([{"x": 1, "y": "zz"}]) + tbl.update([{"x": 2, "y": "a"}]) + assert [cause for cause, _ in calls] == ["group", "group"] + assert removed_indices(calls[0][1]) == [{"y": "a"}] + assert removed_indices(calls[1][1]) == [{"y": "b"}] + view.remove_remove(cid) + + def test_on_remove_group_by_two_levels(self): + tbl = Table(data, index="x") + view = tbl.view(group_by=["y", "x"]) + calls = [] + cid = view.on_remove(lambda port_id, indices, cause: calls.append(indices)) + tbl.remove([2]) + assert len(calls) == 1 + rows = sorted(removed_indices(calls[0]), key=lambda row: row["x"] or 0) + assert rows == [{"y": "b", "x": None}, {"y": "b", "x": 2}] + view.remove_remove(cid) + + def test_on_remove_group_by_aggregate_change_does_not_fire(self): + tbl = Table(data, index="x") + view = tbl.view(group_by=["y"], aggregates={"x": "sum"}) + calls = [] + cid = view.on_remove(lambda port_id, indices, cause: calls.append(cause)) + tbl.update([{"x": 1, "y": "a"}]) + tbl.update([{"x": 4, "y": "a"}]) + assert calls == [] + view.remove_remove(cid) + + def test_on_remove_group_by_limit_eviction(self): + tbl = Table(data, limit=3) + view = tbl.view(group_by=["y"]) + calls = [] + cid = view.on_remove( + lambda port_id, indices, cause: calls.append((cause, indices)) + ) + tbl.update([{"x": 4, "y": "d"}]) + assert [cause for cause, _ in calls] == ["group"] + assert removed_indices(calls[0][1]) == [{"y": "a"}] + view.remove_remove(cid) + + def test_on_remove_filter_reports_rows_leaving(self): + tbl = Table(data, index="x") + view = tbl.view(filter=[["y", "==", "a"]]) + calls = [] + cid = view.on_remove( + lambda port_id, indices, cause: calls.append((cause, indices)) + ) + tbl.update([{"x": 1, "y": "b"}]) + assert [cause for cause, _ in calls] == ["filter"] + assert removed_indices(calls[0][1]) == [{"x": 1}] + assert view.to_records() == [] + tbl.update([{"x": 1, "y": "a"}]) + assert len(calls) == 1 + view.remove_remove(cid) + + def test_on_remove_filter_key_and_filter_are_separate_messages(self): + tbl = Table(data, index="x") + view = tbl.view(filter=[["y", "==", "a"]]) + calls = [] + cid = view.on_remove( + lambda port_id, indices, cause: calls.append((cause, indices)) + ) + tbl.remove([2]) + tbl.update([{"x": 1, "y": "b"}]) + assert [cause for cause, _ in calls] == ["key", "filter"] + assert removed_indices(calls[0][1]) == [{"x": 2}] + assert removed_indices(calls[1][1]) == [{"x": 1}] + view.remove_remove(cid) diff --git a/rust/perspective-python/perspective/tests/table/test_table.py b/rust/perspective-python/perspective/tests/table/test_table.py index 08aefd937e..47ebd7e2cd 100644 --- a/rust/perspective-python/perspective/tests/table/test_table.py +++ b/rust/perspective-python/perspective/tests/table/test_table.py @@ -141,6 +141,24 @@ def test_table_float(self): assert tbl.size() == 2 assert tbl.schema() == {"a": "float", "b": "float"} + def test_table_int_to_float_promotion_preserves_nulls(self): + tbl = Table([{"x": 1}, {"x": None}, {"x": 2.5}]) + assert tbl.schema() == {"x": "float"} + assert tbl.view().to_columns() == {"x": [1, None, 2.5]} + + def test_table_int_to_float_promotion_preserves_nulls_column_oriented(self): + leading = Table({"x": [None, 1, 2.5]}) + assert leading.schema() == {"x": "float"} + assert leading.view().to_columns() == {"x": [None, 1, 2.5]} + interior = Table({"x": [1, 2, None, 2.5]}) + assert interior.schema() == {"x": "float"} + assert interior.view().to_columns() == {"x": [1, 2, None, 2.5]} + + def test_table_int_to_str_promotion_preserves_nulls(self): + tbl = Table([{"x": 1}, {"x": None}, {"x": "a"}]) + assert tbl.schema() == {"x": "string"} + assert tbl.view().to_columns() == {"x": ["1", None, "a"]} + def test_table_str(self): str_data = [{"a": "b", "b": "b"}, {"a": "3", "b": "3"}] tbl = Table(str_data) diff --git a/rust/perspective-python/perspective/tests/table/test_view.py b/rust/perspective-python/perspective/tests/table/test_view.py index 5919683564..7429b2c2b5 100644 --- a/rust/perspective-python/perspective/tests/table/test_view.py +++ b/rust/perspective-python/perspective/tests/table/test_view.py @@ -431,6 +431,75 @@ def test_view_aggregate_mean(self): {"__ROW_PATH__": ["a"], "y": 300 / 2}, ] + def test_view_aggregate_mean_skips_nulls_before_float_promotion(self): + tbl = Table([{"g": "a", "x": 1}, {"g": "a", "x": None}, {"g": "a", "x": 2.5}]) + view = tbl.view(aggregates={"x": "avg"}, group_by=["g"], columns=["x"]) + assert view.to_columns()["x"] == [1.75, 1.75] + + ORDER_AGGREGATES_EXPECTED = { + "avg": 1.5, + "min": 1, + "max": 2, + "low": 1, + "high": 2, + "q1": 1, + "median": 1.5, + "q3": 2, + "high minus low": 1, + } + + def _order_aggregate(self, tbl, agg): + view = tbl.view(aggregates={"x": agg}, group_by=["g"], columns=["x"]) + out = view.to_columns()["x"] + view.delete() + return out + + def test_view_order_aggregates_skip_nulls(self): + tbl = Table({"g": "string", "x": "float"}) + tbl.update({"g": ["a", "a", "a"], "x": [1, None, 2]}) + results = { + agg: self._order_aggregate(tbl, agg) + for agg in self.ORDER_AGGREGATES_EXPECTED + } + assert results == {k: [v, v] for k, v in self.ORDER_AGGREGATES_EXPECTED.items()} + + def test_view_order_aggregates_skip_nulls_from_later_update(self): + tbl = Table({"g": "string", "x": "float"}) + tbl.update({"g": ["a"], "x": [1]}) + tbl.update({"g": ["a", "a"], "x": [None, 2]}) + results = { + agg: self._order_aggregate(tbl, agg) + for agg in self.ORDER_AGGREGATES_EXPECTED + } + assert results == {k: [v, v] for k, v in self.ORDER_AGGREGATES_EXPECTED.items()} + + def test_view_high_low_skip_nulls_with_negative_values(self): + tbl = Table({"g": "string", "x": "float"}) + tbl.update({"g": ["a", "a", "a"], "x": [-1, None, -2]}) + assert self._order_aggregate(tbl, "high") == [-1, -1] + assert self._order_aggregate(tbl, "low") == [-2, -2] + assert self._order_aggregate(tbl, "max") == [-1, -1] + assert self._order_aggregate(tbl, "min") == [-2, -2] + assert self._order_aggregate(tbl, "high minus low") == [1, 1] + + def test_view_order_aggregates_null_for_all_null_group(self): + tbl = Table({"g": "string", "x": "float"}) + tbl.update({"g": ["a", "a"], "x": [None, None]}) + results = { + agg: self._order_aggregate(tbl, agg) + for agg in self.ORDER_AGGREGATES_EXPECTED + } + assert results == {k: [None, None] for k in self.ORDER_AGGREGATES_EXPECTED} + + def test_view_high_low_keep_mark_when_row_updated_to_null(self): + tbl = Table({"g": "string", "x": "float", "k": "integer"}, index="k") + tbl.update({"g": ["a", "a"], "x": [3, 5], "k": [1, 2]}) + high = tbl.view(aggregates={"x": "high"}, group_by=["g"], columns=["x"]) + low = tbl.view(aggregates={"x": "low"}, group_by=["g"], columns=["x"]) + tbl.update({"k": [2], "x": [None]}) + assert high.to_columns()["x"] == [5, 5] + assert low.to_columns()["x"] == [3, 3] + def test_view_aggregate_gmv(self): data = { "division": [ @@ -1954,6 +2023,90 @@ def cb1(port_id, delta, reset): tbl.clear() assert steps == [False, True, True] + def test_view_row_delta_two_arg_callback_omits_reset(self): + data = [{"a": 1, "b": 2}, {"a": 3, "b": 4}] + calls = [] + + def cb1(port_id, delta): + calls.append((port_id, delta is not None)) + + tbl = Table(data, index="a") + view = tbl.view() + view.on_update(cb1, mode="row") + tbl.update([{"a": 1, "b": 5}]) + tbl.replace(data) + tbl.clear() + assert calls == [(0, True), (0, True), (0, True)] + + def test_view_row_delta_keyword_only_reset(self): + data = [{"a": 1, "b": 2}, {"a": 3, "b": 4}] + steps = [] + + def cb1(port_id, delta, *, reset): + steps.append(reset) + + tbl = Table(data, index="a") + view = tbl.view() + view.on_update(cb1, mode="row") + tbl.update([{"a": 1, "b": 5}]) + tbl.replace(data) + assert steps == [False, True] + + def test_view_row_delta_var_args_receives_reset(self): + data = [{"a": 1, "b": 2}, {"a": 3, "b": 4}] + arities = [] + + def cb1(*args): + arities.append(len(args)) + + tbl = Table(data, index="a") + view = tbl.view() + view.on_update(cb1, mode="row") + tbl.update([{"a": 1, "b": 5}]) + assert arities == [3] + + def test_view_row_delta_kwargs_receives_reset(self): + data = [{"a": 1, "b": 2}, {"a": 3, "b": 4}] + kwargs = [] + + def cb1(port_id, delta, **kw): + kwargs.append(kw) + + tbl = Table(data, index="a") + view = tbl.view() + view.on_update(cb1, mode="row") + tbl.clear() + assert kwargs == [{"reset": True}] + + def test_view_row_delta_bound_method_and_partial(self): + data = [{"a": 1, "b": 2}, {"a": 3, "b": 4}] + import functools + + class Collector: + def __init__(self): + self.calls = [] + + def two(self, port_id, delta): + self.calls.append(2) + + def three(self, port_id, delta, reset): + self.calls.append(3) + + collector = Collector() + partial_calls = [] + + def with_extra(extra, port_id, delta, reset): + partial_calls.append((extra, reset)) + + tbl = Table(data, index="a") + view = tbl.view() + view.on_update(collector.two, mode="row") + view.on_update(collector.three, mode="row") + view.on_update(functools.partial(with_extra, "x"), mode="row") + tbl.update([{"a": 1, "b": 5}]) + assert sorted(collector.calls) == [2, 3] + assert partial_calls == [("x", False)] + def test_view_row_delta_zero_column_subset(self, util): data = [{"a": 1, "b": 2}, {"a": 3, "b": 4}] update_data = {"a": [5], "b": [6]} @@ -2029,7 +2182,7 @@ def test_view_row_delta_one(self, util): update_data = {"a": [5], "b": [6]} def cb1(port_id, delta, reset): - compare_delta(delta, {"a": [9, 5], "b": [12, 6]}) + compare_delta(delta, {"a": [None, 5], "a (sum)": [9, 5], "b": [12, 6]}) tbl = Table(data) view = tbl.view(group_by=["a"]) @@ -2045,7 +2198,14 @@ def test_view_row_delta_one_from_schema(self, util): update_data = {"a": [1, 2, 3, 4, 5], "b": [6, 7, 8, 9, 10]} def cb1(port_id, delta, reset): - compare_delta(delta, {"a": [15, 1, 2, 3, 4, 5], "b": [40, 6, 7, 8, 9, 10]}) + compare_delta( + delta, + { + "a": [None, 1, 2, 3, 4, 5], + "a (sum)": [15, 1, 2, 3, 4, 5], + "b": [40, 6, 7, 8, 9, 10], + }, + ) tbl = Table({"a": "integer", "b": "integer"}) view = tbl.view(group_by=["a"]) @@ -2056,7 +2216,14 @@ def test_view_row_delta_one_from_schema_sorted(self, util): update_data = {"a": [1, 2, 3, 4, 5], "b": [6, 7, 8, 9, 10]} def cb1(port_id, delta, reset): - compare_delta(delta, {"a": [15, 5, 4, 3, 2, 1], "b": [40, 10, 9, 8, 7, 6]}) + compare_delta( + delta, + { + "a": [None, 5, 4, 3, 2, 1], + "a (sum)": [15, 5, 4, 3, 2, 1], + "b": [40, 10, 9, 8, 7, 6], + }, + ) tbl = Table({"a": "integer", "b": "integer"}) view = tbl.view(group_by=["a"], sort=[["a", "desc"]]) @@ -2067,7 +2234,9 @@ def test_view_row_delta_one_from_schema_filtered(self, util): update_data = {"a": [1, 2, 3, 4, 5], "b": [6, 7, 8, 9, 10]} def cb1(port_id, delta, reset): - compare_delta(delta, {"a": [9, 4, 5], "b": [19, 9, 10]}) + compare_delta( + delta, {"a": [None, 4, 5], "a (sum)": [9, 4, 5], "b": [19, 9, 10]} + ) tbl = Table({"a": "integer", "b": "integer"}) view = tbl.view(group_by=["a"], filter=[["a", ">", 3]]) @@ -2078,7 +2247,9 @@ def test_view_row_delta_one_from_schema_sorted_filtered(self, util): update_data = {"a": [1, 2, 3, 4, 5], "b": [6, 7, 8, 9, 10]} def cb1(port_id, delta, reset): - compare_delta(delta, {"a": [9, 5, 4], "b": [19, 10, 9]}) + compare_delta( + delta, {"a": [None, 5, 4], "a (sum)": [9, 5, 4], "b": [19, 10, 9]} + ) tbl = Table({"a": "integer", "b": "integer"}) view = tbl.view(group_by=["a"], sort=[["a", "desc"]], filter=[["a", ">", 3]]) @@ -2089,7 +2260,14 @@ def test_view_row_delta_one_from_schema_indexed(self, util): update_data = {"a": [1, 2, 3, 4, 5, 5, 4], "b": [6, 7, 8, 9, 10, 11, 12]} def cb1(port_id, delta, reset): - compare_delta(delta, {"a": [15, 1, 2, 3, 4, 5], "b": [44, 6, 7, 8, 12, 11]}) + compare_delta( + delta, + { + "a": [None, 1, 2, 3, 4, 5], + "a (sum)": [15, 1, 2, 3, 4, 5], + "b": [44, 6, 7, 8, 12, 11], + }, + ) tbl = Table({"a": "integer", "b": "integer"}, index="a") @@ -2102,7 +2280,14 @@ def test_view_row_delta_one_from_schema_sorted_indexed(self, util): update_data = {"a": [1, 2, 3, 4, 5, 5, 4], "b": [6, 7, 8, 9, 10, 11, 12]} def cb1(port_id, delta, reset): - compare_delta(delta, {"a": [15, 4, 5, 3, 2, 1], "b": [44, 12, 11, 8, 7, 6]}) + compare_delta( + delta, + { + "a": [None, 4, 5, 3, 2, 1], + "a (sum)": [15, 4, 5, 3, 2, 1], + "b": [44, 12, 11, 8, 7, 6], + }, + ) tbl = Table({"a": "integer", "b": "integer"}, index="a") @@ -2115,7 +2300,9 @@ def test_view_row_delta_one_from_schema_filtered_indexed(self, util): update_data = {"a": [1, 2, 3, 4, 5, 5, 4], "b": [6, 7, 8, 9, 10, 11, 12]} def cb1(port_id, delta, reset): - compare_delta(delta, {"a": [9, 4, 5], "b": [23, 12, 11]}) + compare_delta( + delta, {"a": [None, 4, 5], "a (sum)": [9, 4, 5], "b": [23, 12, 11]} + ) tbl = Table({"a": "integer", "b": "integer"}, index="a") @@ -2132,11 +2319,12 @@ def cb1(port_id, delta, reset): compare_delta( delta, { - "2|a": [1, None], + "a": [None, 5], + "2|a (sum)": [1, None], "2|b": [2, None], - "4|a": [3, None], + "4|a (sum)": [3, None], "4|b": [4, None], - "6|a": [5, 5], + "6|a (sum)": [5, 5], "6|b": [6, 6], }, ) @@ -2160,9 +2348,10 @@ def cb1(port_id, delta, reset): compare_delta( delta, { - "2|a": [1, 1, None], + "a": [None, 1, 3], + "2|a (sum)": [1, 1, None], "2|b": [2, 2, None], - "4|a": [3, None, 3], + "4|a (sum)": [3, None, 3], "4|b": [4, None, 4], }, ) @@ -2179,9 +2368,10 @@ def cb1(port_id, delta, reset): compare_delta( delta, { - "2|a": [1, 1, None], + "a": [None, 1, 3], + "2|a (sum)": [1, 1, None], "2|b": [2, 2, None], - "5|a": [3, None, 3], + "5|a (sum)": [3, None, 3], "5|b": [5, None, 5], }, ) diff --git a/rust/perspective-python/src/client/client_async.rs b/rust/perspective-python/src/client/client_async.rs index 1301fb2f36..f8bc7d2b64 100644 --- a/rust/perspective-python/src/client/client_async.rs +++ b/rust/perspective-python/src/client/client_async.rs @@ -37,6 +37,89 @@ use super::{pandas, polars, pyarrow}; use crate::py_async::{self, AllowThreads}; use crate::py_err::{PyPerspectiveError, ResultTClientErrorExt}; +/// How a callback receives an argument its signature may or may not declare. +#[derive(Clone, Copy)] +enum ExtraArg { + Positional, + Keyword, + Omit, +} + +/// Decide from `callback`'s signature how to pass the argument `name` beyond +/// the first `legacy_arity` positional ones, passing it positionally to any +/// callable `inspect` cannot describe. +fn negotiate_extra_arg( + py: Python<'_>, + callback: &Py, + legacy_arity: usize, + name: &str, +) -> ExtraArg { + let inspect = || -> PyResult { + let inspect = py.import("inspect")?; + let parameter = inspect.getattr("Parameter")?; + let var_positional = parameter.getattr("VAR_POSITIONAL")?; + let positional_only = parameter.getattr("POSITIONAL_ONLY")?; + let positional_or_keyword = parameter.getattr("POSITIONAL_OR_KEYWORD")?; + let keyword_only = parameter.getattr("KEYWORD_ONLY")?; + let var_keyword = parameter.getattr("VAR_KEYWORD")?; + let signature = inspect.call_method1("signature", (callback.bind(py),))?; + let parameters = signature.getattr("parameters")?.call_method0("values")?; + let mut positional = 0usize; + let mut keyword = false; + for param in parameters.try_iter()? { + let param = param?; + let kind = param.getattr("kind")?; + if kind.eq(&var_positional)? { + return Ok(ExtraArg::Positional); + } else if kind.eq(&positional_only)? || kind.eq(&positional_or_keyword)? { + positional += 1; + } else if kind.eq(&var_keyword)? { + keyword = true; + } else if kind.eq(&keyword_only)? && param.getattr("name")?.extract::()? == name + { + keyword = true; + } + } + + Ok(if positional > legacy_arity { + ExtraArg::Positional + } else if keyword { + ExtraArg::Keyword + } else { + ExtraArg::Omit + }) + }; + + inspect().unwrap_or(ExtraArg::Positional) +} + +/// Call `callback` with `port_id`, `payload` and, as `extra` decided, `value`. +fn call_with_extra<'py, V: IntoPyObject<'py>>( + py: Python<'py>, + callback: &Py, + port_id: u32, + payload: Bound<'py, PyBytes>, + extra: ExtraArg, + name: &str, + value: V, +) -> PyResult<()> { + match extra { + ExtraArg::Positional => { + callback.call1(py, (port_id, payload, value))?; + }, + ExtraArg::Keyword => { + let kwargs = PyDict::new(py); + kwargs.set_item(name, value)?; + callback.call(py, (port_id, payload), Some(&kwargs))?; + }, + ExtraArg::Omit => { + callback.call1(py, (port_id, payload))?; + }, + } + + Ok(()) +} + fn py_to_table_ref_async(val: &Bound<'_, PyAny>) -> PyResult { if let Ok(t) = val.cast::() { let table = t.borrow(); @@ -967,7 +1050,8 @@ impl AsyncView { /// indicating which port the update fired on, and in `"row"` mode also /// with `delta`, the new data that was updated for each row, and `reset`, /// which is `True` when the step began with `clear()` or `replace()` so - /// `delta` is relative to an empty table. + /// `delta` is relative to an empty table. A callback which does not + /// declare `reset`, positionally or as a keyword, is called without it. /// /// # Arguments /// @@ -979,6 +1063,7 @@ impl AsyncView { /// rows. Otherwise `delta` will be [`Option::None`]. #[pyo3(signature=(callback, mode=None))] pub async fn on_update(&self, callback: Py, mode: Option) -> PyResult { + let extra = Python::attach(|py| negotiate_extra_arg(py, &callback, 2, "reset")); let callback = move |x: OnUpdateData| { let callback = Python::attach(|py| Py::clone_ref(&callback, py)); async move { @@ -986,10 +1071,18 @@ impl AsyncView { let callback = Python::attach(|py| Py::clone_ref(&callback, py)); Python::attach(|py| { match &x.delta { - None => callback.call1(py, (x.port_id,))?, - Some(delta) => { - callback.call1(py, (x.port_id, PyBytes::new(py, delta), x.reset))? + None => { + callback.call1(py, (x.port_id,))?; }, + Some(delta) => call_with_extra( + py, + &callback, + x.port_id, + PyBytes::new(py, delta), + extra, + "reset", + x.reset, + )?, }; Ok(()) @@ -1024,34 +1117,49 @@ impl AsyncView { self.view.remove_update(callback_id).await.into_pyerr() } - /// Register a callback which is invoked whenever rows are removed from - /// this [`View`]'s [`Table`] by [`Table::remove`], with two arguments: - /// `port_id`, - /// and `indices`, the removed `index` column values as an Apache Arrow - /// (`bytes`) of one column named after the index. - /// - /// [`Table::replace`] reports the keys it does not re-supply and - /// [`Table::clear`] reports every key. It never fires for a - /// [`Table`] without an `index`. + /// Register a callback which is invoked whenever rows leave this [`View`] + /// with `port_id`, `indices` (their keys as an Apache Arrow (`bytes`) of + /// one column per key column named after it) and `cause`, one of + /// `"key"`, `"filter"` or `"group"`. + /// + /// For a [`View`] without `group_by` the keys are the `index` columns of + /// the [`Table`], which must have one: `"key"` when [`Table::remove`] + /// removed them ([`Table::replace`] reports the keys it does not + /// re-supply and [`Table::clear`] every key) and `"filter"` when + /// [`Table::update`] moved them outside the [`View`]'s `filter`. For a + /// [`View`] with `group_by` the keys are the `group_by` columns of each + /// group which no longer has any rows, `"group"`, `None` below the + /// group's depth. It fires once per cause per update step. A callback + /// which does not declare `cause`, positionally or as a keyword, is + /// called without it. /// /// # Python Examples /// /// ```python - /// def on_remove(port_id, indices): + /// def on_remove(port_id, indices, cause): /// replica.remove(indices) /// /// callback_id = await view.on_remove(on_remove) /// ``` pub async fn on_remove(&self, callback: Py) -> PyResult { + let extra = Python::attach(|py| negotiate_extra_arg(py, &callback, 2, "cause")); let callback = move |x: OnRemoveData| { let callback = Python::attach(|py| Py::clone_ref(&callback, py)); async move { let aggregate_errors: PyResult<()> = Python::attach(|py| { match &x.indices { - None => callback.call1(py, (x.port_id,))?, - Some(indices) => { - callback.call1(py, (x.port_id, PyBytes::new(py, indices)))? + None => { + callback.call1(py, (x.port_id,))?; }, + Some(indices) => call_with_extra( + py, + &callback, + x.port_id, + PyBytes::new(py, indices), + extra, + "cause", + x.cause.as_str(), + )?, }; Ok(()) diff --git a/rust/perspective-python/src/client/client_sync.rs b/rust/perspective-python/src/client/client_sync.rs index 1d08b7fe71..4c1d2ae95e 100644 --- a/rust/perspective-python/src/client/client_sync.rs +++ b/rust/perspective-python/src/client/client_sync.rs @@ -826,7 +826,8 @@ impl View { /// parameter. /// - `options` - If this is provided as `OnUpdateOptions { mode: /// Some(OnUpdateMode::Row) }`, then `delta` is an Arrow of the updated - /// rows. Otherwise `delta` will be [`Option::None`]. + /// rows, followed by `reset` when the callback declares it. Otherwise + /// `delta` will be [`Option::None`]. #[pyo3(signature = (callback, mode=None))] pub fn on_update( &self, @@ -855,20 +856,26 @@ impl View { self.0.remove_update(callback_id).py_block_on(py) } - /// Register a callback which is invoked whenever rows are removed from - /// this [`View`]'s [`Table`] by [`Table::remove`], with two arguments: - /// `port_id`, - /// and `indices`, the removed `index` column values as an Apache Arrow - /// (`bytes`) of one column named after the index. + /// Register a callback which is invoked whenever rows leave this [`View`] + /// with `port_id`, `indices` (their keys as an Apache Arrow (`bytes`) of + /// one column per key column named after it) and `cause`, one of + /// `"key"`, `"filter"` or `"group"`. /// - /// [`Table::replace`] reports the keys it does not re-supply and - /// [`Table::clear`] reports every key. It never fires for a - /// [`Table`] without an `index`. + /// For a [`View`] without `group_by` the keys are the `index` columns of + /// the [`Table`], which must have one: `"key"` when [`Table::remove`] + /// removed them ([`Table::replace`] reports the keys it does not + /// re-supply and [`Table::clear`] every key) and `"filter"` when + /// [`Table::update`] moved them outside the [`View`]'s `filter`. For a + /// [`View`] with `group_by` the keys are the `group_by` columns of each + /// group which no longer has any rows, `"group"`, `None` below the + /// group's depth. It fires once per cause per update step. A callback + /// which does not declare `cause`, positionally or as a keyword, is + /// called without it. /// /// # Python Examples /// /// ```python - /// def on_remove(port_id, indices): + /// def on_remove(port_id, indices, cause): /// replica.remove(indices) /// /// callback_id = view.on_remove(on_remove) diff --git a/rust/perspective-server/cpp/perspective/src/cpp/context_one.cpp b/rust/perspective-server/cpp/perspective/src/cpp/context_one.cpp index f19488264b..e120ba56bd 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/context_one.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/context_one.cpp @@ -518,7 +518,7 @@ t_ctx1::get_row_delta() { PSP_VERBOSE_ASSERT(m_init, "touching uninited object"); std::vector rows = get_rows_changed(); std::vector data = get_data(rows); - t_rowdelta rval(m_rows_changed, rows.size(), data); + t_rowdelta rval(m_rows_changed, rows, data); m_tree->clear_deltas(); return rval; } diff --git a/rust/perspective-server/cpp/perspective/src/cpp/context_two.cpp b/rust/perspective-server/cpp/perspective/src/cpp/context_two.cpp index 574f28587e..8e0bcf59b0 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/context_two.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/context_two.cpp @@ -1131,7 +1131,7 @@ t_rowdelta t_ctx2::get_row_delta() { std::vector rows = get_rows_changed(); std::vector data = get_data(rows); - t_rowdelta rval(true, rows.size(), data); + t_rowdelta rval(true, rows, data); clear_deltas(); return rval; } diff --git a/rust/perspective-server/cpp/perspective/src/cpp/context_zero.cpp b/rust/perspective-server/cpp/perspective/src/cpp/context_zero.cpp index bb02d87466..cffbd6c0c9 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/context_zero.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/context_zero.cpp @@ -61,6 +61,7 @@ t_ctx0::step_begin() { m_deltas = std::make_shared(); m_delta_pkeys.clear(); + m_filtered_out_pkeys.clear(); m_rows_changed = false; m_columns_changed = false; m_traversal->step_begin(); @@ -167,6 +168,9 @@ t_ctx0::notify( ); } else { m_traversal->delete_row(pkey); + if (m_features[CTX_FEAT_REMOVES]) { + m_filtered_out_pkeys.insert(pkey); + } } } else { if (filter_curr) { @@ -955,6 +959,23 @@ t_ctx0::set_deltas_enabled(bool enabled_state) { m_features[CTX_FEAT_DELTA] = enabled_state; } +void +t_ctx0::set_removes_enabled(bool enabled_state) { + m_features[CTX_FEAT_REMOVES] = enabled_state; + if (!enabled_state) { + m_filtered_out_pkeys.clear(); + } +} + +std::vector +t_ctx0::take_filtered_out_pkeys() { + std::vector pkeys( + m_filtered_out_pkeys.begin(), m_filtered_out_pkeys.end() + ); + m_filtered_out_pkeys.clear(); + return pkeys; +} + std::vector t_ctx0::get_trees() { return {}; diff --git a/rust/perspective-server/cpp/perspective/src/cpp/data_slice.cpp b/rust/perspective-server/cpp/perspective/src/cpp/data_slice.cpp index 7d0ddbb8d6..d4fc5661f0 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/data_slice.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/data_slice.cpp @@ -155,6 +155,18 @@ t_data_slice::get_column_indices() const { return m_column_indices; } +template +const std::vector& +t_data_slice::get_row_indices() const { + return m_row_indices; +} + +template +void +t_data_slice::set_row_indices(const std::vector& row_indices) { + m_row_indices = row_indices; +} + template bool t_data_slice::is_column_only() const { diff --git a/rust/perspective-server/cpp/perspective/src/cpp/data_table.cpp b/rust/perspective-server/cpp/perspective/src/cpp/data_table.cpp index 213ac72884..d3f019edf7 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/data_table.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/data_table.cpp @@ -803,6 +803,11 @@ t_data_table::promote_column( if (fill) { for (auto i = 0; i < iter_limit; ++i) { + if (current_col->is_status_enabled() && !current_col->is_valid(i)) { + promoted_col->clear(i); + continue; + } + switch (new_dtype) { case DTYPE_INT64: { auto* val = current_col->get_nth(i); diff --git a/rust/perspective-server/cpp/perspective/src/cpp/server.cpp b/rust/perspective-server/cpp/perspective/src/cpp/server.cpp index 6f17aacbed..dabf589495 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/server.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/server.cpp @@ -878,7 +878,9 @@ ServerResources::table_has_on_remove_subs(const t_id& table_id) { auto range = m_table_to_view.equal_range(table_id); for (auto it = range.first; it != range.second; ++it) { auto subs = m_view_on_remove_subs.find(it->second); - if (subs != m_view_on_remove_subs.end() && !subs->second.empty()) { + auto view = m_views.find(it->second); + if (subs != m_view_on_remove_subs.end() && !subs->second.empty() + && view != m_views.end() && view->second->sides() == 0) { return true; } } @@ -3166,9 +3168,10 @@ ProtoServer::_handle_request(std::uint32_t client_id, Request&& req) { sub_info.id = req.msg_id(); sub_info.client_id = client_id; m_resources.create_view_on_remove_sub(req.entity_id(), sub_info); + m_resources.get_view(req.entity_id())->set_removes_enabled(true); auto table_id = m_resources.get_table_id_for_view(req.entity_id()); m_resources.get_table(table_id)->get_gnode()->set_removes_enabled( - true + m_resources.table_has_on_remove_subs(table_id) ); break; } @@ -3177,6 +3180,12 @@ ProtoServer::_handle_request(std::uint32_t client_id, Request&& req) { m_resources.remove_view_on_remove_sub( req.entity_id(), sub_id, client_id ); + const auto& view_id = req.entity_id(); + if (m_resources.has_view(view_id) + && m_resources.get_view_on_remove_sub(view_id).empty()) { + m_resources.get_view(view_id)->set_removes_enabled(false); + } + auto table_id = m_resources.get_table_id_for_view(req.entity_id()); m_resources.get_table(table_id)->get_gnode()->set_removes_enabled( m_resources.table_has_on_remove_subs(table_id) @@ -4008,37 +4017,14 @@ 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; + std::vector pkeys; + pkeys.reserve(removed->size()); 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]); - } + pkeys.push_back(pkey->get_scalar(ridx)); } - return apachearrow::table_to_arrow_ipc(keys); + return pkeys_to_index_arrow(*table, pkeys); } void @@ -4061,23 +4047,41 @@ ProtoServer::_notify_table( } auto view = m_resources.get_view(view_id); - if (has_removes) { - auto remove_subs = m_resources.get_view_on_remove_sub(view_id); - if (!remove_subs.empty() && !removed_indices) { - removed_indices = removed_keys_to_arrow(table, removed); + auto remove_subs = m_resources.get_view_on_remove_sub(view_id); + if (!remove_subs.empty()) { + const bool flat = view->sides() == 0; + auto emit = [&](const std::shared_ptr& indices, + proto::ViewOnRemoveResp::Cause cause) { + for (auto& subscription : remove_subs) { + Response out; + out.set_msg_id(subscription.id); + out.set_entity_id(view_id); + auto* r = out.mutable_view_on_remove_resp(); + r->set_port_id(port_id); + r->set_cause(cause); + *r->mutable_indices() = *indices; + ProtoServerResp resp2; + resp2.data = std::move(out); + resp2.client_id = subscription.client_id; + outs.emplace_back(std::move(resp2)); + } + }; + + if (flat && has_removes) { + if (!removed_indices) { + removed_indices = removed_keys_to_arrow(table, removed); + } + + emit(removed_indices, proto::ViewOnRemoveResp::KEY); } - for (auto& subscription : remove_subs) { - Response out; - out.set_msg_id(subscription.id); - out.set_entity_id(view_id); - auto* r = out.mutable_view_on_remove_resp(); - r->set_port_id(port_id); - *r->mutable_indices() = *removed_indices; - ProtoServerResp resp2; - resp2.data = std::move(out); - resp2.client_id = subscription.client_id; - outs.emplace_back(std::move(resp2)); + auto left = view->take_removed_rows(); + if (left) { + emit( + left, + flat ? proto::ViewOnRemoveResp::FILTER + : proto::ViewOnRemoveResp::GROUP + ); } } diff --git a/rust/perspective-server/cpp/perspective/src/cpp/sparse_tree.cpp b/rust/perspective-server/cpp/perspective/src/cpp/sparse_tree.cpp index f18fd259dc..170378ec05 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/sparse_tree.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/sparse_tree.cpp @@ -214,6 +214,19 @@ t_stree::get_sortby_value(t_index idx) const { namespace { +/// Removes every non-valid scalar so order statistics see only real values. +void +drop_invalid(std::vector& values) { + values.erase( + std::remove_if( + values.begin(), + values.end(), + [](const t_tscalar& value) { return !value.is_valid(); } + ), + values.end() + ); +} + /// The strand delta dtype for a source column, signed for unsigned sources. t_dtype strand_value_dtype(t_dtype dtype) { @@ -1057,7 +1070,9 @@ t_stree::update_shape_from_static(const t_dtree_ctx& ctx) { m_newids.insert(sptidx); for (const auto& capture : m_captures) { - capture->m_created.push_back(sptidx); + if (!capture->m_dropped_only) { + capture->m_created.push_back(sptidx); + } } if (ndepth == dtree.last_level()) { @@ -1825,6 +1840,13 @@ t_stree::update_agg_table( values ); + drop_invalid(values); + if (values.empty()) { + new_value.set(mknull(dst->get_dtype())); + dst->set_scalar(dst_ridx, new_value); + break; + } + new_value.set(*std::max_element(values.begin(), values.end())); dst->set_scalar(dst_ridx, new_value); } break; @@ -1885,6 +1907,13 @@ t_stree::update_agg_table( values ); + drop_invalid(values); + if (values.empty()) { + new_value.set(mknull(dst->get_dtype())); + dst->set_scalar(dst_ridx, new_value); + break; + } + new_value.set(*std::min_element(values.begin(), values.end())); dst->set_scalar(dst_ridx, new_value); } break; @@ -1932,10 +1961,15 @@ t_stree::update_agg_table( t_tscalar dst_scalar = dst->get_scalar(dst_ridx); old_value.set(dst_scalar); - new_value.set(src_scalar); - if (dst_scalar.is_valid()) { + if (!src_scalar.is_valid()) { + new_value.set( + is_new ? mknull(dst->get_dtype()) : dst_scalar + ); + } else if (!is_new && dst_scalar.is_valid()) { new_value.set(std::max(dst_scalar, src_scalar)); + } else { + new_value.set(src_scalar); } dst->set_scalar(dst_ridx, new_value); @@ -1945,11 +1979,17 @@ t_stree::update_agg_table( t_tscalar dst_scalar = dst->get_scalar(dst_ridx); old_value.set(dst_scalar); - new_value.set(src_scalar); - if (dst_scalar.is_valid()) { + if (!src_scalar.is_valid()) { + new_value.set( + is_new ? mknull(dst->get_dtype()) : dst_scalar + ); + } else if (!is_new && dst_scalar.is_valid()) { new_value.set(std::min(dst_scalar, src_scalar)); + } else { + new_value.set(src_scalar); } + dst->set_scalar(dst_ridx, new_value); } break; case AGGTYPE_HIGH_MINUS_LOW: { @@ -1964,6 +2004,14 @@ t_stree::update_agg_table( pkeys, values ); + + drop_invalid(values); + if (values.empty()) { + new_value.set(mknull(dst->get_dtype())); + dst->set_scalar(dst_ridx, new_value); + break; + } + auto low_high = std::minmax_element(values.begin(), values.end()); t_tscalar first; @@ -2327,7 +2375,7 @@ t_stree::update_agg_table( )); } - if (!m_captures.empty() + if (m_captures_want_changes && (val_neq || old_value.is_valid() != dst->is_valid(dst_ridx))) { capture_change(nidx, idx, old_value); } @@ -2811,6 +2859,7 @@ t_stree::get_aggregate(t_index idx, t_index aggnum) const { t_tscalar t_stree::get_aggregate_median(std::vector& values, float n) const { + drop_invalid(values); const auto size = values.size(); const float raw_size = static_cast(size) * n; bool is_even_size = fabsf(roundf(raw_size) - raw_size) <= 0.00001; @@ -2878,6 +2927,8 @@ t_stree::set_deltas_enabled(bool enabled_state) { void t_stree::add_capture(const std::shared_ptr& capture) { m_captures.push_back(capture); + m_captures_want_changes = + m_captures_want_changes || !capture->m_dropped_only; } void @@ -2886,6 +2937,11 @@ t_stree::remove_capture(const std::shared_ptr& capture) { std::remove(m_captures.begin(), m_captures.end(), capture), m_captures.end() ); + m_captures_want_changes = std::any_of( + m_captures.begin(), + m_captures.end(), + [](const auto& remaining) { return !remaining->m_dropped_only; } + ); } bool @@ -2912,6 +2968,10 @@ t_stree::capture_change( t_uindex nidx, t_uindex aggnum, const t_tscalar& old_value ) { for (const auto& capture : m_captures) { + if (capture->m_dropped_only) { + continue; + } + auto& cells = capture->m_changed[nidx]; bool seen = false; for (const auto& cell : cells) { @@ -2932,9 +2992,11 @@ t_stree::capture_dropped(const t_stnode& node) { dropped.m_depth = node.m_depth; get_path(node.m_idx, dropped.m_path); std::reverse(dropped.m_path.begin(), dropped.m_path.end()); - dropped.m_aggregates.reserve(m_aggcols.size()); - for (t_uindex aggnum = 0; aggnum < m_aggcols.size(); ++aggnum) { - dropped.m_aggregates.push_back(get_aggregate(node.m_idx, aggnum)); + if (m_captures_want_changes) { + dropped.m_aggregates.reserve(m_aggcols.size()); + for (t_uindex aggnum = 0; aggnum < m_aggcols.size(); ++aggnum) { + dropped.m_aggregates.push_back(get_aggregate(node.m_idx, aggnum)); + } } for (const auto& capture : m_captures) { diff --git a/rust/perspective-server/cpp/perspective/src/cpp/step_delta.cpp b/rust/perspective-server/cpp/perspective/src/cpp/step_delta.cpp index 5b7bebf981..14fcf8e5dc 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/step_delta.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/step_delta.cpp @@ -68,6 +68,16 @@ t_rowdelta::t_rowdelta( rows_changed(rows_changed), num_rows_changed(num_rows_changed), data(data) {} + +t_rowdelta::t_rowdelta( + bool rows_changed, + const std::vector& rows, + const std::vector& data +) : + rows_changed(rows_changed), + num_rows_changed(rows.size()), + data(data), + rows(rows) {} } // end namespace perspective namespace std { diff --git a/rust/perspective-server/cpp/perspective/src/cpp/view.cpp b/rust/perspective-server/cpp/perspective/src/cpp/view.cpp index f6267b6279..6e7b6943e5 100644 --- a/rust/perspective-server/cpp/perspective/src/cpp/view.cpp +++ b/rust/perspective-server/cpp/perspective/src/cpp/view.cpp @@ -12,6 +12,14 @@ #include #include +#include + +namespace { +template +constexpr bool has_row_tree = + std::is_same_v + || std::is_same_v; +} #include #include #include @@ -91,6 +99,7 @@ View::View( template View::~View() { + detach_removes_capture(); auto pool = m_table->get_pool(); auto gnode = m_table->get_gnode(); PSP_GIL_UNLOCK(); @@ -873,6 +882,17 @@ View::data_slice_to_batches( auto num_output_row_paths = is_column_only() ? 0 : num_row_paths; + const std::vector& row_indices = data_slice->get_row_indices(); + auto row_path_value = [&](t_uindex rpidx, t_uindex ridx) -> t_tscalar { + t_uindex vidx = row_indices.empty() ? ridx : row_indices.at(ridx); + auto depth = m_ctx->unity_get_row_depth(vidx); + if (rpidx < depth) { + return m_ctx->unity_get_row_path(vidx).at((depth - 1) - rpidx); + } + + return mknone(); + }; + t_uindex write_idx = 0; if (emit_group_by && num_output_row_paths > 0) { auto schema = m_table->get_schema(); @@ -908,14 +928,8 @@ View::data_slice_to_batches( vectors[write_idx] = apachearrow::numeric_col_to_array< arrow::Int8Type, std::int8_t>(extents, [&, rpidx](t_uindex ridx) { - auto depth = m_ctx->unity_get_row_depth(ridx); - if (rpidx < depth) { - return m_ctx->unity_get_row_path(ridx).at( - (depth - 1) - rpidx - ); - } - return mknone(); - }); + return row_path_value(rpidx, ridx); + }); } break; case DTYPE_UINT8: { fields[write_idx] = @@ -923,14 +937,8 @@ View::data_slice_to_batches( vectors[write_idx] = apachearrow::numeric_col_to_array< arrow::UInt8Type, std::uint8_t>(extents, [&, rpidx](t_uindex ridx) { - auto depth = m_ctx->unity_get_row_depth(ridx); - if (rpidx < depth) { - return m_ctx->unity_get_row_path(ridx).at( - (depth - 1) - rpidx - ); - } - return mknone(); - }); + return row_path_value(rpidx, ridx); + }); } break; case DTYPE_INT16: { fields[write_idx] = @@ -938,14 +946,8 @@ View::data_slice_to_batches( vectors[write_idx] = apachearrow::numeric_col_to_array< arrow::Int16Type, std::int16_t>(extents, [&, rpidx](t_uindex ridx) { - auto depth = m_ctx->unity_get_row_depth(ridx); - if (rpidx < depth) { - return m_ctx->unity_get_row_path(ridx).at( - (depth - 1) - rpidx - ); - } - return mknone(); - }); + return row_path_value(rpidx, ridx); + }); } break; case DTYPE_UINT16: { fields[write_idx] = @@ -953,14 +955,8 @@ View::data_slice_to_batches( vectors[write_idx] = apachearrow::numeric_col_to_array< arrow::UInt16Type, std::uint16_t>(extents, [&, rpidx](t_uindex ridx) { - auto depth = m_ctx->unity_get_row_depth(ridx); - if (rpidx < depth) { - return m_ctx->unity_get_row_path(ridx).at( - (depth - 1) - rpidx - ); - } - return mknone(); - }); + return row_path_value(rpidx, ridx); + }); } break; case DTYPE_INT32: { fields[write_idx] = @@ -968,14 +964,8 @@ View::data_slice_to_batches( vectors[write_idx] = apachearrow::numeric_col_to_array< arrow::Int32Type, std::int32_t>(extents, [&, rpidx](t_uindex ridx) { - auto depth = m_ctx->unity_get_row_depth(ridx); - if (rpidx < depth) { - return m_ctx->unity_get_row_path(ridx).at( - (depth - 1) - rpidx - ); - } - return mknone(); - }); + return row_path_value(rpidx, ridx); + }); } break; case DTYPE_UINT32: { fields[write_idx] = @@ -983,14 +973,8 @@ View::data_slice_to_batches( vectors[write_idx] = apachearrow::numeric_col_to_array< arrow::UInt32Type, std::uint32_t>(extents, [&, rpidx](t_uindex ridx) { - auto depth = m_ctx->unity_get_row_depth(ridx); - if (rpidx < depth) { - return m_ctx->unity_get_row_path(ridx).at( - (depth - 1) - rpidx - ); - } - return mknone(); - }); + return row_path_value(rpidx, ridx); + }); } break; case DTYPE_INT64: { fields[write_idx] = @@ -998,14 +982,8 @@ View::data_slice_to_batches( vectors[write_idx] = apachearrow::numeric_col_to_array< arrow::Int64Type, std::int64_t>(extents, [&, rpidx](t_uindex ridx) { - auto depth = m_ctx->unity_get_row_depth(ridx); - if (rpidx < depth) { - return m_ctx->unity_get_row_path(ridx).at( - (depth - 1) - rpidx - ); - } - return mknone(); - }); + return row_path_value(rpidx, ridx); + }); } break; case DTYPE_UINT64: { fields[write_idx] = @@ -1013,14 +991,8 @@ View::data_slice_to_batches( vectors[write_idx] = apachearrow::numeric_col_to_array< arrow::UInt64Type, std::uint64_t>(extents, [&, rpidx](t_uindex ridx) { - auto depth = m_ctx->unity_get_row_depth(ridx); - if (rpidx < depth) { - return m_ctx->unity_get_row_path(ridx).at( - (depth - 1) - rpidx - ); - } - return mknone(); - }); + return row_path_value(rpidx, ridx); + }); } break; case DTYPE_FLOAT32: { fields[write_idx] = @@ -1028,14 +1000,8 @@ View::data_slice_to_batches( vectors[write_idx] = apachearrow::numeric_col_to_array< arrow::FloatType, float>(extents, [&, rpidx](t_uindex ridx) { - auto depth = m_ctx->unity_get_row_depth(ridx); - if (rpidx < depth) { - return m_ctx->unity_get_row_path(ridx).at( - (depth - 1) - rpidx - ); - } - return mknone(); - }); + return row_path_value(rpidx, ridx); + }); } break; case DTYPE_FLOAT64: { fields[write_idx] = @@ -1043,14 +1009,8 @@ View::data_slice_to_batches( vectors[write_idx] = apachearrow::numeric_col_to_array< arrow::DoubleType, double>(extents, [&, rpidx](t_uindex ridx) { - auto depth = m_ctx->unity_get_row_depth(ridx); - if (rpidx < depth) { - return m_ctx->unity_get_row_path(ridx).at( - (depth - 1) - rpidx - ); - } - return mknone(); - }); + return row_path_value(rpidx, ridx); + }); } break; case DTYPE_DATE: { fields[write_idx] = @@ -1058,13 +1018,7 @@ View::data_slice_to_batches( vectors[write_idx] = apachearrow::date_col_to_array( extents, [&, rpidx](t_uindex ridx) { - auto depth = m_ctx->unity_get_row_depth(ridx); - if (rpidx < depth) { - return m_ctx->unity_get_row_path(ridx).at( - (depth - 1) - rpidx - ); - } - return mknone(); + return row_path_value(rpidx, ridx); } ); } break; @@ -1075,13 +1029,7 @@ View::data_slice_to_batches( vectors[write_idx] = apachearrow::timestamp_col_to_array( extents, [&, rpidx](t_uindex ridx) { - auto depth = m_ctx->unity_get_row_depth(ridx); - if (rpidx < depth) { - return m_ctx->unity_get_row_path(ridx).at( - (depth - 1) - rpidx - ); - } - return mknone(); + return row_path_value(rpidx, ridx); } ); } break; @@ -1091,13 +1039,7 @@ View::data_slice_to_batches( vectors[write_idx] = apachearrow::boolean_col_to_array( extents, [&, rpidx](t_uindex ridx) { - auto depth = m_ctx->unity_get_row_depth(ridx); - if (rpidx < depth) { - return m_ctx->unity_get_row_path(ridx).at( - (depth - 1) - rpidx - ); - } - return mknone(); + return row_path_value(rpidx, ridx); } ); } break; @@ -1110,14 +1052,8 @@ View::data_slice_to_batches( apachearrow::string_col_to_dictionary_array( extents, [&, rpidx](t_uindex ridx) { - auto depth = m_ctx->unity_get_row_depth(ridx); - if (rpidx < depth) { - return m_ctx->unity_get_row_path(ridx).at( - (depth - 1) - rpidx - ); - } - return mknone(); - } + return row_path_value(rpidx, ridx); + } ); } break; case DTYPE_OBJECT: @@ -1503,6 +1439,192 @@ View::_set_deltas_enabled(bool enabled_state) { m_ctx->set_deltas_enabled(enabled_state); } +template +void +View::set_removes_enabled(bool enabled_state) { + m_removes_enabled = enabled_state; + if constexpr (std::is_same_v) { + m_ctx->set_removes_enabled(enabled_state); + } else if constexpr (has_row_tree) { + if (enabled_state) { + attach_removes_capture(); + } else { + detach_removes_capture(); + } + } +} + +template +std::shared_ptr +View::take_removed_rows() { + if (!m_removes_enabled) { + return nullptr; + } + + if constexpr (std::is_same_v) { + auto pkeys = m_ctx->take_filtered_out_pkeys(); + if (pkeys.empty()) { + return nullptr; + } + + return pkeys_to_index_arrow(*m_table, pkeys); + } else if constexpr (has_row_tree) { + if (m_removes_generation != m_ctx->get_storage_generation()) { + attach_removes_capture(); + return nullptr; + } + + std::vector dropped; + std::swap(dropped, m_removes_capture->m_dropped); + return dropped_to_arrow(dropped); + } else { + return nullptr; + } +} + +template +void +View::attach_removes_capture() { + detach_removes_capture(); + if constexpr (has_row_tree) { + auto trees = m_ctx->get_trees(); + if (trees.empty()) { + return; + } + + m_removes_capture = std::make_shared(); + m_removes_capture->m_dropped_only = true; + trees[0]->add_capture(m_removes_capture); + m_removes_generation = m_ctx->get_storage_generation(); + } +} + +template +void +View::detach_removes_capture() { + if (!m_removes_capture) { + return; + } + + if constexpr (has_row_tree) { + auto trees = m_ctx->get_trees(); + if (!trees.empty()) { + trees[0]->remove_capture(m_removes_capture); + } + } + + m_removes_capture = nullptr; +} + +template +t_dtype +View::row_pivot_dtype(const std::string& name) const { + const auto& schema = m_table->get_schema(); + if (schema.has_column(name)) { + return schema.get_dtype(name); + } + + for (const auto& expr : m_expressions) { + if (expr->get_expression_alias() == name) { + return expr->get_dtype(); + } + } + + return DTYPE_STR; +} + +template +std::shared_ptr +View::dropped_to_arrow(const std::vector& dropped +) const { + const t_uindex num_pivots = m_row_pivots.size(); + std::vector rows; + rows.reserve(dropped.size()); + for (const auto& node : dropped) { + bool selected = node.m_depth > 0; + if (m_view_config->is_total_only()) { + selected = false; + } else if (m_view_config->is_leaves_only()) { + selected = node.m_depth == num_pivots; + } + + if (selected) { + rows.push_back(&node); + } + } + + if (rows.empty()) { + return nullptr; + } + + std::vector dtypes; + dtypes.reserve(num_pivots); + for (const auto& name : m_row_pivots) { + dtypes.push_back(row_pivot_dtype(name)); + } + + t_schema schema(m_row_pivots, dtypes); + t_data_table keys(schema); + keys.init(); + keys.extend(rows.size()); + for (t_uindex rpidx = 0; rpidx < num_pivots; ++rpidx) { + auto column = keys.get_column(m_row_pivots[rpidx]); + for (t_uindex ridx = 0; ridx < rows.size(); ++ridx) { + const t_stree_dropped& node = *rows[ridx]; + if (rpidx < node.m_depth) { + column->set_scalar(ridx, node.m_path[rpidx]); + } else { + column->set_scalar(ridx, mknull(dtypes[rpidx])); + } + } + } + + return apachearrow::table_to_arrow_ipc(keys); +} + +std::shared_ptr +pkeys_to_index_arrow(const Table& table, const std::vector& pkeys) { + const auto& index = table.get_index(); + std::vector dtypes; + if (index.size() == 1) { + dtypes.push_back(table.get_schema().get_dtype(index[0])); + } else { + dtypes = table.get_pkey_codec()->dtypes(); + } + + t_schema schema(index, dtypes); + t_data_table keys(schema); + keys.init(); + keys.extend(pkeys.size()); + std::vector> cols; + cols.reserve(index.size()); + for (const auto& name : index) { + cols.push_back(keys.get_column(name)); + } + + if (index.size() == 1) { + for (t_uindex ridx = 0; ridx < pkeys.size(); ++ridx) { + cols[0]->set_scalar(ridx, pkeys[ridx]); + } + + return apachearrow::table_to_arrow_ipc(keys); + } + + const auto& codec = *table.get_pkey_codec(); + std::vector storage; + for (t_uindex ridx = 0; ridx < pkeys.size(); ++ridx) { + const auto& key = pkeys[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); +} + // Pivot table operations template bool @@ -1704,7 +1826,7 @@ View::get_row_delta() const { // the eventual Arrow output for this path. t_uindex ncols = paths.size(); - return std::make_shared>( + auto slice = std::make_shared>( m_ctx, 0, num_rows_changed, @@ -1715,6 +1837,9 @@ View::get_row_delta() const { data, paths ); + + slice->set_row_indices(delta.rows); + return slice; } template diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/aggregate.h b/rust/perspective-server/cpp/perspective/src/include/perspective/aggregate.h index 22b1e95b6d..db40b8e105 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/aggregate.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/aggregate.h @@ -32,6 +32,8 @@ class PERSPECTIVE_EXPORT t_aggimpl { typedef ROLLING_T t_rolling; typedef RESULT_T t_result; + static constexpr bool skip_invalid = false; + RESULT_T value(ROLLING_T rs); ROLLING_T reduce(const RAW_DATA_T* biter, const RAW_DATA_T* eiter); ROLLING_T roll_up(const ROLLING_T* biter, const ROLLING_T* eiter); @@ -157,6 +159,8 @@ template class PERSPECTIVE_EXPORT t_aggimpl_hwm : public t_aggimpl { public: + static constexpr bool skip_invalid = true; + RESULT_T value(ROLLING_T rs) { return RESULT_T(rs); } @@ -181,6 +185,8 @@ template class PERSPECTIVE_EXPORT t_aggimpl_lwm : public t_aggimpl { public: + static constexpr bool skip_invalid = true; + RESULT_T value(ROLLING_T rs) { return RESULT_T(rs); } @@ -236,6 +242,30 @@ build_aggregate_helper( ) { typedef typename AGGIMPL_T::t_rolling t_rolling; const t_rolling* base = ocolumn->get_nth_base(); + + if constexpr (AGGIMPL_T::skip_invalid) { + if (ocolumn->is_status_enabled()) { + std::vector sbuf; + sbuf.reserve(ecidx - bcidx); + for (t_index cidx = bcidx; cidx < ecidx; ++cidx) { + const auto uidx = static_cast(cidx); + if (*ocolumn->get_nth_status(uidx) == STATUS_VALID) { + sbuf.push_back(base[cidx]); + } + } + + if (sbuf.empty()) { + ocolumn->set_valid(nidx, false); + return; + } + + t_rolling rolling = + aggimpl.roll_up(sbuf.data(), sbuf.data() + sbuf.size()); + ocolumn->set_nth(nidx, rolling); + return; + } + } + const t_rolling* biter = base + bcidx; const t_rolling* eiter = base + ecidx; t_rolling rolling = aggimpl.roll_up(biter, eiter); @@ -257,19 +287,30 @@ build_aggregate_helper( ) { const t_column* c_ocolumn = ocolumn; typedef typename AGGIMPL_T::t_rolling t_rolling; - std::vector sbuf(ecidx - bcidx); + const bool skip_invalid = + AGGIMPL_T::skip_invalid && ocolumn->is_status_enabled(); + std::vector sbuf; + sbuf.reserve(ecidx - bcidx); for (t_index sidx = bcidx; sidx < ecidx; ++sidx) { + const auto uidx = static_cast(sidx); + if (skip_invalid && *c_ocolumn->get_nth_status(uidx) != STATUS_VALID) { + continue; + } - auto tmpv = c_ocolumn->get_nth(static_cast(sidx)); - sbuf[sidx - bcidx] = tmpv; + sbuf.push_back(c_ocolumn->get_nth(uidx)); } - if (!sbuf.empty()) { + if (sbuf.empty()) { + if (skip_invalid) { + ocolumn->set_valid(nidx, false); + } - t_rolling rolling = - aggimpl.roll_up(sbuf.data(), sbuf.data() + sbuf.size()); - ocolumn->set_nth(nidx, rolling); + return; } + + t_rolling rolling = + aggimpl.roll_up(sbuf.data(), sbuf.data() + sbuf.size()); + ocolumn->set_nth(nidx, rolling); } template @@ -329,6 +370,24 @@ t_aggregate::build_aggregate() { t_raw_data* biter = &buffer[0]; t_raw_data* eiter = biter + leaf_count; + + if constexpr (AGGIMPL_T::skip_invalid) { + if (icptr->is_status_enabled()) { + eiter = biter; + for (t_uindex lidx = 0; lidx < leaf_count; ++lidx) { + if (*icptr->get_nth_status(blptr[lidx]) + == STATUS_VALID) { + *eiter++ = buffer[lidx]; + } + } + + if (biter == eiter && ocolumn->is_status_enabled()) { + ocolumn->set_valid(nidx, false); + continue; + } + } + } + auto tmp = aggimpl.reduce(biter, eiter); ocolumn->set_nth(nidx, tmp); } diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/base.h b/rust/perspective-server/cpp/perspective/src/include/perspective/base.h index 4c47fcb60f..ee092cf2df 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/base.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/base.h @@ -400,6 +400,7 @@ enum t_ctx_feature { CTX_FEAT_DELTA, CTX_FEAT_ALERT, CTX_FEAT_ENABLED, + CTX_FEAT_REMOVES, CTX_FEAT_LAST_FEATURE }; diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/context_zero.h b/rust/perspective-server/cpp/perspective/src/include/perspective/context_zero.h index 34000c6b3a..ec44c31b3f 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/context_zero.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/context_zero.h @@ -55,6 +55,17 @@ class PERSPECTIVE_EXPORT t_ctx0 : public t_ctxbase { */ std::vector get_member_pkeys() const; + /** + * @brief Record the keys of rows an update moves outside this context's + * filter, for `take_filtered_out_pkeys`. + */ + void set_removes_enabled(bool enabled_state); + + /** + * @brief The keys recorded by `set_removes_enabled` since the last call. + */ + std::vector take_filtered_out_pkeys(); + void sort_by(); std::vector get_sort_by() const; @@ -116,6 +127,7 @@ class PERSPECTIVE_EXPORT t_ctx0 : public t_ctxbase { std::shared_ptr m_traversal; std::shared_ptr m_deltas; tsl::hopscotch_set m_delta_pkeys; + tsl::hopscotch_set m_filtered_out_pkeys; std::shared_ptr m_expression_tables; std::shared_ptr m_window_engine; t_symtable m_symtable; diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/data_slice.h b/rust/perspective-server/cpp/perspective/src/include/perspective/data_slice.h index b552a4947c..3c27bde4be 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/data_slice.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/data_slice.h @@ -111,6 +111,13 @@ class PERSPECTIVE_EXPORT t_data_slice { const std::vector& get_slice() const; const std::vector>& get_column_names() const; const std::vector& get_column_indices() const; + + /** + * @brief The view row index of each slice row when the slice is a row + * delta, empty when slice rows are view rows. + */ + const std::vector& get_row_indices() const; + void set_row_indices(const std::vector& row_indices); t_get_data_extents get_data_extents() const; t_uindex get_stride() const; t_uindex num_rows() const; @@ -141,5 +148,6 @@ class PERSPECTIVE_EXPORT t_data_slice { std::vector m_slice; std::vector> m_column_names; std::vector m_column_indices; + std::vector m_row_indices; }; } // end namespace perspective \ No newline at end of file diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/server.h b/rust/perspective-server/cpp/perspective/src/include/perspective/server.h index 7d28b27103..a4f782c450 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/server.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/server.h @@ -251,6 +251,10 @@ namespace server { [[nodiscard]] virtual bool get_deltas_enabled() const = 0; + virtual void set_removes_enabled(bool enabled_state) = 0; + [[nodiscard]] + virtual std::shared_ptr take_removed_rows() = 0; + virtual t_index collapse(std::int32_t row_idx) = 0; virtual t_index expand(std::int32_t row_idx) = 0; @@ -459,7 +463,7 @@ namespace server { get_row_delta_as_arrow() const override { auto delta = m_view->get_row_delta(); return m_view->data_slice_to_arrow( - delta, false, t_arrow_compression::NONE + delta, true, t_arrow_compression::NONE ); } @@ -474,6 +478,17 @@ namespace server { return m_view->get_context()->get_deltas_enabled(); } + void + set_removes_enabled(bool enabled_state) override { + m_view->set_removes_enabled(enabled_state); + } + + [[nodiscard]] + std::shared_ptr + take_removed_rows() override { + return m_view->take_removed_rows(); + } + t_index collapse(std::int32_t row_idx) override { return m_view->collapse(row_idx); diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/sparse_tree.h b/rust/perspective-server/cpp/perspective/src/include/perspective/sparse_tree.h index 1be5e3579a..f45245b59b 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/sparse_tree.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/sparse_tree.h @@ -172,6 +172,10 @@ struct PERSPECTIVE_EXPORT t_stree_capture { m_changed; std::vector m_created; std::vector m_dropped; + + /// Whether only `m_dropped` is wanted, so changes and creations are + /// not recorded and dropped nodes carry no aggregates. + bool m_dropped_only = false; }; struct PERSPECTIVE_EXPORT t_agg_update_info { @@ -549,6 +553,7 @@ class PERSPECTIVE_EXPORT t_stree { t_symtable m_symtable; bool m_has_delta; std::vector> m_captures; + bool m_captures_want_changes = false; std::string m_grand_agg_str; // Used by AGGTYPE_GMV under split_by. For t_ctx1 (group_by only) the diff --git a/rust/perspective-server/cpp/perspective/src/include/perspective/step_delta.h b/rust/perspective-server/cpp/perspective/src/include/perspective/step_delta.h index 53e8f7083b..0e49ab4b7e 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/step_delta.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/step_delta.h @@ -115,9 +115,18 @@ struct PERSPECTIVE_EXPORT t_rowdelta { const std::vector& data ); + t_rowdelta( + bool rows_changed, + const std::vector& rows, + const std::vector& data + ); + bool rows_changed; t_uindex num_rows_changed; std::vector data; + + /// The view row index of each delta row, empty for a flat context. + std::vector rows; }; } // end namespace perspective 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 55bb0e775f..1a452cdc4b 100644 --- a/rust/perspective-server/cpp/perspective/src/include/perspective/view.h +++ b/rust/perspective-server/cpp/perspective/src/include/perspective/view.h @@ -40,6 +40,17 @@ namespace perspective { */ enum class t_arrow_compression : std::uint8_t { NONE, LZ4, ZSTD }; +struct t_stree_capture; +struct t_stree_dropped; + +/** + * @brief Serialize `pkeys` of `table` as an Arrow IPC of one column per + * index column, named after it. + */ +PERSPECTIVE_EXPORT std::shared_ptr pkeys_to_index_arrow( + const Table& table, const std::vector& pkeys +); + template class PERSPECTIVE_EXPORT View { public: @@ -347,6 +358,19 @@ class PERSPECTIVE_EXPORT View { bool _get_deltas_enabled() const; void _set_deltas_enabled(bool enabled_state); + /** + * @brief Record the rows which leave this `View`, for + * `take_removed_rows`. + */ + void set_removes_enabled(bool enabled_state); + + /** + * @brief The keys of the rows which left this `View` since the last call, + * as an Arrow IPC of one column per key column, or `nullptr` when none + * did. + */ + std::shared_ptr take_removed_rows(); + // Pivot table operations /** @@ -447,6 +471,12 @@ class PERSPECTIVE_EXPORT View { void _find_hidden_sort(const std::vector& sort); + void attach_removes_capture(); + void detach_removes_capture(); + t_dtype row_pivot_dtype(const std::string& name) const; + std::shared_ptr + dropped_to_arrow(const std::vector& dropped) const; + std::shared_ptr
m_table; std::shared_ptr m_ctx; std::shared_ptr m_pkey_codec; @@ -468,6 +498,9 @@ class PERSPECTIVE_EXPORT View { bool m_split_rollup; std::shared_ptr m_view_config; + std::shared_ptr m_removes_capture; + t_uindex m_removes_generation = 0; + bool m_removes_enabled = false; }; /** diff --git a/tools/bench/package.json b/tools/bench/package.json index c1ce9b481c..687d201a5d 100644 --- a/tools/bench/package.json +++ b/tools/bench/package.json @@ -31,6 +31,9 @@ "zx": "catalog:" }, "dependencies": { + "perspective-5-8-0": "npm:@perspective-dev/client@5.8.0", + "perspective-5-7-0": "npm:@perspective-dev/client@5.7.0", + "perspective-5-6-0": "npm:@perspective-dev/client@5.6.0", "perspective-5-5-0": "npm:@perspective-dev/client@5.5.0", "perspective-5-4-0": "npm:@perspective-dev/client@5.4.0", "perspective-5-3-0": "npm:@perspective-dev/client@5.3.0", @@ -43,6 +46,12 @@ "perspective-4-2-0": "npm:@perspective-dev/client@4.2.0", "perspective-4-1-0": "npm:@perspective-dev/client@4.1.0", "perspective-4-0-0": "npm:@perspective-dev/client@4.0.0", + "perspective-3-0-0": "npm:@finos/perspective@3.0.0", + "perspective-2-2-0": "npm:@finos/perspective@2.2.0", + "perspective-1-0-0": "npm:@finos/perspective@1.0.0", + "perspective-0-9-0": "npm:@finos/perspective@0.9.0" + }, + "deprecatedDependencies": { "perspective-3-8-0": "npm:@finos/perspective@3.8.0", "perspective-3-7-0": "npm:@finos/perspective@3.7.0", "perspective-3-6-0": "npm:@finos/perspective@3.6.0", @@ -51,23 +60,11 @@ "perspective-3-3-0": "npm:@finos/perspective@3.3.0", "perspective-3-2-0": "npm:@finos/perspective@3.2.0", "perspective-3-1-0": "npm:@finos/perspective@3.1.0", - "perspective-3-0-0": "npm:@finos/perspective@3.0.0", "perspective-2-10-0": "npm:@finos/perspective@2.10.0", - "perspective-2-9-0": "npm:@finos/perspective@2.9.0", - "perspective-2-8-0": "npm:@finos/perspective@2.8.0", "perspective-2-7-0": "npm:@finos/perspective@2.7.0", "perspective-2-6-0": "npm:@finos/perspective@2.6.0", - "perspective-2-5-0": "npm:@finos/perspective@2.5.0", "perspective-2-4-0": "npm:@finos/perspective@2.4.0", "perspective-2-3-0": "npm:@finos/perspective@2.3.0", - "perspective-2-2-0": "npm:@finos/perspective@2.2.0", - "perspective-2-1-0": "npm:@finos/perspective@2.1.0", - "perspective-1-1-0": "npm:@finos/perspective@1.1.0", - "perspective-1-0-0": "npm:@finos/perspective@1.0.0", - "perspective-0-10-0": "npm:@finos/perspective@0.10.0", - "perspective-0-9-0": "npm:@finos/perspective@0.9.0" - }, - "deprecatedDependencies": { "perspective-2-0-0": "npm:@finos/perspective@2.0.0", "perspective-1-9-0": "npm:@finos/perspective@1.9.0", "perspective-1-8-0": "npm:@finos/perspective@1.8.0", @@ -76,6 +73,9 @@ "perspective-1-5-0": "npm:@finos/perspective@1.5.0", "perspective-1-4-0": "npm:@finos/perspective@1.4.0", "perspective-1-3-0": "npm:@finos/perspective@1.3.0", - "perspective-1-2-0": "npm:@finos/perspective@1.2.0" + "perspective-1-2-0": "npm:@finos/perspective@1.2.0", + "perspective-1-1-0": "npm:@finos/perspective@1.1.0", + "perspective-0-10-0": "npm:@finos/perspective@0.10.0" + } } diff --git a/tools/bench/src/html/index.html b/tools/bench/src/html/index.html index 0e32a2b90a..e8575a38d3 100644 --- a/tools/bench/src/html/index.html +++ b/tools/bench/src/html/index.html @@ -30,9 +30,9 @@ const websocket = await perspective.websocket("ws://localhost:8081/subscribe"); const table = await websocket.open_table("benchmarks"); const worker = await perspective.worker(); - const table2 = await worker.table(await table.view()); - el.load(table2); - el.restore(JSON.parse(localStorage.getItem("layout"))); + const table2 = await worker.table(await table.view(), { name: "benchmarks" }); + el.load(worker); + el.restore({ table: "benchmarks", ...JSON.parse(localStorage.getItem("layout")) }); diff --git a/tools/bench/src/js/benchmark.mjs b/tools/bench/src/js/benchmark.mjs index c03f248d91..88882be1f4 100644 --- a/tools/bench/src/js/benchmark.mjs +++ b/tools/bench/src/js/benchmark.mjs @@ -166,6 +166,10 @@ async function benchmark_node_version(version, benchmarks_table) { ...(process.env.PSP_BENCH_WASM32 && { PSP_BENCH_WASM32: process.env.PSP_BENCH_WASM32, }), + ...(process.env.PSP_BENCH_SUPERSTORE_COPIES && { + PSP_BENCH_SUPERSTORE_COPIES: + process.env.PSP_BENCH_SUPERSTORE_COPIES, + }), }, });