feat: emit per-chunk timing telemetry on the FetchResults path

Aggregate initial/slowest/sum chunk latencies in MetricsAggregator,
emit them in the proto chunk_details block, and time each FetchResults
RPC in RowSetProvider so the inline-Arrow path populates chunk
telemetry alongside CloudFetch (mirrors Go's chunkTimingAccumulator).

Also fix MetricsAggregator clobbering accumulated chunkCount and
bytesDownloaded to 0 on STATEMENT_COMPLETE when the event omitted
those fields — this hid chunk_details from every path.

Co-authored-by: Isaac

Author: samikshya-db

lib/DBSQLOperation.ts

@@ -103,6 +103,7 @@ export default class DBSQLOperation implements IOperation {
       this.operationHandle,
       [directResults?.resultSet],
       useOnlyPrefetchedResults,
+      this.id,
     );
     this.closeOperation = directResults?.closeOperation;
     this.context.getLogger().log(LogLevel.debug, `Operation created with id: ${this.id}`);

lib/result/RowSetProvider.ts

@@ -2,6 +2,7 @@ import Int64 from 'node-int64';
 import { TFetchOrientation, TFetchResultsResp, TOperationHandle, TRowSet } from '../../thrift/TCLIService_types';
 import Status from '../dto/Status';
 import IClientContext from '../contracts/IClientContext';
+import { LogLevel } from '../contracts/IDBSQLLogger';
 import IResultsProvider, { ResultsProviderFetchNextOptions } from './IResultsProvider';
 import { getColumnValue } from './utils';
 
@@ -26,6 +27,10 @@ export default class RowSetProvider implements IResultsProvider<TRowSet | undefi
 
   private readonly operationHandle: TOperationHandle;
 
+  private readonly statementId?: string;
+
+  private chunkIndex: number = 0;
+
   private fetchOrientation: TFetchOrientation = TFetchOrientation.FETCH_FIRST;
 
   private prefetchedResults: TFetchResultsResp[] = [];
@@ -48,9 +53,11 @@ export default class RowSetProvider implements IResultsProvider<TRowSet | undefi
     operationHandle: TOperationHandle,
     prefetchedResults: Array<TFetchResultsResp | undefined>,
     returnOnlyPrefetchedResults: boolean,
+    statementId?: string,
   ) {
     this.context = context;
     this.operationHandle = operationHandle;
+    this.statementId = statementId;
     prefetchedResults.forEach((item) => {
       if (item) {
         this.prefetchedResults.push(item);
@@ -83,16 +90,55 @@ export default class RowSetProvider implements IResultsProvider<TRowSet | undefi
     }
 
     const driver = await this.context.getDriver();
+    const startTime = Date.now();
     const response = await driver.fetchResults({
       operationHandle: this.operationHandle,
       orientation: this.fetchOrientation,
       maxRows: new Int64(limit),
       fetchType: FetchType.Data,
     });
+    const latencyMs = Date.now() - startTime;
+
+    this.emitChunkEvent(latencyMs, response);
 
     return this.processFetchResponse(response);
   }
 
+  /**
+   * Emit a chunk telemetry event for one FetchResults page.
+   * CRITICAL: All exceptions swallowed and logged at LogLevel.debug ONLY.
+   */
+  private emitChunkEvent(latencyMs: number, response: TFetchResultsResp): void {
+    try {
+      if (!this.statementId) {
+        return;
+      }
+
+      const { telemetryEmitter } = this.context as any;
+      if (!telemetryEmitter) {
+        return;
+      }
+
+      let bytes = 0;
+      const arrowBatches = response.results?.arrowBatches;
+      if (arrowBatches) {
+        for (const batch of arrowBatches) {
+          bytes += batch.batch?.length ?? 0;
+        }
+      }
+
+      telemetryEmitter.emitCloudFetchChunk({
+        statementId: this.statementId,
+        chunkIndex: this.chunkIndex,
+        latencyMs,
+        bytes,
+      });
+      this.chunkIndex += 1;
+    } catch (error: any) {
+      this.context.getLogger().log(LogLevel.debug, `Error emitting FetchResults chunk event: ${error.message}`);
+    }
+  }
+
   public async hasMore() {
     // If there are prefetched results available - return `true` regardless of
     // the actual state of `hasMoreRows` flag (because we actually have some data)

lib/telemetry/DatabricksTelemetryExporter.ts

@@ -396,6 +396,15 @@ export default class DatabricksTelemetryExporter {
           log.entry.sql_driver_log.sql_operation.chunk_details = {
             total_chunks_present: metric.chunkCount,
             total_chunks_iterated: metric.chunkCount,
+            ...(metric.chunkInitialLatencyMs !== undefined && {
+              initial_chunk_latency_millis: metric.chunkInitialLatencyMs,
+            }),
+            ...(metric.chunkSlowestLatencyMs !== undefined && {
+              slowest_chunk_latency_millis: metric.chunkSlowestLatencyMs,
+            }),
+            ...(metric.chunkSumLatencyMs !== undefined && {
+              sum_chunks_download_time_millis: metric.chunkSumLatencyMs,
+            }),
           };
         }
       }

lib/telemetry/MetricsAggregator.ts

@@ -38,6 +38,9 @@ interface StatementTelemetryDetails {
   executionLatencyMs?: number;
   resultFormat?: string;
   chunkCount: number;
+  chunkInitialLatencyMs?: number;
+  chunkSlowestLatencyMs?: number;
+  chunkSumLatencyMs: number;
   bytesDownloaded: number;
   pollCount: number;
   compressionEnabled?: boolean;
@@ -226,8 +229,14 @@ export default class MetricsAggregator {
       case TelemetryEventType.STATEMENT_COMPLETE:
         details.executionLatencyMs = event.latencyMs;
         details.resultFormat = event.resultFormat;
-        details.chunkCount = event.chunkCount ?? 0;
-        details.bytesDownloaded = event.bytesDownloaded ?? 0;
+        // Only override with event-provided values; otherwise keep counts
+        // accumulated from CLOUDFETCH_CHUNK events.
+        if (event.chunkCount !== undefined) {
+          details.chunkCount = event.chunkCount;
+        }
+        if (event.bytesDownloaded !== undefined) {
+          details.bytesDownloaded = event.bytesDownloaded;
+        }
         details.pollCount = event.pollCount ?? 0;
         break;
 
@@ -237,6 +246,18 @@ export default class MetricsAggregator {
         if (event.compressed !== undefined) {
           details.compressionEnabled = event.compressed;
         }
+        // Per-chunk timing aggregation (mirrors Go's chunkTimingAccumulator).
+        // Only record when latencyMs is positive — keeps prefetched/cached
+        // pages out of the timing stats.
+        if (event.latencyMs !== undefined && event.latencyMs > 0) {
+          if (details.chunkInitialLatencyMs === undefined) {
+            details.chunkInitialLatencyMs = event.latencyMs;
+          }
+          if (details.chunkSlowestLatencyMs === undefined || event.latencyMs > details.chunkSlowestLatencyMs) {
+            details.chunkSlowestLatencyMs = event.latencyMs;
+          }
+          details.chunkSumLatencyMs += event.latencyMs;
+        }
         break;
 
       default:
@@ -258,6 +279,7 @@ export default class MetricsAggregator {
         workspaceId: event.workspaceId,
         startTime: event.timestamp,
         chunkCount: 0,
+        chunkSumLatencyMs: 0,
         bytesDownloaded: 0,
         pollCount: 0,
         errors: [],
@@ -293,6 +315,9 @@ export default class MetricsAggregator {
         latencyMs: details.executionLatencyMs,
         resultFormat: details.resultFormat,
         chunkCount: details.chunkCount,
+        chunkInitialLatencyMs: details.chunkInitialLatencyMs,
+        chunkSlowestLatencyMs: details.chunkSlowestLatencyMs,
+        chunkSumLatencyMs: details.chunkSumLatencyMs > 0 ? details.chunkSumLatencyMs : undefined,
         bytesDownloaded: details.bytesDownloaded,
         pollCount: details.pollCount,
         compressed: details.compressionEnabled,

lib/telemetry/types.ts

@@ -194,6 +194,15 @@ export interface TelemetryMetric {
   /** Number of result chunks */
   chunkCount?: number;
 
+  /** Latency of the first chunk fetch in milliseconds */
+  chunkInitialLatencyMs?: number;
+
+  /** Latency of the slowest chunk fetch in milliseconds */
+  chunkSlowestLatencyMs?: number;
+
+  /** Sum of all chunk fetch latencies in milliseconds */
+  chunkSumLatencyMs?: number;
+
   /** Total bytes downloaded */
   bytesDownloaded?: number;
 
@@ -315,6 +324,15 @@ export interface StatementMetrics {
   /** Number of CloudFetch chunks downloaded */
   chunkCount: number;
 
+  /** Latency of the first chunk fetch in milliseconds */
+  chunkInitialLatencyMs?: number;
+
+  /** Latency of the slowest chunk fetch in milliseconds */
+  chunkSlowestLatencyMs?: number;
+
+  /** Sum of all chunk fetch latencies in milliseconds */
+  chunkSumLatencyMs?: number;
+
   /** Total bytes downloaded */
   totalBytesDownloaded: number;