andygrove commented on PR #5370: URL: https://github.com/apache/datafusion-comet/pull/5370#issuecomment-5308955100
Thanks for this. The disk-side fix is a genuine bug with a clear root cause, and I like that `writer_stream_position` measures the file rather than re-deriving it, since `BufBatchWriter::flush` guarantees everything has reached the file before the second reading. I traced the shared-buffer memory bookkeeping through several scenarios and it holds up: - Clearing `spill_accounted_input_buffers` per `insert_batch` is what protects against allocator address reuse across outer batches. Since the outer batch is held alive for the whole slicing loop, an address cannot be freed and reused within one input batch, so the set cannot produce a false "already counted" hit. That is a subtle hazard the code gets right. - `repeated_spill_buffer_bytes` deliberately persisting across `insert_batch` boundaries is correct. The bytes it refers to are still pinned by leftover slices, so they must stay subtracted until the next spill drains them. - `repeated` is a subset of `new_buffer_bytes`, which is a subset of `mem_growth`, and `mem_growth` always lands in the pot either via a successful `try_grow` or via `unreserved_bytes`, so the subtraction never over-subtracts. - In the common shape where input batches arrive at exactly `batch_size`, `repeated_spill_buffer_bytes` stays zero throughout and the whole mechanism reduces to `reservation.free() + unreserved_bytes`. The complexity only engages for oversized inputs, which seems like the right tradeoff. Marking only `buffered_batches.last()` rather than every buffered batch also looks fine to me, since `RecordBatch::slice` is offset-based across every Arrow array type and all slices of one outer batch expose identical buffer addresses. The Rust tests are well targeted. Comparing one input batch against two in `max_buffer_spills_charge_shared_backing_once_per_input_batch` is exactly the assertion that catches per-slice double counting, and checking `spilled_bytes` against actual `fs::metadata` file lengths pins the disk metric to ground truth rather than to itself. A few things I would like to see addressed. ### 1. The task metric bridge belongs on `CometMetricNode` `CometMetricNode` already has two methods that do precisely this job: `reportScanInputMetrics(ctx)` and `reportNativeWriteOutputMetrics(ctx)`. The second one even carries the doc comment explaining the listener ordering requirement that the new inline block in `CometNativeShuffleWriter.write()` re-explains. `write()` already calls `nativeMetrics.reportScanInputMetrics(...)` a few lines above. Would a `reportSpillMetrics(ctx)` next to those two work here? That keeps all three task metric bridges in one place under one documented ordering contract. A future change to how Comet publishes final native metrics would then have one place to fix rather than three. For what it is worth I checked the ordering claim against Spark master. `TaskContextImpl` keeps completion listeners in a `Stack` and pops them, so registration-before does mean invocation-after. The reasoning in the comment is right. ### 2. The disk metric label is now ambiguous The whole point of the change is separating the two figures, but the UI ends up showing `spilled bytes` next to `memory spilled bytes`. Someone looking at those two rows has no way to tell that the first is the compressed on-disk number. Could the existing label become something like `disk spilled bytes`? The metric key stays `spilled_bytes`, so nothing on the native side changes. ### 3. `docs/source/user-guide/latest/metrics.md` needs updating That page is hand-edited, and its Exchange section currently lists only `native shuffle time`, `repartition time`, `memory pool time`, and `encoding and compression time`. This PR adds two user-visible metrics and, more importantly, changes what Spark's `memoryBytesSpilled` means for native shuffle. Documenting `partition interleaving time` and the memory versus disk spill distinction there would help people comparing Comet's numbers against vanilla Spark. ### 4. Spills from operators inlined under the shuffle writer still do not reach task metrics `shuffleWriterSQLMetrics` filters `detailedMetrics` out of the writer node's own map. When `spec.childNativeOp` is a rich native subtree, a `CometSortExec` or `CometSortMergeJoinExec` inside it reports its own `spill_count` and `spilled_bytes` into `spec.childMetricNode`, and those never make it into `taskMetrics`. So the Stages tab still under-reports spill for those plans. That is not a regression, it was equally true before. But it is directly adjacent to what this PR fixes, and the listener you just added is the natural place to handle it. Since the description scopes this PR to the writer, could you file a tracking issue and link it here? Otherwise it will not get picked up. ### 5. `COMET_SHUFFLE_JVM_BATCH_SIZE` looks like a no-op in the new test In `CometTaskMetricsSuite`, the failed-attempt test sets `COMET_SHUFFLE_JVM_BATCH_SIZE`, but that config is only read by `CometDiskBlockWriter` and the JVM `SpillWriter`, both on the columnar shuffle path. The test runs with `COMET_SHUFFLE_MODE=native`, so it should have no effect, and `COMET_BATCH_SIZE` is doing the real work. Was it left over from an earlier iteration? Worth dropping so a future reader does not assume it matters for reproducing the failure at row 8192. -- 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]
