added logs for cloud fetch download speed (#913)

## Description
<!-- Provide a brief summary of the changes made and the issue they aim
to address.-->
Add logs for cloud fetch download speed
relevant ticket: https://databricks.atlassian.net/browse/XTA-11037



## Testing
<!-- Describe how the changes have been tested-->
verified logs are printed when log level is set to INFO


## Additional Notes to the Reviewer
<!-- Share any additional context or insights that may help the reviewer
understand the changes better. This could include challenges faced,
limitations, or compromises made during the development process.
Also, mention any areas of the code that you would like the reviewer to
focus on specifically. -->

Author: shivam2680

NEXT_CHANGELOG.md

@@ -8,6 +8,7 @@
 - Provide an option to push telemetry logs (using the flag `ForceEnableTelemetry=1`). For more details see [documentation](https://docs.databricks.com/aws/en/integrations/jdbc-oss/properties#-telemetry-collection)
 - Added putFiles methods in DBFSVolumeClient for async multi-file upload.
 - Added validation on UID param to ensure it is either not set or set to 'token'.
+- Added CloudFetch download speed logging at INFO level
 - Added vendor error codes to SQLExceptions raised for incorrect UID, host or token.
 
 ### Updated

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

@@ -422,6 +422,11 @@ public int getCloudFetchThreadPoolSize() {
     return Integer.parseInt(getParameter(DatabricksJdbcUrlParams.CLOUD_FETCH_THREAD_POOL_SIZE));
   }
 
+  @Override
+  public double getCloudFetchSpeedThreshold() {
+    return Double.parseDouble(getParameter(DatabricksJdbcUrlParams.CLOUD_FETCH_SPEED_THRESHOLD));
+  }
+
   @Override
   public String getCatalog() {
     return getParameter(DatabricksJdbcUrlParams.CONN_CATALOG);

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

@@ -147,11 +147,12 @@ public boolean releaseChunk() {
    *
    * @param httpClient the HTTP client to use for downloading
    * @param compressionCodec the compression codec to use for decompression
+   * @param speedThreshold the minimum expected download speed in MB/s for logging warnings
    * @throws DatabricksParsingException if there is an error parsing the data
    * @throws IOException if there is an error downloading or reading the data
    */
   protected abstract void downloadData(
-      IDatabricksHttpClient httpClient, CompressionCodec compressionCodec)
+      IDatabricksHttpClient httpClient, CompressionCodec compressionCodec, double speedThreshold)
       throws DatabricksParsingException, IOException;
 
   /** Handles a failure during the download or processing of this chunk. */

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

@@ -66,7 +66,8 @@ public static Builder builder() {
    * @throws IOException if there is an error during download or data reading
    */
   @Override
-  protected void downloadData(IDatabricksHttpClient httpClient, CompressionCodec compressionCodec)
+  protected void downloadData(
+      IDatabricksHttpClient httpClient, CompressionCodec compressionCodec, double speedThreshold)
       throws DatabricksParsingException, IOException {
     CloseableHttpResponse response = null;
     long startTime = System.nanoTime();
@@ -77,11 +78,15 @@ protected void downloadData(IDatabricksHttpClient httpClient, CompressionCodec c
       // Retry would be done in http client, we should not bother about that here
       response = httpClient.execute(getRequest, true);
       checkHTTPError(response);
+
+      long downloadTimeMs = (System.nanoTime() - startTime) / 1_000_000;
+      long contentLength = response.getEntity().getContentLength();
+      logDownloadMetrics(
+          downloadTimeMs, contentLength, chunkLink.getExternalLink(), speedThreshold);
+
       TelemetryCollector.getInstance()
           .recordChunkDownloadLatency(
-              getStatementIdString(statementId),
-              chunkIndex,
-              ((System.nanoTime() - startTime) / 1000_000)); // Convert nano to millis
+              getStatementIdString(statementId), chunkIndex, downloadTimeMs);
       setStatus(ChunkStatus.DOWNLOAD_SUCCEEDED);
       String decompressionContext =
           String.format(
@@ -133,6 +138,26 @@ private void addHeaders(HttpGet getRequest, Map<String, String> headers) {
     }
   }
 
+  private void logDownloadMetrics(
+      long downloadTimeMs, long contentLength, String url, double speedThreshold) {
+    if (downloadTimeMs > 0 && contentLength > 0) {
+      double speedMBps = (contentLength / 1024.0 / 1024.0) / (downloadTimeMs / 1000.0);
+      String baseUrl = url.split("\\?")[0];
+
+      LOGGER.info(
+          String.format(
+              "CloudFetch download: %.4f MB/s, %d bytes in %dms from %s",
+              speedMBps, contentLength, downloadTimeMs, baseUrl));
+
+      if (speedMBps < speedThreshold) {
+        LOGGER.warn(
+            String.format(
+                "CloudFetch download slower than threshold: %.4f MB/s < %.4f MB/s",
+                speedMBps, speedThreshold));
+      }
+    }
+  }
+
   public static class Builder {
     private long chunkIndex;
     private long numRows;

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

@@ -60,7 +60,10 @@ public Void call() throws DatabricksSQLException, ExecutionException, Interrupte
             chunk.setChunkLink(link);
           }
 
-          chunk.downloadData(httpClient, chunkDownloader.getCompressionCodec());
+          chunk.downloadData(
+              httpClient,
+              chunkDownloader.getCompressionCodec(),
+              connectionContext != null ? connectionContext.getCloudFetchSpeedThreshold() : 0.1);
           downloadSuccessful = true;
         } catch (IOException | DatabricksSQLException e) {
           retries++;

src/main/java/com/databricks/jdbc/api/internal/IDatabricksConnectionContext.java

@@ -132,6 +132,9 @@ public interface IDatabricksConnectionContext {
   /** Returns the number of threads to be used for fetching data from cloud storage */
   int getCloudFetchThreadPoolSize();
 
+  /** Returns the minimum expected download speed threshold in MB/s for CloudFetch operations */
+  double getCloudFetchSpeedThreshold();
+
   Boolean getDirectResultMode();
 
   Boolean shouldRetryTemporarilyUnavailableError();

src/main/java/com/databricks/jdbc/common/DatabricksJdbcUrlParams.java

@@ -145,7 +145,9 @@ public enum DatabricksJdbcUrlParams {
   MAX_CONCURRENT_PRESIGNED_REQUESTS(
       "MaxVolumeOperationConcurrentPresignedRequests",
       "Maximum number of concurrent presigned requests",
-      "50");
+      "50"),
+  CLOUD_FETCH_SPEED_THRESHOLD(
+      "CloudFetchSpeedThreshold", "Minimum expected download speed in MB/s", "0.1");
 
   private final String paramName;
   private final String defaultValue;

src/test/java/com/databricks/jdbc/api/impl/arrow/ChunkDownloadTaskTest.java

@@ -65,11 +65,11 @@ void testRetryLogicWithSocketException() throws Exception {
         .doThrow(throwableError)
         .doNothing()
         .when(chunk)
-        .downloadData(httpClient, CompressionCodec.NONE);
+        .downloadData(httpClient, CompressionCodec.NONE, 0.1);
 
     chunkDownloadTask.call();
 
-    verify(chunk, times(3)).downloadData(httpClient, CompressionCodec.NONE);
+    verify(chunk, times(3)).downloadData(httpClient, CompressionCodec.NONE, 0.1);
     assertTrue(downloadFuture.isDone());
     assertDoesNotThrow(() -> downloadFuture.get());
   }
@@ -88,11 +88,11 @@ void testRetryLogicExhaustedWithSocketException() throws Exception {
                 new SocketException("Connection reset"),
                 DatabricksDriverErrorCode.INVALID_STATE))
         .when(chunk)
-        .downloadData(httpClient, CompressionCodec.NONE);
+        .downloadData(httpClient, CompressionCodec.NONE, 0.1);
 
     assertThrows(DatabricksSQLException.class, () -> chunkDownloadTask.call());
     verify(chunk, times(ChunkDownloadTask.MAX_RETRIES))
-        .downloadData(httpClient, CompressionCodec.NONE);
+        .downloadData(httpClient, CompressionCodec.NONE, 0.1);
     assertTrue(downloadFuture.isDone());
     ExecutionException executionException =
         assertThrows(ExecutionException.class, () -> downloadFuture.get());