You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2023/02/21 13:48:33 UTC
[spark] branch master updated: [SPARK-41765][SQL] Pull out v1 write metrics to WriteFiles
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new a111a02de1a [SPARK-41765][SQL] Pull out v1 write metrics to WriteFiles
a111a02de1a is described below
commit a111a02de1a814c5f335e0bcac4cffb0515557dc
Author: ulysses-you <ul...@gmail.com>
AuthorDate: Tue Feb 21 21:48:00 2023 +0800
[SPARK-41765][SQL] Pull out v1 write metrics to WriteFiles
### What changes were proposed in this pull request?
Pull out the metrics which is in `InsertIntoHiveTable` and `InsertIntoHadoopFsRelationCommand` to `WriteFiles`.
### Why are the changes needed?
Move metrics to the right place.
### Does this PR introduce _any_ user-facing change?
yes, the SQL UI metrics changed from `V1WriteCommand` to `WriteFiles`
with `spark.sql.optimizer.plannedWrite.enabled` disable and enable:
// disable
<img width="314" alt="image" src="https://user-images.githubusercontent.com/12025282/220267296-62d2deef-f8d8-4e71-adc6-23e416f0777c.png">
// enable
<img width="319" alt="image" src="https://user-images.githubusercontent.com/12025282/220267151-dd1e7ed9-eb92-44a5-abdc-75f204ecf97e.png">
### How was this patch tested?
fix and improve test
Closes #39428 from ulysses-you/SPARK-41765.
Authored-by: ulysses-you <ul...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../sql/execution/command/DataWritingCommand.scala | 47 +++++++--------
.../datasources/BasicWriteStatsTracker.scala | 17 +++++-
.../execution/datasources/FileFormatWriter.scala | 69 +++++++++++++---------
.../sql/execution/datasources/WriteFiles.scala | 3 +
.../BasicWriteJobStatsTrackerMetricSuite.scala | 19 +++---
.../sql/execution/metric/SQLMetricsSuite.scala | 28 +++++----
.../sql/execution/metric/SQLMetricsTestUtils.scala | 21 ++++++-
.../hive/execution/InsertIntoHiveDirCommand.scala | 7 +++
.../spark/sql/hive/execution/SQLMetricsSuite.scala | 42 ++++++++-----
9 files changed, 159 insertions(+), 94 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/DataWritingCommand.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/DataWritingCommand.scala
index 338ce8cac42..58c3fca4ad7 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/DataWritingCommand.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/DataWritingCommand.scala
@@ -21,14 +21,13 @@ import java.net.URI
import org.apache.hadoop.conf.Configuration
-import org.apache.spark.SparkContext
import org.apache.spark.sql.{Row, SaveMode, SparkSession}
import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, UnaryCommand}
import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.execution.{SparkPlan, SQLExecution}
+import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.datasources.BasicWriteJobStatsTracker
-import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.util.SerializableConfiguration
@@ -52,11 +51,19 @@ trait DataWritingCommand extends UnaryCommand {
def outputColumns: Seq[Attribute] =
DataWritingCommand.logicalPlanOutputWithNames(query, outputColumnNames)
- lazy val metrics: Map[String, SQLMetric] = BasicWriteJobStatsTracker.metrics
+ lazy val metrics: Map[String, SQLMetric] = {
+ // If planned write is enable, we have pulled out write files metrics from `V1WriteCommand`
+ // to `WriteFiles`. `DataWritingCommand` should only holds the task commit metric and driver
+ // commit metric.
+ if (conf.getConf(SQLConf.PLANNED_WRITE_ENABLED)) {
+ BasicWriteJobStatsTracker.writeCommitMetrics
+ } else {
+ BasicWriteJobStatsTracker.metrics
+ }
+ }
def basicWriteJobStatsTracker(hadoopConf: Configuration): BasicWriteJobStatsTracker = {
- val serializableHadoopConf = new SerializableConfiguration(hadoopConf)
- new BasicWriteJobStatsTracker(serializableHadoopConf, metrics)
+ DataWritingCommand.basicWriteJobStatsTracker(metrics, hadoopConf)
}
def run(sparkSession: SparkSession, child: SparkPlan): Seq[Row]
@@ -79,27 +86,6 @@ object DataWritingCommand {
}
}
- /**
- * When execute CTAS operators, Spark will use [[InsertIntoHadoopFsRelationCommand]]
- * or [[InsertIntoHiveTable]] command to write data, they both inherit metrics from
- * [[DataWritingCommand]], but after running [[InsertIntoHadoopFsRelationCommand]]
- * or [[InsertIntoHiveTable]], we only update metrics in these two command through
- * [[BasicWriteJobStatsTracker]], we also need to propogate metrics to the command
- * that actually calls [[InsertIntoHadoopFsRelationCommand]] or [[InsertIntoHiveTable]].
- *
- * @param sparkContext Current SparkContext.
- * @param command Command to execute writing data.
- * @param metrics Metrics of real DataWritingCommand.
- */
- def propogateMetrics(
- sparkContext: SparkContext,
- command: DataWritingCommand,
- metrics: Map[String, SQLMetric]): Unit = {
- command.metrics.foreach { case (key, metric) => metrics(key).set(metric.value) }
- SQLMetrics.postDriverMetricUpdates(sparkContext,
- sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY),
- metrics.values.toSeq)
- }
/**
* When execute CTAS operators, and the location is not empty, throw [[AnalysisException]].
* For CTAS, the SaveMode is always [[ErrorIfExists]]
@@ -120,4 +106,11 @@ object DataWritingCommand {
}
}
}
+
+ def basicWriteJobStatsTracker(
+ metrics: Map[String, SQLMetric],
+ hadoopConf: Configuration): BasicWriteJobStatsTracker = {
+ val serializableHadoopConf = new SerializableConfiguration(hadoopConf)
+ new BasicWriteJobStatsTracker(serializableHadoopConf, metrics)
+ }
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/BasicWriteStatsTracker.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/BasicWriteStatsTracker.scala
index 47685899784..980659b9fab 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/BasicWriteStatsTracker.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/BasicWriteStatsTracker.scala
@@ -197,6 +197,11 @@ class BasicWriteJobStatsTracker(
this(serializableHadoopConf, metrics - TASK_COMMIT_TIME, metrics(TASK_COMMIT_TIME))
}
+ def writeCommitMetrics(): Map[String, SQLMetric] = {
+ Map(TASK_COMMIT_TIME -> taskCommitTimeMetric,
+ JOB_COMMIT_TIME -> driverSideMetrics(JOB_COMMIT_TIME))
+ }
+
override def newTaskInstance(): WriteTaskStatsTracker = {
new BasicWriteTaskStatsTracker(serializableHadoopConf.value, Some(taskCommitTimeMetric))
}
@@ -239,12 +244,22 @@ object BasicWriteJobStatsTracker {
val FILE_LENGTH_XATTR = "header.x-hadoop-s3a-magic-data-length"
def metrics: Map[String, SQLMetric] = {
+ writeFilesMetrics ++ writeCommitMetrics
+ }
+
+ def writeFilesMetrics: Map[String, SQLMetric] = {
val sparkContext = SparkContext.getActive.get
Map(
NUM_FILES_KEY -> SQLMetrics.createMetric(sparkContext, "number of written files"),
NUM_OUTPUT_BYTES_KEY -> SQLMetrics.createSizeMetric(sparkContext, "written output"),
NUM_OUTPUT_ROWS_KEY -> SQLMetrics.createMetric(sparkContext, "number of output rows"),
- NUM_PARTS_KEY -> SQLMetrics.createMetric(sparkContext, "number of dynamic part"),
+ NUM_PARTS_KEY -> SQLMetrics.createMetric(sparkContext, "number of dynamic part")
+ )
+ }
+
+ def writeCommitMetrics: Map[String, SQLMetric] = {
+ val sparkContext = SparkContext.getActive.get
+ Map(
TASK_COMMIT_TIME -> SQLMetrics.createTimingMetric(sparkContext, "task commit time"),
JOB_COMMIT_TIME -> SQLMetrics.createTimingMetric(sparkContext, "job commit time")
)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
index 8321b1fac71..8f1eaf4ab77 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
@@ -40,6 +40,7 @@ import org.apache.spark.sql.connector.write.WriterCommitMessage
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.{ProjectExec, SortExec, SparkPlan, SQLExecution, UnsafeExternalRowSorter}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
+import org.apache.spark.sql.execution.command.DataWritingCommand
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.util.{SerializableConfiguration, Utils}
@@ -122,29 +123,6 @@ object FileFormatWriter extends Logging {
val outputWriterFactory =
fileFormat.prepareWrite(sparkSession, job, caseInsensitiveOptions, dataSchema)
- val description = new WriteJobDescription(
- uuid = UUID.randomUUID.toString,
- serializableHadoopConf = new SerializableConfiguration(job.getConfiguration),
- outputWriterFactory = outputWriterFactory,
- allColumns = finalOutputSpec.outputColumns,
- dataColumns = dataColumns,
- partitionColumns = partitionColumns,
- bucketSpec = writerBucketSpec,
- path = finalOutputSpec.outputPath,
- customPartitionLocations = finalOutputSpec.customPartitionLocations,
- maxRecordsPerFile = caseInsensitiveOptions.get("maxRecordsPerFile").map(_.toLong)
- .getOrElse(sparkSession.sessionState.conf.maxRecordsPerFile),
- timeZoneId = caseInsensitiveOptions.get(DateTimeUtils.TIMEZONE_OPTION)
- .getOrElse(sparkSession.sessionState.conf.sessionLocalTimeZone),
- statsTrackers = statsTrackers
- )
-
- // We should first sort by dynamic partition columns, then bucket id, and finally sorting
- // columns.
- val requiredOrdering = partitionColumns.drop(numStaticPartitionCols) ++
- writerBucketSpec.map(_.bucketIdExpression) ++ sortColumns
- val writeFilesOpt = V1WritesUtils.getWriteFilesOpt(plan)
-
// SPARK-40588: when planned writing is disabled and AQE is enabled,
// plan contains an AdaptiveSparkPlanExec, which does not know
// its final plan's ordering, so we have to materialize that plan first
@@ -154,18 +132,17 @@ object FileFormatWriter extends Logging {
case p: SparkPlan => p.withNewChildren(p.children.map(materializeAdaptiveSparkPlan))
}
+ // We should first sort by dynamic partition columns, then bucket id, and finally sorting
+ // columns.
+ val requiredOrdering = partitionColumns.drop(numStaticPartitionCols) ++
+ writerBucketSpec.map(_.bucketIdExpression) ++ sortColumns
+ val writeFilesOpt = V1WritesUtils.getWriteFilesOpt(plan)
// the sort order doesn't matter
val actualOrdering = writeFilesOpt.map(_.child)
.getOrElse(materializeAdaptiveSparkPlan(plan))
.outputOrdering
val orderingMatched = V1WritesUtils.isOrderingMatched(requiredOrdering, actualOrdering)
- SQLExecution.checkSQLExecutionId(sparkSession)
-
- // propagate the description UUID into the jobs, so that committers
- // get an ID guaranteed to be unique.
- job.getConfiguration.set("spark.sql.sources.writeJobUUID", description.uuid)
-
// When `PLANNED_WRITE_ENABLED` is true, the optimizer rule V1Writes will add logical sort
// operator based on the required ordering of the V1 write command. So the output
// ordering of the physical plan should always match the required ordering. Here
@@ -176,6 +153,40 @@ object FileFormatWriter extends Logging {
// V1 write command will be empty).
if (Utils.isTesting) outputOrderingMatched = orderingMatched
+ SQLExecution.checkSQLExecutionId(sparkSession)
+
+ val finalStatsTrackers = if (writeFilesOpt.isDefined) {
+ val writeFilesMetrics = writeFilesOpt.get.metrics
+ statsTrackers.map {
+ case tracker: BasicWriteJobStatsTracker =>
+ val finalMetrics = writeFilesMetrics ++ tracker.writeCommitMetrics()
+ DataWritingCommand.basicWriteJobStatsTracker(finalMetrics, hadoopConf)
+ case other => other
+ }
+ } else {
+ statsTrackers
+ }
+ val description = new WriteJobDescription(
+ uuid = UUID.randomUUID.toString,
+ serializableHadoopConf = new SerializableConfiguration(job.getConfiguration),
+ outputWriterFactory = outputWriterFactory,
+ allColumns = finalOutputSpec.outputColumns,
+ dataColumns = dataColumns,
+ partitionColumns = partitionColumns,
+ bucketSpec = writerBucketSpec,
+ path = finalOutputSpec.outputPath,
+ customPartitionLocations = finalOutputSpec.customPartitionLocations,
+ maxRecordsPerFile = caseInsensitiveOptions.get("maxRecordsPerFile").map(_.toLong)
+ .getOrElse(sparkSession.sessionState.conf.maxRecordsPerFile),
+ timeZoneId = caseInsensitiveOptions.get(DateTimeUtils.TIMEZONE_OPTION)
+ .getOrElse(sparkSession.sessionState.conf.sessionLocalTimeZone),
+ statsTrackers = finalStatsTrackers
+ )
+
+ // propagate the description UUID into the jobs, so that committers
+ // get an ID guaranteed to be unique.
+ job.getConfiguration.set("spark.sql.sources.writeJobUUID", description.uuid)
+
if (writeFilesOpt.isDefined) {
// build `WriteFilesSpec` for `WriteFiles`
val concurrentOutputWriterSpecFunc = (plan: SparkPlan) => {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/WriteFiles.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/WriteFiles.scala
index d0ed6b02fef..2437bf4e580 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/WriteFiles.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/WriteFiles.scala
@@ -30,6 +30,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, UnaryNode}
import org.apache.spark.sql.connector.write.WriterCommitMessage
import org.apache.spark.sql.execution.{SparkPlan, UnaryExecNode}
import org.apache.spark.sql.execution.datasources.FileFormatWriter.ConcurrentOutputWriterSpec
+import org.apache.spark.sql.execution.metric.SQLMetric
/**
* The write files spec holds all information of [[V1WriteCommand]] if its provider is
@@ -69,6 +70,8 @@ case class WriteFilesExec(
staticPartitions: TablePartitionSpec) extends UnaryExecNode {
override def output: Seq[Attribute] = Seq.empty
+ override lazy val metrics: Map[String, SQLMetric] = BasicWriteJobStatsTracker.writeFilesMetrics
+
override protected def doExecuteWrite(
writeFilesSpec: WriteFilesSpec): RDD[WriterCommitMessage] = {
val rdd = child.execute()
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BasicWriteJobStatsTrackerMetricSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BasicWriteJobStatsTrackerMetricSuite.scala
index 3e58c225d8c..c538757b2d1 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BasicWriteJobStatsTrackerMetricSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BasicWriteJobStatsTrackerMetricSuite.scala
@@ -17,6 +17,10 @@
package org.apache.spark.sql.execution.datasources
+import org.scalatest.concurrent.Eventually.eventually
+import org.scalatest.concurrent.Futures.timeout
+import org.scalatest.time.SpanSugar.convertIntToGrainOfTime
+
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.{LocalSparkSession, SparkSession}
@@ -44,13 +48,14 @@ class BasicWriteJobStatsTrackerMetricSuite extends SparkFunSuite with LocalSpark
// but the executionId is indeterminate in maven test,
// so the `statusStore.execution(executionId)` API is not used.
assert(statusStore.executionsCount() == 2)
- val executionData = statusStore.executionsList()(1)
- val accumulatorIdOpt =
- executionData.metrics.find(_.name == "number of dynamic part").map(_.accumulatorId)
- assert(accumulatorIdOpt.isDefined)
- val numPartsOpt = executionData.metricValues.get(accumulatorIdOpt.get)
- assert(numPartsOpt.isDefined && numPartsOpt.get == partitions)
-
+ eventually(timeout(10.seconds)) {
+ val executionData = statusStore.executionsList()(1)
+ val accumulatorIdOpt =
+ executionData.metrics.find(_.name == "number of dynamic part").map(_.accumulatorId)
+ assert(accumulatorIdOpt.isDefined)
+ val numPartsOpt = executionData.metricValues.get(accumulatorIdOpt.get)
+ assert(numPartsOpt.isDefined && numPartsOpt.get == partitions)
+ }
} finally {
spark.sql("drop table if exists dynamic_partition")
spark.stop()
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
index 26e61c6b58d..5b61c046941 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala
@@ -34,7 +34,7 @@ import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.adaptive.DisableAdaptiveExecutionSuite
import org.apache.spark.sql.execution.aggregate.HashAggregateExec
import org.apache.spark.sql.execution.command.DataWritingCommandExec
-import org.apache.spark.sql.execution.datasources.{BasicWriteJobStatsTracker, InsertIntoHadoopFsRelationCommand, SQLHadoopMapReduceCommitProtocol, V1WriteCommand}
+import org.apache.spark.sql.execution.datasources.{BasicWriteJobStatsTracker, SQLHadoopMapReduceCommitProtocol, WriteFilesExec}
import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, ShuffleExchangeExec}
import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, ShuffledHashJoinExec}
import org.apache.spark.sql.expressions.Window
@@ -830,29 +830,33 @@ class SQLMetricsSuite extends SharedSparkSession with SQLMetricsTestUtils
test("SPARK-34567: Add metrics for CTAS operator") {
withTable("t") {
- var v1WriteCommand: V1WriteCommand = null
+ var dataWriting: DataWritingCommandExec = null
val listener = new QueryExecutionListener {
override def onFailure(f: String, qe: QueryExecution, e: Exception): Unit = {}
override def onSuccess(funcName: String, qe: QueryExecution, duration: Long): Unit = {
qe.executedPlan match {
case dataWritingCommandExec: DataWritingCommandExec =>
- val createTableAsSelect = dataWritingCommandExec.cmd
- v1WriteCommand = createTableAsSelect.asInstanceOf[InsertIntoHadoopFsRelationCommand]
+ dataWriting = dataWritingCommandExec
case _ =>
}
}
}
spark.listenerManager.register(listener)
try {
- val df = sql("CREATE TABLE t USING PARQUET AS SELECT 1 as a")
+ sql("CREATE TABLE t USING PARQUET AS SELECT 1 as a")
sparkContext.listenerBus.waitUntilEmpty()
- assert(v1WriteCommand != null)
- assert(v1WriteCommand.metrics.contains("numFiles"))
- assert(v1WriteCommand.metrics("numFiles").value == 1)
- assert(v1WriteCommand.metrics.contains("numOutputBytes"))
- assert(v1WriteCommand.metrics("numOutputBytes").value > 0)
- assert(v1WriteCommand.metrics.contains("numOutputRows"))
- assert(v1WriteCommand.metrics("numOutputRows").value == 1)
+ assert(dataWriting != null)
+ val metrics = if (conf.plannedWriteEnabled) {
+ dataWriting.child.asInstanceOf[WriteFilesExec].metrics
+ } else {
+ dataWriting.cmd.metrics
+ }
+ assert(metrics.contains("numFiles"))
+ assert(metrics("numFiles").value == 1)
+ assert(metrics.contains("numOutputBytes"))
+ assert(metrics("numOutputBytes").value > 0)
+ assert(metrics.contains("numOutputRows"))
+ assert(metrics("numOutputRows").value == 1)
} finally {
spark.listenerManager.unregister(listener)
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsTestUtils.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsTestUtils.scala
index 81667d52e16..0952b1c0c10 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsTestUtils.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsTestUtils.scala
@@ -27,6 +27,7 @@ import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.execution.{SparkPlan, SparkPlanInfo}
import org.apache.spark.sql.execution.ui.{SparkPlanGraph, SQLAppStatusStore}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.WHOLESTAGE_CODEGEN_ENABLED
import org.apache.spark.sql.test.SQLTestUtils
@@ -80,7 +81,13 @@ trait SQLMetricsTestUtils extends SQLTestUtils {
assert(executionIds.size == 1)
val executionId = executionIds.head
- val executedNode = statusStore.planGraph(executionId).nodes.head
+ val executedNode = if (conf.plannedWriteEnabled) {
+ val executedNodeOpt = statusStore.planGraph(executionId).nodes.find(_.name == "WriteFiles")
+ assert(executedNodeOpt.isDefined)
+ executedNodeOpt.get
+ } else {
+ statusStore.planGraph(executionId).nodes.head
+ }
val metricsNames = Seq(
"number of written files",
@@ -104,9 +111,17 @@ trait SQLMetricsTestUtils extends SQLTestUtils {
assert(totalNumBytes > 0)
}
+ protected def withPlannedWrite(f: => Unit): Unit = {
+ Seq(true, false).foreach { plannedWrite =>
+ withSQLConf(SQLConf.PLANNED_WRITE_ENABLED.key -> plannedWrite.toString) {
+ f
+ }
+ }
+ }
+
protected def testMetricsNonDynamicPartition(
dataFormat: String,
- tableName: String): Unit = {
+ tableName: String): Unit = withPlannedWrite {
withTable(tableName) {
Seq((1, 2)).toDF("i", "j")
.write.format(dataFormat).mode("overwrite").saveAsTable(tableName)
@@ -126,7 +141,7 @@ trait SQLMetricsTestUtils extends SQLTestUtils {
protected def testMetricsDynamicPartition(
provider: String,
dataFormat: String,
- tableName: String): Unit = {
+ tableName: String): Unit = withPlannedWrite {
withTable(tableName) {
withTempPath { dir =>
spark.sql(
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveDirCommand.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveDirCommand.scala
index bd6278473a7..fed7f2cc7a3 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveDirCommand.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveDirCommand.scala
@@ -30,6 +30,8 @@ import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.command.DDLUtils
+import org.apache.spark.sql.execution.datasources.BasicWriteJobStatsTracker
+import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.hive.client.HiveClientImpl
import org.apache.spark.sql.util.SchemaUtils
@@ -57,6 +59,11 @@ case class InsertIntoHiveDirCommand(
overwrite: Boolean,
outputColumnNames: Seq[String]) extends SaveAsHiveFile with V1WritesHiveUtils {
+ // We did not pull out `InsertIntoHiveDirCommand` to `V1WriteCommand`,
+ // so there is no `WriteFiles`. It should always hold all metrics by itself.
+ override lazy val metrics: Map[String, SQLMetric] =
+ BasicWriteJobStatsTracker.metrics
+
override def run(sparkSession: SparkSession, child: SparkPlan): Seq[Row] = {
assert(storage.locationUri.nonEmpty)
SchemaUtils.checkColumnNameDuplication(
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLMetricsSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLMetricsSuite.scala
index c5a84b930a9..cb19f01a522 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLMetricsSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLMetricsSuite.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.hive.execution
import org.apache.spark.sql.execution.QueryExecution
import org.apache.spark.sql.execution.adaptive.DisableAdaptiveExecutionSuite
import org.apache.spark.sql.execution.command.DataWritingCommandExec
-import org.apache.spark.sql.execution.datasources.{InsertIntoHadoopFsRelationCommand, V1WriteCommand}
+import org.apache.spark.sql.execution.datasources.{InsertIntoHadoopFsRelationCommand, WriteFilesExec}
import org.apache.spark.sql.execution.metric.SQLMetricsTestUtils
import org.apache.spark.sql.hive.HiveUtils
import org.apache.spark.sql.hive.test.TestHiveSingleton
@@ -43,21 +43,22 @@ class SQLMetricsSuite extends SQLMetricsTestUtils with TestHiveSingleton
}
test("SPARK-34567: Add metrics for CTAS operator") {
+ withPlannedWrite {
+ checkWriteMetrics()
+ }
+ }
+
+ private def checkWriteMetrics(): Unit = {
Seq(false, true).foreach { canOptimized =>
withSQLConf(HiveUtils.CONVERT_METASTORE_CTAS.key -> canOptimized.toString) {
withTable("t") {
- var v1WriteCommand: V1WriteCommand = null
+ var dataWriting: DataWritingCommandExec = null
val listener = new QueryExecutionListener {
override def onFailure(f: String, qe: QueryExecution, e: Exception): Unit = {}
override def onSuccess(funcName: String, qe: QueryExecution, duration: Long): Unit = {
qe.executedPlan match {
case dataWritingCommandExec: DataWritingCommandExec =>
- val createTableAsSelect = dataWritingCommandExec.cmd
- v1WriteCommand = if (canOptimized) {
- createTableAsSelect.asInstanceOf[InsertIntoHadoopFsRelationCommand]
- } else {
- createTableAsSelect.asInstanceOf[InsertIntoHiveTable]
- }
+ dataWriting = dataWritingCommandExec
case _ =>
}
}
@@ -66,13 +67,24 @@ class SQLMetricsSuite extends SQLMetricsTestUtils with TestHiveSingleton
try {
sql(s"CREATE TABLE t STORED AS PARQUET AS SELECT 1 as a")
sparkContext.listenerBus.waitUntilEmpty()
- assert(v1WriteCommand != null)
- assert(v1WriteCommand.metrics.contains("numFiles"))
- assert(v1WriteCommand.metrics("numFiles").value == 1)
- assert(v1WriteCommand.metrics.contains("numOutputBytes"))
- assert(v1WriteCommand.metrics("numOutputBytes").value > 0)
- assert(v1WriteCommand.metrics.contains("numOutputRows"))
- assert(v1WriteCommand.metrics("numOutputRows").value == 1)
+ assert(dataWriting != null)
+ val v1WriteCommand = if (canOptimized) {
+ dataWriting.cmd.asInstanceOf[InsertIntoHadoopFsRelationCommand]
+ } else {
+ dataWriting.cmd.asInstanceOf[InsertIntoHiveTable]
+ }
+ val metrics = if (conf.plannedWriteEnabled) {
+ dataWriting.child.asInstanceOf[WriteFilesExec].metrics
+ } else {
+ v1WriteCommand.metrics
+ }
+
+ assert(metrics.contains("numFiles"))
+ assert(metrics("numFiles").value == 1)
+ assert(metrics.contains("numOutputBytes"))
+ assert(metrics("numOutputBytes").value > 0)
+ assert(metrics.contains("numOutputRows"))
+ assert(metrics("numOutputRows").value == 1)
} finally {
spark.listenerManager.unregister(listener)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org