WangGuangxin opened a new pull request, #13606: URL: https://github.com/apache/gluten/pull/13606
## What changes were proposed in this pull request? Move the `ScopedTimer` in `CompressedInputStream` from `Read()` into `RefillDecompressed()`, so the decompress-time metric is measured once per chunk refill instead of once per `Read()` call. Add `CompressedInputStreamTest`. ### 1. Why `CompressedInputStream` is the decompressing input stream used by the sort shuffle reader (`VeloxSortShuffleReaderDeserializer`). That reader deserializes row by row, and for every row it calls `Read()` twice: once for the 4-byte row length and once for the row payload. Before this change, every `Read()` constructed a `ScopedTimer`, i.e. two `steady_clock::now()` calls. So every row paid **four clock reads**, while the work being timed is usually just a small `memcpy` out of the already decompressed buffer (a few ns per row). A vDSO `clock_gettime` costs tens of ns, so on this path the timer cost far more than the work it measured. Actual decompression and raw input reads only happen in `RefillDecompressed()`. Timing there measures exactly the work the metric is meant to describe, and the per-row hot path no longer touches the clock. `blockingTime_` (raw input wait) is accumulated by the nested timer in `EnsureCompressedData()`, which is only called from `RefillDecompressed()`. It therefore stays nested inside `decompressWallTime_`, and `decompressTime() = decompressWallTime_ - blockingTime_` remains correct. ### 2. Benefit TPC-DS SF1K, 103 statements, sort shuffle writer forced via `spark.gluten.sql.columnar.shuffle.sort.partitions.threshold=1` (otherwise the hash shuffle reader is used, which does not go through `CompressedInputStream`). Same commit, toolchain and build options for both sides; the only difference is this patch. One warm-up run, then base, pr, pr, base, base, pr. All runs completed with 0 failed queries. | Metric | before | after | |---|---|---| | Query sum (s), 3 runs | 564.81 ± 1.49 | **536.33 ± 2.79 (-5.0%)** | | Wall time (s), 3 runs | 577.34 ± 1.57 | 549.27 ± 2.87 (-4.9%) | | Executor CPU samples (async-profiler, 4 executors) | 4,994,066 | 4,720,210 (-5.5%) | | `CompressedInputStream::Read` -> `clock_gettime`, share of executor CPU | **8.17%** | 0.02% | | LZ4 decompression under `Read`, share of executor CPU | 0.72% | 0.79% | - 43 of 103 queries are faster by more than 0.1 s. Largest gains: q34 5.05 -> 2.33 s, q95 50.69 -> 47.80 s, q14_part1 17.52 -> 15.18 s, q14_part2 16.93 -> 14.69 s, q93 16.08 -> 14.14 s, q16 10.51 -> 8.90 s. - In the before flame graph, `[vdso]` is 73% of the sort-reader subtree; after the change the subtree is dominated by the copy, LZ4 and row deserialization. - Microbenchmark of the reader pattern (two reads of 4 B and 100 B per row from a decompressed buffer): 82.6 ns/row with a timer around each read, ~4 ns/row without. The real decompression work is unchanged; only the clock reads around buffer copies are removed. ### 3. Metric semantics - **`decompressWallTime_` / "time to decompress" now covers only chunk refills** (decompressing a frame and reading compressed input), with raw input wait still subtracted via `blockingTime_`. - Previously it also included the time spent copying bytes that were already decompressed, for every `Read()` call. That part is no longer counted. It was a copy, not decompression, so the new value is closer to what the metric's name says, and it can be slightly smaller than before for the same work. - A `Read()` that spans a chunk boundary is no longer timed as a whole; only the refill inside it is. - In our runs the task-level "time to decompress" sum was effectively unchanged (24.67 ms before vs 24.78 ms after, summed over all tasks). That total is far below the actual LZ4 CPU time in the profile because the sort reader path mostly does not propagate this metric today; that is a pre-existing issue and is not addressed here. ## How was this patch tested? - Added unit test `compressed_input_stream_test` (`cpp/core/tests/CompressedInputStreamTest.cc`), to be run by CI: cached reads add no decompression time; reads of varied sizes across compressed/decompressed chunk boundaries; concatenated frames and repeated EOF; empty frame and zero-length read; truncated frame reports an `IOError`. - TPC-DS SF1K end-to-end A/B as above, 0 failed queries in all runs. 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
