[ES-1717770] Fix timeout exception during fetch results (databricks#1199)

## Description
- This fix was made in the [PR
databricks#1135](databricks#1135),
but it did not consider the fetch result timeout. This PR fixes that.

## Testing
- unit tests

## Additional Notes to the Reviewer
- Most of the code changes are propogating errors as `SQLException`, but
the main code change is a small one and is present in
`src/main/java/com/databricks/jdbc/common/util/DatabricksThriftUtil.java`
file

OVERRIDE_FREEZE=true

Author: samikshya-db

NEXT_CHANGELOG.md

@@ -18,6 +18,7 @@
 - **Enhanced `enableMultipleCatalogSupport` behavior**: When this parameter is disabled (`enableMultipleCatalogSupport=0`), metadata operations (such as `getSchemas()`, `getTables()`, `getColumns()`, etc.) now return results only when the catalog parameter is either `null` or matches the current catalog. For any other catalog name, an empty result set is returned. This ensures metadata queries are restricted to the current catalog context. When enabled (`enableMultipleCatalogSupport=1`), metadata operations continue to work across all accessible catalogs.
 
 ### Fixed
+- Fixed timeout exception handling to throw `SQLTimeoutException` instead of `DatabricksHttpException` when queries timeout during result fetching phase. This completes the timeout exception fix to handle both query execution polling and result fetching phases.
 - Fixed `getTypeInfo()` and `getClientInfoProperties()` to return fresh ResultSet instances on each call instead of shared static instances. This resolves issues where calling these methods multiple times would fail due to exhausted cursor state (Issue #1178).
 - Fixed complex data type metadata support when retrieving 0 rows in Arrow format
 - Normalized TIMESTAMP_NTZ to TIMESTAMP in Thrift path for consistency with SEA behavior

src/main/java/com/databricks/client/jdbc/DataSource.java

@@ -4,7 +4,6 @@
 
 import com.databricks.jdbc.common.DatabricksJdbcConstants;
 import com.databricks.jdbc.common.DatabricksJdbcUrlParams;
-import com.databricks.jdbc.exception.DatabricksSQLException;
 import com.databricks.jdbc.log.JdbcLogger;
 import com.databricks.jdbc.log.JdbcLoggerFactory;
 import com.databricks.jdbc.pooling.DatabricksPooledConnection;
@@ -37,13 +36,13 @@ public DataSource(Driver driver) {
   }
 
   @Override
-  public Connection getConnection() throws DatabricksSQLException {
+  public Connection getConnection() throws SQLException {
     LOGGER.debug("public Connection getConnection()");
     return getConnection(this.getUsername(), this.getPassword());
   }
 
   @Override
-  public Connection getConnection(String username, String password) throws DatabricksSQLException {
+  public Connection getConnection(String username, String password) throws SQLException {
     LOGGER.debug("public Connection getConnection(String, String)");
     if (username != null) {
       setUsername(username);
@@ -55,14 +54,13 @@ public Connection getConnection(String username, String password) throws Databri
   }
 
   @Override
-  public PooledConnection getPooledConnection() throws DatabricksSQLException {
+  public PooledConnection getPooledConnection() throws SQLException {
     LOGGER.debug("public PooledConnection getPooledConnection()");
     return new DatabricksPooledConnection(getConnection());
   }
 
   @Override
-  public PooledConnection getPooledConnection(String user, String password)
-      throws DatabricksSQLException {
+  public PooledConnection getPooledConnection(String user, String password) throws SQLException {
     LOGGER.debug("public PooledConnection getPooledConnection(String, String)");
     return new DatabricksPooledConnection(getConnection(user, password));
   }

src/main/java/com/databricks/client/jdbc/Driver.java

@@ -47,7 +47,7 @@ public boolean acceptsURL(String url) {
   }
 
   @Override
-  public Connection connect(String url, Properties info) throws DatabricksSQLException {
+  public Connection connect(String url, Properties info) throws SQLException {
     if (!acceptsURL(url)) {
       // Return null connection if URL is not accepted - as per JDBC standard.
       return null;

src/main/java/com/databricks/jdbc/api/impl/DatabricksConnection.java

@@ -63,7 +63,7 @@ public DatabricksConnection(
   }
 
   @Override
-  public void open() throws DatabricksSQLException {
+  public void open() throws SQLException {
     this.session.open();
   }
 
@@ -401,7 +401,7 @@ public void rollback() throws SQLException {
   }
 
   @Override
-  public void close() throws DatabricksSQLException {
+  public void close() throws SQLException {
     LOGGER.debug("public void close()");
     for (IDatabricksStatementInternal statement : statementSet) {
       statement.close(false);

src/main/java/com/databricks/jdbc/api/impl/DatabricksResultSet.java

@@ -84,7 +84,7 @@ public DatabricksResultSet(
       StatementType statementType,
       IDatabricksSession session,
       IDatabricksStatementInternal parentStatement)
-      throws DatabricksSQLException {
+      throws SQLException {
     this.executionStatus = new ExecutionStatus(statementStatus);
     this.statementId = statementId;
     if (resultData != null) {

src/main/java/com/databricks/jdbc/api/impl/DatabricksSession.java

@@ -28,6 +28,7 @@
 import com.databricks.jdbc.telemetry.latency.DatabricksMetricsTimedProcessor;
 import com.databricks.sdk.support.ToStringer;
 import com.google.common.annotations.VisibleForTesting;
+import java.sql.SQLException;
 import java.util.HashMap;
 import java.util.Map;
 import javax.annotation.Nullable;
@@ -131,7 +132,7 @@ public boolean isOpen() {
   }
 
   @Override
-  public void open() throws DatabricksSQLException {
+  public void open() throws SQLException {
     LOGGER.debug("public void open()");
 
     // Skip for tests, it would be already set
@@ -205,7 +206,7 @@ public void open() throws DatabricksSQLException {
   }
 
   @Override
-  public void close() throws DatabricksSQLException {
+  public void close() throws SQLException {
     LOGGER.debug("public void close()");
     synchronized (this) {
       if (isSessionOpen) {
@@ -361,7 +362,7 @@ public void setEmptyMetadataClient() {
   public void forceClose() {
     try {
       this.close();
-    } catch (DatabricksSQLException e) {
+    } catch (SQLException e) {
       LOGGER.error("Error closing session resources, but marking the session as closed.");
     } finally {
       this.isSessionOpen = false;

src/main/java/com/databricks/jdbc/api/impl/ExecutionResultFactory.java

@@ -32,7 +32,7 @@ static IExecutionResult getResultSet(
       StatementId statementId,
       IDatabricksSession session,
       IDatabricksStatementInternal statement)
-      throws DatabricksSQLException {
+      throws SQLException {
     IExecutionResult resultHandler = getResultHandler(data, manifest, statementId, session);
     if (manifest.getIsVolumeOperation() != null && manifest.getIsVolumeOperation()) {
       return new VolumeOperationResult(

src/main/java/com/databricks/jdbc/api/impl/IExecutionResult.java

@@ -1,6 +1,7 @@
 package com.databricks.jdbc.api.impl;
 
 import com.databricks.jdbc.exception.DatabricksSQLException;
+import java.sql.SQLException;
 
 /** Interface to provide methods over an underlying statement result */
 public interface IExecutionResult {
@@ -26,7 +27,7 @@ public interface IExecutionResult {
    *
    * @return true if cursor is moved at next row
    */
-  boolean next() throws DatabricksSQLException;
+  boolean next() throws SQLException;
 
   /** Returns if there is next row in the result set */
   boolean hasNext();

src/main/java/com/databricks/jdbc/api/impl/LazyThriftResult.java

@@ -10,6 +10,7 @@
 import com.databricks.jdbc.log.JdbcLoggerFactory;
 import com.databricks.jdbc.model.client.thrift.generated.TFetchResultsResp;
 import com.databricks.jdbc.model.telemetry.enums.DatabricksDriverErrorCode;
+import java.sql.SQLException;
 
 public class LazyThriftResult implements IExecutionResult {
   private static final JdbcLogger LOGGER = JdbcLoggerFactory.getLogger(LazyThriftResult.class);
@@ -99,10 +100,10 @@ public long getCurrentRow() {
    * Moves the cursor to the next row. Fetches additional data from server if needed.
    *
    * @return true if there is a next row, false if at the end
-   * @throws DatabricksSQLException if an error occurs while fetching data
+   * @throws SQLException if an error occurs while fetching data
    */
   @Override
-  public boolean next() throws DatabricksSQLException {
+  public boolean next() throws SQLException {
     if (isClosed || hasReachedEnd) {
       return false;
     }
@@ -221,9 +222,9 @@ private void loadCurrentBatch() throws DatabricksSQLException {
   /**
    * Fetches the next batch of data from the server and creates columnar views.
    *
-   * @throws DatabricksSQLException if the fetch operation fails
+   * @throws SQLException if the fetch operation fails
    */
-  private void fetchNextBatch() throws DatabricksSQLException {
+  private void fetchNextBatch() throws SQLException {
     try {
       LOGGER.debug("Fetching next batch, current total rows fetched: {}", totalRowsFetched);
       currentResponse = session.getDatabricksClient().getMoreResults(statement);

src/main/java/com/databricks/jdbc/api/impl/arrow/AbstractRemoteChunkProvider.java

@@ -19,6 +19,7 @@
 import com.databricks.jdbc.model.telemetry.enums.DatabricksDriverErrorCode;
 import com.databricks.jdbc.telemetry.TelemetryHelper;
 import com.databricks.sdk.service.sql.BaseChunkInfo;
+import java.sql.SQLException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
@@ -94,7 +95,7 @@ protected AbstractRemoteChunkProvider(
       IDatabricksHttpClient httpClient,
       int maxParallelChunkDownloadsPerQuery,
       CompressionCodec compressionCodec)
-      throws DatabricksSQLException {
+      throws SQLException {
     this.chunkReadyTimeoutSeconds = session.getConnectionContext().getChunkReadyTimeoutSeconds();
     this.maxParallelChunkDownloadsPerQuery = maxParallelChunkDownloadsPerQuery;
     this.session = session;
@@ -264,7 +265,7 @@ private ConcurrentMap<Long, T> initializeChunksMap(
       TFetchResultsResp resultsResp,
       IDatabricksStatementInternal parentStatement,
       IDatabricksSession session)
-      throws DatabricksSQLException {
+      throws SQLException {
     ConcurrentMap<Long, T> chunkIndexMap = new ConcurrentHashMap<>();
     populateChunkIndexMap(resultsResp.getResults(), chunkIndexMap);
     while (resultsResp.hasMoreRows) {