Skip to content

[SPARK-28098][SQL]Support read hive table while LeafDir had multi-level paths #32679

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
wants to merge 1 commit into from
Closed

[SPARK-28098][SQL]Support read hive table while LeafDir had multi-level paths #32679

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
10 changes: 10 additions & 0 deletions sql/catalyst/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 @@ -3210,6 +3210,13 @@ object SQLConf {
.intConf
.createWithDefault(0)

val READ_PARTITION_WITH_SUBDIRECTORY_ENABLED =
buildConf("spark.sql.sources.readPartitionWithSubdirectory.enabled")
.doc("When set to true, Spark SQL could read the files of " +
" partitioned hive table from subdirectories under root path of table")
.booleanConf
.createWithDefault(false)

/**
* Holds information about keys that have been deprecated.
*
Expand Down Expand Up @@ -3908,6 +3915,9 @@ class SQLConf extends Serializable with Logging {

def maxConcurrentOutputFileWriters: Int = getConf(SQLConf.MAX_CONCURRENT_OUTPUT_FILE_WRITERS)

def readPartitionWithSubdirectoryEnabled: Boolean =
getConf(READ_PARTITION_WITH_SUBDIRECTORY_ENABLED)

/** ********************** SQLConf functionality methods ************ */

/** Set Spark SQL configuration properties. */
Expand Down
18 changes: 17 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InMemoryFileIndex.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,9 @@ class InMemoryFileIndex(
override val rootPaths =
rootPathsSpecified.filterNot(FileStreamSink.ancestorIsMetadataDirectory(_, hadoopConf))

val readPartitionWithSubdirectoryEnabled =
sparkSession.sessionState.conf.readPartitionWithSubdirectoryEnabled

@volatile private var cachedLeafFiles: mutable.LinkedHashMap[Path, FileStatus] = _
@volatile private var cachedLeafDirToChildrenFiles: Map[Path, Array[FileStatus]] = _
@volatile private var cachedPartitionSpec: PartitionSpec = _
Expand Down Expand Up @@ -94,10 +97,23 @@ class InMemoryFileIndex(
val files = listLeafFiles(rootPaths)
cachedLeafFiles =
new mutable.LinkedHashMap[Path, FileStatus]() ++= files.map(f => f.getPath -> f)
cachedLeafDirToChildrenFiles = files.toArray.groupBy(_.getPath.getParent)
cachedLeafDirToChildrenFiles =
if (readPartitionWithSubdirectoryEnabled) {
files.toArray.groupBy(file => getRootPathsLeafDir(file.getPath.getParent))
Copy link
Contributor

@zhengchenyu zhengchenyu Oct 22, 2025

Choose a reason for hiding this comment

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

I found that here can not infer partition for non-catalog table. For table location /dir/table/pt=1/file, here the key is /dir/table, the value is /dir/table/pt=1/file. So we can not infer partition from the key /dir/table.

} else {
files.toArray.groupBy(_.getPath.getParent)
}
cachedPartitionSpec = null
}

private def getRootPathsLeafDir(path: Path): Path = {
if (rootPaths.contains(path)) {
path
} else {
getRootPathsLeafDir(path.getParent)
Copy link

@catalinii catalinii Aug 20, 2021

Choose a reason for hiding this comment

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

This fails with NPE if a parquet file is provided directly (instead of a directory).
The fix: lyft@4375b8a

}
}

override def equals(other: Any): Boolean = other match {
case hdfs: InMemoryFileIndex => rootPaths.toSet == hdfs.rootPaths.toSet
case _ => false
Expand Down
16 changes: 16 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileIndexSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -520,6 +520,22 @@ class FileIndexSuite extends SharedSparkSession {
SQLConf.get.setConf(StaticSQLConf.METADATA_CACHE_TTL_SECONDS, previousValue)
}
}

test("SPARK-28098 - supporting read partitioned Hive tables with subdirectories") {
Copy link

@zinking zinking Sep 6, 2021

Choose a reason for hiding this comment

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

not sure what this is testing for

this case also passes on 2.4 without any patched code

withTempPath { dir =>
spark
.range(2)
.select(col("id").as("p"), col("id"))
.write
.partitionBy("p")
.orc(s"${dir.getAbsolutePath}/sub1/sub2")
val path = new Path(dir.getAbsolutePath)
val fileIndex = new InMemoryFileIndex(spark, Seq(path), Map.empty, None)
val partitionValues = fileIndex.partitionSpec().partitions.map(_.values)
assert(partitionValues.length == 2 && partitionValues(0).numFields == 1 &&
partitionValues(1).numFields == 1)
}
}
}

object DeletionRaceFileSystem {
Expand Down