Add new endpoints to api-server

modules/cudf/src/column.cpp

@@ -49,6 +49,8 @@ Napi::Function Column::Init(Napi::Env const& env, Napi::Object exports) {
                        // column/filling.cpp
                        InstanceMethod<&Column::fill>("fill"),
                        InstanceMethod<&Column::fill_in_place>("fillInPlace"),
+                       // column/repeat.cpp
+                       InstanceMethod<&Column::repeat>("repeat"),
                        // column/binaryop.cpp
                        InstanceMethod<&Column::add>("add"),
                        InstanceMethod<&Column::sub>("sub"),

modules/cudf/src/column.ts

@@ -249,6 +249,16 @@ export interface Column<T extends DataType = any> {
    */
   fill(value: Scalar<T>, begin?: number, end?: number, memoryResource?: MemoryResource): Column<T>;
 
+  /**
+   * Repeats the values of this n times.
+   *
+   * @param repeats The number of times to repeat the column.
+   * @param memoryResource The optional MemoryResource used to allocate the result Column's device
+   *   memory.
+   */
+  repeat(value: Scalar<T>, begin?: number, end?: number, memoryResource?: MemoryResource):
+    Column<T>;
+
   /**
    * Fills a range of elements in-place in a column with a scalar value.
    *

modules/cudf/src/node_cudf/column.hpp

@@ -717,6 +717,11 @@ struct Column : public EnvLocalObjectWrap<Column> {
     cudf::scalar const& value,
     rmm::mr::device_memory_resource* mr = rmm::mr::get_current_device_resource());
 
+  // column/filling/repeat.cpp
+  Column::wrapper_t repeat(
+    cudf::size_type repeats,
+    rmm::mr::device_memory_resource* mr = rmm::mr::get_current_device_resource());
+
   // column/replace.cpp
   Column::wrapper_t replace_nulls(
     cudf::column_view const& replacement,

modules/cudf/src/series.ts

@@ -811,6 +811,27 @@ export class AbstractSeries<T extends DataType = any> {
       this._col.fill(new Scalar({type: this.type, value}), begin, end, memoryResource));
   }
 
+  /**
+   * Repeats a Series n times, returning a new Series.
+   *
+   * @param repeats The number of times to repeat this.
+   *
+   * @example
+   * ```typescript
+   * import {Series} from '@rapidsai/cudf';
+   *
+   * // Float64Series
+   * Series.new([1, 2, 3]).repeat(2) // [1, 2, 3, 1, 2, 3]
+   * // StringSeries
+   * Series.new(["foo", "bar", "test"]).repeat(2) // ["foo", "bar", "test", "foo", "bar", "test"]
+   * // Bool8Series
+   * Series.new([true, true, true]).repeat(2) // [true, false, false, true, false, false]
+   * ```
+   */
+  repeat(repeats: T['scalarType'], memoryResource?: MemoryResource): Series<T> {
+    return this.__construct(this._col.repeat(repeats, memoryResource));
+  }
+
   /**
    * Fills a range of elements in-place in a column with a scalar value.
    *

modules/demo/api-server/index.js

@@ -29,5 +29,5 @@ const env = {
 };
 
 spawnSync(process.execPath,
-          [fastify, 'start', '-l', 'info', '-P', '-p', '3010', '-w', 'app.js'],
+          [fastify, 'start', '-l', 'info', '--ignore-watch', '-P', '-p', '3010', 'app.js'],
           {env, cwd: __dirname, stdio: 'inherit'});

modules/demo/api-server/package.json

@@ -12,8 +12,8 @@
   "description": "A fastify-based web server that provides browser-access to GPU resources.",
   "scripts": {
     "test": "tap \"test/**/*.test.js\"",
-    "start": "fastify start -l info -P -p 3010 -w app.js",
-    "dev": "fastify start -w -l info -P -p 3010 app.js"
+    "start": "fastify start --ignore-watch -l info -P -p 3010 app.js",
+    "dev": "fastify start --ignore-watch -l info -P -p 3010 app.js"
   },
   "keywords": [
     "rapids.ai",

modules/demo/api-server/routes/gpu/index.js

@@ -0,0 +1,122 @@
+// Copyright (c) 2022, NVIDIA CORPORATION.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+const {Utf8String, Int32, Uint32, Float32, DataFrame, Series, Float64} = require('@rapidsai/cudf');
+const {RecordBatchStreamWriter, Field, Vector, List}                   = require('apache-arrow');
+const Path                                                             = require('path');
+const {promisify}                                                      = require('util');
+const Fs                                                               = require('fs');
+const Stat                                                             = promisify(Fs.stat);
+const fastifyCors                                                      = require('@fastify/cors');
+const fastify = require('fastify')({logger: true});
+
+const arrowPlugin = require('fastify-arrow');
+const gpu_cache   = require('../../util/gpu_cache.js');
+const root_schema = require('../../util/schema.js');
+
+module.exports = async function(fastify, opts) {
+  fastify.register(arrowPlugin);
+  fastify.register(fastifyCors, {origin: '*'});
+  fastify.decorate('cacheObject', gpu_cache.cacheObject);
+  fastify.decorate('getData', gpu_cache.getData);
+  fastify.decorate('readCSV', gpu_cache.readCSV);
+  fastify.decorate('publicPath', gpu_cache.publicPath);
+
+  const get_schema = {
+    logLevel: 'debug',
+    schema: {
+      response: {
+        200: {
+          type: 'object',
+          properties:
+            {success: {type: 'boolean'}, message: {type: 'string'}, params: {type: 'string'}}
+        }
+      }
+    }
+  };
+
+  fastify.get('/', {...get_schema, handler: () => root_schema['gpu']});
+
+  fastify.route({
+    method: 'POST',
+    url: '/DataFrame/readCSV',
+    schema: {},
+    handler: async (request, reply) => {
+      let message = 'Error';
+      let result  = {'params': request.body, success: false, message: message};
+      try {
+        const path             = Path.join(fastify.publicPath(), request.body.filename);
+        const stats            = await Stat(path);
+        const message          = 'File is available';
+        const currentDataFrame = await fastify.getData(request.body.filename);
+        if (currentDataFrame !== undefined) {
+          console.log('Found existing dataframe.');
+          console.log(request.body);
+          console.log(currentDataFrame);
+          currentDataFrame.dispose();
+        }
+        const cacheObject = await fastify.readCSV({
+          header: 0,
+          sourceType: 'files',
+          sources: [path],
+        });
+        const name        = request.body.filename;  // request.body.replace('/\//g', '_');
+        await fastify.cacheObject(name, cacheObject);
+        result.success    = true;
+        result.message    = 'CSV file in GPU memory.';
+        result.statusCode = 200;
+        await reply.code(200).send(result);
+      } catch (e) {
+        result.message = e.message;
+        if (e.message.search('no such file or directory') !== -1) {
+          await reply.code(404).send(result);
+        } else {
+          await reply.code(500).send(result);
+        }
+      }
+    }
+  });
+
+  fastify.route({
+    method: 'GET',
+    url: '/get_column/:table/:column',
+    schema: {querystring: {table: {type: 'string'}, 'column': {type: 'string'}}},
+    handler: async (request, reply) => {
+      let message = 'Error';
+      let result  = {'params': JSON.stringify(request.params), success: false, message: message};
+      const table = await fastify.getData(request.params.table);
+      if (table == undefined) {
+        result.message = 'Table not found';
+        await reply.code(404).send(result);
+      } else {
+        try {
+          const name        = request.params.column;
+          const column      = table.get(name);
+          const newDfObject = {};
+          newDfObject[name] = column;
+          const result      = new DataFrame(newDfObject);
+          const writer      = RecordBatchStreamWriter.writeAll(result.toArrow());
+          await reply.code(200).send(writer.toNodeStream());
+        } catch (e) {
+          if (e.message.search('Unknown column name') != -1) {
+            result.message = e;
+            await reply.code(404).send(result);
+          } else {
+            await reply.code(500).send(result);
+          }
+        }
+      }
+    }
+  });
+}

modules/demo/api-server/routes/graphology/index.js

@@ -28,13 +28,13 @@ const root_schema = require('../../util/schema.js');
 module.exports = async function(fastify, opts) {
   fastify.register(arrowPlugin);
   fastify.register(fastifyCors, {origin: 'http://localhost:3001'});
-  fastify.decorate('setDataframe', gpu_cache.setDataframe);
-  fastify.decorate('getDataframe', gpu_cache.getDataframe);
+  fastify.decorate('cacheObject', gpu_cache.cacheObject);
+  fastify.decorate('getData', gpu_cache.getData);
   fastify.decorate('listDataframes', gpu_cache.listDataframes);
   fastify.decorate('readGraphology', gpu_cache.readGraphology);
   fastify.decorate('readLargeGraphDemo', gpu_cache.readLargeGraphDemo);
   fastify.decorate('clearDataFrames', gpu_cache.clearDataframes);
-  fastify.get('/', async function(request, reply) { return root_schema; });
+  fastify.get('/', async function(request, reply) { return root_schema['graphology']; });
 
   fastify.route({
     method: 'POST',
@@ -76,9 +76,9 @@ module.exports = async function(fastify, opts) {
           result.success = true;
           try {
             const graphology = await fastify.readLargeGraphDemo(path);
-            await fastify.setDataframe('nodes', graphology['nodes']);
-            await fastify.setDataframe('edges', graphology['edges']);
-            await fastify.setDataframe('options', graphology['options']);
+            await fastify.cacheObject('nodes', graphology['nodes']);
+            await fastify.cacheObject('edges', graphology['edges']);
+            await fastify.cacheObject('options', graphology['options']);
             result.message = 'File read onto GPU.';
           } catch (e) {
             result.success    = false;
@@ -135,10 +135,10 @@ module.exports = async function(fastify, opts) {
           result.success = true;
           try {
             const graphology = await fastify.readGraphology(path);
-            await fastify.setDataframe('nodes', graphology['nodes']);
-            await fastify.setDataframe('edges', graphology['edges']);
-            await fastify.setDataframe('clusters', graphology['clusters']);
-            await fastify.setDataframe('tags', graphology['tags']);
+            await fastify.cacheObject('nodes', graphology['nodes']);
+            await fastify.cacheObject('edges', graphology['edges']);
+            await fastify.cacheObject('clusters', graphology['clusters']);
+            await fastify.cacheObject('tags', graphology['tags']);
             result.message = 'File read onto GPU.';
           } catch (e) {
             result.success    = false;
@@ -173,7 +173,7 @@ module.exports = async function(fastify, opts) {
     handler: async (request, reply) => {
       let message = 'Error';
       let result  = {'params': JSON.stringify(request.params), success: false, message: message};
-      const table = await fastify.getDataframe(request.params.table);
+      const table = await fastify.getData(request.params.table);
       if (table == undefined) {
         result.message = 'Table not found';
         await reply.code(404).send(result);
@@ -208,7 +208,7 @@ module.exports = async function(fastify, opts) {
     handler: async (request, reply) => {
       let message = 'Error';
       let result  = {'params': JSON.stringify(request.params), success: false, message: message};
-      const table = await fastify.getDataframe(request.params.table);
+      const table = await fastify.getData(request.params.table);
       if (table == undefined) {
         result.message = 'Table not found';
         await reply.code(404).send(result);
@@ -225,7 +225,7 @@ module.exports = async function(fastify, opts) {
     handler: async (request, reply) => {
       let message = 'Error';
       let result  = {success: false, message: message};
-      const df    = await fastify.getDataframe('nodes');
+      const df    = await fastify.getData('nodes');
       if (df == undefined) {
         result.message = 'Table not found';
         await reply.code(404).send(result);
@@ -250,7 +250,7 @@ module.exports = async function(fastify, opts) {
     handler: async (request, reply) => {
       let message = 'Error';
       let result  = {success: false, message: message};
-      const df    = await fastify.getDataframe('nodes');
+      const df    = await fastify.getData('nodes');
       if (df == undefined) {
         result.message = 'Table not found';
         await reply.code(404).send(result);
@@ -290,9 +290,9 @@ module.exports = async function(fastify, opts) {
       let message = 'Error';
       let result  = {success: false, message: message};
       /** @type DataFrame<{x: Float32, y: Float32}> */
-      const df = await fastify.getDataframe('nodes');
+      const df = await fastify.getData('nodes');
       /** @type DataFrame<{x: Int32, y: Int32}> */
-      const edges = await fastify.getDataframe('edges');
+      const edges = await fastify.getData('edges');
       if (df == undefined) {
         result.message = 'Table not found';
         await reply.code(404).send(result);