Use async wait timeout for SEA when direct results disabled (fix hybrid-path truncation)

NEXT_CHANGELOG.md

@@ -9,6 +9,7 @@
 ### Fixed
 - Fixed `setCatalog()` and `setSchema()` producing invalid SQL (e.g. `SET CATALOG ``name``) when the catalog or schema name was passed already wrapped in backticks. Backticks are now stripped before wrapping, and `getCatalog()`/`getSchema()` return the bare identifier name.
 - Fixed metadata SQL generation for catalog, schema, and table identifiers containing backticks.
+- Fixed SEA result truncation when direct results are disabled. Large, highly-compressible results that span multiple chunks were delivered inline via the old hybrid path and truncated to the first chunk. The SQL Execution path now uses an async (`0s`) wait timeout when direct results are disabled, so results are returned via external links and fetched in full.
 
 ---
 *Note: When making changes, please add your change under the appropriate section

src/main/java/com/databricks/jdbc/dbclient/impl/sqlexec/DatabricksSdkClient.java

@@ -740,9 +740,9 @@ private ExecuteStatementRequest getRequest(
     if (executeAsync) {
       request.setWaitTimeout(ASYNC_TIMEOUT_VALUE);
     } else {
-      // Only set timeout if direct results mode is not enabled
+      // DirectResults off -> async (0s); avoids truncation (ES-1714092)
       if (!connectionContext.getDirectResultMode()) {
-        request.setWaitTimeout(SYNC_TIMEOUT_VALUE);
+        request.setWaitTimeout(ASYNC_TIMEOUT_VALUE);
       }
       request.setOnWaitTimeout(ExecuteStatementRequestOnWaitTimeout.CONTINUE);
     }

src/test/java/com/databricks/jdbc/dbclient/impl/sqlexec/DatabricksSdkClientTest.java

@@ -1402,4 +1402,62 @@ public void testCheckStatementAlive_exceptionWrapped() throws Exception {
             () -> databricksSdkClient.checkStatementAlive(STATEMENT_ID));
     assertTrue(exception.getMessage().contains("Heartbeat status check failed"));
   }
+
+  @Test
+  public void testWaitTimeout_directResultsDisabled_usesAsyncZero() throws Exception {
+    setupClientMocks(true, false);
+    // EnableDirectResults=0 -> getDirectResultMode() is false
+    IDatabricksConnectionContext connectionContext =
+        DatabricksConnectionContext.parse(JDBC_URL + "EnableDirectResults=0", new Properties());
+    DatabricksSdkClient databricksSdkClient =
+        new DatabricksSdkClient(connectionContext, statementExecutionService, apiClient);
+    DatabricksConnection connection =
+        new DatabricksConnection(connectionContext, databricksSdkClient);
+    connection.open();
+    DatabricksStatement statement = new DatabricksStatement(connection);
+
+    databricksSdkClient.executeStatement(
+        STATEMENT,
+        warehouse,
+        sqlParams,
+        StatementType.QUERY,
+        connection.getSession(),
+        statement,
+        null);
+
+    ArgumentCaptor<ExecuteStatementRequest> captor =
+        ArgumentCaptor.forClass(ExecuteStatementRequest.class);
+    verify(apiClient, atLeastOnce()).serialize(captor.capture());
+    // Direct results disabled -> async (0s), not the hybrid 10s path that truncates (ES-1714092).
+    assertEquals("0s", captor.getValue().getWaitTimeout());
+  }
+
+  @Test
+  public void testWaitTimeout_directResultsEnabled_leftUnset() throws Exception {
+    setupClientMocks(true, false);
+    // Default JDBC_URL has direct results enabled -> getDirectResultMode() is true
+    IDatabricksConnectionContext connectionContext =
+        DatabricksConnectionContext.parse(JDBC_URL, new Properties());
+    DatabricksSdkClient databricksSdkClient =
+        new DatabricksSdkClient(connectionContext, statementExecutionService, apiClient);
+    DatabricksConnection connection =
+        new DatabricksConnection(connectionContext, databricksSdkClient);
+    connection.open();
+    DatabricksStatement statement = new DatabricksStatement(connection);
+
+    databricksSdkClient.executeStatement(
+        STATEMENT,
+        warehouse,
+        sqlParams,
+        StatementType.QUERY,
+        connection.getSession(),
+        statement,
+        null);
+
+    ArgumentCaptor<ExecuteStatementRequest> captor =
+        ArgumentCaptor.forClass(ExecuteStatementRequest.class);
+    verify(apiClient, atLeastOnce()).serialize(captor.capture());
+    // Direct results enabled -> WaitTimeout left unset (true SEA direct results).
+    assertNull(captor.getValue().getWaitTimeout());
+  }
 }