Add debug-level logging for cloud fetch performance diagnostics (#1218)

## Summary
- Adds structured `debug`-level logs across the cloud fetch pipeline
(chunk download, link wait, decompression, task completion) with
`statementId` correlation for end-to-end tracing
- Enriches the `executeStatement` completion log with manifest details
(format, totalBytes, totalChunks, totalRows, compression) replacing the
previous sparse log
- Breaks down `ArrowResultChunk.downloadData` timing into HTTP response,
stream read, and decompress+parse phases

NO_CHANGELOG=true

## Test plan
- [ ] Verify logs appear when debug logging is enabled (e.g.
`log4j.logger.com.databricks.jdbc=DEBUG`)
- [ ] Verify no logs appear at default (INFO) level — no noise for
customers
- [ ] Run existing unit tests to confirm no regressions

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Signed-off-by: Vikrant Puppala <vikrant.puppala@databricks.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: vikrantpuppala

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

@@ -156,6 +156,7 @@ public T getChunk() throws DatabricksSQLException {
 
     T chunk = chunkIndexToChunksMap.get(currentChunkIndex);
 
+    long waitStart = System.nanoTime();
     try {
       chunk.waitForChunkReady();
     } catch (InterruptedException e) {
@@ -174,6 +175,12 @@ public T getChunk() throws DatabricksSQLException {
       throw new DatabricksSQLException(
           "Failed to ready chunk", e.getCause(), DatabricksDriverErrorCode.CHUNK_READY_ERROR);
     }
+    long waitMs = (System.nanoTime() - waitStart) / 1_000_000;
+    LOGGER.debug(
+        "Chunk ready: statementId={}, chunkIndex={}, waitMs={}",
+        statementId,
+        chunk.getChunkIndex(),
+        waitMs);
 
     return chunk;
   }

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

@@ -89,6 +89,7 @@ protected void downloadData(
 
       // Read compressed stream fully (download latency excludes decompression)
       byte[] compressed = IOUtils.toByteArray(response.getEntity().getContent());
+      long readTimeMs = (System.nanoTime() - startTime) / 1_000_000;
 
       // Set status to DOWNLOAD_SUCCEEDED after reading the compressed stream fully
       setStatus(ChunkStatus.DOWNLOAD_SUCCEEDED);
@@ -101,6 +102,7 @@ protected void downloadData(
           speedThreshold);
 
       // Decompress (if needed) and parse
+      long decompressStart = System.nanoTime();
       try {
         String ctx =
             String.format(
@@ -111,6 +113,19 @@ protected void downloadData(
       } catch (Exception e) {
         handleFailure(e, ChunkStatus.PROCESSING_FAILED);
       }
+      long decompressTimeMs = (System.nanoTime() - decompressStart) / 1_000_000;
+      long totalTimeMs = (System.nanoTime() - startTime) / 1_000_000;
+      LOGGER.debug(
+          "Chunk download complete: statementId={}, chunkIndex={}, "
+              + "compressedBytes={}, httpResponseMs={}, streamReadMs={}, decompressAndParseMs={}, "
+              + "totalMs={}",
+          statementId,
+          chunkIndex,
+          compressed.length,
+          downloadTimeMs,
+          readTimeMs - downloadTimeMs,
+          decompressTimeMs,
+          totalTimeMs);
     } catch (Exception e) {
       handleFailure(e, ChunkStatus.DOWNLOAD_FAILED);
     } finally {

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

@@ -48,15 +48,23 @@ public Void call() throws DatabricksSQLException, ExecutionException, Interrupte
     DatabricksThreadContextHolder.setConnectionContext(this.connectionContext);
     DatabricksThreadContextHolder.setStatementId(this.statementId);
 
+    long taskStartTime = System.nanoTime();
     try {
       DatabricksThreadContextHolder.setRetryCount(retries);
       while (!downloadSuccessful) {
         try {
           if (chunk.isChunkLinkInvalid()) {
+            long linkWaitStart = System.nanoTime();
             ExternalLink link =
                 linkDownloadService
                     .getLinkForChunk(chunk.getChunkIndex())
                     .get(); // Block until link is available
+            long linkWaitMs = (System.nanoTime() - linkWaitStart) / 1_000_000;
+            LOGGER.debug(
+                "Link wait: statementId={}, chunkIndex={}, linkWaitMs={}",
+                statementId,
+                chunk.getChunkIndex(),
+                linkWaitMs);
             chunk.setChunkLink(link);
           }
 
@@ -65,6 +73,13 @@ public Void call() throws DatabricksSQLException, ExecutionException, Interrupte
               chunkDownloader.getCompressionCodec(),
               connectionContext != null ? connectionContext.getCloudFetchSpeedThreshold() : 0.1);
           downloadSuccessful = true;
+          long taskTotalMs = (System.nanoTime() - taskStartTime) / 1_000_000;
+          LOGGER.debug(
+              "ChunkDownloadTask complete: statementId={}, chunkIndex={}, totalTaskMs={}, retries={}",
+              statementId,
+              chunk.getChunkIndex(),
+              taskTotalMs,
+              retries);
         } catch (IOException | DatabricksSQLException e) {
           retries++;
           if (retries >= MAX_RETRIES) {

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

@@ -242,15 +242,19 @@ private void triggerNextBatchDownload() {
         CompletableFuture.runAsync(
             () -> {
               try {
+                long fetchStart = System.nanoTime();
                 ChunkLinkFetchResult result =
                     session
                         .getDatabricksClient()
                         .getResultChunks(statementId, batchStartIndex, batchStartRowOffset);
-                LOGGER.info(
-                    "Retrieved {} links for batch starting at {} for statement id {}",
-                    result.getChunkLinks().size(),
+                long fetchMs = (System.nanoTime() - fetchStart) / 1_000_000;
+                LOGGER.debug(
+                    "Link batch fetch: statementId={}, batchStartIndex={}, "
+                        + "linksRetrieved={}, fetchApiMs={}",
+                    statementId,
                     batchStartIndex,
-                    statementId);
+                    result.getChunkLinks().size(),
+                    fetchMs);
 
                 // Complete futures for all chunks in this batch
                 for (ExternalLink link : result.getChunkLinks()) {

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

@@ -34,6 +34,7 @@
 import com.databricks.jdbc.model.core.Disposition;
 import com.databricks.jdbc.model.core.ExternalLink;
 import com.databricks.jdbc.model.core.ResultData;
+import com.databricks.jdbc.model.core.ResultManifest;
 import com.databricks.jdbc.model.telemetry.enums.DatabricksDriverErrorCode;
 import com.databricks.sdk.WorkspaceClient;
 import com.databricks.sdk.core.ApiClient;
@@ -279,12 +280,30 @@ public DatabricksResultSet executeStatement(
       pollCount++;
     }
     long executionEndTime = Instant.now().toEpochMilli();
-    LOGGER.debug(
-        "Executed sql {} with status {}, total time taken {} and pollCount {}",
-        sql,
-        responseState,
-        (executionEndTime - executionStartTime),
-        pollCount);
+    {
+      ResultManifest manifest = response.getManifest();
+      ResultData resultData = response.getResult();
+      int numExternalLinks =
+          (resultData != null && resultData.getExternalLinks() != null)
+              ? resultData.getExternalLinks().size()
+              : 0;
+      LOGGER.debug(
+          "executeStatement complete: statementId={}, state={}, format={}, "
+              + "totalBytes={}, totalChunks={}, totalRows={}, "
+              + "hasInlineAttachment={}, numExternalLinks={}, compression={}, "
+              + "executionTimeMs={}, pollCount={}",
+          statementId,
+          responseState,
+          manifest != null ? manifest.getFormat() : "null",
+          manifest != null ? manifest.getTotalByteCount() : "null",
+          manifest != null ? manifest.getTotalChunkCount() : "null",
+          manifest != null ? manifest.getTotalRowCount() : "null",
+          resultData != null && resultData.getAttachment() != null,
+          numExternalLinks,
+          manifest != null ? manifest.getResultCompression() : "null",
+          (executionEndTime - executionStartTime),
+          pollCount);
+    }
     if (responseState != StatementState.SUCCEEDED && responseState != StatementState.CLOSED) {
       handleFailedExecution(response, statementId, sql);
     }