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..4ba0deef7 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,25 @@ 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. + 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 @@ -928,20 +936,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. + 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 @@ -961,21 +973,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. + 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: >- @@ -996,19 +1012,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..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 @@ -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" - } + // 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 unitForMetric(name: String): String = { - if (name.contains("Time") || name.contains("Wait")) "ms" - else if (name.contains("Bytes")) "bytes" - else "count" - } - - 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 @@ -452,42 +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 } - val (sum, max, avg) = if (isMax) { - (None: Option[Long], - Some(convertValue(name, stats.max)), - None: Option[Long]) - } else { - (Some(convertValue(name, stats.total)), - Some(convertValue(name, stats.max)), - Some(convertValue(name, stats.med))) - } - // 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 } } } @@ -497,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] @@ -516,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/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..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 @@ -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)) } } @@ -1576,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. */ @@ -1591,33 +1609,56 @@ 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") } + /** 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() } /** - * 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( @@ -1632,14 +1673,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 +1708,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..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 @@ -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, 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") } } @@ -502,4 +520,172 @@ 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("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", 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", total = Some(4234820190L), max = Some(7123115846L), count = 3L) + 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