Fix Arrow chunk download telemetry error bucketing

NEXT_CHANGELOG.md

@@ -8,6 +8,9 @@
 - `DatabaseMetaData.getColumns(...)` with a `null` catalog now issues a single `SHOW COLUMNS IN ALL CATALOGS` statement (consistent with `getSchemas`/`getTables`) instead of enumerating every catalog and issuing a per-catalog `SHOW COLUMNS`. Older DBR versions that do not support the syntax transparently fall back to the previous enumerate-and-fan-out behavior.
 
 ### Fixed
+- Fixed Arrow chunk download failures being emitted under the internal `DOWNLOAD_FAILED` lifecycle
+  state instead of the canonical `CHUNK_DOWNLOAD_ERROR` telemetry code.
+
 - Fixed `IdleConnectionEvictor` thread leak in long-running applications. Driver-side resources (HTTP client, background threads) are now always released when `Connection.close()` is called, even if statement cleanup or server-side session termination fails.
 
 - Throw `DatabricksSQLException` instead of an unchecked `ClassCastException` when a complex-type getter (`getArray`, `getStruct`, `getMap`) is called on a column of a different complex type.

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

@@ -16,6 +16,7 @@
 import com.databricks.jdbc.log.JdbcLoggerFactory;
 import com.databricks.jdbc.model.client.thrift.generated.TSparkArrowResultLink;
 import com.databricks.jdbc.model.core.ExternalLink;
+import com.databricks.jdbc.model.telemetry.enums.DatabricksDriverErrorCode;
 import com.databricks.jdbc.telemetry.TelemetryHelper;
 import com.databricks.sdk.service.sql.BaseChunkInfo;
 import java.io.IOException;
@@ -127,6 +128,8 @@ protected void downloadData(
           readTimeMs - downloadTimeMs,
           decompressTimeMs,
           totalTimeMs);
+    } catch (DatabricksParsingException e) {
+      throw e;
     } catch (Exception e) {
       handleFailure(e, ChunkStatus.DOWNLOAD_FAILED);
     } finally {
@@ -156,6 +159,10 @@ protected void handleFailure(Exception exception, ChunkStatus failedStatus)
             this.chunkIndex, this.statementId, exception);
     LOGGER.error(this.errorMessage);
     setStatus(failedStatus);
+    if (failedStatus == ChunkStatus.DOWNLOAD_FAILED) {
+      throw new DatabricksParsingException(
+          errorMessage, exception, DatabricksDriverErrorCode.CHUNK_DOWNLOAD_ERROR);
+    }
     throw new DatabricksParsingException(errorMessage, exception, failedStatus.toString());
   }
 

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

@@ -9,6 +9,7 @@
 import com.databricks.jdbc.exception.DatabricksHttpException;
 import com.databricks.jdbc.exception.DatabricksParsingException;
 import com.databricks.jdbc.model.core.ExternalLink;
+import com.databricks.jdbc.model.telemetry.enums.DatabricksDriverErrorCode;
 import com.databricks.jdbc.telemetry.latency.TelemetryCollectorManager;
 import java.io.ByteArrayInputStream;
 import java.io.FilterInputStream;
@@ -64,10 +65,27 @@ void readError_setsDownloadFailed_notDownloadSucceeded() {
 
     // Act + Assert: downloadData should throw parsing exception and status should be
     // DOWNLOAD_FAILED
-    assertThrows(
-        DatabricksParsingException.class,
-        () -> chunk.downloadData(http, CompressionCodec.NONE, 0.0));
+    DatabricksParsingException exception =
+        assertThrows(
+            DatabricksParsingException.class,
+            () -> chunk.downloadData(http, CompressionCodec.NONE, 0.0));
     assertEquals(ChunkStatus.DOWNLOAD_FAILED, chunk.getStatus());
+    assertEquals(DatabricksDriverErrorCode.CHUNK_DOWNLOAD_ERROR.name(), exception.getSQLState());
+  }
+
+  @Test
+  void processingError_isNotReportedAsDownloadError() {
+    byte[] payload = "not an Arrow stream".getBytes();
+    ArrowResultChunk chunk = newChunk();
+    IDatabricksHttpClient http = httpWithEntity(new ByteArrayInputStream(payload), payload.length);
+
+    DatabricksParsingException exception =
+        assertThrows(
+            DatabricksParsingException.class,
+            () -> chunk.downloadData(http, CompressionCodec.NONE, 0.0));
+
+    assertEquals(ChunkStatus.PROCESSING_FAILED, chunk.getStatus());
+    assertEquals(ChunkStatus.PROCESSING_FAILED.name(), exception.getSQLState());
   }
 
   private static ArrowResultChunk newChunk() {