From da04030deb85fcd36fd463793be08270e0216aad Mon Sep 17 00:00:00 2001
From: leesf <leesf@apache.org>
Date: Fri, 3 Jun 2022 08:35:06 +0800
Subject: [PATCH] [HUDI-4183] Fix using HoodieCatalog to create non-hudi tables

---
 .../spark/sql/hudi/TestCreateTable.scala      | 31 +++++++++++++++++++
 .../sql/hudi/catalog/HoodieCatalog.scala      | 10 ++++--
 2 files changed, 38 insertions(+), 3 deletions(-)

diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestCreateTable.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestCreateTable.scala
index cad30eca2446..7091de4a8ec0 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestCreateTable.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestCreateTable.scala
@@ -781,4 +781,35 @@ class TestCreateTable extends HoodieSparkSqlTestBase {
     val tablePath = s"${dbPath}/${tableName}"
     assertResult(false)(existsPath(tablePath))
   }
+
+  test("Test Create Non-Hudi Table(Parquet Table)") {
+    val databaseName = "test_database"
+    spark.sql(s"create database if not exists $databaseName")
+    spark.sql(s"use $databaseName")
+
+    val tableName = generateTableName
+    // Create a managed table
+    spark.sql(
+      s"""
+         | create table $tableName (
+         |  id int,
+         |  name string,
+         |  price double,
+         |  ts long
+         | ) using parquet
+       """.stripMargin)
+    val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier(tableName))
+    assertResult(tableName)(table.identifier.table)
+    assertResult("parquet")(table.provider.get)
+    assertResult(CatalogTableType.MANAGED)(table.tableType)
+    assertResult(
+      Seq(
+        StructField("id", IntegerType),
+        StructField("name", StringType),
+        StructField("price", DoubleType),
+        StructField("ts", LongType))
+    )(table.schema.fields)
+
+    spark.sql("use default")
+  }
 }
diff --git a/hudi-spark-datasource/hudi-spark3/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala b/hudi-spark-datasource/hudi-spark3/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
index 67012c7723db..e1c2f228fa31 100644
--- a/hudi-spark-datasource/hudi-spark3/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
+++ b/hudi-spark-datasource/hudi-spark3/src/main/scala/org/apache/spark/sql/hudi/catalog/HoodieCatalog.scala
@@ -118,9 +118,13 @@ class HoodieCatalog extends DelegatingCatalogExtension
                            schema: StructType,
                            partitions: Array[Transform],
                            properties: util.Map[String, String]): Table = {
-    val locUriAndTableType = deduceTableLocationURIAndTableType(ident, properties)
-    createHoodieTable(ident, schema, locUriAndTableType, partitions, properties,
-      Map.empty, Option.empty, TableCreationMode.CREATE)
+    if (sparkAdapter.isHoodieTable(properties)) {
+      val locUriAndTableType = deduceTableLocationURIAndTableType(ident, properties)
+      createHoodieTable(ident, schema, locUriAndTableType, partitions, properties,
+        Map.empty, Option.empty, TableCreationMode.CREATE)
+    } else {
+      super.createTable(ident, schema, partitions, properties)
+    }
   }
 
   override def tableExists(ident: Identifier): Boolean = super.tableExists(ident)