You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by vi...@apache.org on 2021/12/13 01:50:51 UTC
[spark] branch master updated: [SPARK-37369][SQL] Avoid redundant ColumnarToRow transistion on InMemoryTableScan
This is an automated email from the ASF dual-hosted git repository.
viirya 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 a30bec1 [SPARK-37369][SQL] Avoid redundant ColumnarToRow transistion on InMemoryTableScan
a30bec1 is described below
commit a30bec17d6187ef6dfe276c67dbd4e023062b92b
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Sun Dec 12 17:49:58 2021 -0800
[SPARK-37369][SQL] Avoid redundant ColumnarToRow transistion on InMemoryTableScan
### What changes were proposed in this pull request?
This patch proposes to let `InMemoryTableScanExec` produces row output directly, if its parent query plan only accepts rows instead of columnar output. In particular, this change adds a new method in `SparkPlan` called `supportsRowBased`, alongside with the existing `supportsColumnar`.
### Why are the changes needed?
We currently have `supportsColumnar` indicating if a physical node can produce columnar output. The current columnar transition rule seems taking an assumption that one node can only produce columnar output but not row-based one if `supportsColumnar` returns true. But actually one node can possibly produce both format, i.e. columnar and row-based. For such node, if we require row-based output, the columnar transition rule will add additional columnar-to-row after it due to the wrong a [...]
So this change introduces `supportsRowBased` which is used to indicates if the node can produce row-based output. The rule can check this method when deciding if a columnar-to-row transition is necessary or not.
For example, `InMemoryTableScanExec` can produce columnar output. So if its parent plan isn't columnar, the rule adds a `ColumnarToRow` between them, e.g.,
```
+- Union
:- ColumnarToRow
: +- InMemoryTableScan i#8, j#9
: +- InMemoryRelation i#8, j#9, StorageLevel(disk, memory, deserialized, 1 replicas)
```
But `InMemoryTableScanExec` is capable of row-based output too. After this change, for such case, we can simply ask `InMemoryTableScanExec` to produce row output instead of a redundant conversion.
```
================================================================================================
Int In-memory
================================================================================================
OpenJDK 64-Bit Server VM 1.8.0_265-b01 on Mac OS X 10.16
Intel(R) Core(TM) i7-9750H CPU 2.60GHz
Int In-Memory scan: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
--------------------------------------------------------------------------------------------------------------------------
columnar deserialization + columnar-to-row 228 245 15 4.4 227.7 1.0X
row-based deserialization 179 187 10 5.6 179.4 1.3X
```
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Existing tests.
Closes #34642 from viirya/SPARK-37369.
Authored-by: Liang-Chi Hsieh <vi...@gmail.com>
Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
---
.../InMemoryColumnarBenchmark-results.txt | 12 ++++
.../org/apache/spark/sql/execution/Columnar.scala | 4 +-
.../org/apache/spark/sql/execution/SparkPlan.scala | 11 +++-
.../sql/execution/columnar/InMemoryRelation.scala | 3 +-
.../execution/columnar/InMemoryTableScanExec.scala | 2 +
.../org/apache/spark/sql/CachedTableSuite.scala | 21 ++++++-
.../spark/sql/DataFrameSetOperationsSuite.scala | 7 +--
.../sql/execution/WholeStageCodegenSuite.scala | 2 +-
.../columnar/CachedBatchSerializerSuite.scala | 5 +-
.../columnar/InMemoryColumnarBenchmark.scala | 66 ++++++++++++++++++++++
.../columnar/InMemoryColumnarQuerySuite.scala | 6 +-
.../spark/sql/execution/debug/DebuggingSuite.scala | 5 +-
.../sql/execution/metric/SQLMetricsSuite.scala | 6 +-
13 files changed, 128 insertions(+), 22 deletions(-)
diff --git a/sql/core/benchmarks/InMemoryColumnarBenchmark-results.txt b/sql/core/benchmarks/InMemoryColumnarBenchmark-results.txt
new file mode 100644
index 0000000..2998d8b
--- /dev/null
+++ b/sql/core/benchmarks/InMemoryColumnarBenchmark-results.txt
@@ -0,0 +1,12 @@
+================================================================================================
+Int In-memory
+================================================================================================
+
+OpenJDK 64-Bit Server VM 1.8.0_312-b07 on Linux 5.11.0-1021-azure
+Intel(R) Xeon(R) Platinum 8272CL CPU @ 2.60GHz
+Int In-Memory scan: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+--------------------------------------------------------------------------------------------------------------------------
+columnar deserialization + columnar-to-row 266 276 12 3.8 265.9 1.0X
+row-based deserialization 197 206 15 5.1 196.7 1.4X
+
+
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala
index d1e9168..f57e7e1 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala
@@ -535,8 +535,8 @@ case class ApplyColumnarRulesAndInsertTransitions(
private def insertTransitions(plan: SparkPlan, outputsColumnar: Boolean): SparkPlan = {
if (outputsColumnar) {
insertRowToColumnar(plan)
- } else if (plan.supportsColumnar) {
- // `outputsColumnar` is false but the plan outputs columnar format, so add a
+ } else if (plan.supportsColumnar && !plan.supportsRowBased) {
+ // `outputsColumnar` is false but the plan only outputs columnar format, so add a
// to-row transition here.
ColumnarToRowExec(insertRowToColumnar(plan))
} else if (!plan.isInstanceOf[ColumnarToRowTransition]) {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala
index 5c4266d..f56beeb 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala
@@ -72,7 +72,16 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
val id: Int = SparkPlan.newPlanId()
/**
- * Return true if this stage of the plan supports columnar execution.
+ * Return true if this stage of the plan supports row-based execution. A plan
+ * can also support columnar execution (see `supportsColumnar`). Spark will decide
+ * which execution to be called during query planning.
+ */
+ def supportsRowBased: Boolean = !supportsColumnar
+
+ /**
+ * Return true if this stage of the plan supports columnar execution. A plan
+ * can also support row-based execution (see `supportsRowBased`). Spark will decide
+ * which execution to be called during query planning.
*/
def supportsColumnar: Boolean = false
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala
index 525653c..89323e7 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala
@@ -256,7 +256,8 @@ case class CachedRDDBuilder(
}
private def buildBuffers(): RDD[CachedBatch] = {
- val cb = if (cachedPlan.supportsColumnar) {
+ val cb = if (cachedPlan.supportsColumnar &&
+ serializer.supportsColumnarInput(cachedPlan.output)) {
serializer.convertColumnarBatchToCachedBatch(
cachedPlan.executeColumnar(),
cachedPlan.output,
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala
index 3f97fbc..da9316e 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala
@@ -55,6 +55,8 @@ case class InMemoryTableScanExec(
override def vectorTypes: Option[Seq[String]] =
relation.cacheBuilder.serializer.vectorTypes(attributes, conf)
+ override def supportsRowBased: Boolean = true
+
/**
* If true, get data from ColumnVector in ColumnarBatch, which are generally faster.
* If false, get data from UnsafeRow build from CachedBatch
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
index bad2815..52803e2 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
@@ -35,7 +35,7 @@ import org.apache.spark.sql.catalyst.analysis.TempTableAlreadyExistsException
import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
import org.apache.spark.sql.catalyst.plans.logical.{BROADCAST, Join, JoinStrategyHint, SHUFFLE_HASH}
import org.apache.spark.sql.catalyst.util.DateTimeConstants
-import org.apache.spark.sql.execution.{ExecSubqueryExpression, RDDScanExec, SparkPlan}
+import org.apache.spark.sql.execution.{ColumnarToRowExec, ExecSubqueryExpression, RDDScanExec, SparkPlan}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.columnar._
import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
@@ -881,8 +881,10 @@ class CachedTableSuite extends QueryTest with SQLTestUtils
test("SPARK-23312: vectorized cache reader can be disabled") {
Seq(true, false).foreach { vectorized =>
withSQLConf(SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> vectorized.toString) {
- val df = spark.range(10).cache()
- df.queryExecution.executedPlan.foreach {
+ val df1 = spark.range(10).cache()
+ val df2 = spark.range(10).cache()
+ val union = df1.union(df2)
+ union.queryExecution.executedPlan.foreach {
case i: InMemoryTableScanExec =>
assert(i.supportsColumnar == vectorized)
case _ =>
@@ -891,6 +893,19 @@ class CachedTableSuite extends QueryTest with SQLTestUtils
}
}
+ test("SPARK-37369: Avoid redundant ColumnarToRow transition on InMemoryTableScan") {
+ Seq(true, false).foreach { vectorized =>
+ withSQLConf(SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> vectorized.toString) {
+ val cache = spark.range(10).cache()
+ val df = cache.filter($"id" > 0)
+ val columnarToRow = df.queryExecution.executedPlan.collect {
+ case c: ColumnarToRowExec => c
+ }
+ assert(columnarToRow.isEmpty)
+ }
+ }
+ }
+
private def checkIfNoJobTriggered[T](f: => T): T = {
var numJobTriggered = 0
val jobListener = new SparkListener {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSetOperationsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSetOperationsSuite.scala
index 26df517..b19e430 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSetOperationsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSetOperationsSuite.scala
@@ -1375,12 +1375,11 @@ class DataFrameSetOperationsSuite extends QueryTest with SharedSparkSession {
val df1 = Seq(1, 2, 3).toDF("i").cache()
val df2 = Seq(4, 5, 6).toDF("j").cache()
- checkIfColumnar(df1.queryExecution.executedPlan,
+ val union = df1.union(df2)
+ checkIfColumnar(union.queryExecution.executedPlan,
_.isInstanceOf[InMemoryTableScanExec], supported)
- checkIfColumnar(df2.queryExecution.executedPlan,
+ checkIfColumnar(union.queryExecution.executedPlan,
_.isInstanceOf[InMemoryTableScanExec], supported)
-
- val union = df1.union(df2)
checkIfColumnar(union.queryExecution.executedPlan, _.isInstanceOf[UnionExec], supported)
checkAnswer(union, Row(1) :: Row(2) :: Row(3) :: Row(4) :: Row(5) :: Row(6) :: Nil)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
index 00ea371..55ca1e8 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
@@ -467,7 +467,7 @@ class WholeStageCodegenSuite extends QueryTest with SharedSparkSession
val planInt = dsIntFilter.queryExecution.executedPlan
assert(planInt.collect {
case WholeStageCodegenExec(FilterExec(_,
- ColumnarToRowExec(InputAdapter(_: InMemoryTableScanExec)))) => ()
+ InputAdapter(_: InMemoryTableScanExec))) => ()
}.length == 1)
assert(dsIntFilter.collect() === Array(1, 2))
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/CachedBatchSerializerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/CachedBatchSerializerSuite.scala
index 099a1aa..9f7304c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/CachedBatchSerializerSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/CachedBatchSerializerSuite.scala
@@ -138,8 +138,9 @@ class CachedBatchSerializerSuite extends QueryTest with SharedSparkSession {
input.write.parquet(workDirPath)
val data = spark.read.parquet(workDirPath)
data.cache()
- assert(data.count() == 3)
- checkAnswer(data, Row(100) :: Row(200) :: Row(300) :: Nil)
+ val df = data.union(data)
+ assert(df.count() == 6)
+ checkAnswer(df, Row(100) :: Row(200) :: Row(300) :: Row(100) :: Row(200) :: Row(300) :: Nil)
}
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarBenchmark.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarBenchmark.scala
new file mode 100644
index 0000000..b975451
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarBenchmark.scala
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.spark.sql.execution.columnar
+
+import org.apache.spark.benchmark.Benchmark
+import org.apache.spark.sql.execution.ColumnarToRowExec
+import org.apache.spark.sql.execution.benchmark.SqlBasedBenchmark
+
+/**
+ * Benchmark to low level memory access using different ways to manage buffers.
+ * To run this benchmark:
+ * {{{
+ * 1. without sbt:
+ * bin/spark-submit --class <this class>
+ * --jars <spark core test jar> <spark sql test jar>
+ * 2. build/sbt "sql/test:runMain <this class>"
+ * 3. generate result: SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "sql/test:runMain <this class>"
+ * Results will be written to "benchmarks/InMemoryColumnarBenchmark-results.txt".
+ * }}}
+ */
+object InMemoryColumnarBenchmark extends SqlBasedBenchmark {
+ def intCache(rowsNum: Int, numIters: Int): Unit = {
+ val data = spark.range(0, rowsNum, 1, 1).toDF("i").cache()
+
+ val inMemoryScan = data.queryExecution.executedPlan.collect {
+ case m: InMemoryTableScanExec => m
+ }
+
+ val columnarScan = ColumnarToRowExec(inMemoryScan(0))
+ val rowScan = inMemoryScan(0)
+
+ assert(inMemoryScan.size == 1)
+
+ val benchmark = new Benchmark("Int In-Memory scan", rowsNum, output = output)
+
+ benchmark.addCase("columnar deserialization + columnar-to-row", numIters) { _ =>
+ columnarScan.executeCollect()
+ }
+
+ benchmark.addCase("row-based deserialization", numIters) { _ =>
+ rowScan.executeCollect()
+ }
+
+ benchmark.run()
+ }
+
+ override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
+ runBenchmark("Int In-memory") {
+ intCache(rowsNum = 1000000, numIters = 3)
+ }
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala
index b8f73f4..2cf12dd 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala
@@ -26,7 +26,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference, AttributeSet, In}
import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
import org.apache.spark.sql.columnar.CachedBatch
-import org.apache.spark.sql.execution.{ColumnarToRowExec, FilterExec, InputAdapter, WholeStageCodegenExec}
+import org.apache.spark.sql.execution.{FilterExec, InputAdapter, WholeStageCodegenExec}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
@@ -504,8 +504,8 @@ class InMemoryColumnarQuerySuite extends QueryTest with SharedSparkSession {
val df2 = df1.where("y = 3")
val planBeforeFilter = df2.queryExecution.executedPlan.collect {
- case FilterExec(_, c: ColumnarToRowExec) => c.child
- case WholeStageCodegenExec(FilterExec(_, ColumnarToRowExec(i: InputAdapter))) => i.child
+ case f: FilterExec => f.child
+ case WholeStageCodegenExec(FilterExec(_, i: InputAdapter)) => i.child
}
assert(planBeforeFilter.head.isInstanceOf[InMemoryTableScanExec])
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala
index 5085c74..1b894dd 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala
@@ -125,8 +125,9 @@ class DebuggingSuite extends DebuggingSuiteBase with DisableAdaptiveExecutionSui
}
test("SPARK-28537: DebugExec cannot debug columnar related queries") {
- val df = spark.range(5)
- df.persist()
+ val base = spark.range(5)
+ base.persist()
+ val df = base.union(base)
val captured = new ByteArrayOutputStream()
Console.withOut(captured) {
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 a51003e..162531ba 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
@@ -743,19 +743,19 @@ class SQLMetricsSuite extends SharedSparkSession with SQLMetricsTestUtils
sql("CREATE TEMPORARY VIEW inMemoryTable AS SELECT 1 AS c1")
sql("CACHE TABLE inMemoryTable")
testSparkPlanMetrics(spark.table("inMemoryTable"), 1,
- Map(1L -> (("Scan In-memory table inMemoryTable", Map.empty)))
+ Map(0L -> (("Scan In-memory table inMemoryTable", Map.empty)))
)
sql("CREATE TEMPORARY VIEW ```a``b``` AS SELECT 2 AS c1")
sql("CACHE TABLE ```a``b```")
testSparkPlanMetrics(spark.table("```a``b```"), 1,
- Map(1L -> (("Scan In-memory table ```a``b```", Map.empty)))
+ Map(0L -> (("Scan In-memory table ```a``b```", Map.empty)))
)
}
// Show InMemoryTableScan on UI
testSparkPlanMetrics(spark.range(1).cache().select("id"), 1,
- Map(1L -> (("InMemoryTableScan", Map.empty)))
+ Map(0L -> (("InMemoryTableScan", Map.empty)))
)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org