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

## Description

When direct results are disabled on the SEA (SQL Execution API) path,
the driver
built the `ExecuteStatement` request with `waitTimeout=10s`
(`SYNC_TIMEOUT_VALUE`)
and `onWaitTimeout=CONTINUE` — the old SEA *hybrid* direct-results path.

For results that span **multiple chunks but compress small**
(highly-compressible
payloads), the server returns only the **first chunk inline** with **no
external
links** (`hasInlineAttachment=true, numExternalLinks=0`). The driver's
inline path
returns just that first chunk and never fetches the rest, so the result
is
**silently truncated**.

This change sets `waitTimeout=0s` (`ASYNC_TIMEOUT_VALUE`) when direct
results are
disabled, avoiding the hybrid inline path. The server then delivers
results via
**external links**, which the driver downloads in full.

Resulting contract (SEA):
- `DirectResults = 0` → `WaitTimeout = 0` (async)
- `DirectResults = 1` → `WaitTimeout` unset (true direct results)

Thrift is unaffected — it has no `waitTimeout` and paginates correctly.

Related: **ES-1714092** (server-side compressed-vs-uncompressed byte
limit on the
hybrid path). This driver change stops using that path; the server-side
fix is
tracked separately.

## Requirement / Motivation

Reported via Slack: SEA with direct results disabled silently truncated
a large,
highly-compressible result (200MB logical → ~0.8MB compressed) to a
handful of
rows, while Thrift returned all rows.

## Testing done

**Repro (manual, against a serverless SQL warehouse)** — query
`SELECT repeat('A', 1024 * 1024) AS payload FROM range(200)` with direct
results
disabled:

| | Before | After |
|---|---|---|
| SEA | 20 / 200 rows, `isCloudFetchUsed=false` (inline) ❌ | 200 / 200
rows, `isCloudFetchUsed=true` (external links) ✅ |
| Thrift | 200 / 200 ✅ | 200 / 200 ✅ |

Additional disambiguation runs (repeating the real table via `UNION ALL`
up to 64×,
9.6M rows, 40s execution) confirmed the trigger is the
**inline-vs-external-links
delivery (compressed size)** — not query time or polling: a fast no-poll
query
truncated while a slow 6-poll query did not.

**Regression — SEA unit + fakeservice suites (all pass, 0 failures):**
- `DatabricksSdkClientTest` (42)
- `SqlExecApiHybridResultsIntegrationTests` (2)
- `DatabricksMetadataQueryClientTest` (57)
- `CommandBuilderTest` (21)
- `DatabricksEmptyMetadataClientTest` (11)
- `SeaCircuitBreakerManagerTest` (13)

## Notes / caveats

- **Latency:** with direct results disabled, queries now always
execute → poll → fetch (one extra round-trip for small results) instead
of
  returning inline on the first response.
- The underlying server-side bug (ES-1714092) is separate; this change
avoids the
  affected path rather than fixing the server.

---------

Signed-off-by: Sreekanth Vadigi <sreekanth.vadigi@databricks.com>

Author: sreekanth-db

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());
+  }
 }