sunchao opened a new pull request, #5445: URL: https://github.com/apache/datafusion-comet/pull/5445
## Why are the changes needed? Closes #5382. A native shuffle plan can execute spill-capable child operators, such as a native sort, beneath its shuffle writer. The existing task-metric bridge reports only the shuffle writer's own spill counters, even though the child operators already expose separate Spark SQL spill metrics. For example, if the shuffle writer spills 32 MiB to disk and its native child sort spills another 128 MiB, Spark currently reports only 32 MiB for the task instead of the actual 160 MiB. This makes Spark's stage/task view disagree with its operator metrics and obscures native memory pressure. Disk spill and memory spill must remain distinct: compressed bytes written to disk cannot be substituted for uncompressed in-memory spill bytes. ## What changes were proposed in this PR? - Build one metric tree containing both the native shuffle writer and its existing child-operator metric tree. - Recursively aggregate existing `spilled_bytes` and `memory_spilled_bytes` metrics, deduplicating shared `SQLMetric` instances so reused accumulators are counted only once. - Register spill reporting from the native shuffle input RDD before child producers are initialized, allowing Spark's task-completion ordering to publish final writer and child metrics before they are read. - Remove the previous writer-only listener to avoid reporting the writer's spill counters twice. - Preserve absent-versus-zero semantics: child memory spill is included only when the child actually exposes a trustworthy `memory_spilled_bytes` metric. The implementation reuses existing Spark-side metric propagation; it does not introduce new native spill accounting, planner changes, or JNI machinery. ## How was this PR tested? - An end-to-end native shuffle regression forces a real child sort to spill and verifies that Spark task disk spill equals the writer spill plus child sort spill exactly once, while task memory spill remains equal to the writer's exposed memory metric. - A deterministic task-completion regression verifies writer and child disk/memory metrics are both reported after nested input producers publish their final values, for successful and failed task attempts. - A nested metric-tree regression verifies shared/reused accumulators are counted only once. - Existing native shuffle spill, failed-attempt, scan-input, output, and task-binary-size coverage remains enabled. The two focused suites pass on Spark 3.5, 4.0, and 4.1, with 12 tests passing for each profile: ```bash SUITES=org.apache.spark.sql.comet.CometTaskMetricsSuite,org.apache.spark.sql.comet.execution.shuffle.CometNativeShuffleInputRDDSuite ./mvnw -Pspark-3.5 test -Dtest=none -DfailIfNoTests=false -Dsuites="$SUITES" ./mvnw -Pspark-4.0 test -Dtest=none -DfailIfNoTests=false -Dsuites="$SUITES" ./mvnw -Pspark-4.1 test -Dtest=none -DfailIfNoTests=false -Dsuites="$SUITES" ``` -- 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]
