Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-32813][SQL] Get default config of ParquetSource vectorized reader if no active SparkSession #29667

Closed
wants to merge 7 commits into from
Closed

[SPARK-32813][SQL] Get default config of ParquetSource vectorized reader if no active SparkSession #29667

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
6170596
Fix needsUnsafeRowConversion.
viirya Sep 8, 2020
3933bb0
Add test.
viirya Sep 8, 2020
070327c
For comment.
viirya Sep 8, 2020
b536303
Use SQLConf.get directly.
viirya Sep 8, 2020
97b094f
Use SparkPlan.sqlContext.
viirya Sep 8, 2020
b89eccc
Stop spark session.
viirya Sep 8, 2020
568176d
Remove redundant empty line.
viirya Sep 8, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -177,7 +177,7 @@ case class FileSourceScanExec(

private lazy val needsUnsafeRowConversion: Boolean = {
if (relation.fileFormat.isInstanceOf[ParquetSource]) {
SparkSession.getActiveSession.get.sessionState.conf.parquetVectorizedReaderEnabled
sqlContext.conf.parquetVectorizedReaderEnabled
} else {
false
}
Expand Down
41 changes: 40 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/execution/SQLExecutionSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,11 +17,17 @@

package org.apache.spark.sql.execution

import java.util.concurrent.Executors

import scala.collection.parallel.immutable.ParRange
import scala.concurrent.{ExecutionContext, Future}
import scala.concurrent.duration._

import org.apache.spark.{SparkConf, SparkContext, SparkFunSuite}
import org.apache.spark.scheduler.{SparkListener, SparkListenerJobStart}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types._
import org.apache.spark.util.ThreadUtils

class SQLExecutionSuite extends SparkFunSuite {

Expand Down Expand Up @@ -119,6 +125,39 @@ class SQLExecutionSuite extends SparkFunSuite {

spark.stop()
}

test("SPARK-32813: Table scan should work in different thread") {
val executor1 = Executors.newSingleThreadExecutor()
val executor2 = Executors.newSingleThreadExecutor()
var session: SparkSession = null
SparkSession.cleanupAnyExistingSession()

withTempDir { tempDir =>
HyukjinKwon marked this conversation as resolved.
Show resolved Hide resolved
try {
val tablePath = tempDir.toString + "/table"
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: why don't we use tempDir directly?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

will complain ... dir was already existing.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

oh, can we try withTempPath?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

#29667 (comment) :-)

val df = ThreadUtils.awaitResult(Future {
session = SparkSession.builder().appName("test").master("local[*]").getOrCreate()

session.createDataFrame(
session.sparkContext.parallelize(Row(Array(1, 2, 3)) :: Nil),
StructType(Seq(
StructField("a", ArrayType(IntegerType, containsNull = false), nullable = false))))
.write.parquet(tablePath)

session.read.parquet(tablePath)
HyukjinKwon marked this conversation as resolved.
Show resolved Hide resolved
}(ExecutionContext.fromExecutorService(executor1)), 1.minute)

ThreadUtils.awaitResult(Future {
assert(df.rdd.collect()(0) === Row(Seq(1, 2, 3)))
}(ExecutionContext.fromExecutorService(executor2)), 1.minute)

viirya marked this conversation as resolved.
Show resolved Hide resolved
} finally {
executor1.shutdown()
HyukjinKwon marked this conversation as resolved.
Show resolved Hide resolved
executor2.shutdown()
session.stop()
}
}
}
}

object SQLExecutionSuite {
Expand Down