chore: separate database dialect and driver dialect

.eslintrc.json

@@ -21,7 +21,12 @@
     "@typescript-eslint/no-explicit-any": "off",
     "@typescript-eslint/no-unused-vars": "off",
     "@typescript-eslint/ban-ts-comment": "off",
-    "prettier/prettier": 2,
+    "prettier/prettier": [
+      "error",
+      {
+        "endOfLine": "auto"
+      }
+    ],
     "header/header": [
       1,
       "block",

common/lib/aws_client.ts

@@ -27,6 +27,7 @@ import { ClientWrapper } from "./client_wrapper";
 import { ConnectionProviderManager } from "./connection_provider_manager";
 import { DefaultTelemetryFactory } from "./utils/telemetry/default_telemetry_factory";
 import { TelemetryFactory } from "./utils/telemetry/telemetry_factory";
+import { DriverDialect } from "./driver_dialect/driver_dialect";
 
 export abstract class AwsClient extends EventEmitter {
   private _defaultPort: number = -1;
@@ -40,7 +41,6 @@ export abstract class AwsClient extends EventEmitter {
   protected _schema: string = "";
   protected _isolationLevel: number = 0;
   protected _errorHandler: ErrorHandler;
-  protected _createClientFunc?: (config: any) => any;
   protected _connectionUrlParser: ConnectionUrlParser;
   readonly properties: Map<string, any>;
   config: any;
@@ -51,7 +51,8 @@ export abstract class AwsClient extends EventEmitter {
     errorHandler: ErrorHandler,
     dbType: DatabaseType,
     knownDialectsByCode: Map<string, DatabaseDialect>,
-    parser: ConnectionUrlParser
+    parser: ConnectionUrlParser,
+    driverDialect: DriverDialect
   ) {
     super();
     this.config = config;
@@ -62,7 +63,7 @@ export abstract class AwsClient extends EventEmitter {
 
     this.telemetryFactory = new DefaultTelemetryFactory(this.properties);
     const container = new PluginServiceManagerContainer();
-    this.pluginService = new PluginService(container, this, dbType, knownDialectsByCode, this.properties);
+    this.pluginService = new PluginService(container, this, dbType, knownDialectsByCode, this.properties, driverDialect);
     this.pluginManager = new PluginManager(
       container,
       this.properties,
@@ -111,10 +112,6 @@ export abstract class AwsClient extends EventEmitter {
     return this._connectionUrlParser;
   }
 
-  getCreateClientFunc<Type>(): ((config: any) => Type) | undefined {
-    return this._createClientFunc;
-  }
-
   abstract updateSessionStateReadOnly(readOnly: boolean): Promise<any | void>;
 
   abstract setReadOnly(readOnly: boolean): Promise<any | void>;

common/lib/aws_pool_client.ts

@@ -17,7 +17,7 @@
 export interface AwsPoolClient {
   connect(): Promise<any>;
 
-  end(poolClient: any): Promise<any>;
+  end(): Promise<any>;
 
   releaseResources(): Promise<void>;
 

common/lib/client_wrapper.ts

@@ -20,4 +20,12 @@ export interface ClientWrapper {
   readonly client: any;
   readonly hostInfo: HostInfo;
   readonly properties: Map<string, any>;
+
+  query(sql: any): Promise<any>;
+
+  end(): Promise<void>;
+
+  rollback(): Promise<void>;
+
+  abort(): Promise<void>;
 }

common/lib/connection_provider.ts

@@ -21,7 +21,6 @@ import { ClientWrapper } from "./client_wrapper";
 
 export interface ConnectionProvider {
   connect(hostInfo: HostInfo, pluginService: PluginService, props: Map<string, any>): Promise<ClientWrapper>;
-  end(pluginService: PluginService, clientWrapper: ClientWrapper | undefined): Promise<void>;
   acceptsUrl(hostInfo: HostInfo, props: Map<string, any>): boolean;
   acceptsStrategy(role: HostRole, strategy: string): boolean;
   getHostInfoByStrategy(hosts: HostInfo[], role: HostRole, strategy: string, props?: Map<string, any>): HostInfo;

common/lib/database_dialect/database_dialect.ts

@@ -33,10 +33,7 @@ export interface DatabaseDialect {
   getServerVersionQuery(): string;
   getDialectUpdateCandidates(): string[];
   isDialect(targetClient: ClientWrapper): Promise<boolean>;
-  getAwsPoolClient(props: any): AwsPoolClient;
   getHostListProvider(props: Map<string, any>, originalUrl: string, hostListProviderService: HostListProviderService): HostListProvider;
-  tryClosingTargetClient(targetClient: ClientWrapper): Promise<void>;
-  rollback(targetClient: ClientWrapper): Promise<any>;
   isClientValid(targetClient: ClientWrapper): Promise<boolean>;
   getDatabaseType(): DatabaseType;
   getDialectName(): string;
@@ -46,7 +43,4 @@ export interface DatabaseDialect {
   doesStatementSetAutoCommit(statement: string): boolean | undefined;
   doesStatementSetSchema(statement: string): string | undefined;
   doesStatementSetCatalog(statement: string): string | undefined;
-  connect(targetClient: any): Promise<any>;
-  end(clientWrapper: ClientWrapper | undefined): Promise<void>;
-  preparePoolClientProperties(props: Map<string, any>, poolConfig: AwsPoolConfig | undefined): any;
 }

common/lib/driver_connection_provider.ts

@@ -31,7 +31,7 @@ import { logger } from "../logutils";
 import { maskProperties } from "./utils/utils";
 import { ClientWrapper } from "./client_wrapper";
 import { RoundRobinHostSelector } from "./round_robin_host_selector";
-import { DatabaseDialect } from "./database_dialect/database_dialect";
+import { DriverDialect } from "./driver_dialect/driver_dialect";
 
 export class DriverConnectionProvider implements ConnectionProvider {
   private static readonly acceptedStrategies: Map<string, HostSelector> = new Map([
@@ -53,9 +53,9 @@ export class DriverConnectionProvider implements ConnectionProvider {
     const resultProps = new Map(props);
     let connectionHostInfo: HostInfo;
 
+    const driverDialect: DriverDialect = pluginService.getDriverDialect();
     try {
-      const targetClient: any = await Promise.resolve(pluginService.createTargetClient(props));
-      await pluginService.getDialect().connect(targetClient);
+      const targetClient: any = await driverDialect.connect(hostInfo, props);
       connectionHostInfo = new HostInfoBuilder({
         hostAvailabilityStrategy: hostInfo.hostAvailabilityStrategy
       })
@@ -92,12 +92,6 @@ export class DriverConnectionProvider implements ConnectionProvider {
       const originalHost: string = hostInfo.host;
       const fixedHost: string = this.rdsUtils.removeGreenInstancePrefix(hostInfo.host);
       resultProps.set(WrapperProperties.HOST.name, fixedHost);
-      connectionHostInfo = new HostInfoBuilder({
-        hostAvailabilityStrategy: hostInfo.hostAvailabilityStrategy
-      })
-        .copyFrom(hostInfo)
-        .withHost(fixedHost)
-        .build();
 
       logger.info(
         "Connecting to " +
@@ -108,23 +102,10 @@ export class DriverConnectionProvider implements ConnectionProvider {
           JSON.stringify(Object.fromEntries(maskProperties(resultProps)))
       );
 
-      const newTargetClient = pluginService.createTargetClient(resultProps);
-      await pluginService.getDialect().connect(newTargetClient);
-      resultTargetClient = newTargetClient;
+      resultTargetClient = driverDialect.connect(hostInfo, resultProps);
     }
 
-    return {
-      client: resultTargetClient,
-      hostInfo: connectionHostInfo,
-      properties: resultProps
-    };
-  }
-
-  async end(pluginService: PluginService, clientWrapper: ClientWrapper | undefined): Promise<void> {
-    if (clientWrapper === undefined) {
-      return;
-    }
-    return await pluginService.getDialect().end(clientWrapper);
+    return resultTargetClient;
   }
 
   getHostInfoByStrategy(hosts: HostInfo[], role: HostRole, strategy: string, props?: Map<string, any>): HostInfo {

common/lib/driver_dialect/driver_dialect.ts

@@ -0,0 +1,30 @@
+/*
+  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+  Licensed under the Apache License, Version 2.0 (the "License").
+  You may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+*/
+
+import { ClientWrapper } from "../client_wrapper";
+import { AwsPoolConfig } from "../aws_pool_config";
+import { AwsPoolClient } from "../aws_pool_client";
+import { HostInfo } from "../host_info";
+
+export interface DriverDialect {
+  getDialectName(): string;
+
+  connect(hostInfo: HostInfo, props: Map<string, any>): Promise<ClientWrapper>;
+
+  preparePoolClientProperties(props: Map<string, any>, poolConfig: AwsPoolConfig | undefined): any;
+
+  getAwsPoolClient(props: any): AwsPoolClient;
+}

common/lib/internal_pooled_connection_provider.ts

@@ -37,6 +37,7 @@ import { logger } from "../logutils";
 import { RoundRobinHostSelector } from "./round_robin_host_selector";
 import { AwsPoolClient } from "./aws_pool_client";
 import { AwsPoolConfig } from "./aws_pool_config";
+import { PoolClientWrapper } from "./pool_client_wrapper";
 
 export class InternalPooledConnectionProvider implements PooledConnectionProvider, CanReleaseResources {
   private static readonly acceptedStrategies: Map<string, HostSelector> = new Map([
@@ -104,32 +105,20 @@ export class InternalPooledConnectionProvider implements PooledConnectionProvide
       }
     }
 
-    const dialect = pluginService.getDialect();
+    const dialect = pluginService.getDriverDialect();
     const preparedConfig = dialect.preparePoolClientProperties(props, this._poolConfig);
 
     this.internalPool = this.databasePools.computeIfAbsent(
-      new PoolKey(hostInfo.url, this.getPoolKey(hostInfo, props)),
+      new PoolKey(connectionHostInfo.url, this.getPoolKey(connectionHostInfo, props)),
       () => dialect.getAwsPoolClient(preparedConfig),
       this.poolExpirationCheckNanos
     );
 
-    const poolClient = await this.getPoolConnection();
-
-    return {
-      client: poolClient,
-      hostInfo: connectionHostInfo,
-      properties: props
-    };
-  }
-
-  async end(pluginService: PluginService, clientWrapper: ClientWrapper | undefined): Promise<void> {
-    if (this.internalPool && clientWrapper) {
-      return this.internalPool.end(clientWrapper.client);
-    }
+    return await this.getPoolConnection(hostInfo, props);
   }
 
-  async getPoolConnection() {
-    return this.internalPool!.connect();
+  async getPoolConnection(hostInfo: HostInfo, props: Map<string, string>) {
+    return new PoolClientWrapper(await this.internalPool!.connect(), hostInfo, props);
   }
 
   public async releaseResources() {
@@ -172,17 +161,12 @@ export class InternalPooledConnectionProvider implements PooledConnectionProvide
   }
 
   logConnections() {
-    for (const [key, val] of this.databasePools.entries) {
-      logger.debug(
-        "Internal Pooled Connection: \t\n[ " +
-          key.getPoolKeyString() +
-          "\n\t {\n\t\t\t" +
-          JSON.stringify(val.item.constructor.name) +
-          "\t(expirationTimeNs: " +
-          val.expirationTimeNs +
-          ")\n\t }\n]"
-      );
+    if (this.databasePools.size === 0) {
+      return;
     }
+
+    const l = Array.from(this.databasePools.entries).map(([v, k]) => [v.getPoolKeyString(), k.item.constructor.name]);
+    logger.debug(`Internal Pooled Connection: [${JSON.stringify(l)}]`);
   }
 
   setDatabasePools(connectionPools: SlidingExpirationCache<PoolKey, any>): void {

common/lib/mysql_client_wrapper.ts

@@ -0,0 +1,61 @@
+/*
+  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+  Licensed under the Apache License, Version 2.0 (the "License").
+  You may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+*/
+
+import { ClientWrapper } from "./client_wrapper";
+import { HostInfo } from "./host_info";
+import { ClientUtils } from "./utils/client_utils";
+
+/*
+This is an internal wrapper class for the target community driver client created by the MySQL2DriverDialect.
+ */
+export class MySQLClientWrapper implements ClientWrapper {
+  readonly client: any;
+  readonly hostInfo: HostInfo;
+  readonly properties: Map<string, string>;
+
+  /**
+   * Creates a wrapper for the target community driver client.
+   *
+   * @param targetClient The community driver client created for an instance.
+   * @param hostInfo Host information for the connected instance.
+   * @param properties Connection properties for the target client.
+   */
+  constructor(targetClient: any, hostInfo: HostInfo, properties: Map<string, any>) {
+    this.client = targetClient;
+    this.hostInfo = hostInfo;
+    this.properties = properties;
+  }
+
+  query(sql: any): Promise<any> {
+    return this.client?.query(sql);
+  }
+
+  end(): Promise<void> {
+    return this.client?.end();
+  }
+
+  rollback(): Promise<void> {
+    return this.client?.rollback();
+  }
+
+  async abort(): Promise<void> {
+    try {
+      return await ClientUtils.queryWithTimeout(this.client?.destroy(), this.properties);
+    } catch (error: any) {
+      // ignore
+    }
+  }
+}

common/lib/pg_client_wrapper.ts

@@ -0,0 +1,60 @@
+/*
+  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+  Licensed under the Apache License, Version 2.0 (the "License").
+  You may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+*/
+
+import { ClientWrapper } from "./client_wrapper";
+import { HostInfo } from "./host_info";
+
+/*
+This an internal wrapper class for a target community driver client created by the NodePostgresPgDriverDialect.
+ */
+export class PgClientWrapper implements ClientWrapper {
+  readonly client: any;
+  readonly hostInfo: HostInfo;
+  readonly properties: Map<string, string>;
+
+  /**
+   * Creates a wrapper for the target community driver client.
+   *
+   * @param targetClient The community driver client created for an instance.
+   * @param hostInfo Host information for the connected instance.
+   * @param properties Connection properties for the target client.
+   */
+  constructor(targetClient: any, hostInfo: HostInfo, properties: Map<string, any>) {
+    this.client = targetClient;
+    this.hostInfo = hostInfo;
+    this.properties = properties;
+  }
+
+  query(sql: any): Promise<any> {
+    return this.client?.query(sql);
+  }
+
+  end(): Promise<void> {
+    return this.client?.end();
+  }
+
+  rollback(): Promise<void> {
+    return this.client?.rollback();
+  }
+
+  async abort(): Promise<void> {
+    try {
+      return await this.end();
+    } catch (error: any) {
+      // Ignore
+    }
+  }
+}