[PECO-930] Add parameterized query support.

lib/DBSQLClient.ts

@@ -1,6 +1,6 @@
+import { EventEmitter } from 'events';
 import thrift, { HttpHeaders } from 'thrift';
 
-import { EventEmitter } from 'events';
 import TCLIService from '../thrift/TCLIService';
 import { TProtocolVersion } from '../thrift/TCLIService_types';
 import IDBSQLClient, { ClientOptions, ConnectionOptions, OpenSessionRequest } from './contracts/IDBSQLClient';

lib/DBSQLSession.ts

@@ -29,6 +29,7 @@ import CloseableCollection from './utils/CloseableCollection';
 import IDBSQLLogger, { LogLevel } from './contracts/IDBSQLLogger';
 import HiveDriverError from './errors/HiveDriverError';
 import globalConfig from './globalConfig';
+import convertToSparkParameters from './utils/ParameterConverter';
 
 const defaultMaxRows = 100000;
 
@@ -140,6 +141,7 @@ export default class DBSQLSession implements IDBSQLSession {
       ...getDirectResultsOptions(options.maxRows),
       ...getArrowOptions(),
       canDownloadResult: options.useCloudFetch ?? globalConfig.useCloudFetch,
+      parameters: options.parameters ? convertToSparkParameters(options.parameters) : undefined,
     });
     const response = await this.handleResponse(operationPromise);
     return this.createOperation(response);

lib/connection/connections/HttpConnection.ts

@@ -1,6 +1,6 @@
-import thrift from 'thrift';
 import https from 'https';
 import http, { IncomingMessage } from 'http';
+import thrift from 'thrift';
 
 import IThriftConnection from '../contracts/IThriftConnection';
 import IConnectionProvider from '../contracts/IConnectionProvider';

lib/contracts/IDBSQLSession.ts

@@ -8,6 +8,7 @@ export type ExecuteStatementOptions = {
   runAsync?: boolean;
   maxRows?: number | null;
   useCloudFetch?: boolean;
+  parameters?: object;
 };
 
 export type TypeInfoRequest = {

lib/utils/ParameterConverter.ts

@@ -0,0 +1,38 @@
+import { TSparkParameter, TSparkParameterValue } from '../../thrift/TCLIService_types';
+import HiveDriverError from '../errors/HiveDriverError';
+
+function getTypeAndValue(value: any): [string, TSparkParameterValue] {
+  switch (typeof value) {
+    case 'object':
+      if (value === null) {
+        return ['VOID', new TSparkParameterValue()];
+      }
+
+      throw new HiveDriverError('Unsupported object type used for parameterized query.');
+
+    case 'boolean':
+      return ['BOOLEAN', new TSparkParameterValue({ booleanValue: value })];
+    case 'number':
+      if (Number.isInteger(value)) {
+        return ['INT', new TSparkParameterValue({ doubleValue: value })];
+      }
+
+      return ['DOUBLE', new TSparkParameterValue({ doubleValue: value })];
+
+    case 'string':
+      return ['STRING', new TSparkParameterValue({ stringValue: value })];
+    default:
+      throw new HiveDriverError('Unsupported object type used for parameterized query.');
+  }
+}
+
+export default function convertToSparkParameters(params: object): TSparkParameter[] {
+  const sparkValueParams = [];
+  for (const e of Object.entries(params)) {
+    const key = e[0];
+    const value = e[1];
+    const typeValueTuple = getTypeAndValue(value);
+    sparkValueParams.push(new TSparkParameter({ name: key, type: typeValueTuple[0], value: typeValueTuple[1] }));
+  }
+  return sparkValueParams;
+}

tests/e2e/parameterized_query.test.js

@@ -0,0 +1,53 @@
+const { expect } = require('chai');
+const config = require('./utils/config');
+const logger = require('./utils/logger')(config.logger);
+
+const DBSQLClient = require('../../dist/DBSQLClient').default;
+const convertToSparkParameters = require('../../dist/utils/ParameterConverter').default;
+const { TSparkParameterValue, TSparkParameter } = require('../../thrift/TCLIService_types');
+const globalConfig = require('../../dist/globalConfig').default;
+
+describe('Parameterized query converter unit test', () => {
+  expect(convertToSparkParameters({ key: null })[0]).to.deep.eq(
+    new TSparkParameter({ name: 'key', type: 'VOID', value: new TSparkParameterValue() }),
+  );
+  expect(convertToSparkParameters({ key: 'value' })[0]).to.deep.eq(
+    new TSparkParameter({ name: 'key', type: 'STRING', value: new TSparkParameterValue({ stringValue: 'value' }) }),
+  );
+  expect(convertToSparkParameters({ key: 1 })[0]).to.deep.eq(
+    new TSparkParameter({ name: 'key', type: 'INT', value: new TSparkParameterValue({ doubleValue: 1 }) }),
+  );
+  expect(convertToSparkParameters({ key: 1.1 })[0]).to.deep.eq(
+    new TSparkParameter({ name: 'key', type: 'DOUBLE', value: new TSparkParameterValue({ doubleValue: 1.1 }) }),
+  );
+  expect(convertToSparkParameters({ key: true })[0]).to.deep.eq(
+    new TSparkParameter({ name: 'key', type: 'BOOLEAN', value: new TSparkParameterValue({ booleanValue: true }) }),
+  );
+});
+
+const openSession = async () => {
+  const client = new DBSQLClient();
+
+  const connection = await client.connect({
+    host: config.host,
+    path: config.path,
+    token: config.token,
+  });
+
+  return connection.openSession({
+    initialCatalog: config.database[0],
+    initialSchema: config.database[1],
+  });
+};
+
+describe('Parameterized Query', async () => {
+  it('should use default socket timeout', async () => {
+    const query = `
+        select * from default.stock_data where open > {{parameter}}
+      `;
+
+    let session = await openSession();
+
+    let result = await session.executeStatement(query, { parameters: 2 });
+  });
+});