Add support for running SQL queries on Boiling

Author: dforsber

package.json

@@ -61,6 +61,7 @@
     "@aws-sdk/client-iam": "^3.533.0",
     "@aws-sdk/client-sts": "^3.533.0",
     "@aws-sdk/credential-provider-cognito-identity": "^3.533.0",
+    "@boilingdata/node-boilingdata": "^1.0.20",
     "amazon-cognito-identity-js": "^6.3.4",
     "aws-jwt-verify": "^4.0.0",
     "chalk": "^4.0.0",

src/bdcli/commands/api/bdcli-api-query.ts

@@ -0,0 +1,38 @@
+import * as cmd from "commander";
+import { ELogLevel, getLogger } from "../../utils/logger_util.js";
+import { spinnerError, spinnerSuccess, updateSpinnerText } from "../../utils/spinner_util.js";
+import { addGlobalOptions } from "../../utils/options_util.js";
+import { getIdToken } from "../../utils/auth_util.js";
+import { combineOptsWithSettings } from "../../utils/config_util.js";
+import { outputResults } from "../../utils/output_util.js";
+import { runBoilingQuery } from "../../../integration/boilingdata/query.js";
+
+const logger = getLogger("bdcli-api");
+logger.setLogLevel(ELogLevel.WARN);
+
+async function query(options: any, _command: cmd.Command): Promise<void> {
+  try {
+    options = await combineOptsWithSettings(options, logger);
+
+    updateSpinnerText("Authenticating");
+    const { idToken: token, cached, region } = await getIdToken(logger);
+    updateSpinnerText(cached ? "Authenticating: cached" : "Authenticating: success");
+    spinnerSuccess();
+
+    updateSpinnerText("Sending Query to Boiling API");
+    const results = await runBoilingQuery(options.sql, token, region, logger);
+    spinnerSuccess();
+    await outputResults(results, options.disableSpinner);
+  } catch (err: any) {
+    spinnerError(err?.message);
+  }
+}
+
+const program = new cmd.Command("bdcli api query")
+  .addOption(new cmd.Option("--sql <sqlQuery>", "SQL clause").makeOptionMandatory())
+  .action(async (options, command) => await query(options, command));
+
+(async () => {
+  await addGlobalOptions(program, logger);
+  await program.parseAsync(process.argv);
+})();

src/bdcli/commands/bdcli-api.ts

@@ -2,6 +2,7 @@ import { Command } from "commander";
 
 const program = new Command("bdcli api")
   .executableDir("api")
-  .command("wssurl", "Get signed WebSocket (wss://) URL for connecting to Boiling Data");
+  .command("wssurl", "Get signed WebSocket (wss://) URL for connecting to Boiling Data")
+  .command("query", "Run SQL query on Boiling cloud");
 
 program.parse(process.argv);

src/integration/boilingdata/query.ts

@@ -0,0 +1,27 @@
+import { BoilingData } from "@boilingdata/node-boilingdata";
+import { ILogger } from "../../bdcli/utils/logger_util.js";
+import { BDAWSRegion } from "@boilingdata/node-boilingdata/dist/cjs/boilingdata/boilingdata.js";
+
+export async function runBoilingQuery(sql: string, idToken: string, region: string, logger: ILogger): Promise<any[]> {
+  try {
+    logger.debug({ sql, idToken, region });
+    const bdInstance = new BoilingData({
+      logLevel: "error",
+      region: <BDAWSRegion>region,
+      authcontext: { idToken: { jwtToken: idToken } },
+    });
+    await bdInstance.connect();
+    const start = Date.now();
+    //const sql = `SELECT COUNT(*) FROM parquet_scan('s3://boilingdata-demo/demo.parquet');`;
+    const rows = await bdInstance.execQueryPromise({ sql });
+    const stop = Date.now();
+    const parsedRows = JSON.parse(JSON.stringify(rows));
+    logger.debug(JSON.parse(JSON.stringify(rows)));
+    logger.debug("Query time measured e2e (ms):", stop - start);
+    await bdInstance.close();
+    return parsedRows;
+  } catch (err) {
+    logger.error(err);
+  }
+  return [];
+}