refactor: move session state to client wrapper

common/lib/aws_client.ts

@@ -38,8 +38,6 @@ export abstract class AwsClient extends EventEmitter {
   protected pluginManager: PluginManager;
   protected pluginService: PluginService;
   protected isConnected: boolean = false;
-  protected _isReadOnly: boolean = false;
-  protected _isolationLevel: number = 0;
   protected _connectionUrlParser: ConnectionUrlParser;
   readonly properties: Map<string, any>;
   config: any;
@@ -105,8 +103,6 @@ export abstract class AwsClient extends EventEmitter {
     return this._connectionUrlParser;
   }
 
-  abstract updateSessionStateReadOnly(readOnly: boolean): Promise<any | void>;
-
   abstract setReadOnly(readOnly: boolean): Promise<any | void>;
 
   abstract isReadOnly(): boolean;
@@ -133,8 +129,6 @@ export abstract class AwsClient extends EventEmitter {
 
   abstract rollback(): Promise<any>;
 
-  abstract resetState(): void;
-
   async isValid(): Promise<boolean> {
     if (!this.targetClient) {
       return Promise.resolve(false);

common/lib/client_wrapper.ts

@@ -15,12 +15,14 @@
 */
 
 import { HostInfo } from "./host_info";
+import { SessionState } from "./session_state";
 
 export interface ClientWrapper {
   readonly client: any;
   readonly hostInfo: HostInfo;
   readonly properties: Map<string, any>;
   readonly id: string;
+  readonly sessionState: SessionState;
 
   query(sql: any): Promise<any>;
 
@@ -31,4 +33,6 @@ export interface ClientWrapper {
   abort(): Promise<void>;
 
   queryWithTimeout(sql: string): Promise<any>;
+
+  setSessionStateDefault(): void;
 }

common/lib/database_dialect/database_dialect.ts

@@ -18,20 +18,30 @@ import { HostListProvider } from "../host_list_provider/host_list_provider";
 import { HostListProviderService } from "../host_list_provider_service";
 import { ClientWrapper } from "../client_wrapper";
 import { FailoverRestriction } from "../plugins/failover/failover_restriction";
-import { AwsPoolClient } from "../aws_pool_client";
-import { AwsPoolConfig } from "../aws_pool_config";
 import { ErrorHandler } from "../error_handler";
+import { SessionState } from "../session_state";
 
 export enum DatabaseType {
   MYSQL,
   POSTGRES
 }
 
 export interface DatabaseDialect {
+  readonly defaultAutoCommit?: boolean;
+  readonly defaultReadOnly?: boolean;
+  readonly defaultTransactionIsolation?: number;
+  readonly defaultCatalog?: string;
+  readonly defaultSchema?: string;
+
   getDefaultPort(): number;
   getHostAliasQuery(): string;
   getHostAliasAndParseResults(targetClient: ClientWrapper): Promise<string>;
   getServerVersionQuery(): string;
+  getSetReadOnlyQuery(readOnly: boolean): string;
+  getSetAutoCommitQuery(autoCommit: boolean): string;
+  getSetTransactionIsolationQuery(level: number): string;
+  getSetCatalogQuery(catalog: string): string;
+  getSetSchemaQuery(schema: string): string;
   getDialectUpdateCandidates(): string[];
   getErrorHandler(): ErrorHandler;
   isDialect(targetClient: ClientWrapper): Promise<boolean>;
@@ -45,4 +55,5 @@ export interface DatabaseDialect {
   doesStatementSetAutoCommit(statement: string): boolean | undefined;
   doesStatementSetSchema(statement: string): string | undefined;
   doesStatementSetCatalog(statement: string): string | undefined;
+  setDefaultSessionState(sessionState: SessionState): void;
 }

common/lib/mysql_client_wrapper.ts

@@ -18,6 +18,8 @@ import { ClientWrapper } from "./client_wrapper";
 import { HostInfo } from "./host_info";
 import { ClientUtils } from "./utils/client_utils";
 import { uniqueId } from "../logutils";
+import { SessionState } from "./session_state";
+import { TransactionIsolationLevel } from "./utils/transaction_isolation_level";
 
 /*
 This is an internal wrapper class for the target community driver client created by the MySQL2DriverDialect.
@@ -27,6 +29,7 @@ export class MySQLClientWrapper implements ClientWrapper {
   readonly hostInfo: HostInfo;
   readonly properties: Map<string, string>;
   readonly id: string;
+  readonly sessionState: SessionState = new SessionState();
 
   /**
    * Creates a wrapper for the target community driver client.
@@ -40,6 +43,8 @@ export class MySQLClientWrapper implements ClientWrapper {
     this.hostInfo = hostInfo;
     this.properties = properties;
     this.id = uniqueId("MySQLClient_");
+
+    this.setSessionStateDefault();
   }
 
   query(sql: any): Promise<any> {
@@ -65,4 +70,11 @@ export class MySQLClientWrapper implements ClientWrapper {
       // ignore
     }
   }
+
+  setSessionStateDefault() {
+    this.sessionState.readOnly.value = false;
+    this.sessionState.autoCommit.value = true;
+    this.sessionState.catalog.value = "";
+    this.sessionState.transactionIsolation.value = TransactionIsolationLevel.TRANSACTION_REPEATABLE_READ;
+  }
 }

common/lib/pg_client_wrapper.ts

@@ -17,6 +17,8 @@
 import { ClientWrapper } from "./client_wrapper";
 import { HostInfo } from "./host_info";
 import { uniqueId } from "../logutils";
+import { SessionState } from "./session_state";
+import { TransactionIsolationLevel } from "./utils/transaction_isolation_level";
 
 /*
 This an internal wrapper class for a target community driver client created by the NodePostgresPgDriverDialect.
@@ -26,6 +28,7 @@ export class PgClientWrapper implements ClientWrapper {
   readonly hostInfo: HostInfo;
   readonly properties: Map<string, string>;
   readonly id: string;
+  readonly sessionState = new SessionState();
 
   /**
    * Creates a wrapper for the target community driver client.
@@ -38,7 +41,10 @@ export class PgClientWrapper implements ClientWrapper {
     this.client = targetClient;
     this.hostInfo = hostInfo;
     this.properties = properties;
+    this.sessionState = new SessionState();
     this.id = uniqueId("PgClient_");
+
+    this.setSessionStateDefault();
   }
 
   query(sql: any): Promise<any> {
@@ -64,4 +70,10 @@ export class PgClientWrapper implements ClientWrapper {
       // Ignore
     }
   }
+
+  setSessionStateDefault() {
+    this.sessionState.readOnly.value = false;
+    this.sessionState.schema.value = "";
+    this.sessionState.transactionIsolation.value = TransactionIsolationLevel.TRANSACTION_READ_COMMITTED;
+  }
 }

common/lib/plugin_service.ts

@@ -43,6 +43,7 @@ import { DatabaseDialectCodes } from "./database_dialect/database_dialect_codes"
 import { getWriter } from "./utils/utils";
 import { TelemetryFactory } from "./utils/telemetry/telemetry_factory";
 import { DriverDialect } from "./driver_dialect/driver_dialect";
+import { SessionState } from "./session_state";
 
 export class PluginService implements ErrorHandler, HostListProviderService {
   private readonly _currentClient: AwsClient;
@@ -74,10 +75,9 @@ export class PluginService implements ErrorHandler, HostListProviderService {
     this.dbDialectProvider = new DatabaseDialectManager(knownDialectsByCode, dbType, this.props);
     this.driverDialect = driverDialect;
     this.initialHost = props.get(WrapperProperties.HOST.name);
-    this.sessionStateService = new SessionStateServiceImpl(this, this.props);
     container.pluginService = this;
-
     this.dialect = this.dbDialectProvider.getDialect(this.props);
+    this.sessionStateService = new SessionStateServiceImpl(this, this.props);
   }
 
   isInTransaction(): boolean {
@@ -332,7 +332,6 @@ export class PluginService implements ErrorHandler, HostListProviderService {
           this.sessionStateService.begin();
 
           try {
-            this.getCurrentClient().resetState();
             this.getCurrentClient().targetClient = newClient;
             this._currentHostInfo = hostInfo;
             await this.sessionStateService.applyCurrentSessionState(this.getCurrentClient());
@@ -385,6 +384,8 @@ export class PluginService implements ErrorHandler, HostListProviderService {
   async abortCurrentClient(): Promise<void> {
     if (this._currentClient.targetClient) {
       await this._currentClient.targetClient.abort();
+      // this.setInTransaction(false);
+      // this.getSessionStateService().reset();
     }
   }
 
@@ -431,35 +432,35 @@ export class PluginService implements ErrorHandler, HostListProviderService {
   private async updateReadOnly(statements: string[]) {
     const updateReadOnly = SqlMethodUtils.doesSetReadOnly(statements, this.getDialect());
     if (updateReadOnly !== undefined) {
-      await this.getCurrentClient().setReadOnly(updateReadOnly);
+      this.getSessionStateService().setReadOnly(updateReadOnly);
     }
   }
 
   private async updateAutoCommit(statements: string[]) {
     const updateAutoCommit = SqlMethodUtils.doesSetAutoCommit(statements, this.getDialect());
     if (updateAutoCommit !== undefined) {
-      await this.getCurrentClient().setAutoCommit(updateAutoCommit);
+      this.getSessionStateService().setAutoCommit(updateAutoCommit);
     }
   }
 
   private async updateCatalog(statements: string[]) {
     const updateCatalog = SqlMethodUtils.doesSetCatalog(statements, this.getDialect());
     if (updateCatalog !== undefined) {
-      await this.getCurrentClient().setCatalog(updateCatalog);
+      this.getSessionStateService().setCatalog(updateCatalog);
     }
   }
 
   private async updateSchema(statements: string[]) {
     const updateSchema = SqlMethodUtils.doesSetSchema(statements, this.getDialect());
     if (updateSchema !== undefined) {
-      await this.getCurrentClient().setSchema(updateSchema);
+      this.getSessionStateService().setSchema(updateSchema);
     }
   }
 
   private async updateTransactionIsolation(statements: string[]) {
     const updateTransactionIsolation = SqlMethodUtils.doesSetTransactionIsolation(statements, this.getDialect());
     if (updateTransactionIsolation !== undefined) {
-      await this.getCurrentClient().setTransactionIsolation(updateTransactionIsolation);
+      this.getSessionStateService().setTransactionIsolation(updateTransactionIsolation);
     }
   }
 

common/lib/pool_client_wrapper.ts

@@ -18,12 +18,14 @@ import { ClientWrapper } from "./client_wrapper";
 import { HostInfo } from "./host_info";
 import { uniqueId } from "../logutils";
 import { ClientUtils } from "./utils/client_utils";
+import { SessionState } from "./session_state";
 
 export class PoolClientWrapper implements ClientWrapper {
   readonly client: any;
   readonly hostInfo: HostInfo;
   readonly properties: Map<string, string>;
   readonly id: string;
+  readonly sessionState = new SessionState();
 
   constructor(targetClient: any, hostInfo: HostInfo, properties: Map<string, any>) {
     this.client = targetClient;
@@ -55,4 +57,6 @@ export class PoolClientWrapper implements ClientWrapper {
   rollback(): Promise<void> {
     return this.client?.rollback();
   }
+
+  setSessionStateDefault() {}
 }

common/lib/session_state.ts

@@ -85,6 +85,26 @@ export class SessionState {
   schema: SessionStateField<string> = new SessionStateField<string>();
   transactionIsolation: SessionStateField<number> = new SessionStateField<number>();
 
+  setAutoCommit(sessionState: SessionState): void {
+    this.autoCommit.value = sessionState.autoCommit.value;
+  }
+
+  setReadOnly(sessionState: SessionState): void {
+    this.readOnly.value = sessionState.readOnly.value;
+  }
+
+  setCatalog(sessionState: SessionState): void {
+    this.catalog.value = sessionState.catalog.value;
+  }
+
+  setSchema(sessionState: SessionState): void {
+    this.schema.value = sessionState.schema.value;
+  }
+
+  setTransactionIsolation(sessionState: SessionState): void {
+    this.transactionIsolation.value = sessionState.transactionIsolation.value;
+  }
+
   copy(): SessionState {
     const newSessionState = new SessionState();
     newSessionState.autoCommit = this.autoCommit.copy();

common/lib/session_state_service.ts

@@ -28,6 +28,7 @@ export interface SessionStateService {
   setReadOnly(readOnly: boolean): void;
   setupPristineReadOnly(): boolean | undefined;
   setupPristineReadOnly(readOnly: boolean): boolean | undefined;
+  updateReadOnly(readOnly: boolean): void;
 
   // catalog
   getCatalog(): string | undefined;