Merge pull request #47 from oslabs-beta/ja/createTableAs

Ja/create table as

Author: jonathanluu17

.github/workflows/integration-tests.yml

@@ -15,6 +15,7 @@ on:
   pull_request:
     branches: 
     - dev
+    - master
 
 # A workflow run is made up of one or more jobs that can run sequentially or in parallel
 jobs:

.github/workflows/publish.yml

@@ -0,0 +1,17 @@
+name: Publish Package to npmjs
+on:
+  release:
+    types: [published]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      # Setup .npmrc file to publish to npm
+      - uses: actions/setup-node@v3
+        with:
+          node-version: '16.x'
+          registry-url: 'https://registry.npmjs.org'
+      - run: npm publish
+        env:
+          NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}

ksqljs/ksqldb/ksqldb.js

@@ -52,7 +52,7 @@ class ksqldb {
     validateInputs([query, 'string', 'query']);
 
     const validatedQuery = builder.build(query);
-    
+
     return axios
       .post(this.ksqldbURL + "/query-stream",
         {
@@ -70,7 +70,7 @@ class ksqldb {
         })
       .then((res) => res.data)
       .catch((error) => {
-        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error;
       });
   }
 
@@ -173,7 +173,7 @@ class ksqldb {
       })
       .then(res => res.data[0])
       .catch(error => {
-        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error;
       });
   }
 
@@ -191,7 +191,7 @@ class ksqldb {
     validateInputs([query, 'string', 'query']);
 
     const validatedQuery = builder.build(query);
-    
+
     return axios.post(this.ksqldbURL + '/ksql',
       {
         ksql: validatedQuery
@@ -208,7 +208,7 @@ class ksqldb {
       })
       .then(res => res.data[0])
       .catch(error => {
-        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error;
       });
   }
 
@@ -244,7 +244,7 @@ class ksqldb {
       httpsAgent: this.httpsAgentAxios ? this.httpsAgentAxios : null,
     })
       .catch(error => {
-        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error;
       });
   }
 
@@ -300,7 +300,7 @@ class ksqldb {
     })
       .then(res => res.data[0].commandStatus.queryId)
       .catch(error => {
-        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error;
       });
   }
 
@@ -340,7 +340,7 @@ class ksqldb {
         httpsAgent: this.httpsAgentAxios ? this.httpsAgentAxios : null,
       })
       .catch(error => {
-        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error;
       });
   }
 
@@ -397,18 +397,17 @@ class ksqldb {
         i += 2;
         conditionsArr.shift()
       }
-      conditionQuery = builder.build('??????', sqlClauses[0], sqlClauses[1], sqlClauses[2], sqlClauses[3], sqlClauses[4], sqlClauses[5]);
+      conditionQuery = builder.build(`${sqlClauses[0][0]}${sqlClauses[1][0]}????`, sqlClauses[2], sqlClauses[3], sqlClauses[4], sqlClauses[5]);
     }
 
-
     // reformat for builder
     tableName = [tableName];
     selectColStr = [selectColStr];
     source = [source];
     conditionQuery = [conditionQuery]
 
 
-    const query = builder.build(`CREATE TABLE ? WITH (kafka_topic=?, value_format=?, partitions=?) AS SELECT ? FROM ? ?EMIT CHANGES;`, tableName, defaultProps.topic, defaultProps.value_format, defaultProps.partitions, selectColStr, source, conditionQuery)
+    const query = builder.build(`CREATE TABLE ? WITH (kafka_topic=?, value_format=?, partitions=?) AS SELECT ? FROM ? ${conditionQuery} EMIT CHANGES;`, tableName, defaultProps.topic, defaultProps.value_format, defaultProps.partitions, selectColStr, source)
     return axios.post(this.ksqldbURL + '/ksql', { ksql: query }, {
       headers:
         this.API && this.secret ?
@@ -541,7 +540,7 @@ class ksqldb {
     return axios.get(this.ksqldbURL + `/status/${commandId}`)
       .then(response => response)
       .catch(error => {
-        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error; 
+        throw error.response?.data['@type'] ? new ksqlDBError(error.response.data) : error;
       });
   }