Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/14607#discussion_r75949718
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala ---
@@ -518,21 +550,87 @@ case class AlterTableRecoverPartitionsCommand(
// TODO: Validate the value
val value = PartitioningUtils.unescapePathName(ps(1))
// comparing with case-insensitive, but preserve the case
- if (columnName == partitionNames(0)) {
- scanPartitions(
- spark, fs, filter, st.getPath, spec ++ Map(columnName ->
value), partitionNames.drop(1))
+ if (columnName == partitionNames.head) {
+ scanPartitions(spark, fs, filter, st.getPath, spec ++
Map(columnName -> value),
+ partitionNames.drop(1), threshold)
} else {
- logWarning(s"expect partition column ${partitionNames(0)}, but
got ${ps(0)}, ignore it")
+ logWarning(s"expect partition column ${partitionNames.head}, but
got ${ps(0)}, ignore it")
Seq()
}
} else {
- if (name != "_SUCCESS" && name != "_temporary" &&
!name.startsWith(".")) {
- logWarning(s"ignore ${new Path(path, name)}")
- }
+ logWarning(s"ignore ${new Path(path, name)}")
Seq()
}
}
}
+
+ private def gatherPartitionStats(
+ spark: SparkSession,
+ partitionSpecsAndLocs: GenSeq[(TablePartitionSpec, Path)],
+ fs: FileSystem,
+ pathFilter: PathFilter,
+ threshold: Int): GenMap[String, (Int, Long)] = {
+ if (partitionSpecsAndLocs.length > threshold) {
+ val hadoopConf = spark.sparkContext.hadoopConfiguration
+ val serializableConfiguration = new
SerializableConfiguration(hadoopConf)
+ val serializedPaths =
partitionSpecsAndLocs.map(_._2.toString).toArray
+
+ // Set the number of parallelism to prevent following file listing
from generating many tasks
+ // in case of large #defaultParallelism.
+ val numParallelism = Math.min(serializedPaths.length,
+ Math.min(spark.sparkContext.defaultParallelism, 10000))
+ // gather the fast stats for all the partitions otherwise Hive
metastore will list all the
+ // files for all the new partitions in sequential way, which is
super slow.
+ logInfo(s"Gather the fast stats in parallel using $numParallelism
tasks.")
+ spark.sparkContext.parallelize(serializedPaths, numParallelism)
+ .mapPartitions { paths =>
+ val pathFilter = getPathFilter(serializableConfiguration.value)
+ paths.map(new Path(_)).map{ path =>
+ val fs = path.getFileSystem(serializableConfiguration.value)
+ val statuses = fs.listStatus(path, pathFilter)
+ (path.toString, (statuses.length, statuses.map(_.getLen).sum))
+ }
+ }.collectAsMap()
+ } else {
+ partitionSpecsAndLocs.map { case (_, location) =>
+ val statuses = fs.listStatus(location, pathFilter)
+ (location.toString, (statuses.length, statuses.map(_.getLen).sum))
+ }.toMap
+ }
+ }
+
+ private def addPartitions(
+ spark: SparkSession,
+ table: CatalogTable,
+ partitionSpecsAndLocs: GenSeq[(TablePartitionSpec, Path)],
+ partitionStats: GenMap[String, (Int, Long)]): Unit = {
+ val total = partitionSpecsAndLocs.length
+ var done = 0L
+ // Hive metastore may not have enough memory to handle millions of
partitions in single RPC,
+ // we should split them into smaller batches. Since Hive client is not
thread safe, we cannot
+ // do this in parallel.
+ partitionSpecsAndLocs.toIterator.grouped(1000).foreach { batch =>
--- End diff --
Changed to 100 to avoid another config, any version of metastore should not
OOM for 100 partitions, right?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
