YannByron commented on a change in pull request #3936:
URL: https://github.com/apache/hudi/pull/3936#discussion_r747940349
##########
File path:
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/CreateHoodieTableCommand.scala
##########
@@ -38,92 +42,125 @@ import org.apache.spark.sql.hudi.HoodieOptionConfig
import org.apache.spark.sql.hudi.HoodieSqlUtils._
import
org.apache.spark.sql.hudi.command.CreateHoodieTableCommand.{initTableIfNeed,
isEmptyPath}
import
org.apache.spark.sql.internal.StaticSQLConf.SCHEMA_STRING_LENGTH_THRESHOLD
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.types.{StructField, StructType}
import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
import org.apache.spark.{SPARK_VERSION, SparkConf}
-import java.util.{Locale, Properties}
-import org.apache.hudi.exception.HoodieException
-import org.apache.hudi.keygen.ComplexKeyGenerator
-import org.apache.hudi.keygen.factory.HoodieSparkKeyGeneratorFactory
+import java.util.{Locale, Properties}
import scala.collection.JavaConverters._
import scala.collection.mutable
+import scala.util.control.NonFatal
/**
* Command for create hoodie table.
*/
case class CreateHoodieTableCommand(table: CatalogTable, ignoreIfExists:
Boolean)
extends RunnableCommand with SparkAdapterSupport {
- override def run(sparkSession: SparkSession): Seq[Row] = {
- val tableName = table.identifier.unquotedString
+ val tableName = formatName(table.identifier.table)
+
+ val tblProperties = table.storage.properties ++ table.properties
+ override def run(sparkSession: SparkSession): Seq[Row] = {
val tableIsExists =
sparkSession.sessionState.catalog.tableExists(table.identifier)
if (tableIsExists) {
if (ignoreIfExists) {
// scalastyle:off
return Seq.empty[Row]
// scalastyle:on
} else {
- throw new IllegalArgumentException(s"Table $tableName already exists.")
+ throw new IllegalArgumentException(s"Table
${table.identifier.unquotedString} already exists.")
}
}
- // Create table in the catalog
- val createTable = createTableInCatalog(sparkSession)
+
+ // get schema with meta fields, table config if hudi table exists, options
including
+ // table configs and properties of the catalog table
+ val path = getTableLocation(table, sparkSession)
+ val (finalSchema, existingTableConfig, tableSqlOptions) =
parseSchemaAndConfigs(sparkSession, path)
+
// Init the hoodie.properties
- initTableIfNeed(sparkSession, createTable)
+ initTableIfNeed(sparkSession, tableName, path, finalSchema,
Review comment:
i'll optimize this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
