Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/22514#discussion_r238909363
--- Diff:
sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveStrategies.scala ---
@@ -181,62 +180,39 @@ case class RelationConversions(
conf: SQLConf,
sessionCatalog: HiveSessionCatalog) extends Rule[LogicalPlan] {
private def isConvertible(relation: HiveTableRelation): Boolean = {
- val serde =
relation.tableMeta.storage.serde.getOrElse("").toLowerCase(Locale.ROOT)
- serde.contains("parquet") &&
conf.getConf(HiveUtils.CONVERT_METASTORE_PARQUET) ||
- serde.contains("orc") &&
conf.getConf(HiveUtils.CONVERT_METASTORE_ORC)
+ isConvertible(relation.tableMeta)
}
- // Return true for Apache ORC and Hive ORC-related configuration names.
- // Note that Spark doesn't support configurations like
`hive.merge.orcfile.stripe.level`.
- private def isOrcProperty(key: String) =
- key.startsWith("orc.") || key.contains(".orc.")
-
- private def isParquetProperty(key: String) =
- key.startsWith("parquet.") || key.contains(".parquet.")
-
- private def convert(relation: HiveTableRelation): LogicalRelation = {
- val serde =
relation.tableMeta.storage.serde.getOrElse("").toLowerCase(Locale.ROOT)
-
- // Consider table and storage properties. For properties existing in
both sides, storage
- // properties will supersede table properties.
- if (serde.contains("parquet")) {
- val options =
relation.tableMeta.properties.filterKeys(isParquetProperty) ++
- relation.tableMeta.storage.properties +
(ParquetOptions.MERGE_SCHEMA ->
-
conf.getConf(HiveUtils.CONVERT_METASTORE_PARQUET_WITH_SCHEMA_MERGING).toString)
- sessionCatalog.metastoreCatalog
- .convertToLogicalRelation(relation, options,
classOf[ParquetFileFormat], "parquet")
- } else {
- val options =
relation.tableMeta.properties.filterKeys(isOrcProperty) ++
- relation.tableMeta.storage.properties
- if (conf.getConf(SQLConf.ORC_IMPLEMENTATION) == "native") {
- sessionCatalog.metastoreCatalog.convertToLogicalRelation(
- relation,
- options,
-
classOf[org.apache.spark.sql.execution.datasources.orc.OrcFileFormat],
- "orc")
- } else {
- sessionCatalog.metastoreCatalog.convertToLogicalRelation(
- relation,
- options,
- classOf[org.apache.spark.sql.hive.orc.OrcFileFormat],
- "orc")
- }
- }
+ private def isConvertible(tableMeta: CatalogTable): Boolean = {
+ val serde =
tableMeta.storage.serde.getOrElse("").toLowerCase(Locale.ROOT)
+ serde.contains("parquet") &&
SQLConf.get.getConf(HiveUtils.CONVERT_METASTORE_PARQUET) ||
+ serde.contains("orc") &&
SQLConf.get.getConf(HiveUtils.CONVERT_METASTORE_ORC)
}
+ private val metastoreCatalog = sessionCatalog.metastoreCatalog
+
override def apply(plan: LogicalPlan): LogicalPlan = {
plan resolveOperators {
// Write path
case InsertIntoTable(r: HiveTableRelation, partition, query,
overwrite, ifPartitionNotExists)
// Inserting into partitioned table is not supported in
Parquet/Orc data source (yet).
if query.resolved && DDLUtils.isHiveTable(r.tableMeta) &&
!r.isPartitioned && isConvertible(r) =>
- InsertIntoTable(convert(r), partition, query, overwrite,
ifPartitionNotExists)
+ InsertIntoTable(metastoreCatalog.convert(r), partition,
+ query, overwrite, ifPartitionNotExists)
// Read path
case relation: HiveTableRelation
if DDLUtils.isHiveTable(relation.tableMeta) &&
isConvertible(relation) =>
- convert(relation)
+ metastoreCatalog.convert(relation)
+
+ // CTAS
+ case CreateTable(tableDesc, mode, Some(query))
+ if DDLUtils.isHiveTable(tableDesc) &&
tableDesc.partitionColumnNames.isEmpty &&
+ isConvertible(tableDesc) =>
--- End diff --
ok.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
