Fix Creating an external table fails when it is given the same options as for the underlying data #809

Signed-off-by: Lucian Neghina <[email protected]>

Author: nflucian

core/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaCatalog.scala

@@ -199,12 +199,13 @@ class DeltaCatalog extends DelegatingCatalogExtension
       partitions: Array[Transform],
       properties: util.Map[String, String]): Table = {
     if (DeltaSourceUtils.isDeltaDataSourceName(getProvider(properties))) {
+      val catalogProperties = DeltaTableProperties(properties)
       createDeltaTable(
         ident,
         schema,
         partitions,
-        properties,
-        Map.empty,
+        catalogProperties.properties.asJava,
+        catalogProperties.options,
         sourceQuery = None,
         TableCreationModes.Create)
     } else {
@@ -350,42 +351,13 @@ class DeltaCatalog extends DelegatingCatalogExtension
     private var writeOptions: Map[String, String] = Map.empty
 
     override def commitStagedChanges(): Unit = {
-      val conf = spark.sessionState.conf
-      val props = new util.HashMap[String, String]()
-      // Options passed in through the SQL API will show up both with an "option." prefix and
-      // without in Spark 3.1, so we need to remove those from the properties
-      val optionsThroughProperties = properties.asScala.collect {
-        case (k, _) if k.startsWith("option.") => k.stripPrefix("option.")
-      }.toSet
-      val sqlWriteOptions = new util.HashMap[String, String]()
-      properties.asScala.foreach { case (k, v) =>
-        if (!k.startsWith("option.") && !optionsThroughProperties.contains(k)) {
-          // Do not add to properties
-          props.put(k, v)
-        } else if (optionsThroughProperties.contains(k)) {
-          sqlWriteOptions.put(k, v)
-        }
-      }
-      if (writeOptions.isEmpty && !sqlWriteOptions.isEmpty) {
-        writeOptions = sqlWriteOptions.asScala.toMap
-      }
-      if (conf.getConf(DeltaSQLConf.DELTA_LEGACY_STORE_WRITER_OPTIONS_AS_PROPS)) {
-        // Legacy behavior
-        writeOptions.foreach { case (k, v) => props.put(k, v) }
-      } else {
-        writeOptions.foreach { case (k, v) =>
-          // Continue putting in Delta prefixed options to avoid breaking workloads
-          if (k.toLowerCase(Locale.ROOT).startsWith("delta.")) {
-            props.put(k, v)
-          }
-        }
-      }
+      val catalogProp = DeltaTableProperties(properties, writeOptions)
       createDeltaTable(
         ident,
         schema,
         partitions,
-        props,
-        writeOptions,
+        catalogProp.properties.asJava,
+        catalogProp.options,
         asSelectQuery,
         operation)
     }

core/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaTableProperties.scala

@@ -0,0 +1,72 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.delta.catalog
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.delta.sources.DeltaSQLConf
+
+import java.util
+import java.util.Locale
+import scala.collection.JavaConverters._
+
+case class DeltaTableProperties(properties: Map[String, String], options: Map[String, String])
+
+object DeltaTableProperties {
+
+  def apply(properties: util.Map[String, String],
+            deltaOptions: Map[String, String] = Map.empty): DeltaTableProperties = {
+    val (options, props) = split(properties.asScala.toMap)
+    val writeOptions = if (deltaOptions.isEmpty && options.nonEmpty) options else deltaOptions
+    val tableProperties = props ++ deltaProperties(writeOptions)
+
+    new DeltaTableProperties(tableProperties, writeOptions)
+  }
+
+  private def deltaProperties(writeOptions: Map[String, String]): Map[String, String] = {
+    val conf = SparkSession.active.sessionState.conf
+
+    if (conf.getConf(DeltaSQLConf.DELTA_LEGACY_STORE_WRITER_OPTIONS_AS_PROPS)) {
+      // Legacy behavior
+      writeOptions
+    } else {
+      writeOptions.filter { case (k, _) =>
+        k.toLowerCase(Locale.ROOT).startsWith("delta.") }
+    }
+  }
+
+  private def split(properties: Map[String, String]): (Map[String, String], Map[String, String]) = {
+    // Options passed in through the SQL API will show up both with an "option." prefix and
+    // without in Spark 3.1, so we need to remove those from the properties
+    val optionsThroughProperties = properties.collect {
+      case (k, _) if k.startsWith("option.") => k.stripPrefix("option.")
+    }.toSet
+
+    val options = new util.HashMap[String, String]()
+    val props = new util.HashMap[String, String]()
+
+    properties.foreach { case (k, v) =>
+      if (!k.startsWith("option.") && !optionsThroughProperties.contains(k) && !k.equals("path")) {
+        // Do not add to properties
+        props.put(k, v)
+      } else if (optionsThroughProperties.contains(k)) {
+        options.put(k, v)
+      }
+    }
+
+    (options.asScala.toMap, props.asScala.toMap)
+  }
+}

core/src/main/scala/org/apache/spark/sql/delta/sources/DeltaDataSource.scala

@@ -16,6 +16,8 @@
 
 package org.apache.spark.sql.delta.sources
 
+import org.apache.spark.sql.delta.catalog.DeltaTableProperties
+
 import scala.collection.JavaConverters._
 import scala.collection.mutable
 import scala.util.{Failure, Success, Try}
@@ -145,12 +147,13 @@ class DeltaDataSource
       .getOrElse(Nil)
 
     val deltaLog = DeltaLog.forTable(sqlContext.sparkSession, path, parameters)
+    val catalogProp = DeltaTableProperties(parameters.asJava)
     WriteIntoDelta(
       deltaLog = deltaLog,
       mode = mode,
       new DeltaOptions(parameters, sqlContext.sparkSession.sessionState.conf),
       partitionColumns = partitionColumns,
-      configuration = Map.empty,
+      configuration = catalogProp.properties,
       data = data).run(sqlContext.sparkSession)
 
     deltaLog.createRelation()

core/src/test/scala/io/delta/tables/DeltaTableBuilderSuite.scala

@@ -365,4 +365,47 @@ class DeltaTableBuilderSuite extends QueryTest with SharedSparkSession with Delt
       assert(e.getMessage.equals("Database 'parquet' not found"))
     }
   }
+
+  test("create table - different PROPERTIES") {
+    withTempDir {
+      tempDir =>
+        val path = tempDir.getCanonicalPath
+        val options = Map(
+          "header"->"true",
+          "foo"->"bar"
+        )
+        spark.range(10)
+          .write
+          .format("delta")
+          .options(options)
+          .save(path)
+
+        val e = intercept[AnalysisException] {
+          sql(s"""CREATE TABLE test (id LONG)
+                       | USING delta LOCATION '$path'
+                       | TBLPROPERTIES ('foo'='bar')""".stripMargin)
+        }
+        assert(e.getMessage.startsWith(
+          "The specified properties do not match the existing properties"))
+    }
+  }
+
+  test("create table - same existing properties") {
+    withTempDir { tempDir =>
+      val path = tempDir.getCanonicalPath
+      spark.range(10)
+        .write
+        .format("delta")
+        .option("header", "true")
+        .save(path)
+      sql(
+        s"""CREATE TABLE testTable (id LONG)
+           | USING delta LOCATION '$path'
+           | TBLPROPERTIES ('header'='true')""".stripMargin)
+
+      val deltaLog = DeltaLog.forTable(spark, path)
+      val expected = Map("header" -> "true")
+      assert(deltaLog.snapshot.metadata.configuration === expected)
+    }
+  }
 }

core/src/test/scala/org/apache/spark/sql/delta/DeltaSuite.scala

@@ -1663,4 +1663,36 @@ class DeltaSuite extends QueryTest
     dir2.renameTo(dir1)
     checkAnswer(spark.read.format("delta").load(dir1.getCanonicalPath), spark.range(10).toDF)
   }
+
+  test("set metadata upon write") {
+    withTempDir { tempDir =>
+      val path = tempDir.getCanonicalPath
+      val options = Map(
+        "header"-> "true",
+        "foo" -> "bar"
+      )
+      spark.range(10).write
+        .format("delta")
+        .options(options)
+        .save(path)
+
+      val deltaLog = DeltaLog.forTable(spark, path)
+      assert(deltaLog.snapshot.metadata.configuration === options)
+    }
+  }
+
+  test("set metadata upon write using v2") {
+    withTempDir { tempDir =>
+      val path = tempDir.getCanonicalPath
+      spark.range(10)
+        .writeTo(s"delta.`$path`")
+        .using("delta")
+        .tableProperty("header", "true")
+        .createOrReplace()
+
+      val deltaLog = DeltaLog.forTable(spark, path)
+      val expected = Map("header" -> "true")
+      assert(deltaLog.snapshot.metadata.configuration === expected)
+    }
+  }
 }