From bb16f970f9cf084d4609b833a221774200fe01fd Mon Sep 17 00:00:00 2001 From: "Ahmed Hussein (amahussein)" Date: Wed, 2 Sep 2026 20:59:23 -0500 Subject: [PATCH 1/2] Fix GPU task metric unit inference Fixes #2136 - add metric registry to configure how to handle metrics. Reads unit, scale and aggregation from the metric catalog instead of inferring them from substrings of a metric's name. - Remove the second unit conversion. `parseAccumFieldToLong` already yields milliseconds, and `convertValue` divided by 1e6 again, flooring every timing metric to zero so its row was dropped. - Stop matching `Wait` inside `Waiting`, which labelled three counters as durations. - Label `gpuMaxTaskFootprint` as bytes; the old rule required `Bytes` in the name. - Populate `avg` for max-aggregated metrics. `sum` stays empty, since per-task peaks never coexist. - Read `gpuOnGpuTasksWaitingGPUAvgCount`, a Double no parse branch could read. It published a fabricated zero; now stored as fixed-point thousandths and divided out at render. - Gate GPU reporting on the catalog's `family`, so a non-GPU metric declared later cannot leak into the `gpu_*` files. Report an unreadable accumulable once per process. Testing - Full core suite: 846 tests, 0 failures. Scalastyle and the Scala 2.13 cross-build clean. - New `parseAccumFieldToLong` coverage: per-branch units, fixed-point scaling, and rejection of `NaN`/`Infinity`/`1e9`/`3d`/`5f` including the overflow boundary. `AnalysisSuite` now reads the catalog rather than re-implementing the rule, and asserts every non-zero GPU accumulable produces a row. - On a mirrored event log the stage file goes from 9 to 17 rows, and `gpuOnGpuTasksWaitingGPUAvgCount` reads `max=2.5, avg=0.714` where it previously showed `0,0,0,0`. Row counts unchanged; a CPU log still emits no `gpu_*` files. Signed-off-by: Ahmed Hussein (amahussein) --- .../configs/metrics/metricCatalog.yaml | 384 ++++++++++++++ .../configs/reports/coreRawMetricsReport.yaml | 55 +- .../analysis/AppSparkMetricsAnalyzer.scala | 56 +-- .../rapids/tool/analysis/MetricCatalog.scala | 339 +++++++++++++ .../tool/analysis/MetricDefinition.scala | 156 ++++++ .../profiling/ProfileClassWarehouse.scala | 59 ++- .../sql/rapids/tool/store/AccumInfo.scala | 31 +- .../sql/rapids/tool/store/AccumMetaRef.scala | 45 +- .../sql/rapids/tool/util/EventUtils.scala | 88 ++++ .../tool/analysis/MetricCatalogSuite.scala | 472 ++++++++++++++++++ .../rapids/tool/profiling/AnalysisSuite.scala | 165 +++++- .../rapids/tool/util/ToolUtilsSuite.scala | 62 +++ 12 files changed, 1802 insertions(+), 110 deletions(-) create mode 100644 core/src/main/resources/configs/metrics/metricCatalog.yaml create mode 100644 core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/MetricCatalog.scala create mode 100644 core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/MetricDefinition.scala create mode 100644 core/src/test/scala/com/nvidia/spark/rapids/tool/analysis/MetricCatalogSuite.scala diff --git a/core/src/main/resources/configs/metrics/metricCatalog.yaml b/core/src/main/resources/configs/metrics/metricCatalog.yaml new file mode 100644 index 000000000..eff43f3c5 --- /dev/null +++ b/core/src/main/resources/configs/metrics/metricCatalog.yaml @@ -0,0 +1,384 @@ +# Copyright (c) 2026, NVIDIA CORPORATION. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Declarative catalog of per-task accumulable metrics. +# +# This table replaces a set of substring rules that inferred a metric's unit and its +# aggregation from its *name*. That information belongs to the accumulator that emits the +# metric, not to the string naming it, so it is declared here instead. +# +# Fields +# ------ +# name The normalized accumulable name, i.e. what AccumMetaRef.getName() returns. +# EventUtils.normalizeMetricName is the identity on every name declared here, +# but consumers look up by the normalized form, so that is the form declared. +# family Which family the metric belongs to: gpu | perfio | spark. This is what +# decides GPU reporting. Being declared in this catalog is NOT enough to make +# a metric a GPU metric -- the catalog is deliberately not GPU-specific, so +# that CPU and taskModel-sourced metrics can be declared here later without +# leaking into the gpu_* output files. The gpu_* files carry the gpu and +# perfio families; perfio is separate so it can be queried, and later reported, +# on its own without redefining what a GPU metric is. +# source Where the per-task values come from. +# accumulable - an event-log accumulable, folded by AccumInfo. +# taskModel - a field already retained on TaskModel. Not yet consumed. +# unit The unit the value is reported in: bytes | ms | count. This is a LABEL +# only. No scaling is derived from it -- see "no source scale" below. +# valueForm How the value is serialized in the event log. +# integer - parses via the plain-integer, duration or memory branch. +# decimal - MAY be a Double, e.g. "0.5". The accumulator's zero case still +# serializes as a plain "0", so ingest must accept both forms. +# Needs storageScale to survive a Long store. +# storageScale Fixed-point multiplier applied on ingest so sub-integer values survive the +# Long store. Must be 1, or a power of ten for a decimal metric. FULLY WIRED: +# EventUtils.parseAccumFieldToLong multiplies by it on ingest, every stored +# statistic for that metric is then in those units, and it is divided out only +# at render, in convertToSeq/convertToCSVSeq via MetricCatalog.formatStoredValue. +# A decimal metric MUST declare a scale > 1; validation rejects it otherwise, +# because without one every non-integral sample is silently dropped. +# aggregation How per-task values combine into a stage value: sum | max. +# NOT derivable from the accumulator class -- gpuMaxTaskFootprint and +# gpuSpillToHostBytes are both SizeInBytesAccumulator but one is a per-task +# high-water mark and the other a running total. +# includeInDiagnostics Whether the metric appears in the per-stage distribution report. +# description One line, for the report column documentation. +# +# There is deliberately NO source-scale field +# ------------------------------------------- +# EventUtils.parseAccumFieldToLong already normalizes every serialized form to a canonical +# unit: a plain integer stays raw, "00:00:01.773" becomes 1773 milliseconds, and +# "3.28GB (3526702303 bytes)" becomes bytes. Nothing downstream should convert again. The +# serialized form is also not stable across plugin releases -- the memory metrics moved from a +# plain integer to the human-readable string -- so recording it here would go stale. Absorbing +# that variation is the parser's job. +# +# Metrics absent from this table +# ------------------------------ +# An undeclared metric falls back to the legacy name-based heuristic for its unit LABEL ONLY: +# a name containing Time or Wait is labelled ms, one containing Bytes is labelled bytes, anything +# else count. It is unscaled and aggregated by sum. +# +# That heuristic is the very thing this table replaces, so relying on it as a fallback needs a +# word of justification. It is right for any metric whose name follows the plugin's convention -- +# a new NanoSecondAccumulator called gpuFooTime is correctly labelled ms -- and wrong only for the +# names that motivated this table (gpuMaxTaskFootprint is bytes without saying so; +# ...WaitingGPUMaxCount is a count that says Wait). Crucially it is now label-only: the value is +# no longer scaled by it, so a bad guess costs a wrong column header rather than a destroyed +# number. Declaring the metric here is still preferred and is the only way to get its +# aggregation, value form and storage scale right. +# +# Discovery works like this: a DECLARED metric is reported in the gpu_* files if and only if its +# family is gpu or perfio, and an UNDECLARED metric falls back to the legacy prefix rule -- a name +# starting with "gpu" or "perfio.", or the literal multithreadReaderMaxParallelism. Note the +# fallback prefix is "perfio." and not "perfio.s3.": widening it is what makes an undeclared +# perfio.gcs.* or perfio.abfs.* metric discoverable rather than silently dropped. Declaring a +# metric here still matters even when it would match the prefix, because only a declaration +# carries its unit, aggregation and value form. + +metrics: + # --- timing: NanoSecondAccumulator, serialized as "HH:MM:SS.mmm", parsed to milliseconds ---- + - name: gpuTime + family: gpu + source: accumulable + unit: ms + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: true + description: Time the task held the GPU semaphore. + - name: gpuSemaphoreWait + family: gpu + source: accumulable + unit: ms + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: true + description: Time the task spent blocked waiting to acquire the GPU semaphore. + - name: gpuRetryBlockTime + family: gpu + source: accumulable + unit: ms + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Time the task spent blocked during an out-of-memory retry. + - name: gpuRetryComputationTime + family: gpu + source: accumulable + unit: ms + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Time spent recomputing work discarded by an out-of-memory retry. + - name: gpuSpillToHostTime + family: gpu + source: accumulable + unit: ms + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Time spent spilling device memory to host memory. + - name: gpuSpillToDiskTime + family: gpu + source: accumulable + unit: ms + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Time spent spilling memory to disk. + - name: gpuReadSpillFromHostTime + family: gpu + source: accumulable + unit: ms + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Time spent reading spilled data back from host memory. + - name: gpuReadSpillFromDiskTime + family: gpu + source: accumulable + unit: ms + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Time spent reading spilled data back from disk. + - name: perfio.s3.requestLimiter.totalWaitTime + family: perfio + source: accumulable + unit: ms + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Time PerfIO S3 requests spent waiting on the request limiter. + + # --- byte totals: SizeInBytesAccumulator / LongAccumulator, summed across tasks ------------- + - name: gpuSpillToHostBytes + family: gpu + source: accumulable + unit: bytes + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: true + description: Bytes spilled from device memory to host memory. + - name: gpuSpillToDiskBytes + family: gpu + source: accumulable + unit: bytes + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: true + description: Bytes spilled to disk. + - name: gpuDiskWriteSavedBytes + family: gpu + source: accumulable + unit: bytes + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Disk writes avoided by the spillable partial-file handle. + + # --- byte high-water marks: per-task peaks, so aggregated by max, never summed -------------- + - name: gpuMaxTaskFootprint + family: gpu + source: accumulable + unit: bytes + valueForm: integer + storageScale: 1 + aggregation: max + includeInDiagnostics: true + description: Peak GPU memory footprint of the task. The value the runtime estimator samples. + - name: gpuMaxDeviceMemoryBytes + family: gpu + source: accumulable + unit: bytes + valueForm: integer + storageScale: 1 + aggregation: max + includeInDiagnostics: true + description: Peak device memory allocated by the task. + - name: gpuMaxHostMemoryBytes + family: gpu + source: accumulable + unit: bytes + valueForm: integer + storageScale: 1 + aggregation: max + includeInDiagnostics: true + description: Peak host memory allocated by the task. + - name: gpuMaxPinnedMemoryBytes + family: gpu + source: accumulable + unit: bytes + valueForm: integer + storageScale: 1 + aggregation: max + includeInDiagnostics: true + description: Peak pinned host memory allocated by the task. + - name: gpuMaxPageableMemoryBytes + family: gpu + source: accumulable + unit: bytes + valueForm: integer + storageScale: 1 + aggregation: max + includeInDiagnostics: true + description: Peak pageable host memory allocated by the task. + - name: gpuMaxDiskMemoryBytes + family: gpu + source: accumulable + unit: bytes + valueForm: integer + storageScale: 1 + aggregation: max + includeInDiagnostics: true + description: Peak disk space used by spilled data for the task. + + # --- counts -------------------------------------------------------------------------------- + - name: gpuMaxConcurrentGpuTasks + family: gpu + source: accumulable + unit: count + valueForm: integer + storageScale: 1 + aggregation: max + includeInDiagnostics: true + description: Peak number of tasks concurrently holding the GPU semaphore. + - name: gpuOnGpuTasksWaitingGPUMaxCount + family: gpu + source: accumulable + unit: count + valueForm: integer + storageScale: 1 + aggregation: max + includeInDiagnostics: true + description: Peak number of tasks queued waiting for the GPU. + - name: gpuOnGpuTasksWaitingGPUAvgCount + family: gpu + source: accumulable + unit: count + valueForm: decimal + storageScale: 1000 + aggregation: max + includeInDiagnostics: false + description: >- + Highest per-task average queue depth for tasks waiting on the GPU. Each task reports its own + average, and the stage reports the maximum of those averages -- so this is neither a mean + across tasks nor a peak queue depth, and it does not pair with the Max sibling over the same + population. Emitted by AvgLongAccumulator as a Double, so it is stored as fixed-point + thousandths. + - name: multithreadReaderMaxParallelism + family: gpu + source: accumulable + unit: count + valueForm: integer + storageScale: 1 + aggregation: max + includeInDiagnostics: true + description: Peak parallelism reached by the multithreaded reader. + - name: gpuRetryCount + family: gpu + source: accumulable + unit: count + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Number of out-of-memory retries the task performed. + - name: gpuSplitAndRetryCount + family: gpu + source: accumulable + unit: count + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Number of split-and-retry attempts the task performed. + # Aggregated by max, unlike today's behaviour. It is a MaxLongAccumulator, so each task reports + # its own deepest observed queue; summing those gives Sigma(per-task peaks), e.g. 800 for a + # limiter that never exceeded 8. No available event log exercises this metric, so the change is + # invisible on current fixtures. + - name: perfio.s3.requestLimiter.maxWaitingRequests + family: perfio + source: accumulable + unit: count + valueForm: integer + storageScale: 1 + aggregation: max + includeInDiagnostics: false + description: Peak number of PerfIO S3 requests queued on the request limiter. + - name: perfio.s3.netty.executors + family: perfio + source: accumulable + unit: count + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Executors using the PerfIO S3 Netty backend. + - name: perfio.s3.crt.executors + family: perfio + source: accumulable + unit: count + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Executors using the PerfIO S3 CRT backend. + - name: perfio.s3.s3a.executors + family: perfio + source: accumulable + unit: count + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Executors falling back to the S3A backend. + - name: perfio.s3.iceberg.fallbacks + family: perfio + source: accumulable + unit: count + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Iceberg reads that fell back off the PerfIO S3 path. + - name: perfio.gcs.http.executors + family: perfio + source: accumulable + unit: count + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Executors using the PerfIO GCS HTTP transport. + - name: perfio.gcs.grpc.executors + family: perfio + source: accumulable + unit: count + valueForm: integer + storageScale: 1 + aggregation: sum + includeInDiagnostics: false + description: Executors using the PerfIO GCS gRPC transport. diff --git a/core/src/main/resources/configs/reports/coreRawMetricsReport.yaml b/core/src/main/resources/configs/reports/coreRawMetricsReport.yaml index 62bf74555..885fe0627 100644 --- a/core/src/main/resources/configs/reports/coreRawMetricsReport.yaml +++ b/core/src/main/resources/configs/reports/coreRawMetricsReport.yaml @@ -875,8 +875,10 @@ reportDefinitions: - label: coreRawGpuStageLevelAggregatedTaskMetricsCSV description: >- GPU task metric aggregations at stage level. Long-format: one row per - (stageId, metricName). Auto-discovered from accumulator names starting - with gpu, perfio.s3., or equal to multithreadReaderMaxParallelism. + (stageId, metricName). Discovered from configs/metrics/metricCatalog.yaml, plus the + legacy name prefixes gpu and perfio. and the literal + multithreadReaderMaxParallelism, so a metric added after a tools release is still + reported. fileName: gpu_stage_level_aggregated_task_metrics.csv scope: per-app columns: @@ -895,19 +897,24 @@ reportDefinitions: - name: unit dataType: String description: >- - ms for time/wait metrics, bytes for byte metrics, count otherwise. + bytes, ms or count. Declared per metric in configs/metrics/metricCatalog.yaml; + a metric absent from that catalog falls back to a name-based heuristic for the + label only, and no value is scaled by it. - name: sum dataType: Long description: >- Total across tasks in the stage. Empty for max-aggregated metrics. - name: max - dataType: Long + dataType: Double description: >- Peak per-task value in the stage. - name: avg - dataType: Long + dataType: Double description: >- - Rolling per-task average. Empty for max-aggregated metrics. + Rolling per-task average. For a max-aggregated metric this is the mean across + tasks of each task's own peak, which is a different quantity from sum and is + populated. sum stays empty for those metrics, because adding per-task peaks is + meaningless when the peaks never coexist. - label: coreRawGpuSqlLevelAggregatedTaskMetricsCSV description: >- GPU task metric aggregations at SQL level. Rolls up stage-level GPU @@ -928,20 +935,24 @@ reportDefinitions: - name: unit dataType: String description: >- - ms / bytes / count. + bytes, ms or count. Declared per metric in configs/metrics/metricCatalog.yaml; + a metric absent from that catalog falls back to a name-based heuristic for the + label only, and no value is scaled by it. - name: sum dataType: Long description: >- Sum of stage sums. Empty for max-aggregated metrics. - name: max - dataType: Long + dataType: Double description: >- Max of stage maxes. - name: avg - dataType: Long + dataType: Double description: >- - Task-weighted average across stages. Empty for max-aggregated - metrics. + Task-weighted average across the stages in the SQL. Populated for + max-aggregated metrics too, where it is the mean across tasks of each task's + own peak -- a different quantity from sum, which stays empty for those + metrics because per-task peaks never coexist. - label: coreRawGpuAppLevelAggregatedTaskMetricsCSV description: >- GPU task metric aggregations at application level. One row per @@ -961,21 +972,25 @@ reportDefinitions: - name: unit dataType: String description: >- - ms / bytes / count. + bytes, ms or count. Declared per metric in configs/metrics/metricCatalog.yaml; + a metric absent from that catalog falls back to a name-based heuristic for the + label only, and no value is scaled by it. - name: sum dataType: Long description: >- Sum across all stages in the app. Empty for max-aggregated metrics. - name: max - dataType: Long + dataType: Double description: >- Overall peak reading any task produced during the run. - name: avg - dataType: Long + dataType: Double description: >- - Task-weighted average across all stages. Empty for - max-aggregated metrics. + Task-weighted average across all stages in the app. Populated for + max-aggregated metrics too, where it is the mean across tasks of each task's + own peak -- a different quantity from sum, which stays empty for those + metrics because per-task peaks never coexist. # AccumProfileResults - label: coreRawStageLevelAllMetricsCSV description: >- @@ -996,19 +1011,19 @@ reportDefinitions: description: >- TBD - name: min - dataType: Long + dataType: Double description: >- TBD - name: median - dataType: Long + dataType: Double description: >- TBD - name: max - dataType: Long + dataType: Double description: >- TBD - name: total - dataType: Long + dataType: Double description: >- TBD # RapidsPropertyProfileResult diff --git a/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSparkMetricsAnalyzer.scala b/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSparkMetricsAnalyzer.scala index f0ed6bf60..d44c2a850 100644 --- a/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSparkMetricsAnalyzer.scala +++ b/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSparkMetricsAnalyzer.scala @@ -410,30 +410,18 @@ class AppSparkMetricsAnalyzer(app: AppBase) extends AppAnalysisBase(app) with Lo // GPU task metric aggregations (Stage / SQL / App) // --------------------------------------------------------------------------- // - // Discovery convention: an accumulator is a GPU task metric if its name - // - starts with "gpu", OR - // - starts with "perfio.s3.", OR - // - equals "multithreadReaderMaxParallelism". - // Unit convention (from name): contains Time|Wait → ms (raw ns / 1e6); - // contains Bytes → bytes; otherwise → count. - // Max-aggregated metrics are discriminated via AccumMetaRef.isAggregateByMax; - // for those, sum and avg are empty and only max is meaningful. - - private def isGpuMetric(name: String): Boolean = { - name.startsWith("gpu") || - name.startsWith("perfio.s3.") || - name == "multithreadReaderMaxParallelism" - } - - private def unitForMetric(name: String): String = { - if (name.contains("Time") || name.contains("Wait")) "ms" - else if (name.contains("Bytes")) "bytes" - else "count" - } + // Discovery lives in MetricCatalog.isGpuReportedMetric and is cached per accumulator id on + // AccumMetaRef, so the rule is stated once rather than reconstructed at each call site. + // + // Unit comes from the catalog, which declares it per metric. An undeclared metric falls back + // to the legacy name heuristic, which is label-only: nothing is scaled by it. + // + // There is deliberately no value conversion here. EventUtils.parseAccumFieldToLong already + // normalizes every serialized form to a canonical unit -- a plain integer stays raw, + // "00:00:01.773" becomes 1773 milliseconds, "3.28GB (3526702303 bytes)" becomes bytes -- so + // converting again is what silently deleted every timing metric from this report. - private def convertValue(name: String, raw: Long): Long = { - if (name.contains("Time") || name.contains("Wait")) raw / 1000000L else raw - } + private def unitForMetric(name: String): String = MetricCatalog.DEFAULT.unitFor(name) /** * Aggregate GPU task accumulators by stage. Emits one row per (stageId, @@ -442,7 +430,7 @@ class AppSparkMetricsAnalyzer(app: AppBase) extends AppAnalysisBase(app) with Lo */ def aggregateGpuMetricsByStage(index: Int): Seq[StageAggGpuMetricsProfileResult] = { val gpuAccums = app.accumManager.accumInfoMap.values.filter { ai => - isGpuMetric(ai.infoRef.getName()) + ai.infoRef.isGpuReportedMetric }.toSeq if (gpuAccums.isEmpty) { return Seq.empty @@ -467,14 +455,20 @@ class AppSparkMetricsAnalyzer(app: AppBase) extends AppAnalysisBase(app) with Lo s"This will be excluded from task-weighted averages at SQL/app level.") 0 } - val (sum, max, avg) = if (isMax) { - (None: Option[Long], - Some(convertValue(name, stats.max)), - None: Option[Long]) + // The arithmetic mean, from the raw sum and count rather than from `med`. `med` is a + // rolling mean recomputed with integer division on every update, so it ratchets toward + // the floor: on a real log it reports 1 for a metric whose true mean is 5.89, and 5 for + // one whose true mean is 12.03. Dividing once is correct to within truncation. + val avg = ai.getRawStatsForStage(stageId).flatMap { raw => + if (raw.count > 0L) Some(raw.total / raw.count) else None + } + // sum stays empty for a max-aggregated metric: adding up per-task peaks is meaningless + // because those peaks never coexist. avg does not -- it is the mean across tasks of + // each task's own peak, which is a genuinely different quantity. + val (sum, max) = if (isMax) { + (None: Option[Long], Some(stats.max)) } else { - (Some(convertValue(name, stats.total)), - Some(convertValue(name, stats.max)), - Some(convertValue(name, stats.med))) + (Some(stats.total), Some(stats.max)) } // Skip rows carrying no signal (both sum and max zero/absent). val zeroSum = sum.forall(_ == 0L) diff --git a/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/MetricCatalog.scala b/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/MetricCatalog.scala new file mode 100644 index 000000000..356704922 --- /dev/null +++ b/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/MetricCatalog.scala @@ -0,0 +1,339 @@ +/* + * Copyright (c) 2026, NVIDIA CORPORATION. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.nvidia.spark.rapids.tool.analysis + +import scala.beans.BeanProperty +import scala.collection.JavaConverters._ + +import org.apache.spark.internal.Logging +import org.apache.spark.sql.rapids.tool.util.{PropertiesLoader, UTF8Source, ValidatableProperties} + +/** + * Catalog of per-task and per-stage metric declarations, loaded from + * `configs/metrics/metricCatalog.yaml`. + * + * The catalog exists so that a metric's unit, its serialized form and its aggregation are + * declared once rather than inferred from substrings of its name. Adding a metric is a YAML + * edit; no Scala change is required. + * + * A metric that is not declared here falls back to the legacy name-based heuristic for its unit + * label (see [[MetricCatalog.legacyUnitFor]]), is unscaled, and is aggregated by sum. Note that + * discovery itself is a separate concern owned by the analyzer, and its prefix rule is narrower + * than this table -- see the resource file's header. + * + * @param metrics the declared metrics + * @see [[org.apache.spark.sql.rapids.tool.util.PropertiesLoader]] + */ +class MetricCatalog( + @BeanProperty var metrics: java.util.List[MetricDefinition] +) extends ValidatableProperties with Logging { + + /** No-arg constructor required by SnakeYAML for deserialization. */ + def this() = this(new java.util.ArrayList[MetricDefinition]()) + + /** + * Index by metric name. Lazy because SnakeYAML populates `metrics` through the setter after + * the no-arg constructor has run, so this must not be forced before deserialization ends. + */ + private lazy val byName: Map[String, MetricDefinition] = { + if (metrics == null) { + Map.empty[String, MetricDefinition] + } else { + metrics.asScala.map(e => e.name -> e).toMap + } + } + + /** + * Validates the catalog. + * + * Called from the `ValidatableProperties` constructor, which runs before SnakeYAML has + * populated the list, so an empty or null list is not an error here. The companion's loader + * checks for emptiness explicitly once deserialization has finished. + */ + override def validate(): Unit = { + if (metrics == null || metrics.isEmpty) { + return // nothing populated yet; the loader validates after deserialization + } + val names = metrics.asScala.map(_.name) + val duplicates = names.groupBy(identity).filter(_._2.size > 1).keys + if (duplicates.nonEmpty) { + throw new IllegalArgumentException( + s"Duplicate metric names in the metric catalog: ${duplicates.mkString(", ")}") + } + metrics.asScala.foreach(validateEntry) + } + + private def validateEntry(entry: MetricDefinition): Unit = { + def reject(field: String, value: String, allowed: Set[String]): Unit = { + throw new IllegalArgumentException( + s"Invalid $field '$value' for metric '${entry.name}'. " + + s"Expected one of: ${allowed.toSeq.sorted.mkString(", ")}") + } + if (entry == null) { + throw new IllegalArgumentException("Null entry in the metric catalog") + } + if (entry.name == null || entry.name.isEmpty) { + throw new IllegalArgumentException("Metric name cannot be null or empty") + } + if (entry.description == null || entry.description.isEmpty) { + throw new IllegalArgumentException( + s"Description cannot be null or empty for metric '${entry.name}'") + } + if (!MetricDefinition.Families.all.contains(entry.family)) { + reject("family", entry.family, MetricDefinition.Families.all) + } + if (!MetricDefinition.Sources.all.contains(entry.source)) { + reject("source", entry.source, MetricDefinition.Sources.all) + } + if (!MetricDefinition.Units.all.contains(entry.unit)) { + reject("unit", entry.unit, MetricDefinition.Units.all) + } + if (!MetricDefinition.ValueForms.all.contains(entry.valueForm)) { + reject("valueForm", entry.valueForm, MetricDefinition.ValueForms.all) + } + if (!MetricDefinition.Aggregations.all.contains(entry.aggregation)) { + reject("aggregation", entry.aggregation, MetricDefinition.Aggregations.all) + } + if (entry.storageScale < 1L) { + throw new IllegalArgumentException( + s"storageScale must be >= 1 for metric '${entry.name}', found ${entry.storageScale}") + } + if (entry.isDecimalValued && entry.storageScale <= 1L) { + throw new IllegalArgumentException( + s"storageScale must be > 1 for the decimal-valued metric '${entry.name}'; without it " + + "the value is routed to the integer parser and every non-integral sample is dropped") + } + if (entry.storageScale > 1L && !MetricCatalog.isPowerOfTen(entry.storageScale)) { + throw new IllegalArgumentException( + s"storageScale must be a power of ten for metric '${entry.name}', " + + s"found ${entry.storageScale}") + } + if (!entry.isDecimalValued && entry.storageScale != 1L) { + throw new IllegalArgumentException( + s"storageScale must be 1 for the integer-valued metric '${entry.name}', " + + s"found ${entry.storageScale}") + } + } + + /** The declaration for a metric, or None when it is not declared. */ + def lookup(name: String): Option[MetricDefinition] = byName.get(name) + + /** True when the metric is declared in the catalog. */ + def isDeclared(name: String): Boolean = byName.contains(name) + + /** The declared family of a metric, or None when it is not declared. */ + def familyOf(name: String): Option[String] = byName.get(name).map(_.family) + + /** True when the metric is declared in the given family. */ + def isInFamily(name: String, family: String): Boolean = { + byName.get(name).exists(_.family == family) + } + + /** Every declared metric name in the given family. */ + def namesInFamily(family: String): Set[String] = { + byName.values.filter(_.family == family).map(_.name).toSet + } + + /** True when the metric is declared in the GPU family. */ + def isGpuFamily(name: String): Boolean = isInFamily(name, MetricDefinition.Families.Gpu) + + /** True when the metric is declared in the PerfIO family. */ + def isPerfioFamily(name: String): Boolean = isInFamily(name, MetricDefinition.Families.Perfio) + + /** + * True when the metric belongs in the `gpu_*` output files. + * + * This is the single place the rule lives, so no caller has to reconstruct it. A DECLARED + * metric qualifies on its family alone -- being in the catalog is not sufficient, because the + * catalog is not GPU-specific and a Spark-family metric declared here must not leak into the + * GPU files. An UNDECLARED metric falls back to the legacy name prefixes, so a plugin metric + * added after a tools release is still reported rather than silently dropped. + */ + def isGpuReportedMetric(name: String): Boolean = { + byName.get(name) match { + case Some(definition) => definition.isGpuReportedMetric + case None => MetricCatalog.matchesLegacyGpuPrefix(name) + } + } + + /** + * The reported unit of a metric. An undeclared metric falls back to the legacy name-based + * heuristic, see [[MetricCatalog.legacyUnitFor]]. + * + * These accessors are on the hot path -- one call per accumulable per task-end event -- so each + * is a single map lookup and allocates nothing. + */ + def unitFor(name: String): String = { + byName.get(name).map(_.unit).getOrElse(MetricCatalog.legacyUnitFor(name)) + } + + /** + * True when the stage value is the maximum of the per-task values rather than their sum. + * An undeclared metric is summed, which is what the previous hardcoded set did. + */ + def isAggregatedByMax(name: String): Boolean = byName.get(name).exists(_.isAggregatedByMax) + + /** True when the metric appears in the per-stage distribution report. Never for undeclared. */ + def includedInDiagnostics(name: String): Boolean = byName.get(name).exists(_.includeInDiagnostics) + + /** True when the value may be a decimal and needs fixed-point storage. */ + def isDecimalValued(name: String): Boolean = byName.get(name).exists(_.isDecimalValued) + + /** Fixed-point multiplier applied to a metric on ingest. 1 unless the metric is decimal. */ + def storageScaleFor(name: String): Long = { + byName.get(name).map(_.storageScale).getOrElse(1L) + } + + /** + * Renders a stored value for output, undoing the fixed-point scale of a decimal metric. + * + * Values are stored as integers, so a metric declared with `storageScale: 1000` is held in + * thousandths and must be divided before it is shown. Unscaled metrics render exactly as + * before, so this is a no-op for 30 of the 31 declared metrics. + */ + def formatValue(name: String, value: Long): String = { + MetricCatalog.formatStoredValue(value, storageScaleFor(name)) + } + + /** Every declared metric name. */ + lazy val declaredNames: Set[String] = byName.keySet + + /** Names of the metrics that appear in the per-stage distribution report. */ + lazy val diagnosticsMetricNames: Set[String] = { + byName.values.filter(_.includeInDiagnostics).map(_.name).toSet + } + + /** Names of the metrics whose stage value is a maximum rather than a sum. */ + lazy val maxAggregatedNames: Set[String] = { + byName.values.filter(_.isAggregatedByMax).map(_.name).toSet + } + + override def toString: String = s"MetricCatalog(${byName.size} metrics)" +} + +/** + * Companion providing the catalog loaded from the packaged resource. + */ +object MetricCatalog extends Logging { + /** Path to the catalog inside the jar. */ + private val DEFAULT_CONFIG_PATH = "configs/metrics/metricCatalog.yaml" + + /** True when the value is a positive power of ten. `formatStoredValue` relies on this. */ + def isPowerOfTen(value: Long): Boolean = { + var v = value + while (v > 1L && v % 10L == 0L) { + v /= 10L + } + v == 1L + } + + /** + * The legacy name-prefix rule for recognising a plugin metric, kept as the fallback for a + * metric the catalog does not declare. + */ + def matchesLegacyGpuPrefix(name: String): Boolean = { + name.startsWith("gpu") || + name.startsWith("perfio.") || + name == "multithreadReaderMaxParallelism" + } + + /** + * Renders a stored value, dividing out a fixed-point scale. + * + * Deliberately integer arithmetic rather than `String.format`/`f"%.3f"`: those use the default + * JVM Locale, and a comma-decimal locale such as de_DE or fr_FR would emit "0,714" -- a comma + * inside a comma-delimited CSV field, which shifts every later column of the row. Working in + * Longs also avoids the Double round-trip entirely. + * + * `storageScale` is validated to be a power of ten, so the number of fractional digits is + * exactly its digit count minus one. + */ + def formatStoredValue(value: Long, storageScale: Long): String = { + if (storageScale <= 1L) { + value.toString + } else { + val whole = value / storageScale + val fraction = Math.abs(value % storageScale) + if (fraction == 0L) { + whole.toString + } else { + // Integer division truncates toward zero, so a negative value whose whole part is zero + // would otherwise lose its sign. + val sign = if (value < 0 && whole == 0L) "-" else "" + val digits = storageScale.toString.length - 1 + val padded = fraction.toString.reverse.padTo(digits, '0').reverse + val trimmed = padded.reverse.dropWhile(_ == '0').reverse + s"$sign$whole.$trimmed" + } + } + } + + /** + * The legacy name-based unit heuristic, kept as the fallback for a metric that is not declared. + * + * This is the rule the catalog replaces, so using it as a fallback deserves justification. It + * is correct for any metric whose name follows the plugin's convention -- a new + * `NanoSecondAccumulator` named `gpuFooTime` is labelled `ms`, which is what the parser + * produces -- and wrong only for the names that motivated the catalog in the first place + * (`gpuMaxTaskFootprint` is bytes without saying so; `...WaitingGPUMaxCount` is a count whose + * name contains `Wait`). + * + * It is now LABEL-ONLY. The value is no longer scaled by it, so a wrong guess costs a wrong + * column header rather than a destroyed number, which is what made the old rule dangerous. + */ + def legacyUnitFor(name: String): String = { + if (name.contains("Time") || name.contains("Wait")) { + MetricDefinition.Units.Millis + } else if (name.contains("Bytes")) { + MetricDefinition.Units.Bytes + } else { + MetricDefinition.Units.Count + } + } + + /** The catalog packaged with the tools. */ + lazy val DEFAULT: MetricCatalog = loadFromResources() + + /** + * Loads the catalog from the packaged resource. + * + * Throws `IllegalStateException` if the resource is missing or cannot be parsed, and + * `IllegalArgumentException` if the parsed catalog declares no metrics. + */ + @throws[IllegalStateException] + @throws[IllegalArgumentException] + def loadFromResources(): MetricCatalog = { + // PropertiesLoader.loadFromContent calls validate() once the bean is populated, so a bad + // declaration surfaces from inside this call. Everything is wrapped so that a parse failure + // reports the resource path rather than escaping as a bare SnakeYAML exception. + val catalog = try { + val source = UTF8Source.fromResource(DEFAULT_CONFIG_PATH) + val content = try source.mkString finally source.close() + PropertiesLoader[MetricCatalog].loadFromContent(content).orNull + } catch { + case e: IllegalArgumentException => throw e // a validation failure is already specific + case e: Exception => + throw new IllegalStateException( + s"Could not load the metric catalog: $DEFAULT_CONFIG_PATH", e) + } + if (catalog == null || catalog.metrics == null || catalog.metrics.isEmpty) { + throw new IllegalArgumentException( + s"The metric catalog declares no metrics: $DEFAULT_CONFIG_PATH") + } + catalog + } +} diff --git a/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/MetricDefinition.scala b/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/MetricDefinition.scala new file mode 100644 index 000000000..919c538e8 --- /dev/null +++ b/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/MetricDefinition.scala @@ -0,0 +1,156 @@ +/* + * Copyright (c) 2026, NVIDIA CORPORATION. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.nvidia.spark.rapids.tool.analysis + +import scala.beans.BeanProperty + +/** + * Declaration of a single per-task metric. + * + * These properties describe the accumulator that emits the metric. They are declared rather + * than inferred from the metric's name, because the name does not carry them: for example + * `gpuMaxTaskFootprint` is measured in bytes without containing "Bytes", and + * `gpuOnGpuTasksWaitingGPUMaxCount` is a count whose name contains "Wait". + * + * Uses JavaBean properties for YAML deserialization by SnakeYAML. + * + * @param name the normalized accumulable name, i.e. what `AccumMetaRef.getName()` + * returns. `EventUtils.normalizeMetricName` is the identity on every name + * declared today, but consumers look up by the normalized form, so that is + * the form declared here + * @param family which family of metrics this belongs to, see + * [[MetricDefinition.Families]]. This is what decides whether a metric + * is reported in the GPU output files; being present in the catalog is + * NOT sufficient, since the catalog is not GPU-specific + * @param source where the per-task values come from, see [[MetricDefinition.Sources]] + * @param unit reported unit, see [[MetricDefinition.Units]]. A label only; no + * scaling is derived from it + * @param valueForm how the value is serialized, see [[MetricDefinition.ValueForms]]. Note + * `decimal` means the value MAY be a Double: the accumulator's zero case + * still serializes as a plain `"0"`, so ingest must accept both forms + * @param storageScale fixed-point multiplier applied on ingest so that sub-integer values + * survive a Long store. 1 for everything except decimal metrics + * @param aggregation how per-task values combine into a stage value, see + * [[MetricDefinition.Aggregations]] + * @param includeInDiagnostics whether the metric appears in the per-stage distribution report + * @param description one-line description, used for report column documentation + */ +class MetricDefinition( + @BeanProperty var name: String, + @BeanProperty var family: String, + @BeanProperty var source: String, + @BeanProperty var unit: String, + @BeanProperty var valueForm: String, + @BeanProperty var storageScale: Long, + @BeanProperty var aggregation: String, + @BeanProperty var includeInDiagnostics: Boolean, + @BeanProperty var description: String) { + + /** + * No-arg constructor required by SnakeYAML for deserialization. + * + * The vocabulary fields default to null rather than to a legal value on purpose: SnakeYAML + * leaves an omitted YAML key at its default, and a legal default would let an entry that forgot + * to declare, say, its aggregation load silently as `sum` -- reintroducing exactly the silent + * guess this table exists to remove. Null fails validation with a message naming the field. + * `storageScale` and `includeInDiagnostics` are genuinely optional and keep real defaults. + */ + def this() = this(null, null, null, null, null, 1L, null, false, null) + + /** + * True when the metric belongs to the GPU family and so belongs in the `gpu_*` output files. + * Membership of the catalog on its own does not make a metric a GPU metric. + */ + def isGpuFamily: Boolean = family == MetricDefinition.Families.Gpu + + /** True when the metric belongs to the PerfIO family. */ + def isPerfioFamily: Boolean = family == MetricDefinition.Families.Perfio + + /** True when the metric is reported in the `gpu_*` output files. */ + def isGpuReportedMetric: Boolean = MetricDefinition.Families.gpuReported.contains(family) + + /** True when the stage value is the maximum of the per-task values rather than their sum. */ + def isAggregatedByMax: Boolean = aggregation == MetricDefinition.Aggregations.Max + + /** + * True when the serialized value may be a decimal and so needs fixed-point storage. The zero + * case of such a metric still arrives as a plain integer. + */ + def isDecimalValued: Boolean = valueForm == MetricDefinition.ValueForms.Decimal + + override def toString: String = { + s"MetricDefinition(name=$name, family=$family, source=$source, unit=$unit, " + + s"valueForm=$valueForm, " + + s"storageScale=$storageScale, aggregation=$aggregation, " + + s"includeInDiagnostics=$includeInDiagnostics)" + } +} + +object MetricDefinition { + /** + * Which family a metric belongs to. This is the field that decides GPU reporting: the catalog + * itself is deliberately not GPU-specific, so a declaration alone must not put a metric into + * the `gpu_*` files. + */ + object Families { + /** GPU execution metrics emitted by the cuDF plugin. */ + val Gpu = "gpu" + /** PerfIO metrics emitted by the plugin's accelerated readers. */ + val Perfio = "perfio" + /** Emitted by Spark itself. Declared for later use; nothing is in this family yet. */ + val Spark = "spark" + val all: Set[String] = Set(Gpu, Perfio, Spark) + + /** + * The families carried by the `gpu_*` output files. PerfIO metrics have always been reported + * there alongside the GPU ones, so they stay together; they are a distinct family so that + * they can be queried and, later, reported separately without changing what a GPU metric is. + */ + val gpuReported: Set[String] = Set(Gpu, Perfio) + } + + /** Where the per-task values come from. */ + object Sources { + val Accumulable = "accumulable" + /** A field already retained on TaskModel. Declared for later use; not yet consumed. */ + val TaskModel = "taskModel" + val all: Set[String] = Set(Accumulable, TaskModel) + } + + /** The unit a metric is reported in. This is a label; no conversion is derived from it. */ + object Units { + val Bytes = "bytes" + val Millis = "ms" + val Count = "count" + val all: Set[String] = Set(Bytes, Millis, Count) + } + + /** How the value is serialized in the event log. */ + object ValueForms { + val Integer = "integer" + /** May be a Double. The accumulator's zero case still serializes as a plain "0". */ + val Decimal = "decimal" + val all: Set[String] = Set(Integer, Decimal) + } + + /** How per-task values combine into a stage value. */ + object Aggregations { + val Sum = "sum" + val Max = "max" + val all: Set[String] = Set(Sum, Max) + } +} diff --git a/core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ProfileClassWarehouse.scala b/core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ProfileClassWarehouse.scala index a73987a6a..de5246411 100644 --- a/core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ProfileClassWarehouse.scala +++ b/core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ProfileClassWarehouse.scala @@ -18,7 +18,7 @@ package com.nvidia.spark.rapids.tool.profiling import scala.collection.Map -import com.nvidia.spark.rapids.tool.analysis.StatisticsMetrics +import com.nvidia.spark.rapids.tool.analysis.{MetricCatalog, StatisticsMetrics} import com.nvidia.spark.rapids.tool.views.OutHeaderRegistry import org.apache.spark.resource.{ExecutorResourceRequest, TaskResourceRequest} @@ -377,24 +377,32 @@ case class AccumProfileResults( OutHeaderRegistry.outputHeaders("AccumProfileResults") } + /** + * Values are stored as integers, so a metric the catalog declares as decimal is held in + * fixed-point units and must be divided before display. A no-op for every other metric. + */ + private def render(value: Long): String = { + MetricCatalog.formatStoredValue(value, accMetaRef.storageScale) + } + override def convertToSeq(): Array[String] = { Array(stageId.toString, accMetaRef.id.toString, accMetaRef.getName(), - min.toString, - median.toString, - max.toString, - total.toString) + render(min), + render(median), + render(max), + render(total)) } override def convertToCSVSeq(): Array[String] = { Array(stageId.toString, accMetaRef.id.toString, accMetaRef.name.csvValue, - min.toString, - median.toString, - max.toString, - total.toString) + render(min), + render(median), + render(max), + render(total)) } } @@ -1599,15 +1607,20 @@ case class StageAggGpuMetricsProfileResult( OutHeaderRegistry.outputHeaders("StageAggGpuMetricsProfileResult") } + /** Divides out the fixed-point scale of a decimal metric; a no-op for every other metric. */ + private def render(value: Option[Long]): String = { + value.map(MetricCatalog.DEFAULT.formatValue(metricName, _)).getOrElse("") + } + override def convertToSeq(): Array[String] = { Array( stageId.toString, numTasks.toString, metricName, unit, - sum.map(_.toString).getOrElse(""), - max.map(_.toString).getOrElse(""), - avg.map(_.toString).getOrElse("")) + render(sum), + render(max), + render(avg)) } override def convertToCSVSeq(): Array[String] = convertToSeq() @@ -1632,14 +1645,19 @@ case class SQLAggGpuMetricsProfileResult( OutHeaderRegistry.outputHeaders("SQLAggGpuMetricsProfileResult") } + /** Divides out the fixed-point scale of a decimal metric; a no-op for every other metric. */ + private def render(value: Option[Long]): String = { + value.map(MetricCatalog.DEFAULT.formatValue(metricName, _)).getOrElse("") + } + override def convertToSeq(): Array[String] = { Array( sqlId.toString, metricName, unit, - sum.map(_.toString).getOrElse(""), - max.map(_.toString).getOrElse(""), - avg.map(_.toString).getOrElse("")) + render(sum), + render(max), + render(avg)) } override def convertToCSVSeq(): Array[String] = convertToSeq() @@ -1662,14 +1680,19 @@ case class AppAggGpuMetricsProfileResult( OutHeaderRegistry.outputHeaders("AppAggGpuMetricsProfileResult") } + /** Divides out the fixed-point scale of a decimal metric; a no-op for every other metric. */ + private def render(value: Option[Long]): String = { + value.map(MetricCatalog.DEFAULT.formatValue(metricName, _)).getOrElse("") + } + override def convertToSeq(): Array[String] = { Array( appId, metricName, unit, - sum.map(_.toString).getOrElse(""), - max.map(_.toString).getOrElse(""), - avg.map(_.toString).getOrElse("")) + render(sum), + render(max), + render(avg)) } override def convertToCSVSeq(): Array[String] = convertToSeq() diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/store/AccumInfo.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/store/AccumInfo.scala index 24c16fc31..93e435a95 100644 --- a/core/src/main/scala/org/apache/spark/sql/rapids/tool/store/AccumInfo.scala +++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/store/AccumInfo.scala @@ -1,5 +1,5 @@ /* - * Copyright (c) 2024-2025, NVIDIA CORPORATION. + * Copyright (c) 2024-2026, NVIDIA CORPORATION. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,8 +21,10 @@ import scala.collection.mutable import com.nvidia.spark.rapids.tool.analysis.StatisticsMetrics import org.apache.spark.scheduler.AccumulableInfo +import org.apache.spark.sql.rapids.tool.util.EventUtils import org.apache.spark.sql.rapids.tool.util.EventUtils.parseAccumFieldToLong + /** * Maintains the accumulator information for a single accumulator. * This maintains following information: @@ -57,7 +59,7 @@ class AccumInfo(val infoRef: AccumMetaRef) { def addAccumToStage(stageId: Int, accumulableInfo: AccumulableInfo, update: Option[Long] = None): Unit = { - val parsedValue = accumulableInfo.value.flatMap(parseAccumFieldToLong) + val parsedValue = accumulableInfo.value.flatMap(parseValue) // in case there is an out of order event, the value showing up later could be // lower-than the previous value. In that case we should take the maximum. val existingEntry = stagesStatMap.getOrElse(stageId, @@ -97,7 +99,7 @@ class AccumInfo(val infoRef: AccumMetaRef) { // 5. Increase total by adding the incoming update for a task // 6. Create final object and update map // TODO: update nomenclature from med to rolling average - val parsedUpdateValue = accumulableInfo.update.flatMap(parseAccumFieldToLong) + val parsedUpdateValue = accumulableInfo.update.flatMap(parseValue) // we need to update the stageMap if the stageId does not exist in the map parsedUpdateValue.foreach { value => val stats = stagesStatMap.getOrElse(stageId, @@ -113,6 +115,20 @@ class AccumInfo(val infoRef: AccumMetaRef) { } } + /** + * Parses a raw accumulable value, applying the fixed-point storage scale the metric catalog + * declares for this metric. A value no branch can read is dropped and reported once per + * process -- this used to be silent, which is how a Double-valued accumulator went unnoticed + * while its metric published zeros. + */ + private def parseValue(rawValue: Any): Option[Long] = { + val parsed = parseAccumFieldToLong(rawValue, infoRef.storageScale) + if (parsed.isEmpty) { + EventUtils.reportUnparseableAccum(infoRef.getName(), rawValue) + } + parsed + } + // Getters for stage-specific metrics /** @@ -191,6 +207,15 @@ class AccumInfo(val infoRef: AccumMetaRef) { readjustTotalStats(reduced_val) } + /** + * The unadjusted record for a stage, before `readjustTotalStats` masks `total`. + * + * Needed to compute a real arithmetic mean: `med` is a rolling mean recomputed with integer + * division on every update, so it ratchets toward the floor and is badly wrong for + * small-valued metrics. `total / count` truncates once instead of once per task. + */ + def getRawStatsForStage(stageId: Int): Option[StatisticsMetrics] = stagesStatMap.get(stageId) + /** * Retrieves statistical metrics for a specific stage */ diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/store/AccumMetaRef.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/store/AccumMetaRef.scala index 12eaad03d..f137d0be7 100644 --- a/core/src/main/scala/org/apache/spark/sql/rapids/tool/store/AccumMetaRef.scala +++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/store/AccumMetaRef.scala @@ -16,6 +16,8 @@ package org.apache.spark.sql.rapids.tool.store +import com.nvidia.spark.rapids.tool.analysis.{MetricCatalog, MetricDefinition} + /** * Accumulator Meta Reference @@ -25,40 +27,47 @@ package org.apache.spark.sql.rapids.tool.store * @param name - Reference to the accumulator name */ case class AccumMetaRef(id: Long, name: AccumNameRef) { + /** + * The catalog declaration for this metric, resolved once per accumulator id rather than once + * per event. `AccumMetaRef` instances are created by `AccumManager.getOrCreateAccumInfo`, so + * this keeps the catalog off the per-task-end hot path entirely. + */ + val definition: Option[MetricDefinition] = MetricCatalog.DEFAULT.lookup(name.value) + // An integer bitWise operator that represents different classification of the metric. // For metrics that are representing max values across agregates, it will be classified as 1; // otherwise 0. // Later, we can expand this classification to other categories if needed. val metricCategory: Int = - if (AccumMetaRef.isMetricAggregateByMax(name.value)) { + if (definition.exists(_.isAggregatedByMax)) { 1 } else { 0 } + + /** + * Fixed-point multiplier the value is stored with. 1 for everything except a metric the + * catalog declares as decimal, whose values would otherwise not survive an integer store. + */ + val storageScale: Long = definition.map(_.storageScale).getOrElse(1L) + + /** + * True when the metric belongs in the `gpu_*` output files. Resolved once per accumulator id + * for the same reason as `metricCategory`, so callers do not repeat the catalog lookup. + */ + val isGpuReportedMetric: Boolean = MetricCatalog.DEFAULT.isGpuReportedMetric(name.value) + def isAggregateByMax: Boolean = metricCategory == 1 def getName(): String = name.value } object AccumMetaRef { - // Metrics for which the stage level accumulable value does not reflect the sum but the max value. - private val METRICS_WITH_MAX_AGGREGATES = Set( - "gpuMaxPageableMemoryBytes", - "gpuMaxDeviceMemoryBytes", - "gpuMaxHostMemoryBytes", - "gpuMaxPinnedMemoryBytes", - "gpuMaxDiskMemoryBytes", - "gpuMaxTaskFootprint", - "gpuOnGpuTasksWaitingGPUMaxCount", - "gpuMaxConcurrentGpuTasks", - "multithreadReaderMaxParallelism" - ) + // Which metrics aggregate by max rather than by sum is declared in + // `configs/metrics/metricCatalog.yaml`, not hardcoded here. The declaration is a property of + // the accumulator that emits the metric, and deriving it from the metric's name -- which is + // what the hardcoded set amounted to -- is what this table replaces. val EMPTY_ACCUM_META_REF: AccumMetaRef = new AccumMetaRef(0L, AccumNameRef.EMPTY_ACC_NAME_REF) - // Used to decide on setting the metricCategory - private def isMetricAggregateByMax(metricName: String): Boolean = { - METRICS_WITH_MAX_AGGREGATES.contains(metricName) - } - def apply(id: Long, name: Option[String]): AccumMetaRef = new AccumMetaRef(id, AccumNameRef.getOrCreateAccumNameRef(name)) } diff --git a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/EventUtils.scala b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/EventUtils.scala index fd7b679c9..d26870def 100644 --- a/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/EventUtils.scala +++ b/core/src/main/scala/org/apache/spark/sql/rapids/tool/util/EventUtils.scala @@ -33,6 +33,12 @@ import org.apache.spark.sql.execution.ui.SparkListenerSQLExecutionStart * Utility containing the implementation of helpers used for parsing data from event. */ object EventUtils extends Logging { + // A decimal literal, optionally in exponent form. Double.toString switches to E-notation + // below 1e-3, so excluding it would silently drop the small averages this exists to read. + // Type suffixes ("3d", "5f"), NaN and Infinity still fail the digits-first shape, and the + // range check in parseScaledDecimalToLong rejects anything that would not fit a Long. + private val DECIMAL_LITERAL_REGEX = """^-?\d+(\.\d+)?([eE][-+]?\d+)?$""".r + // A valid Spark catalog name must start with a letter, can contain letters, digits, // underscores, or dashes, and must not end with a dot. val SPARK_CATALOG_REGEX: Regex = """spark\\.sql\\.catalog\\.([A-Za-z][A-Za-z0-9_-]*)$""".r @@ -60,6 +66,14 @@ object EventUtils extends Logging { // The key is the exception message, and the value is the count of occurences private val cachedUnknownExceptions = new java.util.concurrent.ConcurrentHashMap[String, Int]() + // A cache to avoid reporting the same unparseable accumulable more than once. The key is the + // accumulable name and the value is the number of occurrences. This is process-global rather + // than per application on purpose: the profiler parses many event logs concurrently, and a + // metric the tool cannot read is a property of the metric, not of the log, so one line per + // distinct name is the useful amount of noise. + private val cachedUnparseableAccums = + new java.util.concurrent.ConcurrentHashMap[String, Int]() + private def reportMissingEventClass(className: String): Unit = { if (!missingEventClasses.contains(className)) { missingEventClasses.add(className) @@ -67,6 +81,26 @@ object EventUtils extends Logging { } } + /** + * Reports, once per process, that no parser could read a value for an accumulable. + * + * A silent drop here is what let a Double-valued accumulator go unnoticed while its metric + * published zeros, so the failure is now visible -- but only the first occurrence, since the + * same metric fails on every task of every application in a run. + * + * @param accumName the accumulable name + * @param rawValue the value that could not be parsed, quoted into the message + */ + def reportUnparseableAccum(accumName: String, rawValue: Any): Unit = { + val occurrences = + cachedUnparseableAccums.compute(accumName, (_, v) => Option(v).map(_ + 1).getOrElse(1)) + if (occurrences == 1) { + logWarning(s"No parser could read a value for the accumulable '$accumName'; " + + s"first unreadable value was '$rawValue'. This accumulable will be missing from the " + + "report. Only the first occurrence is logged; subsequent ones are suppressed.") + } + } + /** * Handles AssertionError exceptions, specifically to catch and log the * "assertion failed: expected hostname or IPv6" error only once. If the error keeps occuring we @@ -199,6 +233,60 @@ object EventUtils extends Logging { } } + /** + * Parses an accumulable value that the metric catalog declares as decimal, storing it as a + * fixed-point integer scaled by `storageScale`. + * + * Deliberately strict rather than delegating to `toDouble`: `java.lang.Double.parseDouble` + * accepts `"1e9"`, `"3d"`, `"5f"`, `"NaN"` and `"Infinity"`, and an infinite value would + * overflow the running total to a negative number. Only a plain decimal literal is admitted. + * + * Note the zero case of such a metric still arrives as a plain `"0"`, which this accepts and + * scales like any other value so that stored units stay consistent. + * + * Rounding is `Math.round`, i.e. half-up rather than half-away-from-zero, so an exact + * negative half tie rounds toward zero. Unreachable for the only decimal metric declared + * today, which is a non-negative queue depth. + */ + def parseScaledDecimalToLong(data: Any, storageScale: Long): Option[Long] = { + val strData = data.toString.trim + if (!DECIMAL_LITERAL_REGEX.pattern.matcher(strData).matches()) { + None + } else { + try { + val scaled = java.lang.Double.parseDouble(strData) * storageScale.toDouble + // Reject on magnitude, symmetrically. Long.MaxValue.toDouble rounds UP to 2^63, so a + // scaled value that reaches it is already out of range and Math.round would clamp + // rather than fail. The same has to apply to the negative side: double spacing near + // 2^63 is 2048, so a true value slightly beyond -2^63 rounds ONTO the representable + // -2^63 and would be clamped to Long.MinValue -- an in-range-looking number that is + // not the input. This errs conservatively at both ends: a true value within Long range + // but within one double-step of the boundary is rejected rather than fabricated. + if (scaled.isNaN || scaled.isInfinite || + Math.abs(scaled) >= Long.MaxValue.toDouble) { + None + } else { + Some(Math.round(scaled)) + } + } catch { + case _: NumberFormatException => None + } + } + } + + /** + * Parses an accumulable value, applying the metric's fixed-point storage scale. + * + * A scale of 1, which is every metric but one, is the pre-existing behaviour unchanged. + */ + def parseAccumFieldToLong(data: Any, storageScale: Long): Option[Long] = { + if (storageScale <= 1L) { + parseAccumFieldToLong(data) + } else { + parseScaledDecimalToLong(data, storageScale) + } + } + // A utility function used to read Spark properties and compare it to a given target. // Note that it takes a default argument as well in case the property is not available. def isPropertyMatch(properties: collection.Map[String, String], propKey: String, diff --git a/core/src/test/scala/com/nvidia/spark/rapids/tool/analysis/MetricCatalogSuite.scala b/core/src/test/scala/com/nvidia/spark/rapids/tool/analysis/MetricCatalogSuite.scala new file mode 100644 index 000000000..d04cc3e98 --- /dev/null +++ b/core/src/test/scala/com/nvidia/spark/rapids/tool/analysis/MetricCatalogSuite.scala @@ -0,0 +1,472 @@ +/* + * Copyright (c) 2026, NVIDIA CORPORATION. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.nvidia.spark.rapids.tool.analysis + +import scala.collection.JavaConverters._ + +import org.scalatest.funsuite.AnyFunSuite + +import org.apache.spark.sql.rapids.tool.util.PropertiesLoader + +class MetricCatalogSuite extends AnyFunSuite { + + private val catalog = MetricCatalog.DEFAULT + + private def entry( + name: String, + family: String = MetricDefinition.Families.Gpu, + unit: String = MetricDefinition.Units.Count, + valueForm: String = MetricDefinition.ValueForms.Integer, + storageScale: Long = 1L, + aggregation: String = MetricDefinition.Aggregations.Sum, + source: String = MetricDefinition.Sources.Accumulable): MetricDefinition = { + new MetricDefinition(name, family, source, unit, valueForm, storageScale, aggregation, + false, "d") + } + + /** + * Builds a catalog directly. Note this validates eagerly: ValidatableProperties calls + * validate() during construction and the constructor parameter is already visible, so an + * invalid table throws here rather than on a later validate() call. The SnakeYAML path is + * different -- it uses the no-arg constructor and populates through the setter, which is why + * the loader re-validates once deserialization has finished. + */ + private def catalogOf(entries: MetricDefinition*): MetricCatalog = { + new MetricCatalog(new java.util.ArrayList[MetricDefinition](entries.asJava)) + } + + test("packaged catalog loads and declares metrics") { + assert(catalog.metrics != null, "catalog should have been populated by the loader") + assert(catalog.declaredNames.nonEmpty, "catalog should declare at least one metric") + } + + test("packaged catalog has no duplicate names") { + val names = catalog.metrics.asScala.map(_.name) + assert(names.size == names.distinct.size, + s"duplicate metric names: ${names.diff(names.distinct).distinct.mkString(", ")}") + } + + test("every declared entry uses a known vocabulary and a coherent storage scale") { + catalog.metrics.asScala.foreach { e => + assert(e.name.nonEmpty, s"empty metric name in $e") + assert(MetricDefinition.Families.all.contains(e.family), s"bad family in $e") + assert(MetricDefinition.Sources.all.contains(e.source), s"bad source in $e") + assert(MetricDefinition.Units.all.contains(e.unit), s"bad unit in $e") + assert(MetricDefinition.ValueForms.all.contains(e.valueForm), s"bad valueForm in $e") + assert(MetricDefinition.Aggregations.all.contains(e.aggregation), s"bad aggregation in $e") + assert(e.storageScale >= 1L, s"non-positive storageScale in $e") + if (!e.isDecimalValued) { + assert(e.storageScale == 1L, s"integer-valued metric must not be scaled: $e") + } + assert(e.description != null && e.description.nonEmpty, s"missing description in $e") + } + } + + test("lookup resolves a declared metric and rejects an undeclared one") { + val footprint = catalog.lookup("gpuMaxTaskFootprint") + assert(footprint.isDefined, "gpuMaxTaskFootprint should be declared") + assert(footprint.get.unit == MetricDefinition.Units.Bytes, + "gpuMaxTaskFootprint is measured in bytes even though its name has no 'Bytes'") + assert(footprint.get.isAggregatedByMax, "gpuMaxTaskFootprint is a per-task high-water mark") + assert(catalog.lookup("someMetricThatDoesNotExist").isEmpty) + assert(!catalog.isDeclared("someMetricThatDoesNotExist")) + } + + test("count metrics whose names contain Wait are not declared as durations") { + // These are the metrics the previous substring rule mislabelled, because "Wait" is a + // substring of "Waiting". Their unit is a count. + Seq("gpuOnGpuTasksWaitingGPUMaxCount", + "gpuOnGpuTasksWaitingGPUAvgCount", + "perfio.s3.requestLimiter.maxWaitingRequests").foreach { name => + val e = catalog.lookup(name) + assert(e.isDefined, s"$name should be declared") + assert(e.get.unit == MetricDefinition.Units.Count, s"$name is a count, not a duration") + } + } + + test("the decimal-valued metric is the only one carrying a fixed-point scale") { + val scaled = catalog.metrics.asScala.filter(_.storageScale != 1L).map(_.name).toSet + assert(scaled == Set("gpuOnGpuTasksWaitingGPUAvgCount"), s"unexpected scaled metrics: $scaled") + val decimal = catalog.metrics.asScala.filter(_.isDecimalValued).map(_.name).toSet + assert(decimal == Set("gpuOnGpuTasksWaitingGPUAvgCount"), s"unexpected decimals: $decimal") + assert(catalog.lookup("gpuOnGpuTasksWaitingGPUAvgCount").get.storageScale == 1000L) + } + + test("max-aggregated metrics are exactly the declared set") { + // Two entries here differ from the hardcoded set this replaces, both deliberately: + // gpuOnGpuTasksWaitingGPUAvgCount (an average, but a per-task one, so the stage value is the + // max of per-task averages) and perfio.s3.requestLimiter.maxWaitingRequests (a genuine + // MaxLongAccumulator that was previously summed). + val expected = Set( + "gpuMaxPageableMemoryBytes", + "gpuMaxDeviceMemoryBytes", + "gpuMaxHostMemoryBytes", + "gpuMaxPinnedMemoryBytes", + "gpuMaxDiskMemoryBytes", + "gpuMaxTaskFootprint", + "gpuOnGpuTasksWaitingGPUMaxCount", + "gpuOnGpuTasksWaitingGPUAvgCount", + "gpuMaxConcurrentGpuTasks", + "multithreadReaderMaxParallelism", + "perfio.s3.requestLimiter.maxWaitingRequests") + assert(catalog.maxAggregatedNames == expected, + s"unexpected: ${catalog.maxAggregatedNames.diff(expected)}, " + + s"missing: ${expected.diff(catalog.maxAggregatedNames)}") + } + + test("diagnostics metrics are exactly the declared set") { + val expected = Set( + "gpuMaxTaskFootprint", + "gpuMaxDeviceMemoryBytes", + "gpuMaxHostMemoryBytes", + "gpuMaxPinnedMemoryBytes", + "gpuMaxPageableMemoryBytes", + "gpuMaxDiskMemoryBytes", + "gpuSpillToHostBytes", + "gpuSpillToDiskBytes", + "gpuMaxConcurrentGpuTasks", + "gpuOnGpuTasksWaitingGPUMaxCount", + "multithreadReaderMaxParallelism", + "gpuSemaphoreWait", + "gpuTime") + assert(catalog.diagnosticsMetricNames == expected, + s"unexpected: ${catalog.diagnosticsMetricNames.diff(expected)}, " + + s"missing: ${expected.diff(catalog.diagnosticsMetricNames)}") + assert(catalog.diagnosticsMetricNames.subsetOf(catalog.declaredNames)) + } + + test("metrics are split between the gpu and perfio families, and all are gpu-reported") { + val perfio = catalog.namesInFamily(MetricDefinition.Families.Perfio) + val gpu = catalog.namesInFamily(MetricDefinition.Families.Gpu) + assert(perfio == catalog.declaredNames.filter(_.startsWith("perfio.")), + s"perfio family should be exactly the perfio.* names, got $perfio") + assert(gpu ++ perfio == catalog.declaredNames, "every metric is gpu or perfio today") + assert(catalog.namesInFamily(MetricDefinition.Families.Spark).isEmpty) + // both families land in the gpu_* files, which is what the analyzer asks about + catalog.declaredNames.foreach { n => + assert(catalog.isGpuReportedMetric(n), s"$n should be reported in the gpu files") + } + } + + test("isGpuReportedMetric is the one place the discovery rule lives") { + // declared: family decides, and being declared is not enough + val c = catalogOf( + entry("someGpuMetric", family = MetricDefinition.Families.Gpu), + entry("perfio.x.y", family = MetricDefinition.Families.Perfio), + entry("someSparkMetric", family = MetricDefinition.Families.Spark)) + assert(c.isGpuReportedMetric("someGpuMetric")) + assert(c.isGpuReportedMetric("perfio.x.y")) + assert(c.isDeclared("someSparkMetric"), "precondition: the metric is declared") + assert(!c.isGpuReportedMetric("someSparkMetric"), + "a declared spark-family metric must not leak into the gpu files") + // undeclared: the legacy prefix rule, so a newly added plugin metric is still reported + assert(c.isGpuReportedMetric("gpuBrandNewMetric")) + assert(c.isGpuReportedMetric("perfio.gcs.something")) + assert(c.isGpuReportedMetric("multithreadReaderMaxParallelism")) + assert(!c.isGpuReportedMetric("internal.metrics.memoryBytesSpilled")) + } + + test("family queries") { + assert(catalog.familyOf("gpuMaxTaskFootprint").contains(MetricDefinition.Families.Gpu)) + assert(catalog.familyOf("perfio.s3.netty.executors") + .contains(MetricDefinition.Families.Perfio)) + assert(catalog.familyOf("neverHeardOfIt").isEmpty) + assert(catalog.isGpuFamily("gpuMaxTaskFootprint")) + assert(!catalog.isGpuFamily("perfio.s3.netty.executors")) + assert(catalog.isPerfioFamily("perfio.s3.netty.executors")) + assert(!catalog.isPerfioFamily("gpuMaxTaskFootprint")) + } + + test("validate rejects an unknown family") { + val ex = intercept[IllegalArgumentException](catalogOf(entry("m", family = "quantum"))) + assert(ex.getMessage.contains("Invalid family")) + } + + test("validate rejects duplicate metric names") { + val ex = intercept[IllegalArgumentException](catalogOf(entry("dup"), entry("dup"))) + assert(ex.getMessage.contains("Duplicate metric names")) + } + + test("validate rejects an unknown unit") { + val ex = intercept[IllegalArgumentException](catalogOf(entry("m", unit = "furlongs"))) + assert(ex.getMessage.contains("Invalid unit")) + } + + test("validate rejects an unknown aggregation") { + val ex = intercept[IllegalArgumentException](catalogOf(entry("m", aggregation = "median"))) + assert(ex.getMessage.contains("Invalid aggregation")) + } + + test("validate rejects a scale on an integer-valued metric") { + val ex = intercept[IllegalArgumentException](catalogOf(entry("m", storageScale = 1000L))) + assert(ex.getMessage.contains("storageScale must be 1")) + } + + test("validate rejects a non-positive scale") { + val ex = intercept[IllegalArgumentException](catalogOf( + entry("m", valueForm = MetricDefinition.ValueForms.Decimal, storageScale = 0L))) + assert(ex.getMessage.contains("storageScale must be >= 1")) + } + + test("validate rejects an empty metric name") { + val ex = intercept[IllegalArgumentException](catalogOf(entry(""))) + assert(ex.getMessage.contains("cannot be null or empty")) + } + + test("the declared name set is exactly the plugin's task metric map") { + // Pinned explicitly: without this, deleting a row or typing a metric name to match the + // plugin's private field rather than its accumulable name leaves the suite green and + // silently downgrades the metric to the undeclared fallback. + val expected = Set( + "gpuTime", + "gpuSemaphoreWait", + "gpuRetryCount", + "gpuSplitAndRetryCount", + "gpuRetryBlockTime", + "gpuRetryComputationTime", + "gpuSpillToHostTime", + "gpuSpillToDiskTime", + "gpuReadSpillFromHostTime", + "gpuReadSpillFromDiskTime", + "gpuSpillToHostBytes", + "gpuSpillToDiskBytes", + "gpuMaxDeviceMemoryBytes", + "gpuMaxHostMemoryBytes", + "gpuMaxDiskMemoryBytes", + "gpuMaxPageableMemoryBytes", + "gpuMaxPinnedMemoryBytes", + "gpuOnGpuTasksWaitingGPUAvgCount", + "gpuOnGpuTasksWaitingGPUMaxCount", + "gpuMaxTaskFootprint", + "multithreadReaderMaxParallelism", + "gpuMaxConcurrentGpuTasks", + "gpuDiskWriteSavedBytes", + "perfio.s3.netty.executors", + "perfio.s3.crt.executors", + "perfio.s3.s3a.executors", + "perfio.s3.iceberg.fallbacks", + "perfio.gcs.http.executors", + "perfio.gcs.grpc.executors", + "perfio.s3.requestLimiter.totalWaitTime", + "perfio.s3.requestLimiter.maxWaitingRequests") + assert(catalog.declaredNames == expected, + s"unexpected: ${catalog.declaredNames.diff(expected)}, " + + s"missing: ${expected.diff(catalog.declaredNames)}") + assert(catalog.metrics.size == expected.size) + } + + test("every metric's declared unit is pinned") { + val millis = Set( + "gpuTime", + "gpuSemaphoreWait", + "gpuRetryBlockTime", + "gpuRetryComputationTime", + "gpuSpillToHostTime", + "gpuSpillToDiskTime", + "gpuReadSpillFromHostTime", + "gpuReadSpillFromDiskTime", + "perfio.s3.requestLimiter.totalWaitTime") + val bytes = Set( + "gpuSpillToHostBytes", + "gpuSpillToDiskBytes", + "gpuDiskWriteSavedBytes", + "gpuMaxTaskFootprint", + "gpuMaxDeviceMemoryBytes", + "gpuMaxHostMemoryBytes", + "gpuMaxPinnedMemoryBytes", + "gpuMaxPageableMemoryBytes", + "gpuMaxDiskMemoryBytes") + def named(u: String): Set[String] = + catalog.metrics.asScala.filter(_.unit == u).map(_.name).toSet + assert(named(MetricDefinition.Units.Millis) == millis, + s"ms mismatch: ${named(MetricDefinition.Units.Millis).diff(millis)} / " + + s"${millis.diff(named(MetricDefinition.Units.Millis))}") + assert(named(MetricDefinition.Units.Bytes) == bytes, + s"bytes mismatch: ${named(MetricDefinition.Units.Bytes).diff(bytes)} / " + + s"${bytes.diff(named(MetricDefinition.Units.Bytes))}") + // everything else is a count + assert(named(MetricDefinition.Units.Count) == catalog.declaredNames.diff(millis).diff(bytes)) + } + + test("an undeclared metric falls back to the legacy name heuristic for its unit") { + // The fallback is label-only: nothing is scaled by it, so a wrong guess costs a column + // header rather than a value. + assert(catalog.unitFor("gpuFooTime") == MetricDefinition.Units.Millis) + assert(catalog.unitFor("gpuFooWait") == MetricDefinition.Units.Millis) + assert(catalog.unitFor("gpuFooBytes") == MetricDefinition.Units.Bytes) + assert(catalog.unitFor("gpuFooCount") == MetricDefinition.Units.Count) + // and everything else defaults conservatively + Seq("gpuFooTime", "gpuFooBytes", "gpuFooCount").foreach { n => + assert(!catalog.isDeclared(n)) + assert(!catalog.isAggregatedByMax(n), s"$n should default to sum") + assert(!catalog.includedInDiagnostics(n)) + assert(!catalog.isDecimalValued(n)) + assert(catalog.storageScaleFor(n) == 1L) + } + } + + test("the legacy fallback agrees with the declaration for convention-following names") { + // Where a declared name follows the plugin's naming convention the fallback would have got + // the unit right anyway; the interesting rows are the ones where it would not. + val disagreeing = catalog.metrics.asScala + .filter(e => MetricCatalog.legacyUnitFor(e.name) != e.unit) + .map(e => s"${e.name}: declared ${e.unit}, heuristic ${MetricCatalog.legacyUnitFor(e.name)}") + .toSet + assert(disagreeing == Set( + "gpuMaxTaskFootprint: declared bytes, heuristic count", + "gpuOnGpuTasksWaitingGPUMaxCount: declared count, heuristic ms", + "gpuOnGpuTasksWaitingGPUAvgCount: declared count, heuristic ms", + "perfio.s3.requestLimiter.maxWaitingRequests: declared count, heuristic ms"), + s"unexpected disagreements: $disagreeing") + } + + test("the single-lookup accessors agree with the declared entries") { + catalog.metrics.asScala.foreach { e => + assert(catalog.unitFor(e.name) == e.unit, s"unitFor disagrees for ${e.name}") + assert(catalog.isAggregatedByMax(e.name) == e.isAggregatedByMax, + s"isAggregatedByMax disagrees for ${e.name}") + assert(catalog.includedInDiagnostics(e.name) == e.includeInDiagnostics, + s"includedInDiagnostics disagrees for ${e.name}") + assert(catalog.storageScaleFor(e.name) == e.storageScale, + s"storageScaleFor disagrees for ${e.name}") + assert(catalog.isDecimalValued(e.name) == e.isDecimalValued, + s"isDecimalValued disagrees for ${e.name}") + } + } + + test("a YAML table with an omitted field is rejected rather than silently defaulted") { + // SnakeYAML leaves an omitted key at the no-arg constructor's default, so the vocabulary + // fields default to null in order to be caught here. + val yaml = + """|metrics: + | - name: someMetric + | family: gpu + | source: accumulable + | unit: count + | valueForm: integer + | storageScale: 1 + | includeInDiagnostics: false + | description: aggregation is missing on purpose + |""".stripMargin + val ex = intercept[IllegalArgumentException]( + PropertiesLoader[MetricCatalog].loadFromContent(yaml)) + assert(ex.getMessage.contains("Invalid aggregation"), ex.getMessage) + } + + test("a YAML table with a bad vocabulary value is rejected on load") { + val yaml = + """|metrics: + | - name: someMetric + | family: gpu + | source: accumulable + | unit: furlongs + | valueForm: integer + | storageScale: 1 + | aggregation: sum + | includeInDiagnostics: false + | description: bad unit + |""".stripMargin + val ex = intercept[IllegalArgumentException]( + PropertiesLoader[MetricCatalog].loadFromContent(yaml)) + assert(ex.getMessage.contains("Invalid unit"), ex.getMessage) + } + + test("validate rejects an entry with no description") { + val e = new MetricDefinition("m", MetricDefinition.Families.Gpu, + MetricDefinition.Sources.Accumulable, MetricDefinition.Units.Count, + MetricDefinition.ValueForms.Integer, 1L, MetricDefinition.Aggregations.Sum, false, "") + val ex = intercept[IllegalArgumentException](catalogOf(e)) + assert(ex.getMessage.contains("Description cannot be null or empty")) + } + + test("formatStoredValue divides out a fixed-point scale") { + val f = MetricCatalog.formatStoredValue _ + // unscaled metrics render exactly as before + assert(f(12345L, 1L) == "12345") + assert(f(0L, 1L) == "0") + assert(f(-7L, 1L) == "-7") + // scaled: integral results carry no fractional part, fractions are trimmed not padded + assert(f(0L, 1000L) == "0") + assert(f(1000L, 1000L) == "1") + assert(f(2500L, 1000L) == "2.5") + assert(f(714L, 1000L) == "0.714") + assert(f(1200L, 1000L) == "1.2") + assert(f(100L, 1000L) == "0.1") + assert(f(10L, 1000L) == "0.01") + assert(f(1L, 1000L) == "0.001") + assert(f(20000L, 1000L) == "20") + // negatives, including the case where the whole part truncates to zero and would lose the sign + assert(f(-1500L, 1000L) == "-1.5") + assert(f(-500L, 1000L) == "-0.5") + assert(f(-1000L, 1000L) == "-1") + } + + test("formatStoredValue is locale independent") { + // String.format and the f-interpolator use the default Locale: under de_DE they render + // "0,714", and a comma inside a comma-delimited CSV shifts every later column. + val original = java.util.Locale.getDefault + try { + Seq(java.util.Locale.GERMANY, java.util.Locale.FRANCE, java.util.Locale.US).foreach { loc => + java.util.Locale.setDefault(loc) + assert(MetricCatalog.formatStoredValue(714L, 1000L) == "0.714", s"broken under $loc") + assert(MetricCatalog.formatStoredValue(2500L, 1000L) == "2.5", s"broken under $loc") + assert(!MetricCatalog.formatStoredValue(714L, 1000L).contains(","), s"comma under $loc") + } + } finally { + java.util.Locale.setDefault(original) + } + } + + test("validate rejects a decimal metric that declares no scale") { + // Without a scale the value is routed to the integer parser and every non-integral sample is + // dropped -- the original defect, silently reintroduced. + val ex = intercept[IllegalArgumentException](catalogOf( + entry("m", valueForm = MetricDefinition.ValueForms.Decimal, storageScale = 1L))) + assert(ex.getMessage.contains("storageScale must be > 1")) + } + + test("validate rejects a scale that is not a power of ten") { + // formatStoredValue derives its fractional digit count from the scale's digit count. + val ex = intercept[IllegalArgumentException](catalogOf( + entry("m", valueForm = MetricDefinition.ValueForms.Decimal, storageScale = 1024L))) + assert(ex.getMessage.contains("power of ten")) + assert(MetricCatalog.isPowerOfTen(1000L) && MetricCatalog.isPowerOfTen(1L)) + assert(!MetricCatalog.isPowerOfTen(1024L) && !MetricCatalog.isPowerOfTen(500L)) + } + + test("the legacy discovery prefix is perfio., not perfio.s3.") { + // Widened from the deleted isGpuMetric so an undeclared perfio.gcs.* or perfio.abfs.* + // metric is discoverable rather than silently dropped. Pinned because two shipped + // documents describe this rule. + assert(MetricCatalog.matchesLegacyGpuPrefix("perfio.s3.brand.new")) + assert(MetricCatalog.matchesLegacyGpuPrefix("perfio.gcs.brand.new")) + assert(MetricCatalog.matchesLegacyGpuPrefix("perfio.abfs.brand.new")) + assert(MetricCatalog.matchesLegacyGpuPrefix("gpuBrandNew")) + assert(MetricCatalog.matchesLegacyGpuPrefix("multithreadReaderMaxParallelism")) + assert(!MetricCatalog.matchesLegacyGpuPrefix("perfioNoDot")) + assert(!MetricCatalog.matchesLegacyGpuPrefix("internal.metrics.memoryBytesSpilled")) + assert(!MetricCatalog.matchesLegacyGpuPrefix("GPU decode time")) + } + + test("the packaged catalog renders its one decimal metric correctly") { + val name = "gpuOnGpuTasksWaitingGPUAvgCount" + assert(catalog.formatValue(name, 2500L) == "2.5") + assert(catalog.formatValue(name, 714L) == "0.714") + // and an unscaled metric is untouched + assert(catalog.formatValue("gpuMaxTaskFootprint", 7123115846L) == "7123115846") + } +} diff --git a/core/src/test/scala/com/nvidia/spark/rapids/tool/profiling/AnalysisSuite.scala b/core/src/test/scala/com/nvidia/spark/rapids/tool/profiling/AnalysisSuite.scala index 9b89d8a17..454a5f41d 100644 --- a/core/src/test/scala/com/nvidia/spark/rapids/tool/profiling/AnalysisSuite.scala +++ b/core/src/test/scala/com/nvidia/spark/rapids/tool/profiling/AnalysisSuite.scala @@ -19,10 +19,13 @@ package com.nvidia.spark.rapids.tool.profiling import java.io.File import com.nvidia.spark.rapids.tool.{PlatformNames, ToolTestUtils} +import com.nvidia.spark.rapids.tool.analysis.MetricCatalog import com.nvidia.spark.rapids.tool.views.{ProfDataSourceView, RawMetricProfilerView} import org.scalatest.funsuite.AnyFunSuite +import org.apache.spark.scheduler.AccumulableInfo import org.apache.spark.sql.{DataFrame, SparkSession} +import org.apache.spark.sql.rapids.tool.store.{AccumInfo, AccumInfoWithMaxAgg, AccumMetaRef} import org.apache.spark.sql.types._ case class TestStageDiagnosticResult( @@ -414,23 +417,37 @@ class AnalysisSuite extends AnyFunSuite { assert(agg.gpuSqlAggs.nonEmpty, "expected SQL-level GPU rows") assert(agg.gpuAppAggs.nonEmpty, "expected app-level GPU rows") - // Discovery: only gpu* / perfio.s3.* / multithreadReaderMaxParallelism names. + // Discovery: the catalog owns the rule; a declared metric qualifies on its family and an + // undeclared one on the legacy prefixes. agg.gpuStageAggs.foreach { row => - val n = row.metricName - assert(n.startsWith("gpu") || n.startsWith("perfio.s3.") || - n == "multithreadReaderMaxParallelism", s"unexpected metric name: $n") + assert(MetricCatalog.DEFAULT.isGpuReportedMetric(row.metricName), + s"unexpected metric name: ${row.metricName}") } - // Unit convention is internally consistent. + // Unit comes from the catalog. Asserted against the catalog rather than re-derived from the + // name, because re-deriving it here is what let the name-based rules ship broken. agg.gpuStageAggs.foreach { row => - val expected = - if (row.metricName.contains("Time") || row.metricName.contains("Wait")) "ms" - else if (row.metricName.contains("Bytes")) "bytes" - else "count" - assert(row.unit == expected, - s"unit mismatch for ${row.metricName}: got ${row.unit}, expected $expected") + assert(row.unit == MetricCatalog.DEFAULT.unitFor(row.metricName), + s"unit mismatch for ${row.metricName}: got ${row.unit}") } + // Every GPU accumulable that carries a non-zero value produces a row. The absence of this + // assertion is why eight of seventeen metrics could vanish unnoticed: they were reduced to + // zero by a spurious unit conversion and then dropped by the zero-signal filter. A metric + // that is genuinely all-zero in the log is still legitimately suppressed. + val expectedNames = apps.head.accumManager.accumInfoMap.values + .filter { ai => + ai.infoRef.isGpuReportedMetric && ai.getStageIds.exists { sId => + ai.calculateAccStatsForStage(sId).exists(st => st.max != 0L || st.total != 0L) + } + } + .map(_.infoRef.getName()) + .toSet + val emittedNames = agg.gpuStageAggs.map(_.metricName).toSet + assert(expectedNames.diff(emittedNames).isEmpty, + s"GPU accumulables with non-zero values but missing from the report: " + + s"${expectedNames.diff(emittedNames)}") + // Rollup math: SQL.sum == Σ stage.sum across the SQL's stages, per metric. val stageRowsByMetric = agg.gpuStageAggs.groupBy(_.metricName) val sqlRowsByMetric = agg.gpuSqlAggs.groupBy(_.metricName) @@ -459,15 +476,13 @@ class AnalysisSuite extends AnyFunSuite { } } - test("GPU metric aggregation: max-aggregated metrics carry only max") { + test("GPU metric aggregation: max-aggregated metrics carry max and avg but no sum") { val logs = Array(s"$logDir/gpu_oom_eventlog.zstd") val apps = ToolTestUtils.processProfileApps(logs, sparkSession) val agg = RawMetricProfilerView.getAggMetrics(apps.toSeq) - val maxOnlyNames = Set( - "gpuMaxDeviceMemoryBytes", "gpuMaxHostMemoryBytes", "gpuMaxPageableMemoryBytes", - "gpuMaxPinnedMemoryBytes", "gpuMaxDiskMemoryBytes", "gpuMaxTaskFootprint", - "gpuOnGpuTasksWaitingGPUMaxCount", "gpuMaxConcurrentGpuTasks", - "multithreadReaderMaxParallelism") + // Read from the catalog rather than duplicating the set here: the local copy this replaces + // drifted silently against the real one. + val maxOnlyNames = MetricCatalog.DEFAULT.maxAggregatedNames val maxRows = agg.gpuStageAggs.filter(r => maxOnlyNames.contains(r.metricName)) ++ agg.gpuSqlAggs.filter(r => maxOnlyNames.contains(r.metricName)) ++ agg.gpuAppAggs.filter(r => maxOnlyNames.contains(r.metricName)) @@ -488,9 +503,12 @@ class AnalysisSuite extends AnyFunSuite { case s: SQLAggGpuMetricsProfileResult => s.max case s: AppAggGpuMetricsProfileResult => s.max } - assert(sum.isEmpty, s"max-only metric should have empty sum: $r") - assert(avg.isEmpty, s"max-only metric should have empty avg: $r") - assert(max.isDefined, s"max-only metric should have a max value: $r") + // sum stays empty: adding per-task peaks is meaningless because they never coexist. + assert(sum.isEmpty, s"max-aggregated metric should have empty sum: $r") + // avg is populated: it is the mean across tasks of each task's own peak, which is a + // different quantity from the sum and was previously discarded on the way out. + assert(avg.isDefined, s"max-aggregated metric should carry an avg: $r") + assert(max.isDefined, s"max-aggregated metric should have a max value: $r") } } @@ -502,4 +520,111 @@ class AnalysisSuite extends AnyFunSuite { assert(agg.gpuSqlAggs.isEmpty, "CPU-only log should produce no GPU SQL rows") assert(agg.gpuAppAggs.isEmpty, "CPU-only log should produce no GPU app rows") } + + test("GPU metric avg is the arithmetic mean, not the ratcheting rolling mean") { + // `med` is a rolling mean recomputed with integer division on every update, so it drifts to + // the floor: on a real log it reports 1 for a metric whose true mean is 5.89. avg must be + // the raw sum over the raw count instead. Oracle computed here from the store's own totals, + // independently of how the analyzer builds the row. + val logs = Array(s"$logDir/gpu_oom_eventlog.zstd") + val apps = ToolTestUtils.processProfileApps(logs, sparkSession) + val agg = RawMetricProfilerView.getAggMetrics(apps.toSeq) + // Keyed by (name, stageId), NOT by name: accumInfoMap is keyed by accumulator id and each + // metric gets a distinct id per stage, so a name-keyed map keeps one entry and silently + // skips every other stage's rows. + val rawByNameAndStage = apps.head.accumManager.accumInfoMap.values + .filter(_.infoRef.isGpuReportedMetric) + .flatMap(ai => ai.getStageIds.flatMap(sId => + ai.getRawStatsForStage(sId).map(raw => (ai.infoRef.getName(), sId) -> raw))) + .toMap + var checked = 0 + agg.gpuStageAggs.foreach { row => + val raw = rawByNameAndStage.get((row.metricName, row.stageId)) + assert(raw.isDefined, s"no raw stats for ${row.metricName} stage ${row.stageId}") + val expected = if (raw.get.count > 0L) Some(raw.get.total / raw.get.count) else None + assert(row.avg == expected, + s"${row.metricName} stage ${row.stageId}: avg ${row.avg} != sum/count $expected") + checked += 1 + } + // every emitted row is checked, not an arbitrary subset + assert(checked == agg.gpuStageAggs.size, s"checked $checked of ${agg.gpuStageAggs.size}") + assert(checked > 0, "expected at least one GPU row to check") + } + + test("a max-aggregated metric publishes max and avg but never sum") { + val logs = Array(s"$logDir/gpu_oom_eventlog.zstd") + val apps = ToolTestUtils.processProfileApps(logs, sparkSession) + val agg = RawMetricProfilerView.getAggMetrics(apps.toSeq) + val maxRows = agg.gpuStageAggs.filter(r => + MetricCatalog.DEFAULT.isAggregatedByMax(r.metricName)) + assert(maxRows.nonEmpty) + maxRows.foreach { r => + assert(r.sum.isEmpty, s"per-task peaks must not be summed: $r") + assert(r.max.isDefined && r.avg.isDefined, s"max and avg both expected: $r") + // and avg must not have collapsed onto max or onto the floor + assert(r.avg.get <= r.max.get, s"avg above max: $r") + } + } + + test("a scaled metric renders divided in the emitted row") { + // Pins the published strings rather than the stored Longs: the store holds thousandths. + val row = StageAggGpuMetricsProfileResult( + stageId = 1, numTasks = 72, metricName = "gpuOnGpuTasksWaitingGPUAvgCount", + unit = "count", sum = None, max = Some(2500L), avg = Some(714L)) + assert(row.convertToCSVSeq().toSeq == Seq("1", "72", "gpuOnGpuTasksWaitingGPUAvgCount", + "count", "", "2.5", "0.714")) + // an unscaled metric is byte-identical to before + val plain = StageAggGpuMetricsProfileResult( + stageId = 1, numTasks = 72, metricName = "gpuMaxTaskFootprint", + unit = "bytes", sum = None, max = Some(7123115846L), avg = Some(1411606730L)) + assert(plain.convertToCSVSeq().toSeq == Seq("1", "72", "gpuMaxTaskFootprint", + "bytes", "", "7123115846", "1411606730")) + } + + test("every render site divides out the scale, not just the stage-level one") { + // Four independent copies of render() exist. The SQL and app rows look up the scale by + // metric name; AccumProfileResults reads it from its AccumMetaRef instead, a different + // source, and had no test at all. + val name = "gpuOnGpuTasksWaitingGPUAvgCount" + val sqlRow = SQLAggGpuMetricsProfileResult( + sqlId = 0, metricName = name, unit = "count", + sum = None, max = Some(2500L), avg = Some(714L)) + assert(sqlRow.convertToCSVSeq().toSeq.takeRight(3) == Seq("", "2.5", "0.714")) + val appRow = AppAggGpuMetricsProfileResult( + appId = "app-1", metricName = name, unit = "count", + sum = None, max = Some(2500L), avg = Some(714L)) + assert(appRow.convertToCSVSeq().toSeq.takeRight(3) == Seq("", "2.5", "0.714")) + // AccumProfileResults, which feeds stage_level_all_metrics.csv + val ref = AccumMetaRef(116L, Some(name)) + assert(ref.storageScale == 1000L, "precondition: the metric is scaled") + val accRow = AccumProfileResults(1, ref, min = 0L, median = 714L, max = 2500L, total = 2500L) + assert(accRow.convertToCSVSeq().toSeq.takeRight(4) == Seq("0", "0.714", "2.5", "2.5")) + // and an unscaled accumulable is unchanged + val plainRef = AccumMetaRef(104L, Some("gpuMaxTaskFootprint")) + assert(plainRef.storageScale == 1L) + val plainAcc = AccumProfileResults(1, plainRef, 1L, 2L, 3L, 6L) + assert(plainAcc.convertToCSVSeq().toSeq.takeRight(4) == Seq("1", "2", "3", "6")) + } + + test("a decimal metric round-trips from raw event value to rendered cell") { + // End to end through the store: parse -> fixed-point storage -> render. The only GPU + // fixture emits "0" for this metric on every task, so without this the whole path is + // exercised nowhere above the isolated unit tests. + val ref = AccumMetaRef(116L, Some("gpuOnGpuTasksWaitingGPUAvgCount")) + val info = AccumInfo(ref) + assert(info.isInstanceOf[AccumInfoWithMaxAgg], "declared aggregation: max") + Seq("0.5", "1.0", "2.5", "0").foreach { v => + info.addAccumToTask(7, AccumulableInfo(116L, Some("gpuOnGpuTasksWaitingGPUAvgCount"), + Some(v), None, internal = false, countFailedValues = false, None)) + } + val raw = info.getRawStatsForStage(7) + assert(raw.isDefined) + // stored in thousandths: 500 + 1000 + 2500 + 0 + assert(raw.get.total == 4000L, s"stored total ${raw.get.total}") + assert(raw.get.count == 4L) + assert(raw.get.max == 2500L) + val row = AccumProfileResults(7, ref, raw.get.min, raw.get.total / raw.get.count, + raw.get.max, raw.get.max) + assert(row.convertToCSVSeq().toSeq.takeRight(4) == Seq("0", "1", "2.5", "2.5")) + } } diff --git a/core/src/test/scala/com/nvidia/spark/rapids/tool/util/ToolUtilsSuite.scala b/core/src/test/scala/com/nvidia/spark/rapids/tool/util/ToolUtilsSuite.scala index c912f9822..79391af7d 100644 --- a/core/src/test/scala/com/nvidia/spark/rapids/tool/util/ToolUtilsSuite.scala +++ b/core/src/test/scala/com/nvidia/spark/rapids/tool/util/ToolUtilsSuite.scala @@ -164,6 +164,68 @@ class ToolUtilsSuite extends AnyFunSuite with Logging { } } + test("parseAccumFieldToLong normalizes each serialized form to a canonical unit") { + // The three branches produce Longs in three DIFFERENT units, and the function does not say + // which. That is the reason a downstream caller must not convert again. + EventUtils.parseAccumFieldToLong("1773") shouldBe Some(1773L) // raw + EventUtils.parseAccumFieldToLong("00:00:01.773") shouldBe Some(1773L) // MILLISECONDS + EventUtils.parseAccumFieldToLong("1.01GB (1085796899 bytes)") shouldBe + Some(1085796899L) // BYTES + // a decimal is not readable without a declared storage scale + EventUtils.parseAccumFieldToLong("0.5") shouldBe None + EventUtils.parseAccumFieldToLong("[]") shouldBe None + } + + test("parseScaledDecimalToLong stores decimals as fixed point") { + EventUtils.parseScaledDecimalToLong("0.5", 1000L) shouldBe Some(500L) + EventUtils.parseScaledDecimalToLong("2.6666666666666665", 1000L) shouldBe Some(2667L) + EventUtils.parseScaledDecimalToLong("3.0", 1000L) shouldBe Some(3000L) + // the accumulator's zero case arrives as a plain integer and must scale the same way, so + // that every value of the metric is stored in the same units + EventUtils.parseScaledDecimalToLong("0", 1000L) shouldBe Some(0L) + EventUtils.parseScaledDecimalToLong("7", 1000L) shouldBe Some(7000L) + EventUtils.parseScaledDecimalToLong("-1.5", 1000L) shouldBe Some(-1500L) + } + + test("parseScaledDecimalToLong accepts exponent form, which Double.toString emits") { + // java.lang.Double.toString switches to E-notation below 1e-3, and this metric is an + // average that can legitimately land there. Rejecting the shape would drop the sample and + // remove the task from the denominator of the mean. + EventUtils.parseScaledDecimalToLong("5.0E-4", 1000L) shouldBe Some(1L) + EventUtils.parseScaledDecimalToLong("2.5E-4", 1000L) shouldBe Some(0L) + EventUtils.parseScaledDecimalToLong("1.0E4", 1000L) shouldBe Some(10000000L) + EventUtils.parseScaledDecimalToLong("1.5e2", 1000L) shouldBe Some(150000L) + } + + test("parseScaledDecimalToLong rejects everything Double.parseDouble would wrongly accept") { + // Double.parseDouble accepts all of these. "Infinity" in particular would become + // Long.MaxValue and overflow the running total to a negative number. + Seq("NaN", "Infinity", "-Infinity", "3d", "5f", "0x1p3", "", " ", + "1.2.3", "[]", "null", "1,5", "1e", "e5", "--1").foreach { bad => + EventUtils.parseScaledDecimalToLong(bad, 1000L) shouldBe None + } + // A value that would overflow after scaling is rejected rather than wrapping, and the + // bound is symmetric: the negative mirror must not be clamped to Long.MinValue, which is + // what a strict lower bound did -- double spacing near 2^63 is 2048, so a true value beyond + // the range rounds onto the representable -2^63. + EventUtils.parseScaledDecimalToLong("9223372036854775.0", 1000L) shouldBe None + EventUtils.parseScaledDecimalToLong("-9223372036854775.0", 1000L) shouldBe None + EventUtils.parseScaledDecimalToLong("-9223372036854776.0", 1000L) shouldBe None + EventUtils.parseScaledDecimalToLong("1e300", 1000L) shouldBe None + EventUtils.parseScaledDecimalToLong("-1e300", 1000L) shouldBe None + // negatives well inside range still work + EventUtils.parseScaledDecimalToLong("-1.5", 1000L) shouldBe Some(-1500L) + } + + test("parseAccumFieldToLong with a scale routes to the right parser") { + // scale 1 is the pre-existing behaviour, unchanged + EventUtils.parseAccumFieldToLong("00:00:01.773", 1L) shouldBe Some(1773L) + EventUtils.parseAccumFieldToLong("0.5", 1L) shouldBe None + // a scaled metric reads decimals and scales plain integers alike + EventUtils.parseAccumFieldToLong("0.5", 1000L) shouldBe Some(500L) + EventUtils.parseAccumFieldToLong("0", 1000L) shouldBe Some(0L) + } + test("convertMemorySizeToBytes should correctly parse memory sizes") { // Test basic unit conversions with default ByteUnit.BYTE StringUtils.convertMemorySizeToBytes("1024b", None) shouldBe 1024L From b2c5f8fd450f0fbbf2ca7cf558cec6d7ab23d959 Mon Sep 17 00:00:00 2001 From: "Ahmed Hussein (amahussein)" Date: Fri, 4 Sep 2026 09:38:13 -0500 Subject: [PATCH 2/2] Fix SQL/app avg for sparsely reported GPU metrics The rollup weighted stage means by the stage task count, which includes tasks that never reported the metric. Pool the totals and counts instead. Signed-off-by: Ahmed Hussein (amahussein) --- .../configs/reports/coreRawMetricsReport.yaml | 25 +++--- .../analysis/AppSparkMetricsAnalyzer.scala | 79 ++++++++----------- .../profiling/ProfileClassWarehouse.scala | 48 ++++++++--- .../rapids/tool/profiling/AnalysisSuite.scala | 69 +++++++++++++++- 4 files changed, 150 insertions(+), 71 deletions(-) diff --git a/core/src/main/resources/configs/reports/coreRawMetricsReport.yaml b/core/src/main/resources/configs/reports/coreRawMetricsReport.yaml index 885fe0627..4ba0deef7 100644 --- a/core/src/main/resources/configs/reports/coreRawMetricsReport.yaml +++ b/core/src/main/resources/configs/reports/coreRawMetricsReport.yaml @@ -911,10 +911,11 @@ reportDefinitions: - name: avg dataType: Double description: >- - Rolling per-task average. For a max-aggregated metric this is the mean across - tasks of each task's own peak, which is a different quantity from sum and is - populated. sum stays empty for those metrics, because adding per-task peaks is - meaningless when the peaks never coexist. + Arithmetic mean, over the tasks that reported this metric, of each task's + reported value. Computed once as total over count, so it is not the rolling + mean the accumulator store maintains. Populated for max-aggregated metrics + too; sum stays empty for those, because adding per-task peaks is meaningless + when the peaks never coexist. - label: coreRawGpuSqlLevelAggregatedTaskMetricsCSV description: >- GPU task metric aggregations at SQL level. Rolls up stage-level GPU @@ -949,10 +950,10 @@ reportDefinitions: - name: avg dataType: Double description: >- - Task-weighted average across the stages in the SQL. Populated for - max-aggregated metrics too, where it is the mean across tasks of each task's - own peak -- a different quantity from sum, which stays empty for those - metrics because per-task peaks never coexist. + Arithmetic mean over every task in the SQL that reported this metric, pooled + across stages as total over count rather than by re-averaging stage means. + Populated for max-aggregated metrics too; sum stays empty for those, because + adding per-task peaks is meaningless when the peaks never coexist. - label: coreRawGpuAppLevelAggregatedTaskMetricsCSV description: >- GPU task metric aggregations at application level. One row per @@ -987,10 +988,10 @@ reportDefinitions: - name: avg dataType: Double description: >- - Task-weighted average across all stages in the app. Populated for - max-aggregated metrics too, where it is the mean across tasks of each task's - own peak -- a different quantity from sum, which stays empty for those - metrics because per-task peaks never coexist. + Arithmetic mean over every task in the app that reported this metric, pooled + across stages as total over count rather than by re-averaging stage means. + Populated for max-aggregated metrics too; sum stays empty for those, because + adding per-task peaks is meaningless when the peaks never coexist. # AccumProfileResults - label: coreRawStageLevelAllMetricsCSV description: >- diff --git a/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSparkMetricsAnalyzer.scala b/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSparkMetricsAnalyzer.scala index d44c2a850..6037ae4ef 100644 --- a/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSparkMetricsAnalyzer.scala +++ b/core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSparkMetricsAnalyzer.scala @@ -440,48 +440,36 @@ class AppSparkMetricsAnalyzer(app: AppBase) extends AppAnalysisBase(app) with Lo gpuAccums.foreach { ai => val name = ai.infoRef.getName() val unit = unitForMetric(name) - val isMax = ai.infoRef.isAggregateByMax ai.getStageIds.foreach { stageId => ai.calculateAccStatsForStage(stageId).foreach { stats => // Invariant: stages with GPU accumulators are tracked by stageManager // and therefore cached. The fallback to 0 is defensive for edge cases // (e.g. driver-side accumulators) where the stage is absent from the - // task-metrics cache. A 0 here would also exclude the stage from the - // task-weighted avg in rollupGpuRows while sum/max still accumulate, - // so log a warning so the inconsistency is visible. + // task-metrics cache. It affects only the emitted numTasks column: the + // SQL/app rollups pool total/count and never read numTasks. val numTasks = stageCache.get(stageId).map(_.numTasks).getOrElse { logWarning(s"GPU accumulator '$name' references stage $stageId which " + - s"is not in the stage-task metrics cache; using numTasks = 0. " + - s"This will be excluded from task-weighted averages at SQL/app level.") + s"is not in the stage-task metrics cache; using numTasks = 0.") 0 } - // The arithmetic mean, from the raw sum and count rather than from `med`. `med` is a - // rolling mean recomputed with integer division on every update, so it ratchets toward - // the floor: on a real log it reports 1 for a metric whose true mean is 5.89, and 5 for - // one whose true mean is 12.03. Dividing once is correct to within truncation. - val avg = ai.getRawStatsForStage(stageId).flatMap { raw => - if (raw.count > 0L) Some(raw.total / raw.count) else None - } - // sum stays empty for a max-aggregated metric: adding up per-task peaks is meaningless - // because those peaks never coexist. avg does not -- it is the mean across tasks of - // each task's own peak, which is a genuinely different quantity. - val (sum, max) = if (isMax) { - (None: Option[Long], Some(stats.max)) - } else { - (Some(stats.total), Some(stats.max)) - } - // Skip rows carrying no signal (both sum and max zero/absent). - val zeroSum = sum.forall(_ == 0L) - val zeroMax = max.forall(_ == 0L) - if (!(zeroSum && zeroMax)) { - rows += StageAggGpuMetricsProfileResult( - stageId = stageId, - numTasks = numTasks, - metricName = name, - unit = unit, - sum = sum, - max = max, - avg = avg) + // The row carries what the store holds -- the accumulated total and the number of + // tasks that reported the metric -- and derives the published sum and avg from them. + // `stats` has been through readjustTotalStats, which replaces total with max for a + // max-aggregated metric, so the unadjusted record is the only place the real sum + // survives, and that sum is the numerator every mean needs. + val rawStats = ai.getRawStatsForStage(stageId) + val max = Some(stats.max) + val row = StageAggGpuMetricsProfileResult( + stageId = stageId, + numTasks = numTasks, + metricName = name, + unit = unit, + total = rawStats.map(_.total), + max = max, + count = rawStats.map(_.count).getOrElse(0L)) + // Skip rows carrying no signal (both the published sum and max zero/absent). + if (!(row.sum.forall(_ == 0L) && max.forall(_ == 0L))) { + rows += row } } } @@ -491,11 +479,10 @@ class AppSparkMetricsAnalyzer(app: AppBase) extends AppAnalysisBase(app) with Lo /** * Rollup helper: groups stage-level GPU rows by metric name and reduces to - * (unit, sum, max, avg). sum is Σ stage.sum (None for max metrics); max is - * max stage.max; avg is task-weighted Σ(stage.avg * stage.numTasks) - * / Σ stage.numTasks over the stages that recorded the metric. numTasks is - * intentionally not propagated — see SQLAggGpuMetricsProfileResult / - * AppAggGpuMetricsProfileResult docstrings. + * (unit, sum, max, avg). sum adds the stage sums (None for max metrics); max is + * the largest stage max; avg divides the pooled stage totals by the pooled stage + * counts over the stages that recorded the metric. numTasks is intentionally not + * propagated: see SQLAggGpuMetricsProfileResult / AppAggGpuMetricsProfileResult. */ private def rollupGpuRows( rows: Seq[StageAggGpuMetricsProfileResult] @@ -510,14 +497,16 @@ class AppSparkMetricsAnalyzer(app: AppBase) extends AppAnalysisBase(app) with Lo val xs = group.flatMap(_.max) if (xs.isEmpty) None else Some(xs.max) } + // Pooled as sum-of-totals over sum-of-counts, not a weighted average of the stage + // averages. Two reasons: stage `avg` is already integer-truncated, so re-averaging + // truncates twice; and the correct weight is the number of tasks that reported the + // metric, not the stage's task count. GPU accumulables are frequently sparse -- spill + // and retry metrics land on a handful of tasks in a large stage -- so weighting by + // numTasks skews the result toward the stages that reported it least, without bound. val avgOpt: Option[Long] = { - val weighted = group.flatMap { r => r.avg.map(a => (a, r.numTasks)) } - val weightTasks = weighted.map(_._2).sum - if (weighted.isEmpty || weightTasks == 0) { - None - } else { - Some(weighted.map { case (a, n) => a * n }.sum / weightTasks) - } + val reporting = group.filter(_.count > 0L) + val totalCount = reporting.map(_.count).sum + if (totalCount <= 0L) None else Some(reporting.flatMap(_.total).sum / totalCount) } (metricName, unit, sumOpt, maxOpt, avgOpt) }.toSeq diff --git a/core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ProfileClassWarehouse.scala b/core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ProfileClassWarehouse.scala index de5246411..fda14a306 100644 --- a/core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ProfileClassWarehouse.scala +++ b/core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ProfileClassWarehouse.scala @@ -1584,13 +1584,23 @@ object UnixExitCode { } /** - * GPU task metric aggregation at stage level — one row per (stageId, metricName). - * Long/transposed schema: unit and sum/max/avg vary by metric. Empty `sum` / `avg` - * denote max-aggregated metrics (e.g. `gpuMaxDeviceMemoryBytes`). + * GPU task metric aggregation at stage level: one row per (stageId, metricName). + * Long/transposed schema: unit and the three numeric cells vary by metric. + * + * The row stores what the accumulator store actually holds, `total` and `count`, and + * derives the two published quantities from them. `sum` is `total` suppressed for + * max-aggregated metrics, whose per-task peaks never coexist and so must not be added; + * `avg` is `total / count`. Keeping the derivation here rather than at construction means + * the SQL and app rollups can pool the raw totals and divide once, instead of re-averaging + * already-truncated stage means against a denominator (`numTasks`) that counts tasks which + * never reported the metric. + * + * `count` is the number of tasks that reported the metric, which is at most `numTasks` and + * is often far below it: spill and retry metrics land on a handful of tasks in a stage. * * Note on stage attempts: unlike StageAggTaskMetricsProfileResult, this class has * no aggregateStageProfileMetric helper because attempt merging happens upstream - * at the AccumInfo layer — `AccumInfo.stagesStatMap` is keyed by stageId only + * at the AccumInfo layer. `AccumInfo.stagesStatMap` is keyed by stageId only * (not stageId + attemptNumber), so calculateAccStatsForStage already returns * the merged result across attempts. */ @@ -1599,9 +1609,25 @@ case class StageAggGpuMetricsProfileResult( numTasks: Int, metricName: String, unit: String, - sum: Option[Long], + total: Option[Long], max: Option[Long], - avg: Option[Long]) extends ProfileResult { + count: Long) extends ProfileResult { + + /** + * The published total. Empty for a max-aggregated metric: adding up per-task peaks is + * meaningless because those peaks never coexist. `total` still holds the sum, which is a + * valid numerator for `avg` even where it is not a valid figure to publish. + */ + def sum: Option[Long] = { + if (MetricCatalog.DEFAULT.isAggregatedByMax(metricName)) None else total + } + + /** + * Arithmetic mean over the tasks that reported the metric. Divided once from the raw total + * rather than read from the store's `med`, which is a rolling mean recomputed with integer + * division on every update and therefore ratchets toward the floor. + */ + def avg: Option[Long] = if (count > 0L) total.map(_ / count) else None override def outputHeaders: Array[String] = { OutHeaderRegistry.outputHeaders("StageAggGpuMetricsProfileResult") @@ -1627,10 +1653,12 @@ case class StageAggGpuMetricsProfileResult( } /** - * GPU task metric aggregation at SQL level — one row per (sqlId, metricName). - * Rolled up from stage-level rows: sum = Σ stage.sum, max = max stage.max, - * avg = task-weighted average over stage.avg. numTasks is intentionally not - * carried — it would be a constant per SQL across every metric row (the non-GPU + * GPU task metric aggregation at SQL level: one row per (sqlId, metricName). + * Rolled up from stage-level rows. sum adds the stage sums (None for max metrics); + * max is the largest stage max; avg divides the pooled stage totals by the pooled + * stage counts, so it is a mean over the tasks that reported the metric rather than + * a re-average of the stage means. numTasks is intentionally not carried: it would + * be a constant per SQL across every metric row (the non-GPU * sql_level_aggregated_task_metrics.csv already has it once per SQL). */ case class SQLAggGpuMetricsProfileResult( diff --git a/core/src/test/scala/com/nvidia/spark/rapids/tool/profiling/AnalysisSuite.scala b/core/src/test/scala/com/nvidia/spark/rapids/tool/profiling/AnalysisSuite.scala index 454a5f41d..908e53e20 100644 --- a/core/src/test/scala/com/nvidia/spark/rapids/tool/profiling/AnalysisSuite.scala +++ b/core/src/test/scala/com/nvidia/spark/rapids/tool/profiling/AnalysisSuite.scala @@ -505,8 +505,8 @@ class AnalysisSuite extends AnyFunSuite { } // sum stays empty: adding per-task peaks is meaningless because they never coexist. assert(sum.isEmpty, s"max-aggregated metric should have empty sum: $r") - // avg is populated: it is the mean across tasks of each task's own peak, which is a - // different quantity from the sum and was previously discarded on the way out. + // avg is populated: it is the mean, over the tasks that reported the metric, of each + // task's reported value -- a different quantity from the sum, and previously discarded. assert(avg.isDefined, s"max-aggregated metric should carry an avg: $r") assert(max.isDefined, s"max-aggregated metric should have a max value: $r") } @@ -566,17 +566,78 @@ class AnalysisSuite extends AnyFunSuite { } } + test("SQL and app avg pool the reporting tasks rather than re-averaging stage means") { + // The rollup used to weight each stage mean by the stage's task count. That denominator + // counts tasks which never reported the metric, and GPU accumulables are often sparse, so + // the published mean drifted toward the stages that reported it least. It also truncated + // twice, once at stage level and again in the rollup. + val logs = Array(s"$logDir/gpu_oom_eventlog.zstd") + val apps = ToolTestUtils.processProfileApps(logs, sparkSession) + val agg = RawMetricProfilerView.getAggMetrics(apps.toSeq) + // Oracle read straight from the store, keyed by (name, stageId) because accumInfoMap is + // keyed by accumulator id and each metric gets a distinct id per stage. + val rawByNameAndStage = apps.head.accumManager.accumInfoMap.values + .filter(_.infoRef.isGpuReportedMetric) + .flatMap(ai => ai.getStageIds.flatMap(sId => + ai.getRawStatsForStage(sId).map(raw => (ai.infoRef.getName(), sId) -> raw))) + .toMap + // Restricted to the stages that actually produced a row, so the zero-signal filter in + // aggregateGpuMetricsByStage does not make the oracle disagree for the wrong reason. + val pooled = agg.gpuStageAggs.groupBy(_.metricName).map { case (name, group) => + val stats = group.flatMap(r => rawByNameAndStage.get((name, r.stageId))) + .filter(_.count > 0L) + val pooledTotal = stats.map(_.total).sum + val pooledCount = stats.map(_.count).sum + name -> ((pooledTotal, pooledCount)) + } + assert(agg.gpuAppAggs.nonEmpty, "expected app-level GPU rows") + agg.gpuAppAggs.foreach { row => + val (total, count) = pooled(row.metricName) + assert(count > 0L, s"no reporting tasks for ${row.metricName}") + assert(row.avg.contains(total / count), + s"${row.metricName}: avg ${row.avg} != pooled $total/$count") + } + // SQL rows pool over that SQL's own stage set, derived here rather than reusing the + // app-level pooling above. + val sqlToStages = apps.head.sqlIdToStages + assert(agg.gpuSqlAggs.nonEmpty, "expected SQL-level GPU rows") + agg.gpuSqlAggs.foreach { row => + val stageIds = sqlToStages.getOrElse(row.sqlId, Seq.empty).toSet + val stats = agg.gpuStageAggs + .filter(r => r.metricName == row.metricName && stageIds.contains(r.stageId)) + .flatMap(r => rawByNameAndStage.get((r.metricName, r.stageId))) + .filter(_.count > 0L) + val sqlCount = stats.map(_.count).sum + assert(sqlCount > 0L, s"no reporting tasks for ${row.metricName} in SQL ${row.sqlId}") + assert(row.avg.contains(stats.map(_.total).sum / sqlCount), + s"SQL ${row.sqlId} ${row.metricName}: avg ${row.avg} is not the pooled mean") + } + + // Regression guard: on this fixture the two formulas genuinely disagree for at least one + // metric, so reverting to the task-count weighting fails here instead of passing silently. + val byNumTasks = agg.gpuStageAggs.groupBy(_.metricName).map { case (name, group) => + val weighted = group.flatMap(r => r.avg.map(a => (a, r.numTasks.toLong))) + val tasks = weighted.map(_._2).sum + name -> (if (tasks == 0L) None else Some(weighted.map(p => p._1 * p._2).sum / tasks)) + } + assert(agg.gpuAppAggs.exists(row => byNumTasks(row.metricName) != row.avg), + "fixture no longer distinguishes the two rollup formulas; the guard is now vacuous") + } + test("a scaled metric renders divided in the emitted row") { // Pins the published strings rather than the stored Longs: the store holds thousandths. + // total and count are the stored inputs; sum and avg are derived on the way out. val row = StageAggGpuMetricsProfileResult( stageId = 1, numTasks = 72, metricName = "gpuOnGpuTasksWaitingGPUAvgCount", - unit = "count", sum = None, max = Some(2500L), avg = Some(714L)) + unit = "count", total = Some(3570L), max = Some(2500L), count = 5L) + // the stored total is present; it is the publishing of it that is suppressed + assert(row.total.isDefined && row.sum.isEmpty, "max-aggregated total must stay unpublished") assert(row.convertToCSVSeq().toSeq == Seq("1", "72", "gpuOnGpuTasksWaitingGPUAvgCount", "count", "", "2.5", "0.714")) // an unscaled metric is byte-identical to before val plain = StageAggGpuMetricsProfileResult( stageId = 1, numTasks = 72, metricName = "gpuMaxTaskFootprint", - unit = "bytes", sum = None, max = Some(7123115846L), avg = Some(1411606730L)) + unit = "bytes", total = Some(4234820190L), max = Some(7123115846L), count = 3L) assert(plain.convertToCSVSeq().toSeq == Seq("1", "72", "gpuMaxTaskFootprint", "bytes", "", "7123115846", "1411606730")) }