rui-mo commented on code in PR #12126:
URL: https://github.com/apache/gluten/pull/12126#discussion_r3558206561
##########
backends-velox/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala:
##########
@@ -16,17 +16,132 @@
*/
package org.apache.gluten.metrics
+import org.apache.gluten.exception.GlutenException
import org.apache.gluten.execution._
import org.apache.gluten.substrait.{AggregationParams, JoinParams}
import org.apache.spark.internal.Logging
import org.apache.spark.metrics.TaskStatsAccumulator
import org.apache.spark.sql.execution.SparkPlan
+import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
+
import java.lang.{Long => JLong}
import java.util.{ArrayList => JArrayList, List => JList, Map => JMap}
+import scala.collection.JavaConverters._
+
object MetricsUtil extends Logging {
+ private val objectMapper = new ObjectMapper()
+
+ private def value(node: JsonNode, fieldName: String): Long =
+ Option(node.get(fieldName)).map(_.asLong()).getOrElse(0L)
+
+ private def customMetricSum(node: JsonNode, metricName: String): Long =
+ Option(node.get("customStats"))
+ .flatMap(customStats => Option(customStats.get(metricName)))
+ .flatMap(metric => Option(metric.get("sum")))
+ .map(_.asLong())
+ .getOrElse(0L)
+
+ private def customMetricCount(node: JsonNode, metricName: String): Long =
+ Option(node.get("customStats"))
+ .flatMap(customStats => Option(customStats.get(metricName)))
+ .flatMap(metric => Option(metric.get("count")))
+ .map(_.asLong())
+ .getOrElse(0L)
+
+ private def operatorMetricFromJson(node: JsonNode): OperatorMetrics = {
+ val metrics = new OperatorMetrics()
+ metrics.inputRows = value(node, "inputRows")
+ metrics.inputVectors = value(node, "inputVectors")
+ metrics.inputBytes = value(node, "inputBytes")
+ metrics.rawInputRows = value(node, "rawInputRows")
+ metrics.rawInputBytes = value(node, "rawInputBytes")
+ metrics.outputRows = value(node, "outputRows")
+ metrics.outputVectors = value(node, "outputVectors")
+ metrics.outputBytes = value(node, "outputBytes")
+ metrics.cpuCount = value(node, "cpuCount")
+ metrics.wallNanos = value(node, "wallNanos")
+ metrics.peakMemoryBytes = value(node, "peakMemoryBytes")
+ metrics.numMemoryAllocations = value(node, "numMemoryAllocations")
+ metrics.spilledInputBytes = value(node, "spilledInputBytes")
+ metrics.spilledBytes = value(node, "spilledBytes")
+ metrics.spilledRows = value(node, "spilledRows")
+ metrics.spilledPartitions = value(node, "spilledPartitions")
+ metrics.spilledFiles = value(node, "spilledFiles")
+ metrics.numDynamicFiltersProduced = customMetricSum(node,
"dynamicFiltersProduced")
+ metrics.numDynamicFiltersAccepted = customMetricSum(node,
"dynamicFiltersAccepted")
+ metrics.numReplacedWithDynamicFilterRows =
+ customMetricSum(node, "replacedWithDynamicFilterRows")
+ metrics.numDynamicFilterInputRows = customMetricSum(node,
"dynamicFilterInputRows")
+ metrics.flushRowCount = customMetricSum(node, "flushRowCount")
+ metrics.abandonedPartialAggregationRows =
+ customMetricSum(node, "abandonedPartialAggregationRows")
+ metrics.loadedToValueHook = customMetricSum(node, "loadedToValueHook")
+ metrics.bloomFilterBlocksByteSize = customMetricSum(node,
"bloomFilterSize")
+ metrics.scanTime = customMetricSum(node, "totalScanTime")
+ metrics.skippedSplits = customMetricSum(node, "skippedSplits")
+ metrics.processedSplits = customMetricSum(node, "processedSplits")
+ metrics.skippedStrides = customMetricSum(node, "skippedStrides")
+ metrics.processedStrides = customMetricSum(node, "processedStrides")
+ metrics.remainingFilterTime = customMetricSum(node,
"totalRemainingFilterWallNanos")
+ metrics.ioWaitTime = customMetricSum(node, "ioWaitWallNanos")
+ metrics.storageReadBytes = customMetricSum(node, "storageReadBytes")
+ metrics.storageReads = customMetricCount(node, "storageReadBytes")
+ metrics.localReadBytes = customMetricSum(node, "localReadBytes")
+ metrics.ramReadBytes = customMetricSum(node, "ramReadBytes")
+ metrics.preloadSplits = customMetricSum(node, "readyPreloadedSplits")
+ metrics.pageLoadTime = customMetricSum(node, "pageLoadTimeNs")
+ metrics.dataSourceAddSplitTime = customMetricSum(node,
"dataSourceAddSplitWallNanos") +
+ customMetricSum(node, "waitForPreloadSplitNanos")
+ metrics.dataSourceReadTime = customMetricSum(node,
"dataSourceReadWallNanos")
+ metrics.physicalWrittenBytes = value(node, "physicalWrittenBytes")
+ metrics.writeIOTime = customMetricSum(node, "writeIOWallNanos")
+ metrics.numWrittenFiles = customMetricSum(node, "numWrittenFiles")
+ metrics
+ }
+
+ private def parseNativeOperatorMetrics(metrics: Metrics):
JArrayList[OperatorMetrics] = {
+ val operatorMetrics = new JArrayList[OperatorMetrics]()
+ if (metrics.numMetrics == 0 || metrics.metricsJson == null ||
metrics.metricsJson.isEmpty) {
+ return operatorMetrics
+ }
+
+ val root = objectMapper.readTree(metrics.metricsJson)
+ val omittedNodeIds =
root.path("omittedNodeIds").elements().asScala.map(_.asText()).toSet
+ val nodeStats = root.path("nodeStats")
+ val orderedNodeIds = root.path("orderedNodeIds")
+ orderedNodeIds.elements().asScala.foreach {
+ nodeIdNode =>
+ val nodeId = nodeIdNode.asText()
+ val node = nodeStats.get(nodeId)
+ if (node == null || node.isNull || node.isMissingNode) {
+ if (omittedNodeIds.contains(nodeId)) {
+ operatorMetrics.add(new OperatorMetrics())
+ } else {
+ throw new GlutenException(s"Node id cannot be found in native
metrics: $nodeId.")
+ }
+ } else {
+ node.path("operatorStats").elements().asScala.foreach {
+ opStats => operatorMetrics.add(operatorMetricFromJson(opStats))
+ }
+ }
+ }
+
+ val loadLazyVectorMetricsIdx = orderedNodeIds.size() - 1
+ if (loadLazyVectorMetricsIdx >= 0 && loadLazyVectorMetricsIdx <
operatorMetrics.size()) {
+ operatorMetrics.get(loadLazyVectorMetricsIdx).loadLazyVectorTime =
+ root.path("loadLazyVectorTime").asLong(0L)
+ }
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
