Merge pull request #961 from finos/bucket-fix

Fully clear queued updates before adding a new computed column

cpp/perspective/src/cpp/computed.cpp

@@ -203,6 +203,9 @@ t_computed_column::apply_computation(
     if (end == 0) {
         end = flattened_columns[0]->size();
     }
+
+    output_column->reserve(end);
+
     auto arity = table_columns.size();
 
     std::function<t_tscalar(t_tscalar)> function_1;

cpp/perspective/src/cpp/emscripten.cpp

@@ -963,16 +963,9 @@ namespace binding {
                 method_name, input_types);
             t_dtype output_column_type = computation.m_return_type;
 
-            // don't double create output column
-            auto schema = flattened->get_schema();
-            std::shared_ptr<t_column> output_column;
-            if (schema.has_column(output_column_name)) {
-                output_column = flattened->get_column(output_column_name);
-            } else {
-                output_column = flattened->add_column_sptr(
+            std::shared_ptr<t_column> output_column = 
+                flattened->add_column_sptr(
                     output_column_name, output_column_type, true);
-                output_column->reserve(table_columns[0]->size());
-            }
 
             t_computed_column::apply_computation(
                 table_columns,

examples/simple/streaming.js

@@ -13,11 +13,12 @@ var CLIENTS = ["Homer", "Marge", "Bart", "Lisa", "Maggie", "Moe", "Lenny", "Carl
 
 function newRows() {
     var rows = [];
-    for (var x = 0; x < 1000; x++) {
+    for (var x = 0; x < 5; x++) {
         rows.push({
             name: SECURITIES[Math.floor(Math.random() * SECURITIES.length)],
             client: CLIENTS[Math.floor(Math.random() * CLIENTS.length)],
             lastUpdate: new Date(),
+            date: new Date(),
             chg: Math.random() * 20 - 10,
             bid: Math.random() * 10 + 90,
             ask: Math.random() * 10 + 100,
@@ -30,9 +31,21 @@ function newRows() {
 window.addEventListener("WebComponentsReady", function() {
     var elem = document.getElementsByTagName("perspective-viewer")[0];
     // eslint-disable-next-line no-undef
-    var table = perspective.worker().table(newRows(), {
-        limit: 100000
-    });
+    var table = perspective.worker().table(
+        {
+            name: "string",
+            client: "string",
+            date: "date",
+            lastUpdate: "datetime",
+            chg: "float",
+            bid: "float",
+            ask: "float",
+            vol: "float"
+        },
+        {
+            limit: 5000
+        }
+    );
     elem.load(table);
 
     (function postRow() {

packages/perspective/src/js/perspective.js

@@ -1100,7 +1100,7 @@ export default function(Module) {
      * @returns {Promise<number>} The number of accumulated rows.
      */
     table.prototype.size = function() {
-        _call_process(this.table_id);
+        _call_process(this._Table.get_id());
         return this._Table.size();
     };
 
@@ -1240,7 +1240,7 @@ export default function(Module) {
      * supplied configuration, bound to this table
      */
     table.prototype.view = function(_config = {}) {
-        _call_process(this.get_id());
+        _call_process(this._Table.get_id());
         let config = {};
         for (const key of Object.keys(_config)) {
             if (defaults.CONFIG_ALIASES[key]) {
@@ -1454,6 +1454,7 @@ export default function(Module) {
         }
 
         try {
+            _call_process(this._Table.get_id());
             _Table = __MODULE__.make_computed_table(this._Table, computed);
             if (this.computed.length > 0) {
                 computed = this.computed.concat(computed);

packages/perspective/test/js/computed.js

@@ -817,6 +817,78 @@ module.exports = perspective => {
                 table.delete();
             });
 
+            it("Bucket (m), date with updates", async function() {
+                const table = perspective.table({
+                    a: "date"
+                });
+
+                const table2 = table.add_computed([
+                    {
+                        column: "bucket",
+                        computed_function_name: "Bucket (m)",
+                        inputs: ["a"]
+                    }
+                ]);
+
+                let view = table2.view();
+
+                const schema = await table2.schema();
+                expect(schema).toEqual({
+                    a: "date",
+                    bucket: "date"
+                });
+
+                table2.update({
+                    a: [new Date(2020, 0, 15, 1, 30, 15), new Date(2020, 1, 27, 1, 30, 30), new Date(2020, 2, 28, 1, 30, 45), new Date(2020, 3, 29, 1, 30, 0), new Date(2020, 4, 30, 1, 30, 15)]
+                });
+
+                table2.update({
+                    a: [new Date(2020, 0, 15, 1, 30, 15), new Date(2020, 1, 27, 1, 30, 30), new Date(2020, 2, 28, 1, 30, 45), new Date(2020, 3, 29, 1, 30, 0), new Date(2020, 4, 30, 1, 30, 15)]
+                });
+
+                let result = await view.to_columns();
+                expect(result.bucket).toEqual(result.a);
+                view.delete();
+                table2.delete();
+                table.delete();
+            });
+
+            it("Bucket (m), datetime with updates", async function() {
+                const table = perspective.table({
+                    a: "datetime"
+                });
+
+                const table2 = table.add_computed([
+                    {
+                        column: "bucket",
+                        computed_function_name: "Bucket (m)",
+                        inputs: ["a"]
+                    }
+                ]);
+
+                let view = table2.view();
+
+                const schema = await table2.schema();
+                expect(schema).toEqual({
+                    a: "datetime",
+                    bucket: "datetime"
+                });
+
+                table2.update({
+                    a: [new Date(2020, 0, 15, 1, 30, 15), new Date(2020, 1, 27, 1, 30, 30), new Date(2020, 2, 28, 1, 30, 45), new Date(2020, 3, 29, 1, 30, 0), new Date(2020, 4, 30, 1, 30, 15)]
+                });
+
+                table2.update({
+                    a: [new Date(2020, 0, 15, 1, 30, 15), new Date(2020, 1, 27, 1, 30, 30), new Date(2020, 2, 28, 1, 30, 45), new Date(2020, 3, 29, 1, 30, 0), new Date(2020, 4, 30, 1, 30, 15)]
+                });
+
+                let result = await view.to_columns();
+                expect(result.bucket.map(x => (x ? new Date(x) : null))).toEqual(result.a.map(x => minute_bucket(x)));
+                view.delete();
+                table2.delete();
+                table.delete();
+            });
+
             it("Computed schema returns names and metadata", async function() {
                 const func = (x, y) => x - y;
                 const computation = {