fix: parameterized queries

Author: karenc-bq

.github/workflows/integration_tests_latest.yml

@@ -5,6 +5,7 @@ on:
   push:
     branches:
       - main
+      - fix/parameterized-queries
     paths-ignore:
       - "**/*.md"
       - "**/*.jpg"

common/lib/AllowedAndBlockedHosts.ts renamed to common/lib/allowed_and_blocked_hosts.ts

@@ -22,7 +22,6 @@ import { HostListProvider } from "./host_list_provider/host_list_provider";
 import { PluginManager } from "./plugin_manager";
 
 import pkgStream from "stream";
-import { DriverConnectionProvider } from "./driver_connection_provider";
 import { ClientWrapper } from "./client_wrapper";
 import { ConnectionProviderManager } from "./connection_provider_manager";
 import { DefaultTelemetryFactory } from "./utils/telemetry/default_telemetry_factory";
@@ -35,10 +34,12 @@ import { AwsWrapperError } from "./utils/errors";
 import { Messages } from "./utils/messages";
 import { TransactionIsolationLevel } from "./utils/transaction_isolation_level";
 import { HostListProviderService } from "./host_list_provider_service";
+import { SessionStateClient } from "./session_state_client";
+import { ConnectionProvider } from "./connection_provider";
 
 const { EventEmitter } = pkgStream;
 
-export abstract class AwsClient extends EventEmitter {
+export abstract class AwsClient extends EventEmitter implements SessionStateClient {
   private _defaultPort: number = -1;
   protected telemetryFactory: TelemetryFactory;
   protected pluginManager: PluginManager;
@@ -55,7 +56,8 @@ export abstract class AwsClient extends EventEmitter {
     dbType: DatabaseType,
     knownDialectsByCode: Map<string, DatabaseDialect>,
     parser: ConnectionUrlParser,
-    driverDialect: DriverDialect
+    driverDialect: DriverDialect,
+    connectionProvider: ConnectionProvider
   ) {
     super();
     this.config = config;
@@ -110,7 +112,7 @@ export abstract class AwsClient extends EventEmitter {
     this.pluginManager = new PluginManager(
       container,
       this.properties,
-      new ConnectionProviderManager(new DriverConnectionProvider(), WrapperProperties.CONNECTION_PROVIDER.get(this.properties)),
+      new ConnectionProviderManager(connectionProvider, WrapperProperties.CONNECTION_PROVIDER.get(this.properties)),
       this.telemetryFactory
     );
   }

common/lib/aws_client.ts

@@ -22,7 +22,9 @@ export interface ClientWrapper {
   readonly properties: Map<string, any>;
   readonly id: string;
 
+  // Internal method, executes wrapper-specific queries like the topology query.
   query(sql: string): Promise<any>;
+  query(config: any, values?: any, callback?: any): Promise<any>;
 
   end(): Promise<void>;
 

common/lib/client_wrapper.ts

@@ -33,4 +33,6 @@ export interface DriverDialect {
   setQueryTimeout(props: Map<string, any>, sql?: any, wrapperConnectTimeout?: any): void;
 
   setKeepAliveProperties(props: Map<string, any>, keepAliveProps: any): void;
+
+  getQueryFromMethodArg(methodArg: any): string;
 }

common/lib/driver_dialect/driver_dialect.ts

@@ -21,7 +21,7 @@ import { DatabaseDialect } from "./database_dialect/database_dialect";
 import { HostInfoBuilder } from "./host_info_builder";
 import { ConnectionUrlParser } from "./utils/connection_url_parser";
 import { TelemetryFactory } from "./utils/telemetry/telemetry_factory";
-import { AllowedAndBlockedHosts } from "./AllowedAndBlockedHosts";
+import { AllowedAndBlockedHosts } from "./allowed_and_blocked_hosts";
 
 export interface HostListProviderService {
   getHostListProvider(): HostListProvider | null;

common/lib/host_list_provider_service.ts

@@ -41,8 +41,8 @@ export class PgClientWrapper implements ClientWrapper {
     this.id = uniqueId("PgClient_");
   }
 
-  query(sql: string): Promise<any> {
-    return this.client?.query(sql);
+  query(config: any, values?: any, callback?: any): Promise<any> {
+    return this.client?.query(config, values, callback);
   }
 
   queryWithTimeout(sql: string): Promise<any> {

common/lib/pg_client_wrapper.ts

@@ -43,7 +43,7 @@ import { DatabaseDialectCodes } from "./database_dialect/database_dialect_codes"
 import { getWriter, logTopology } from "./utils/utils";
 import { TelemetryFactory } from "./utils/telemetry/telemetry_factory";
 import { DriverDialect } from "./driver_dialect/driver_dialect";
-import { AllowedAndBlockedHosts } from "./AllowedAndBlockedHosts";
+import { AllowedAndBlockedHosts } from "./allowed_and_blocked_hosts";
 import { ConnectionPlugin } from "./connection_plugin";
 
 export interface PluginService extends ErrorHandler {

common/lib/plugin_service.ts

@@ -20,7 +20,7 @@ import { PluginService } from "../../plugin_service";
 import { TelemetryCounter } from "../../utils/telemetry/telemetry_counter";
 import { logger } from "../../../logutils";
 import { CustomEndpointInfo } from "./custom_endpoint_info";
-import { AllowedAndBlockedHosts } from "../../AllowedAndBlockedHosts";
+import { AllowedAndBlockedHosts } from "../../allowed_and_blocked_hosts";
 import { CacheMap } from "../../utils/cache_map";
 import { MemberListType } from "./member_list_type";
 import { Messages } from "../../utils/messages";

common/lib/plugins/custom_endpoint/custom_endpoint_monitor_impl.ts

@@ -142,7 +142,7 @@ export class ReadWriteSplittingPlugin extends AbstractConnectionPlugin implement
   }
 
   override async execute<T>(methodName: string, executeFunc: () => Promise<T>, methodArgs: any): Promise<T> {
-    const statement = methodArgs.sql ?? methodArgs;
+    const statement = SqlMethodUtils.parseMethodArgs(methodArgs, this.pluginService.getDriverDialect());
     const statements = SqlMethodUtils.parseMultiStatementQueries(statement);
 
     const updateReadOnly: boolean | undefined = SqlMethodUtils.doesSetReadOnly(statements, this.pluginService.getDialect());