[CARMEL-3030] Convert a temporary table to delta is not allowed (delt…

…a-io#77)
LantaoJin · Jun 10, 2020 · 9a6a0ae · 9a6a0ae
1 parent 05db466
commit 9a6a0ae
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 1 deletion.
diff --git a/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala b/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala
@@ -668,6 +668,10 @@ object DeltaErrors
       s"convert a $sourceName source: $ident")
   }
 
+  def convertTemporaryTablesException(ident: TableIdentifier): Throwable = {
+    new AnalysisException(s"CONVERT TO DELTA doesn't support TEMPORARY/VOLATILE table $ident")
+  }
+
   def unexpectedPartitionColumnFromFileNameException(
       path: String, parsedCol: String, expectedCol: String): Throwable = {
     new AnalysisException(s"Expecting partition column ${formatColumn(expectedCol)}, but" +

diff --git a/src/main/scala/org/apache/spark/sql/delta/commands/ConvertToDeltaCommand.scala b/src/main/scala/org/apache/spark/sql/delta/commands/ConvertToDeltaCommand.scala
@@ -38,7 +38,7 @@ import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogUtils}
 import org.apache.spark.sql.catalyst.analysis.Analyzer
 import org.apache.spark.sql.catalyst.expressions.Cast
 import org.apache.spark.sql.delta.services.{ConvertToDeltaEvent, DeltaTableMetadata}
-import org.apache.spark.sql.execution.command.RunnableCommand
+import org.apache.spark.sql.execution.command.{DDLUtils, RunnableCommand}
 import org.apache.spark.sql.execution.datasources.parquet.{ParquetFileFormat, ParquetToSparkSchemaConverter}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{StringType, StructField, StructType}
@@ -93,6 +93,12 @@ abstract class ConvertToDeltaCommandBase(
         throw DeltaErrors.missingProviderForConvertException(convertProperties.targetDir)
     }
 
+    convertProperties.catalogTable match {
+      case Some(table) if DDLUtils.isTemporaryTable(table) =>
+        throw DeltaErrors.convertTemporaryTablesException(tableIdentifier)
+      case None =>
+    }
+
     val deltaLog = DeltaLog.forTable(spark, deltaPath.getOrElse(convertProperties.targetDir))
     deltaLog.update()
     val txn = deltaLog.startTransaction()

diff --git a/src/test/scala/org/apache/spark/sql/delta/services/SQLQueryTest.scala b/src/test/scala/org/apache/spark/sql/delta/services/SQLQueryTest.scala
@@ -1389,4 +1389,20 @@ class SQLQuerySuite extends QueryTest
       }
     }
   }
+
+  test("convert temporary table is not allowed") {
+    withTempTable("tt1", "tt2") {
+      sql("CREATE TEMP TABLE tt1 (id INT) USING parquet")
+      val e1 = intercept[AnalysisException] {
+        sql("CONVERT TO DELTA tt1")
+      }.getMessage()
+      assert(e1.contains("CONVERT TO DELTA doesn't support TEMPORARY/VOLATILE table"))
+
+      sql("CREATE VOLATILE TABLE tt2 USING parquet AS SELECT 1 AS id")
+      val e2 = intercept[AnalysisException] {
+        sql("CONVERT TO DELTA tt2")
+      }.getMessage()
+      assert(e2.contains("CONVERT TO DELTA doesn't support TEMPORARY/VOLATILE table"))
+    }
+  }
 }