Merged in createTableAs

Author: Jonathan Luu

__tests__/integrationtests.js

@@ -1,3 +1,4 @@
+const { default: waitForExpect } = require('wait-for-expect');
 const ksqljs = require('../ksqljs/ksqlJS.js');
 // Pre-requisite: start a docker container
 /* To add to README: Prior to running test with 'npm test', please start the ksqlDB
@@ -86,14 +87,15 @@ describe('--Integration Tests--', () => {
   // describe('--Materialized Views Test--', () => {
   //   beforeAll( async () => {
   //     client = new ksqljs({ ksqldbURL: 'http://localhost:8088'});
-  //     await client.ksql('CREATE STREAM TESTJESTSTREAM (NAME VARCHAR, AGE INTEGER, LOCATION VARCHAR, WEIGHT INTEGER) WITH (kafka_topic= \'testJestTopic\', value_format=\'json\', partitions=1);')
+  //     const waitForExpect = require('wait-for-expect');
+  //     await client.ksql('CREATE STREAM NEWTESTSTREAM (NAME VARCHAR, AGE INTEGER, LOCATION VARCHAR, WEIGHT INTEGER) WITH (kafka_topic= \'testJestTopic2\', value_format=\'json\', partitions=1);')
   //   });
   //   afterAll(async () => {
   //     await client.ksql('DROP TABLE IF EXISTS TABLEOFSTREAM DELETE TOPIC;')
-  //     await client.ksql('DROP STREAM IF EXISTS TESTJESTSTREAM DELETE TOPIC;')
+  //     await client.ksql('DROP STREAM IF EXISTS NEWTESTSTREAM DELETE TOPIC;')
   //   })
   //   it('creates a materialized table view of a stream', async () => {
-  //     await client.createTableAs('TABLEOFSTREAM', 'TESTJESTSTREAM', ['name', 'LATEST_BY_OFFSET(age) AS recentAge', 'LATEST_BY_OFFSET(weight) AS recentweight'], {topic:'newTopic'},{WHERE: 'age >= 21', GROUP_BY: 'name'});
+  //     await client.createTableAs('TABLEOFSTREAM', 'NEWTESTSTREAM', ['name', 'LATEST_BY_OFFSET(age) AS recentAge', 'LATEST_BY_OFFSET(weight) AS recentweight'], {topic:'newTopic'},{WHERE: 'age >= 21', GROUP_BY: 'name'});
   //     const tables = await client.ksql('LIST TABLES;');
   //     const allTables = tables.tables;
   //     let tableCheck = false;
@@ -109,20 +111,17 @@ describe('--Integration Tests--', () => {
   //   it('materialized table view updates with source stream', async () => {
   //     let rowCheck = false;
   //     // push query for the table
+  //     // console.log('testing materialized view')
   //     await client.push('SELECT * FROM TABLEOFSTREAM EMIT CHANGES LIMIT 1;', async (data) => {
+  //       console.log('QUERY INFO',data)
   //       if (Array.isArray(JSON.parse(data))){
-  //         console.log(JSON.parse(data))
-  //         if (JSON.parse(data)[0] === "firstTester" && JSON.parse(data)[1] === 25 && JSON.parse(data)[2] === "Seattle" && JSON.parse(data)[3] === 130){
+  //         if (JSON.parse(data)[0] === "firstTester" && JSON.parse(data)[1] === 25 &&  JSON.parse(data)[2] === 130){
   //           rowCheck = true;
   //         }
-  //         expect
   //       }
   //     })
-  //     await client.insertStream('TESTJESTSTREAM', [{"NAME":"firstTester", "AGE":25, "LOCATION": "Seattle", "WEIGHT": 130}])      
-  //     const matTable = await client.pull('SELECT * FROM TABLEOFSTREAM;');
-  //     // console.log('table list', matTable)
-
-  //     expect(rowCheck).toEqual(true);
+  //     await client.insertStream('NEWTESTSTREAM', [{"NAME":"firstTester", "AGE":25, "LOCATION": "Seattle", "WEIGHT": 130}]);
+  //     await waitForExpect(() => expect(rowCheck).toEqual(true))
   //   })
   // })
 

docker-compose.yml

@@ -1,5 +1,5 @@
 ---
-version: "2"
+version: "3.2"
 
 services:
   zookeeper:

ksqljs/ksqlJS.js

@@ -306,6 +306,20 @@ class ksqljs {
   }
 
   //---------------------Create tables as select-----------------
+  /**
+   * Execute a query to create a new materialized table view of an existing table or stream
+   * 
+   * <p>This method is used to create a materialized table view
+   * 
+   * <p>This method is sql injection protected with the use of queryBuilder.
+   * 
+   * @param {string} tableName name of the table to be created 
+   * @param {string} source name of the source stream / table materialized view is based on
+   * @param {array} selectArray an array that contains the values (strings, aggregate functions) of the columns for the materialized view table
+   * @param {object} propertiesObj an object containing key value pairs for supported table properties e.g {topic: 'myTopic', value_format: 'json', partitions: '1'}. {} for default values
+   * @param {object} conditionsObj an object containing key value pairs for supported query conditions e.g {WHERE: 'a is not null', GROUP_BY: 'profileID', HAVING: 'COUNT(a) > 5' }
+   * @returns {Promise} a promise that completes once the server response is received, returning a response object
+   */
   createTableAs = (tableName, source, selectArray, propertiesObj, conditionsObj) => {
     let selectColStr = selectArray.reduce((result, current) => result + ', ' + current);
 
@@ -317,6 +331,7 @@ class ksqljs {
       partitions: 1
     };
     Object.assign(defaultProps, propertiesObj);
+    
     // if there's no properties Obj, assign them all default values
 
     // expect user to input a conditions object of format {WHERE: condition, GROUP_BY: condition, HAVING: condition};

ksqljs/queryBuilder.js

@@ -33,11 +33,7 @@ class queryBuilder {
         return param;
       case "object":
         if (Array.isArray(param)) {
-          //check if spaces
-          if (param[0].includes(" ")) {
-            throw new InappropriateStringParamError("string params not wrapped in quotes should not include spaces");
-          }
-          else if (param[0].includes(";")) {
+          if (param[0].includes(";")) {
             throw new InappropriateStringParamError("string params not wrapped in quotes should not include semi-colons");
           }
           return `${param[0].replaceAll("'", "''")}`

ksqljsTest.js

@@ -87,12 +87,17 @@ createTableTest(); */
 //---------------------Test Table Create As-------------------
 const createTableAsTest = async () => {
 
-    await client.ksql('CREATE STREAM RIDERLOCATIONS (PROFILEID VARCHAR, LATITUDE DOUBLE, LONGITUDE DOUBLE) WITH (KAFKA_TOPIC=\'locations\', value_format=\'json\', partitions=1);')
-    await client.createTableAs('currentLocation', 'riderlocations', ['profileId', 'LATEST_BY_OFFSET(latitude) AS la', 'LATEST_BY_OFFSET(longitude) AS lo'], {}, {GROUP_BY: 'profileId'})
+    // await client.ksql('CREATE STREAM RIDERLOCATIONS (PROFILEID VARCHAR, LATITUDE DOUBLE, LONGITUDE DOUBLE) WITH (KAFKA_TOPIC=\'locations\', value_format=\'json\', partitions=1);')
+    // await client.createTableAs('currentlocation', 'riderlocations', ['profileid','LATEST_BY_OFFSET(latitude) AS la', 'LATEST_BY_OFFSET(longitude) AS lo'], {}, {GROUP_BY: 'profileId'})
     // let x;
-    // x = await client.push('SELECT * FROM CURRENTLOCATION EMIT CHANGES LIMIT 20;', async (data) => {
-    //     console.log('push query data', JSON.parse(data))
+    await client.push('SELECT * FROM CURRENTLOCATION EMIT CHANGES LIMIT 3;', async (data) => {
+        console.log('push query data', JSON.parse(data))
+    })
+    //     // x = await client.ksql('LIST QUERIES;');
+    //     // console.log('querylist', x.queries)
     // })
+
+    // console.log(await client.ksql('LIST QUERIES;'));
     // console.log('first x', x)
     // await client.insertStream('RIDERLOCATIONS', [{'PROFILEID':'abc123abcddzzz', 'latitude':9999999,'longitude':999999}])
     // console.log('second x', x)