feat: auto-recover Reyden Thrift connections onto the kernel backend

An unconfigured connection to a Reyden / Real-Time SQL warehouse defaults to
the Thrift backend, which the SQL Gateway proxy rejects with SQLSTATE KP001.
Detect that rejection (StatusError.sqlState === "KP001") in
ThriftBackend.openSession and transparently re-open the session on the
KernelBackend (SEA), remembering the warehouse in a process-wide cache keyed
by (host, warehouse_id) with a ~6h TTL so later connects skip the doomed
Thrift attempt. Only the default path auto-recovers; an explicit backend
choice (routed upstream in the client) is unaffected. On a double failure the
kernel error is surfaced with the original Thrift rejection preserved as its
cause.

Also re-throw the original error unchanged on the non-recovery paths:
StatusError implements Error but does not extend it, so the previous
`error instanceof Error ? error : new Error(String(error))` normalization
wrapped every StatusError into Error("[object Object]"), losing its sqlState
and the double-failure cause.

Co-authored-by: Isaac <no-reply@databricks.com>
Signed-off-by: Rahul Singhal <rahul.singhal@databricks.com>

Author: rahuls-db

lib/ReydenWarehouseCache.ts

@@ -0,0 +1,106 @@
+/**
+ * Process-wide cache for tracking Reyden (Real-Time SQL) warehouses.
+ *
+ * When a Thrift OpenSession fails with SQLSTATE KP001, the driver falls back
+ * to the SEA (Statement Execution API) backend. This cache avoids retrying
+ * the same failed Thrift path on subsequent connections by recording which
+ * warehouses are known to require SEA.
+ *
+ * The cache is keyed by (host_lowercased, warehouse_id) to handle multi-tenant
+ * safety — the same warehouse ID on different hosts may have different support.
+ *
+ * TTL is ~6 hours to allow the server side to update warehouse routing without
+ * requiring a process restart. Expired entries are opportunistically evicted on
+ * access (no background GC thread — Node is single-threaded).
+ */
+
+const TTL_MS = 6 * 60 * 60 * 1000; // 6 hours
+
+interface CacheEntry {
+  timestamp: number;
+  isReyden: boolean;
+}
+
+class ReydenWarehouseCache {
+  private static instance?: ReydenWarehouseCache;
+
+  private cache: Map<string, CacheEntry> = new Map();
+
+  // Singleton: constructor is private to enforce getInstance() usage
+  // eslint-disable-next-line @typescript-eslint/no-empty-function
+  private constructor() {}
+
+  public static getInstance(): ReydenWarehouseCache {
+    if (!ReydenWarehouseCache.instance) {
+      ReydenWarehouseCache.instance = new ReydenWarehouseCache();
+    }
+    return ReydenWarehouseCache.instance;
+  }
+
+  /**
+   * Constructs a cache key from host and warehouse ID.
+   * Host is lowercased for case-insensitive comparison.
+   */
+  private getKey(host: string, warehouseId: string): string {
+    return `${host.toLowerCase()}:${warehouseId}`;
+  }
+
+  /**
+   * Check if an entry is expired based on TTL.
+   */
+  private isExpired(entry: CacheEntry): boolean {
+    return Date.now() - entry.timestamp > TTL_MS;
+  }
+
+  /**
+   * Checks if a warehouse is known to be Reyden (requiring SEA fallback).
+   * Returns undefined if the warehouse is not in the cache or the entry has expired.
+   */
+  public isKnownReyden(host: string, warehouseId: string): boolean | undefined {
+    const key = this.getKey(host, warehouseId);
+    const entry = this.cache.get(key);
+
+    if (!entry) {
+      return undefined;
+    }
+
+    // Opportunistically evict expired entries on access
+    if (this.isExpired(entry)) {
+      this.cache.delete(key);
+      return undefined;
+    }
+
+    return entry.isReyden;
+  }
+
+  /**
+   * Mark a warehouse as being Reyden (KP001 rejection detected).
+   */
+  public markReyden(host: string, warehouseId: string): void {
+    const key = this.getKey(host, warehouseId);
+    this.cache.set(key, {
+      timestamp: Date.now(),
+      isReyden: true,
+    });
+  }
+
+  /**
+   * Clears the cache. Intended for testing only.
+   *
+   * @internal
+   */
+  public clear(): void {
+    this.cache.clear();
+  }
+
+  /**
+   * Returns the current cache size. Intended for testing/observability.
+   *
+   * @internal
+   */
+  public size(): number {
+    return this.cache.size;
+  }
+}
+
+export default ReydenWarehouseCache.getInstance();

lib/errors/StatusError.ts

@@ -7,12 +7,15 @@ export default class StatusError implements Error {
 
   public code: number;
 
+  public sqlState?: string;
+
   public stack?: string;
 
   constructor(status: TStatus) {
     this.name = 'Status Error';
     this.message = status.errorMessage || '';
     this.code = status.errorCode || -1;
+    this.sqlState = status.sqlState;
 
     if (Array.isArray(status.infoMessages)) {
       this.stack = status.infoMessages.join('\n');

lib/thrift-backend/ThriftBackend.ts

@@ -2,11 +2,15 @@ import Int64 from 'node-int64';
 import IBackend from '../contracts/IBackend';
 import ISessionBackend from '../contracts/ISessionBackend';
 import IClientContext from '../contracts/IClientContext';
-import { OpenSessionRequest } from '../contracts/IDBSQLClient';
+import { ConnectionOptions, OpenSessionRequest } from '../contracts/IDBSQLClient';
 import { TProtocolVersion } from '../../thrift/TCLIService_types';
 import Status from '../dto/Status';
 import { definedOrError, serializeQueryTags } from '../utils';
 import ThriftSessionBackend from './ThriftSessionBackend';
+import StatusError from '../errors/StatusError';
+import reydenCache from '../ReydenWarehouseCache';
+import KernelBackend from '../kernel/KernelBackend';
+import { LogLevel } from '../contracts/IDBSQLLogger';
 
 function getInitialNamespaceOptions(catalogName?: string, schemaName?: string) {
   if (!catalogName && !schemaName) {
@@ -31,12 +35,36 @@ export default class ThriftBackend implements IBackend {
 
   private readonly onConnectionEvent: ThriftBackendOptions['onConnectionEvent'];
 
+  private connectionOptions?: ConnectionOptions;
+
   constructor({ context, onConnectionEvent }: ThriftBackendOptions) {
     this.context = context;
     this.onConnectionEvent = onConnectionEvent;
   }
 
-  public async connect(): Promise<void> {
+  /**
+   * Extracts warehouse/endpoint ID from the HTTP path.
+   * Matches patterns like `/sql/1.0/warehouses/<id>` or `/sql/1.0/endpoints/<id>`.
+   * Returns undefined if no ID can be extracted.
+   */
+  private static extractWarehouseId(httpPath: string | undefined): string | undefined {
+    if (!httpPath) {
+      return undefined;
+    }
+
+    // Stop at query string
+    const pathOnly = httpPath.split('?')[0];
+
+    // Match `/warehouses/<id>` or `/endpoints/<id>`
+    // Stop at `/` or end of string
+    const match = pathOnly.match(/\/(warehouses|endpoints)\/([^/]+)/);
+    return match ? match[2] : undefined;
+  }
+
+  public async connect(options: ConnectionOptions): Promise<void> {
+    // Store connection options for warehouse ID extraction in openSession
+    this.connectionOptions = options;
+
     // The connection provider is owned by DBSQLClient (it implements IClientContext).
     // We only need to wire the EventEmitter listeners through this backend.
     const connectionProvider = await this.context.getConnectionProvider();
@@ -60,6 +88,57 @@ export default class ThriftBackend implements IBackend {
   }
 
   public async openSession(request: OpenSessionRequest): Promise<ISessionBackend> {
+    const logger = this.context.getLogger();
+
+    // Extract warehouse ID for cache lookups
+    const warehouseId = ThriftBackend.extractWarehouseId(this.connectionOptions?.path);
+    const host = this.connectionOptions?.host;
+
+    // Check if this warehouse is known to be Reyden (requires SEA backend)
+    if (host && warehouseId && reydenCache.isKnownReyden(host, warehouseId)) {
+      logger.log(LogLevel.debug, `Reyden: warehouse ${warehouseId} is known to require SEA fallback; skipping Thrift`);
+      return this.openSessionWithKernelBackend(request);
+    }
+
+    // Try Thrift first (default path).
+    try {
+      return await this.openSessionWithThrift(request);
+    } catch (error) {
+      // Only a Reyden KP001 rejection triggers fallback. Every other error
+      // propagates unchanged — note StatusError is NOT an Error subclass
+      // (it only `implements Error`), so it must be re-thrown as-is rather
+      // than normalized, or its sqlState/message would be lost.
+      if (error instanceof StatusError && error.sqlState === 'KP001') {
+        logger.log(LogLevel.debug, `Reyden: detected KP001 on warehouse ${warehouseId}; falling back to SEA backend`);
+
+        // Mark this warehouse as Reyden for future connections.
+        if (host && warehouseId) {
+          reydenCache.markReyden(host, warehouseId);
+        }
+
+        // Fall back to the kernel (SEA) backend exactly once. If it also fails,
+        // surface the kernel error but keep the original Thrift rejection as its
+        // cause for diagnosis.
+        try {
+          return await this.openSessionWithKernelBackend(request);
+        } catch (kernelError) {
+          if (kernelError && typeof kernelError === 'object') {
+            (kernelError as { cause?: unknown }).cause = error;
+          }
+          logger.log(LogLevel.error, 'Reyden: both Thrift (KP001) and SEA fallback failed');
+          throw kernelError;
+        }
+      }
+
+      // Not a Reyden rejection — surface the original error unchanged.
+      throw error;
+    }
+  }
+
+  /**
+   * Opens a session using the Thrift backend.
+   */
+  private async openSessionWithThrift(request: OpenSessionRequest): Promise<ISessionBackend> {
     const driver = await this.context.getDriver();
     const config = this.context.getConfig();
 
@@ -93,6 +172,24 @@ export default class ThriftBackend implements IBackend {
     });
   }
 
+  /**
+   * Opens a session using the KernelBackend (SEA).
+   * Called as a fallback when Thrift returns KP001 (Reyden rejection).
+   */
+  private async openSessionWithKernelBackend(request: OpenSessionRequest): Promise<ISessionBackend> {
+    if (!this.connectionOptions) {
+      throw new Error('KernelBackend fallback: connection options not available');
+    }
+
+    const logger = this.context.getLogger();
+    logger.log(LogLevel.debug, 'Reyden: opening session via KernelBackend (SEA)');
+
+    // Create a new KernelBackend instance and connect/open
+    const kernelBackend = new KernelBackend({ context: this.context });
+    await kernelBackend.connect(this.connectionOptions);
+    return kernelBackend.openSession(request);
+  }
+
   public async close(): Promise<void> {
     // DBSQLClient owns the connection lifecycle and clears its own state
     // (connectionProvider, authProvider, thrift client) after this returns.

tests/unit/thrift-backend/ReydenThriftRecovery.test.ts

@@ -0,0 +1,131 @@
+import { expect } from 'chai';
+import reydenCache from '../../../lib/ReydenWarehouseCache';
+import ThriftBackend from '../../../lib/thrift-backend/ThriftBackend';
+import StatusError from '../../../lib/errors/StatusError';
+import { TStatusCode } from '../../../thrift/TCLIService_types';
+
+describe('Reyden Warehouse Cache', () => {
+  beforeEach(() => {
+    reydenCache.clear();
+  });
+
+  afterEach(() => {
+    reydenCache.clear();
+  });
+
+  describe('Warehouse ID Extraction', () => {
+    it('should extract warehouse ID from /warehouses/<id> path', () => {
+      const extractWarehouseId = (ThriftBackend as any).extractWarehouseId;
+      expect(extractWarehouseId('/sql/1.0/warehouses/abc123')).to.equal('abc123');
+    });
+
+    it('should extract endpoint ID from /endpoints/<id> path', () => {
+      const extractWarehouseId = (ThriftBackend as any).extractWarehouseId;
+      expect(extractWarehouseId('/sql/1.0/endpoints/xyz789')).to.equal('xyz789');
+    });
+
+    it('should stop at query string when extracting warehouse ID', () => {
+      const extractWarehouseId = (ThriftBackend as any).extractWarehouseId;
+      expect(extractWarehouseId('/sql/1.0/warehouses/abc123?o=12345')).to.equal('abc123');
+    });
+
+    it('should return undefined if no warehouse ID is found', () => {
+      const extractWarehouseId = (ThriftBackend as any).extractWarehouseId;
+      expect(extractWarehouseId('/some/other/path')).to.be.undefined;
+    });
+
+    it('should return undefined for undefined path', () => {
+      const extractWarehouseId = (ThriftBackend as any).extractWarehouseId;
+      expect(extractWarehouseId(undefined)).to.be.undefined;
+    });
+  });
+
+  describe('Cache Operations', () => {
+    it('should mark a warehouse as Reyden', () => {
+      const host = 'example.com';
+      const warehouseId = 'warehouse-123';
+
+      expect(reydenCache.isKnownReyden(host, warehouseId)).to.be.undefined;
+      reydenCache.markReyden(host, warehouseId);
+      expect(reydenCache.isKnownReyden(host, warehouseId)).to.be.true;
+    });
+
+    it('should be case-insensitive on host', () => {
+      const warehouseId = 'warehouse-123';
+
+      reydenCache.markReyden('Example.COM', warehouseId);
+
+      expect(reydenCache.isKnownReyden('example.com', warehouseId)).to.be.true;
+      expect(reydenCache.isKnownReyden('EXAMPLE.COM', warehouseId)).to.be.true;
+    });
+
+    it('should isolate entries by warehouse ID', () => {
+      const host = 'example.com';
+
+      reydenCache.markReyden(host, 'warehouse-1');
+
+      expect(reydenCache.isKnownReyden(host, 'warehouse-1')).to.be.true;
+      expect(reydenCache.isKnownReyden(host, 'warehouse-2')).to.be.undefined;
+    });
+
+    it('should isolate entries by host', () => {
+      const warehouseId = 'warehouse-123';
+
+      reydenCache.markReyden('host1.com', warehouseId);
+
+      expect(reydenCache.isKnownReyden('host1.com', warehouseId)).to.be.true;
+      expect(reydenCache.isKnownReyden('host2.com', warehouseId)).to.be.undefined;
+    });
+
+    it('should have cache size method', () => {
+      expect(reydenCache.size()).to.equal(0);
+
+      reydenCache.markReyden('host1.com', 'warehouse-1');
+      expect(reydenCache.size()).to.equal(1);
+
+      reydenCache.markReyden('host1.com', 'warehouse-2');
+      expect(reydenCache.size()).to.equal(2);
+    });
+
+    it('should clear cache', () => {
+      reydenCache.markReyden('host1.com', 'warehouse-1');
+      reydenCache.markReyden('host2.com', 'warehouse-2');
+      expect(reydenCache.size()).to.equal(2);
+
+      reydenCache.clear();
+      expect(reydenCache.size()).to.equal(0);
+      expect(reydenCache.isKnownReyden('host1.com', 'warehouse-1')).to.be.undefined;
+    });
+  });
+});
+
+describe('StatusError SQLSTATE Support', () => {
+  it('should capture SQLSTATE in StatusError', () => {
+    const error = new StatusError({
+      statusCode: TStatusCode.ERROR_STATUS,
+      errorMessage: 'Some error',
+      sqlState: 'KP001',
+    });
+
+    expect(error.sqlState).to.equal('KP001');
+  });
+
+  it('should handle undefined SQLSTATE', () => {
+    const error = new StatusError({
+      statusCode: TStatusCode.ERROR_STATUS,
+      errorMessage: 'Some error',
+    });
+
+    expect(error.sqlState).to.be.undefined;
+  });
+
+  it('should detect KP001 errors correctly', () => {
+    const kp001Error = new StatusError({
+      statusCode: TStatusCode.ERROR_STATUS,
+      errorMessage: 'Lakehouse/RT is not supported for Thrift protocol',
+      sqlState: 'KP001',
+    });
+
+    expect(kp001Error.sqlState).to.equal('KP001');
+  });
+});