Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/14618#discussion_r74549761
--- Diff:
sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveStrategies.scala ---
@@ -74,3 +86,321 @@ private[hive] trait HiveStrategies {
}
}
}
+
+/**
+ * Creates any tables required for query execution.
+ * For example, because of a CREATE TABLE X AS statement.
+ */
+class CreateTables(sparkSession: SparkSession) extends Rule[LogicalPlan] {
+ def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+ // Wait until children are resolved.
+ case p: LogicalPlan if !p.childrenResolved => p
+
+ case CreateTable(tableDesc, mode, Some(query)) if
tableDesc.provider.get == "hive" =>
+ val catalog = sparkSession.sessionState.catalog
+ val dbName =
tableDesc.identifier.database.getOrElse(catalog.getCurrentDatabase).toLowerCase
+
+ val newTableDesc = if (tableDesc.storage.serde.isEmpty) {
+ // add default serde
+ tableDesc.withNewStorage(
+ serde =
Some("org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe"))
+ } else {
+ tableDesc
+ }
+
+ // Currently we will never hit this branch, as SQL string API can
only use `Ignore` or
+ // `ErrorIfExists` mode, and `DataFrameWriter.saveAsTable` doesn't
support hive serde
+ // tables yet.
+ if (mode == SaveMode.Append || mode == SaveMode.Overwrite) {
+ throw new AnalysisException("" +
+ "CTAS for hive serde tables does not support append or overwrite
semantics.")
+ }
+
+ execution.CreateHiveTableAsSelectCommand(
+ newTableDesc.copy(identifier =
TableIdentifier(tableDesc.identifier.table, Some(dbName))),
+ query,
+ mode == SaveMode.Ignore)
+ }
+}
+
+
+/**
+ * When scanning Metastore ORC tables, convert them to ORC data source
relations
+ * for better performance.
+ */
+class OrcConversions(sparkSession: SparkSession) extends Rule[LogicalPlan]
{
+ private def shouldConvertMetastoreOrc(relation: MetastoreRelation):
Boolean = {
+ relation.tableDesc.getSerdeClassName.toLowerCase.contains("orc") &&
+ sparkSession.sessionState.conf.getConf(HiveUtils.CONVERT_METASTORE_ORC)
+ }
+
+ private def convertToOrcRelation(relation: MetastoreRelation):
LogicalRelation = {
+ val defaultSource = new OrcFileFormat()
+ val fileFormatClass = classOf[OrcFileFormat]
+ val options = Map[String, String]()
+
+ ConvertMetastoreTablesUtils.convertToLogicalRelation(
+ sparkSession, relation, options, defaultSource, fileFormatClass,
"orc")
+ }
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ if (!plan.resolved || plan.analyzed) {
+ return plan
+ }
+
+ plan transformUp {
+ // Write path
+ case InsertIntoTable(r: MetastoreRelation, partition, child,
overwrite, ifNotExists)
+ // Inserting into partitioned table is not supported in Orc data
source (yet).
+ if !r.hiveQlTable.isPartitioned && shouldConvertMetastoreOrc(r) =>
+ InsertIntoTable(convertToOrcRelation(r), partition, child,
overwrite, ifNotExists)
+
+ // Read path
+ case relation: MetastoreRelation if
shouldConvertMetastoreOrc(relation) =>
+ val orcRelation = convertToOrcRelation(relation)
+ SubqueryAlias(relation.tableName, orcRelation)
+ }
+ }
+}
+
+/**
+ * When scanning or writing to non-partitioned Metastore Parquet tables,
convert them to Parquet
+ * data source relations for better performance.
+ */
+class ParquetConversions(sparkSession: SparkSession) extends
Rule[LogicalPlan] {
+ private def shouldConvertMetastoreParquet(relation: MetastoreRelation):
Boolean = {
+ relation.tableDesc.getSerdeClassName.toLowerCase.contains("parquet") &&
+
sparkSession.sessionState.conf.getConf(HiveUtils.CONVERT_METASTORE_PARQUET)
+ }
+
+ private def convertToParquetRelation(relation: MetastoreRelation):
LogicalRelation = {
+ val defaultSource = new ParquetFileFormat()
+ val fileFormatClass = classOf[ParquetFileFormat]
+
+ val mergeSchema = sparkSession.sessionState.conf.getConf(
+ HiveUtils.CONVERT_METASTORE_PARQUET_WITH_SCHEMA_MERGING)
+ val options = Map(ParquetOptions.MERGE_SCHEMA -> mergeSchema.toString)
+
+ ConvertMetastoreTablesUtils.convertToLogicalRelation(
+ sparkSession, relation, options, defaultSource, fileFormatClass,
"parquet")
+ }
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ if (!plan.resolved || plan.analyzed) {
+ return plan
+ }
+
+ plan transformUp {
+ // Write path
+ case InsertIntoTable(r: MetastoreRelation, partition, child,
overwrite, ifNotExists)
+ // Inserting into partitioned table is not supported in Parquet
data source (yet).
+ if !r.hiveQlTable.isPartitioned &&
shouldConvertMetastoreParquet(r) =>
+ InsertIntoTable(convertToParquetRelation(r), partition, child,
overwrite, ifNotExists)
+
+ // Read path
+ case relation: MetastoreRelation if
shouldConvertMetastoreParquet(relation) =>
+ val parquetRelation = convertToParquetRelation(relation)
+ SubqueryAlias(relation.tableName, parquetRelation)
+ }
+ }
+}
+
+object ConvertMetastoreTablesUtils extends Logging {
--- End diff --
If we can combine the rules `OrcConversions` and `ParquetConversions` into
the same rule, this object can be removed.
---
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]
