drop on full refresh

Author: sryza

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala

@@ -178,12 +178,14 @@ object DatasetManager extends Logging {
     }
 
     // Wipe the data if we need to
-    if ((isFullRefresh || !table.isStreamingTable) && existingTableOpt.isDefined) {
-      context.spark.sql(s"TRUNCATE TABLE ${table.identifier.quotedString}")
+    val dropTable = (isFullRefresh || !table.isStreamingTable) && existingTableOpt.isDefined
+    if (dropTable) {
+      catalog.dropTable(identifier)
+//      context.spark.sql(s"DROP TABLE ${table.identifier.quotedString}")
     }
 
     // Alter the table if we need to
-    if (existingTableOpt.isDefined) {
+    if (existingTableOpt.isDefined && !dropTable) {
       val existingSchema = existingTableOpt.get.schema()
 
       val targetSchema = if (table.isStreamingTable && !isFullRefresh) {
@@ -198,7 +200,7 @@ object DatasetManager extends Logging {
     }
 
     // Create the table if we need to
-    if (existingTableOpt.isEmpty) {
+    if (dropTable || existingTableOpt.isEmpty) {
       catalog.createTable(
         identifier,
         new TableInfo.Builder()

sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/MaterializeTablesSuite.scala

@@ -434,8 +434,9 @@ class MaterializeTablesSuite extends BaseCoreExecutionTest {
 
     val table2 = catalog.loadTable(identifier)
     assert(
-      table2.columns() sameElements CatalogV2Util
-        .structTypeToV2Columns(new StructType().add("y", IntegerType).add("x", BooleanType))
+      table2.columns().toSet == CatalogV2Util
+        .structTypeToV2Columns(new StructType().add("x", BooleanType).add("y", IntegerType))
+        .toSet
     )
     assert(table2.partitioning().toSeq == Seq(Expressions.identity("x")))
 
@@ -456,8 +457,9 @@ class MaterializeTablesSuite extends BaseCoreExecutionTest {
 
     val table3 = catalog.loadTable(identifier)
     assert(
-      table3.columns() sameElements CatalogV2Util
-        .structTypeToV2Columns(new StructType().add("y", IntegerType).add("x", BooleanType))
+      table3.columns().toSet == CatalogV2Util
+        .structTypeToV2Columns(new StructType().add("x", BooleanType).add("y", IntegerType))
+        .toSet
     )
     assert(table3.partitioning().toSeq == Seq(Expressions.identity("x")))
   }