databricks
diff --git a/‎lib/DBSQLClient.ts‎
Lines changed: 24 additions & 3 deletions b/‎lib/DBSQLClient.ts‎
Lines changed: 24 additions & 3 deletions
diff --git a/‎lib/DBSQLOperation.ts‎
Lines changed: 27 additions & 8 deletions b/‎lib/DBSQLOperation.ts‎
Lines changed: 27 additions & 8 deletions
diff --git a/‎lib/DBSQLSession.ts‎
Lines changed: 51 additions & 46 deletions b/‎lib/DBSQLSession.ts‎
Lines changed: 51 additions & 46 deletions
@@ -242,22 +242,43 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I
     // doesn't ship in the public `.d.ts`. Mirrors Python's `kwargs.get("use_sea")`
     // pattern (see databricks-sql-python/src/databricks/sql/session.py).
     const internalOptions = options as ConnectionOptions & InternalConnectionOptions;
-    this.backend = internalOptions.useSEA
+    const backend = internalOptions.useSEA
       ? new SeaBackend()
       : new ThriftBackend({
           context: this,
           onConnectionEvent: (event, payload) => this.forwardConnectionEvent(event, payload),
         });
 
-    await this.backend.connect(options);
+    // Publish `this.backend` only after a successful `connect()`. Otherwise a
+    // failed connect would leave a half-initialized backend in place, and the
+    // next `openSession()` would slip past the `!this.backend` guard and
+    // surface a misleading "backend not implemented" / partial-state error
+    // instead of the accurate "DBSQLClient: not connected".
+    try {
+      await backend.connect(options);
+    } catch (err) {
+      // `IBackend.close()` is documented as safe on a partially-initialized
+      // backend; best-effort cleanup so we don't leak sockets / state.
+      try {
+        await backend.close();
+      } catch (closeErr) {
+        // Swallow; the original error is what the caller needs to see.
+      }
+      throw err;
+    }
+    this.backend = backend;
 
     return this;
   }
 
   private forwardConnectionEvent(event: 'error' | 'reconnecting' | 'close' | 'timeout', payload?: unknown): void {
     switch (event) {
       case 'error': {
-        const error = payload as Error;
+        // `payload` is typed `unknown` because the cross-backend
+        // `IBackend.onConnectionEvent` doesn't constrain the error shape.
+        // Normalize to `Error` so the stack/name/message access below is safe
+        // for any backend that emits a non-Error value (e.g. a bare string).
+        const error = payload instanceof Error ? payload : new Error(String(payload));
         this.logger.log(LogLevel.error, error.stack || `${error.name}: ${error.message}`);
         try {
           this.emit('error', error);
 
@@ -19,10 +19,10 @@ import { LogLevel } from './contracts/IDBSQLLogger';
 import OperationStateError, { OperationStateErrorCode } from './errors/OperationStateError';
 import { OperationChunksIterator, OperationRowsIterator } from './utils/OperationIterator';
 import IClientContext from './contracts/IClientContext';
-import IOperationBackend from './contracts/IOperationBackend';
+import IOperationBackend, { IOperationBackendWaitOptions } from './contracts/IOperationBackend';
 import { ResultMetadata } from './contracts/ResultMetadata';
 import ThriftOperationBackend from './thrift-backend/ThriftOperationBackend';
-import { synthesizeThriftStatus, synthesizeThriftResultSetMetadata } from './utils/thriftWireSynthesis';
+import { synthesizeThriftStatus, synthesizeThriftResultSetMetadata } from './thrift-backend/wireSynthesis';
 
 interface DBSQLOperationConstructorOptions {
   backend: IOperationBackend;
@@ -114,7 +114,7 @@ export default class DBSQLOperation implements IOperation {
   public async fetchChunk(options?: FetchOptions): Promise<Array<object>> {
     await this.failIfClosed();
 
-    if (!this.backend.hasResultSet) {
+    if (!this.backend.hasResultSet()) {
       return [];
     }
 
@@ -123,7 +123,11 @@ export default class DBSQLOperation implements IOperation {
 
     const defaultMaxRows = this.context.getConfig().fetchChunkDefaultMaxRows;
     const limit = options?.maxRows ?? defaultMaxRows;
-    const result = await this.backend.fetchChunk({ limit, disableBuffering: options?.disableBuffering });
+    const result = await this.backend.fetchChunk({
+      limit,
+      disableBuffering: options?.disableBuffering,
+      isClosed: () => this.closed || this.cancelled,
+    });
     await this.failIfClosed();
 
     this.context.getLogger().log(LogLevel.debug, `Fetched chunk of size: ${limit} from operation with id: ${this.id}`);
@@ -192,7 +196,7 @@ export default class DBSQLOperation implements IOperation {
       return false;
     }
 
-    if (this.backend.hasResultSet) {
+    if (this.backend.hasResultSet()) {
       await this.waitUntilReadyThroughBackend();
     }
 
@@ -202,7 +206,7 @@ export default class DBSQLOperation implements IOperation {
   public async getSchema(options?: GetSchemaOptions): Promise<TTableSchema | null> {
     await this.failIfClosed();
 
-    if (!this.backend.hasResultSet) {
+    if (!this.backend.hasResultSet()) {
       return null;
     }
 
@@ -227,7 +231,9 @@ export default class DBSQLOperation implements IOperation {
    * fields (`cacheLookupResult`, `uncompressedBytes`, `compressedBytes`,
    * `status`) left undefined / defaulted.
    *
-   * Prefer {@link DBSQLOperation.getResultMetadata} in new code.
+   * @deprecated Use {@link DBSQLOperation.getResultMetadata}; this method
+   * synthesizes Thrift-only fields as `undefined` on non-Thrift backends and
+   * couples callers to the Thrift wire shape.
    */
   public async getMetadata(): Promise<TGetResultSetMetadataResp> {
     await this.failIfClosed();
@@ -248,8 +254,21 @@ export default class DBSQLOperation implements IOperation {
   }
 
   private async waitUntilReadyThroughBackend(options?: WaitUntilReadyOptions) {
+    // The backend-facing `waitUntilReady` takes a neutral
+    // `IOperationBackendWaitOptions` whose `callback` receives an
+    // `OperationStatus`. The public `WaitUntilReadyOptions.callback` is
+    // Thrift-shaped — synthesize the wire response from the neutral status
+    // at this boundary so the backend impl doesn't have to know about Thrift
+    // IDL.
+    const userCallback = options?.callback;
+    const backendOptions: IOperationBackendWaitOptions = {
+      progress: options?.progress,
+      callback: userCallback
+        ? (status) => userCallback(synthesizeThriftStatus(status))
+        : undefined,
+    };
     try {
-      await this.backend.waitUntilReady(options);
+      await this.backend.waitUntilReady(backendOptions);
     } catch (err) {
       // Reflect terminal states back into facade flags so subsequent calls
       // short-circuit via failIfClosed().
 
@@ -26,12 +26,20 @@ import StagingError from './errors/StagingError';
 import IClientContext from './contracts/IClientContext';
 import ISessionBackend from './contracts/ISessionBackend';
 import IOperationBackend from './contracts/IOperationBackend';
+import { numberToInt64 as numberToInt64Impl } from './thrift-backend/ThriftSessionBackend';
 
 // Explicitly promisify a callback-style `pipeline` because `node:stream/promises` is not available in Node 14
 const pipeline = util.promisify(stream.pipeline);
 
-// Re-export for back-compat with existing imports.
-export { numberToInt64 } from './thrift-backend/ThriftSessionBackend';
+/**
+ * Convert a JS number to a Thrift-wire `node-int64`.
+ *
+ * @deprecated Thrift-only utility re-exported for back-compat with existing
+ * external consumers. Backends other than Thrift do not use `node-int64`;
+ * new code should not import this from `DBSQLSession`. It will be removed
+ * when the public API stops exposing Thrift wire types.
+ */
+export const numberToInt64 = numberToInt64Impl;
 
 interface DBSQLSessionConstructorOptions {
   backend: ISessionBackend;
@@ -68,10 +76,7 @@ export default class DBSQLSession implements IDBSQLSession {
    * const response = await session.getInfo(thrift.TCLIService_types.TGetInfoType.CLI_DBMS_VER);
    */
   public async getInfo(infoType: number): Promise<InfoValue> {
-    await this.failIfClosed();
-    const result = await this.backend.getInfo(infoType);
-    await this.failIfClosed();
-    return result;
+    return this.runBackend(() => this.backend.getInfo(infoType));
   }
 
   /**
@@ -84,12 +89,16 @@ export default class DBSQLSession implements IDBSQLSession {
    * const operation = await session.executeStatement(query);
    */
   public async executeStatement(statement: string, options: ExecuteStatementOptions = {}): Promise<IOperation> {
-    await this.failIfClosed();
-    const opBackend = await this.backend.executeStatement(statement, options);
-    await this.failIfClosed();
+    const opBackend = await this.runBackend(() => this.backend.executeStatement(statement, options));
     const operation = this.wrapOperation(opBackend);
 
-    // Staging detection: only run when stagingAllowedLocalPath is provided.
+    // If `stagingAllowedLocalPath` is provided - assume that operation possibly may be a staging operation.
+    // To know for sure, fetch metadata and check a `isStagingOperation` flag. If it happens that it wasn't
+    // a staging operation - not a big deal, we just fetched metadata earlier, but operation is still usable
+    // and user can get data from it.
+    // If `stagingAllowedLocalPath` is not provided - don't do anything to the operation. In a case of regular
+    // operation, everything will work as usual. In a case of staging operation, it will be processed like any
+    // other query - it will be possible to get data from it as usual, or use other operation methods.
     if (options.stagingAllowedLocalPath !== undefined) {
       const metadata = await operation.getResultMetadata();
       if (metadata.isStagingOperation) {
@@ -174,6 +183,13 @@ export default class DBSQLSession implements IDBSQLSession {
     const agent = await connectionProvider.getAgent();
 
     const response = await fetch(presignedUrl, { method: 'DELETE', headers, agent });
+    // Looks that AWS and Azure have a different behavior of HTTP `DELETE` for non-existing files.
+    // AWS assumes that - since file already doesn't exist - the goal is achieved, and returns HTTP 200.
+    // Azure, on the other hand, is somewhat stricter and check if file exists before deleting it. And if
+    // file doesn't exist - Azure returns HTTP 404.
+    //
+    // For us, it's totally okay if file didn't exist before removing. So when we get an HTTP 404 -
+    // just ignore it and report success. This way we can have a uniform library behavior for all clouds
     if (!response.ok && response.status !== 404) {
       throw new StagingError(`HTTP error ${response.status} ${response.statusText}`);
     }
@@ -198,6 +214,7 @@ export default class DBSQLSession implements IDBSQLSession {
       method: 'PUT',
       headers: {
         ...headers,
+        // This header is required by server
         'Content-Length': fileInfo.size.toString(),
       },
       agent,
@@ -215,10 +232,7 @@ export default class DBSQLSession implements IDBSQLSession {
    * @returns DBSQLOperation
    */
   public async getTypeInfo(request: TypeInfoRequest = {}): Promise<IOperation> {
-    await this.failIfClosed();
-    const opBackend = await this.backend.getTypeInfo(request);
-    await this.failIfClosed();
-    return this.wrapOperation(opBackend);
+    return this.wrapOperation(await this.runBackend(() => this.backend.getTypeInfo(request)));
   }
 
   /**
@@ -228,10 +242,7 @@ export default class DBSQLSession implements IDBSQLSession {
    * @returns DBSQLOperation
    */
   public async getCatalogs(request: CatalogsRequest = {}): Promise<IOperation> {
-    await this.failIfClosed();
-    const opBackend = await this.backend.getCatalogs(request);
-    await this.failIfClosed();
-    return this.wrapOperation(opBackend);
+    return this.wrapOperation(await this.runBackend(() => this.backend.getCatalogs(request)));
   }
 
   /**
@@ -241,10 +252,7 @@ export default class DBSQLSession implements IDBSQLSession {
    * @returns DBSQLOperation
    */
   public async getSchemas(request: SchemasRequest = {}): Promise<IOperation> {
-    await this.failIfClosed();
-    const opBackend = await this.backend.getSchemas(request);
-    await this.failIfClosed();
-    return this.wrapOperation(opBackend);
+    return this.wrapOperation(await this.runBackend(() => this.backend.getSchemas(request)));
   }
 
   /**
@@ -254,10 +262,7 @@ export default class DBSQLSession implements IDBSQLSession {
    * @returns DBSQLOperation
    */
   public async getTables(request: TablesRequest = {}): Promise<IOperation> {
-    await this.failIfClosed();
-    const opBackend = await this.backend.getTables(request);
-    await this.failIfClosed();
-    return this.wrapOperation(opBackend);
+    return this.wrapOperation(await this.runBackend(() => this.backend.getTables(request)));
   }
 
   /**
@@ -267,10 +272,7 @@ export default class DBSQLSession implements IDBSQLSession {
    * @returns DBSQLOperation
    */
   public async getTableTypes(request: TableTypesRequest = {}): Promise<IOperation> {
-    await this.failIfClosed();
-    const opBackend = await this.backend.getTableTypes(request);
-    await this.failIfClosed();
-    return this.wrapOperation(opBackend);
+    return this.wrapOperation(await this.runBackend(() => this.backend.getTableTypes(request)));
   }
 
   /**
@@ -280,10 +282,7 @@ export default class DBSQLSession implements IDBSQLSession {
    * @returns DBSQLOperation
    */
   public async getColumns(request: ColumnsRequest = {}): Promise<IOperation> {
-    await this.failIfClosed();
-    const opBackend = await this.backend.getColumns(request);
-    await this.failIfClosed();
-    return this.wrapOperation(opBackend);
+    return this.wrapOperation(await this.runBackend(() => this.backend.getColumns(request)));
   }
 
   /**
@@ -293,17 +292,11 @@ export default class DBSQLSession implements IDBSQLSession {
    * @returns DBSQLOperation
    */
   public async getFunctions(request: FunctionsRequest): Promise<IOperation> {
-    await this.failIfClosed();
-    const opBackend = await this.backend.getFunctions(request);
-    await this.failIfClosed();
-    return this.wrapOperation(opBackend);
+    return this.wrapOperation(await this.runBackend(() => this.backend.getFunctions(request)));
   }
 
   public async getPrimaryKeys(request: PrimaryKeysRequest): Promise<IOperation> {
-    await this.failIfClosed();
-    const opBackend = await this.backend.getPrimaryKeys(request);
-    await this.failIfClosed();
-    return this.wrapOperation(opBackend);
+    return this.wrapOperation(await this.runBackend(() => this.backend.getPrimaryKeys(request)));
   }
 
   /**
@@ -313,10 +306,7 @@ export default class DBSQLSession implements IDBSQLSession {
    * @returns DBSQLOperation
    */
   public async getCrossReference(request: CrossReferenceRequest): Promise<IOperation> {
-    await this.failIfClosed();
-    const opBackend = await this.backend.getCrossReference(request);
-    await this.failIfClosed();
-    return this.wrapOperation(opBackend);
+    return this.wrapOperation(await this.runBackend(() => this.backend.getCrossReference(request)));
   }
 
   /**
@@ -346,6 +336,21 @@ export default class DBSQLSession implements IDBSQLSession {
     return operation;
   }
 
+  /**
+   * Bracket a backend call with `failIfClosed()` on both sides. The pre-call
+   * check rejects work against an already-closed session; the post-call check
+   * rejects results that came back after a concurrent close (server-side
+   * close doesn't error out the in-flight RPC). Centralizing the pattern
+   * keeps the 10+ delegation methods readable and makes the contract
+   * impossible to forget.
+   */
+  private async runBackend<T>(fn: () => Promise<T>): Promise<T> {
+    await this.failIfClosed();
+    const result = await fn();
+    await this.failIfClosed();
+    return result;
+  }
+
   private async failIfClosed(): Promise<void> {
     if (!this.isOpen) {
       throw new HiveDriverError('The session was closed or has expired');