Skip to content

[SPARK-19833][SQL]remove SQLConf.HIVE_VERIFY_PARTITION_PATH, always return empty when the location does not exists #17176

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

Closed
windpiger wants to merge 6 commits into from
Closed

[SPARK-19833][SQL]remove SQLConf.HIVE_VERIFY_PARTITION_PATH, always return empty when the location does not exists #17176

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
95aa931
[SPARK-19833][SQL]remove SQLConf.HIVE_VERIFY_PARTITION_PATH, always r…
windpiger Mar 6, 2017
8128567
fix a bug
windpiger Mar 6, 2017
22b1f53
add log to find why jenkins failed
windpiger Mar 7, 2017
5fd0e20
Merge branch 'master' of github.com:apache/spark into removeHiveVerfi…
windpiger Mar 7, 2017
262e2f2
fix test failed
windpiger Mar 7, 2017
3a15e5d
remove log
windpiger Mar 7, 2017
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
8 changes: 0 additions & 8 deletions sql/core/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -264,12 +264,6 @@ object SQLConf {
.booleanConf
.createWithDefault(false)

val HIVE_VERIFY_PARTITION_PATH = buildConf("spark.sql.hive.verifyPartitionPath")
.doc("When true, check all the partition paths under the table\'s root directory " +
"when reading data stored in HDFS.")
.booleanConf
.createWithDefault(false)

val HIVE_METASTORE_PARTITION_PRUNING =
buildConf("spark.sql.hive.metastorePartitionPruning")
.doc("When true, some predicates will be pushed down into the Hive metastore so that " +
Expand Down Expand Up @@ -768,8 +762,6 @@ private[sql] class SQLConf extends Serializable with CatalystConf with Logging {

def orcFilterPushDown: Boolean = getConf(ORC_FILTER_PUSHDOWN_ENABLED)

def verifyPartitionPath: Boolean = getConf(HIVE_VERIFY_PARTITION_PATH)

def metastorePartitionPruning: Boolean = getConf(HIVE_METASTORE_PARTITION_PRUNING)

def manageFilesourcePartitions: Boolean = getConf(HIVE_MANAGE_FILESOURCE_PARTITIONS)
Expand Down
53 changes: 27 additions & 26 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,36 +159,37 @@ class HadoopTableReader(
def verifyPartitionPath(
partitionToDeserializer: Map[HivePartition, Class[_ <: Deserializer]]):
Map[HivePartition, Class[_ <: Deserializer]] = {
if (!sparkSession.sessionState.conf.verifyPartitionPath) {
partitionToDeserializer
} else {
var existPathSet = collection.mutable.Set[String]()
var pathPatternSet = collection.mutable.Set[String]()
partitionToDeserializer.filter {
case (partition, partDeserializer) =>
def updateExistPathSetByPathPattern(pathPatternStr: String) {
val pathPattern = new Path(pathPatternStr)
val fs = pathPattern.getFileSystem(hadoopConf)
val matches = fs.globStatus(pathPattern)
matches.foreach(fileStatus => existPathSet += fileStatus.getPath.toString)
}
// convert /demo/data/year/month/day to /demo/data/*/*/*/
def getPathPatternByPath(parNum: Int, tempPath: Path): String = {
var existPathSet = collection.mutable.Set[String]()
Copy link
Contributor Author

@windpiger windpiger Mar 7, 2017

Choose a reason for hiding this comment

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

after this pr https://github.com/apache/spark/pull/17187， read hive table which does not use stored by will not use HiveTableScanExec.

this function has a bug ,that if the partition path is custom path

  1. it will still do filter for all partition path in the parameter partitionToDeserializer,
  2. it will scan the path which does not belong to the table ,e.g. custom path is /root/a
    and the partitionSpec is b=1/c=2, this will lead to scan / because of the getPathPatternByPath

var pathPatternSet = collection.mutable.Set[String]()
partitionToDeserializer.filter {
case (partition, partDeserializer) =>
def updateExistPathSetByPathPattern(pathPatternStr: String) {
val pathPattern = new Path(pathPatternStr)
val fs = pathPattern.getFileSystem(hadoopConf)
val matches = fs.globStatus(pathPattern)
matches.foreach(fileStatus => existPathSet += fileStatus.getPath.toString)
}
// convert /demo/data/year/month/day to /demo/data/*/*/*/
def getPathPatternByPath(parNum: Int, tempPath: Path, partitionName: String): String = {
// if the partition path does not end with partition name, we should not
Copy link
Contributor Author

@windpiger windpiger Mar 6, 2017

Choose a reason for hiding this comment

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

if the partition location has been altered to another location, we should not do this pattern, or we will list pattern files which does not belong to the partition

// generate the pattern, return the partition path directly
if (tempPath.toString.endsWith(partitionName)) {
var path = tempPath
for (i <- (1 to parNum)) path = path.getParent
val tails = (1 to parNum).map(_ => "*").mkString("/", "/", "/")
path.toString + tails
}

val partPath = partition.getDataLocation
val partNum = Utilities.getPartitionDesc(partition).getPartSpec.size();
var pathPatternStr = getPathPatternByPath(partNum, partPath)
if (!pathPatternSet.contains(pathPatternStr)) {
pathPatternSet += pathPatternStr
updateExistPathSetByPathPattern(pathPatternStr)
}
existPathSet.contains(partPath.toString)
}
} else tempPath.toString
}

val partPath = partition.getDataLocation
val partitionName = partition.getName
val partNum = Utilities.getPartitionDesc(partition).getPartSpec.size();
var pathPatternStr = getPathPatternByPath(partNum, partPath, partitionName)
if (!pathPatternSet.contains(pathPatternStr)) {
pathPatternSet += pathPatternStr
updateExistPathSetByPathPattern(pathPatternStr)
}
existPathSet.contains(partPath.toString)
}
}

Expand Down
58 changes: 28 additions & 30 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/QueryPartitionSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,40 +32,38 @@ class QueryPartitionSuite extends QueryTest with SQLTestUtils with TestHiveSingl
import spark.implicits._

test("SPARK-5068: query data when path doesn't exist") {
withSQLConf((SQLConf.HIVE_VERIFY_PARTITION_PATH.key, "true")) {
val testData = sparkContext.parallelize(
(1 to 10).map(i => TestData(i, i.toString))).toDF()
testData.createOrReplaceTempView("testData")
val testData = sparkContext.parallelize(
(1 to 10).map(i => TestData(i, i.toString))).toDF()
testData.createOrReplaceTempView("testData")

val tmpDir = Files.createTempDir()
// create the table for test
sql(s"CREATE TABLE table_with_partition(key int,value string) " +
s"PARTITIONED by (ds string) location '${tmpDir.toURI}' ")
sql("INSERT OVERWRITE TABLE table_with_partition partition (ds='1') " +
"SELECT key,value FROM testData")
sql("INSERT OVERWRITE TABLE table_with_partition partition (ds='2') " +
"SELECT key,value FROM testData")
sql("INSERT OVERWRITE TABLE table_with_partition partition (ds='3') " +
"SELECT key,value FROM testData")
sql("INSERT OVERWRITE TABLE table_with_partition partition (ds='4') " +
"SELECT key,value FROM testData")
val tmpDir = Files.createTempDir()
// create the table for test
sql(s"CREATE TABLE table_with_partition(key int,value string) " +
s"PARTITIONED by (ds string) location '${tmpDir.toURI}' ")
sql("INSERT OVERWRITE TABLE table_with_partition partition (ds='1') " +
"SELECT key,value FROM testData")
sql("INSERT OVERWRITE TABLE table_with_partition partition (ds='2') " +
"SELECT key,value FROM testData")
sql("INSERT OVERWRITE TABLE table_with_partition partition (ds='3') " +
"SELECT key,value FROM testData")
sql("INSERT OVERWRITE TABLE table_with_partition partition (ds='4') " +
"SELECT key,value FROM testData")

// test for the exist path
checkAnswer(sql("select key,value from table_with_partition"),
testData.toDF.collect ++ testData.toDF.collect
++ testData.toDF.collect ++ testData.toDF.collect)
// test for the exist path
checkAnswer(sql("select key,value from table_with_partition"),
testData.toDF.collect ++ testData.toDF.collect
++ testData.toDF.collect ++ testData.toDF.collect)

// delete the path of one partition
tmpDir.listFiles
.find { f => f.isDirectory && f.getName().startsWith("ds=") }
.foreach { f => Utils.deleteRecursively(f) }
// delete the path of one partition
tmpDir.listFiles
.find { f => f.isDirectory && f.getName().startsWith("ds=") }
.foreach { f => Utils.deleteRecursively(f) }

// test for after delete the path
checkAnswer(sql("select key,value from table_with_partition"),
testData.toDF.collect ++ testData.toDF.collect ++ testData.toDF.collect)
// test for after delete the path
checkAnswer(sql("select key,value from table_with_partition"),
testData.toDF.collect ++ testData.toDF.collect ++ testData.toDF.collect)

sql("DROP TABLE IF EXISTS table_with_partition")
sql("DROP TABLE IF EXISTS createAndInsertTest")
}
sql("DROP TABLE IF EXISTS table_with_partition")
sql("DROP TABLE IF EXISTS createAndInsertTest")
}
}