NO-SNOW: Fix chunk download/parse timing split in metrics log (#2640)

Author: sfc-gh-rkowalski

CHANGELOG.md

@@ -24,6 +24,7 @@
     - Fixed OAuth token requests sending `scope=session:role:null` when no scope is configured (or scope is empty/blank); the `scope` parameter is now omitted entirely in those cases (snowflakedb/snowflake-jdbc#2646).
     - Fixed Okta native SSO federated login sending malformed JSON to `/api/v1/authn` (HTTP 400 from Okta) when the username or password contained JSON-special characters such as double quotes or backslashes; the request body is now serialized with Jackson instead of string concatenation.
     - Added one in-band telemetry record per successful login describing which connection-identifier fields the user supplied (`account_provided`, `account_with_region`, `account_org_provided`, `region_provided`, `host_provided`). No hostname or account value is included. This is gated by the existing server-side `CLIENT_TELEMETRY_ENABLED` parameter and can additionally be disabled locally by setting `SF_TELEMETRY_DISABLE_CONNECTION_SHAPE=true`. The telemetry collection is time-boxed and will be removed in a future release.
+    - Fixed `SnowflakeChunkDownloader` per-chunk metrics log misattributing the response body transfer to `parseTime`: `getResultStreamProvider().getInputStream()` returns once headers come back and streams the body lazily during the parser's `read()` calls, so the old code billed only HTTP/TLS setup to `downloadTime` and the entire body read+parse to `parseTime`. When the metrics logger is at `FINE`/debug level, the InputStream is now wrapped in a `TimingInputStream` that accumulates time blocked inside `read()`, so `downloadTime` reflects true network read time and `parseTime` reflects only CPU parse cost; at higher log levels the original stream is used unchanged to avoid the per-`read()` overhead (snowflakedb/snowflake-jdbc#2640).
     - Fixed `Connection.isValid()` silently swallowing thread interruption: when the underlying heartbeat is interrupted, the connection's interrupt flag is now restored via `Thread.currentThread().interrupt()` so connection pools and Thread shutdown mechanisms can react to the interruption (snowflakedb/snowflake-jdbc#2314).
     - Fixed non-retryable HTTP 400 response bodies always being logged as `"Failed to read content due to exception: Attempted read from closed stream"`. The response entity is now buffered before `RestRequest#checkForDPoPNonceError` and `SnowflakeUtil#logResponseDetails` consume it so both readers see the body (snowflakedb/snowflake-jdbc#2631).
     - Added defense-in-depth canonical-path validation in the S3, Azure, and GCS download clients to ensure resolved local download paths cannot escape the user's GET target directory via traversal segments, absolute paths, or symlink redirection (snowflakedb/snowflake-jdbc#2623).

src/main/java/net/snowflake/client/internal/jdbc/SnowflakeChunkDownloader.java

@@ -7,6 +7,7 @@
 import com.fasterxml.jackson.databind.MappingJsonFactory;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.ByteArrayOutputStream;
+import java.io.FilterInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.PrintWriter;
@@ -953,18 +954,26 @@ private static Callable<Void> getDownloadChunkCallable(
        * @throws SnowflakeSQLException
        */
       private void downloadAndParseChunk(InputStream inputStream) throws SnowflakeSQLException {
-        // remember the download time
-        resultChunk.setDownloadTime(System.currentTimeMillis() - startTime);
-        downloader.addDownloadTime(resultChunk.getDownloadTime());
-
-        startTime = System.currentTimeMillis();
-
-        // parse the result json
+        // getResultStreamProvider().getInputStream() returns once headers come back, but the
+        // response body is streamed lazily during the parser's read*() calls below — so most of
+        // the true download time is spent inside read(), not before it. When the metrics log is
+        // active (debug/trace), wrap the stream in a TimingInputStream so we can split the
+        // download/parse times accurately. Otherwise, skip the per-read instrumentation and
+        // fall back to the cheaper, less-accurate split (treating everything before the parse
+        // call as download and everything inside it as parse) so we don't pay the wrapping cost
+        // when nothing reads the result.
+        boolean splitDownloadAndParseTimes = logger.isDebugEnabled() || logger.isTraceEnabled();
+        long connectionSetupMs = System.currentTimeMillis() - startTime;
+        TimingInputStream timedStream =
+            splitDownloadAndParseTimes ? new TimingInputStream(inputStream) : null;
+        InputStream parseStream = timedStream != null ? timedStream : inputStream;
+
+        long parseStart = System.currentTimeMillis();
         try {
           if (downloader.queryResultFormat == QueryResultFormat.ARROW) {
-            ((ArrowResultChunk) resultChunk).readArrowStream(inputStream);
+            ((ArrowResultChunk) resultChunk).readArrowStream(parseStream);
           } else {
-            parseJsonToChunkV2(inputStream, resultChunk);
+            parseJsonToChunkV2(parseStream, resultChunk);
           }
         } catch (Exception ex) {
           logger.debug(
@@ -997,9 +1006,22 @@ private void downloadAndParseChunk(InputStream inputStream) throws SnowflakeSQLE
           }
         }
 
-        // add parsing time
-        resultChunk.setParseTime(System.currentTimeMillis() - startTime);
-        downloader.addParsingTime(resultChunk.getParseTime());
+        long readPlusParseMs = System.currentTimeMillis() - parseStart;
+        long downloadMs;
+        long parseMs;
+        if (timedStream != null) {
+          long networkReadMs = TimeUnit.NANOSECONDS.toMillis(timedStream.getNanosBlocked());
+          downloadMs = connectionSetupMs + networkReadMs;
+          parseMs = Math.max(0L, readPlusParseMs - networkReadMs);
+        } else {
+          downloadMs = connectionSetupMs;
+          parseMs = readPlusParseMs;
+        }
+
+        resultChunk.setDownloadTime(downloadMs);
+        downloader.addDownloadTime(downloadMs);
+        resultChunk.setParseTime(parseMs);
+        downloader.addParsingTime(parseMs);
       }
 
       private long startTime;
@@ -1189,4 +1211,36 @@ public DownloaderMetrics terminate() {
       return null;
     }
   }
+
+  /**
+   * InputStream wrapper that accumulates the time spent blocked inside read() calls. Used to
+   * separate true network read time from CPU-bound parsing time when reading a chunk.
+   */
+  private static final class TimingInputStream extends FilterInputStream {
+    private long nanosBlocked = 0L;
+
+    TimingInputStream(InputStream in) {
+      super(in);
+    }
+
+    long getNanosBlocked() {
+      return nanosBlocked;
+    }
+
+    @Override
+    public int read() throws IOException {
+      long t0 = System.nanoTime();
+      int b = in.read();
+      nanosBlocked += System.nanoTime() - t0;
+      return b;
+    }
+
+    @Override
+    public int read(byte[] b, int off, int len) throws IOException {
+      long t0 = System.nanoTime();
+      int n = in.read(b, off, len);
+      nanosBlocked += System.nanoTime() - t0;
+      return n;
+    }
+  }
 }