[Kernel] Allow setting arbitrary properties when creating/updating the table

kernel/kernel-api/src/main/java/io/delta/kernel/internal/TableConfig.java

@@ -204,16 +204,23 @@ public static Map<String, String> validateProperties(Map<String, String> configu
     for (Map.Entry<String, String> kv : configurations.entrySet()) {
       String key = kv.getKey().toLowerCase(Locale.ROOT);
       String value = kv.getValue();
-      if (key.startsWith("delta.") && VALID_PROPERTIES.containsKey(key)) {
+
+      if (key.startsWith("delta.")) {
+        // If it is a delta table property, make sure it is a supported property and editable
+        if (!VALID_PROPERTIES.containsKey(key)) {
+          throw DeltaErrors.unknownConfigurationException(kv.getKey());
+        }
+
         TableConfig<?> tableConfig = VALID_PROPERTIES.get(key);
-        if (tableConfig.editable) {
-          tableConfig.validate(value);
-          validatedConfigurations.put(tableConfig.getKey(), value);
-        } else {
+        if (!tableConfig.editable) {
           throw DeltaErrors.cannotModifyTableProperty(kv.getKey());
         }
+
+        tableConfig.validate(value);
+        validatedConfigurations.put(tableConfig.getKey(), value);
       } else {
-        throw DeltaErrors.unknownConfigurationException(kv.getKey());
+        // allow unknown properties to be set
+        validatedConfigurations.put(key, value);
       }
     }
     return validatedConfigurations;

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableWritesSuite.scala

@@ -292,29 +292,6 @@ class DeltaTableWritesSuite extends DeltaTableWriteSuiteBase with ParquetSuiteBa
     }
   }
 
-  test("create table - invalid properties - expect failure") {
-    withTempDirAndEngine { (tablePath, engine) =>
-      val ex1 = intercept[UnknownConfigurationException] {
-        createTxn(
-          engine, tablePath, isNewTable = true, testSchema, Seq.empty, Map("invalid key" -> "10"))
-      }
-      assert(ex1.getMessage.contains("Unknown configuration was specified: invalid key"))
-
-      val ex2 = intercept[InvalidConfigurationValueException] {
-        createTxn(
-          engine,
-          tablePath,
-          isNewTable = true,
-          testSchema, Seq.empty, Map(TableConfig.CHECKPOINT_INTERVAL.getKey -> "-1"))
-      }
-      assert(
-        ex2.getMessage.contains(
-          String.format(
-            "Invalid value for table property '%s': '%s'. %s",
-            TableConfig.CHECKPOINT_INTERVAL.getKey, "-1", "needs to be a positive integer.")))
-    }
-  }
-
   test("create partitioned table - partition column is not part of the schema") {
     withTempDirAndEngine { (tablePath, engine) =>
       val table = Table.forPath(engine, tablePath)

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/TablePropertiesSutie.scala

@@ -0,0 +1,91 @@
+/*
+ * 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 io.delta.kernel.defaults
+
+import io.delta.kernel.Table
+import io.delta.kernel.exceptions.UnknownConfigurationException
+import io.delta.kernel.internal.SnapshotImpl
+import io.delta.kernel.utils.CloseableIterable.emptyIterable
+
+import scala.collection.immutable.Seq
+
+/**
+ * Suite to set or get table properties.
+ * TODO: for now we just have the support for `set`. API `get` will be added in the next PRs.
+ */
+class TablePropertiesSutie extends DeltaTableWriteSuiteBase {
+  test("create/update table - allow arbitrary properties") {
+    withTempDir { tempFile =>
+      val tablePath = tempFile.getAbsolutePath
+
+      // create table with arbitrary properties and check if they are set
+      createUpdateTableWithProps(
+        tablePath,
+        createTable = true,
+        props = Map("my key" -> "10", "my key2" -> "20")
+      )
+      assertHasProp(tablePath, expProps = Map("my key" -> "10", "my key2" -> "20"))
+
+      // update table by modifying the arbitrary properties and check if they are updated
+      createUpdateTableWithProps(tablePath, props = Map("my key" -> "30"))
+      assertHasProp(tablePath, expProps = Map("my key" -> "30", "my key2" -> "20"))
+
+      // update table without any new properties and check if the existing properties are retained
+      createUpdateTableWithProps(tablePath)
+      assertHasProp(tablePath, expProps = Map("my key" -> "30", "my key2" -> "20"))
+
+      // update table by adding new arbitrary properties and check if they are set
+      createUpdateTableWithProps(tablePath, props = Map("new key3" -> "str"))
+      assertHasProp(
+        tablePath,
+        expProps = Map("my key" -> "30", "my key2" -> "20", "new key3" -> "str"))
+    }
+  }
+
+  test("create/update table - disallow unknown delta.* properties to Kernel") {
+    withTempDir { tempFile =>
+      val tablePath = tempFile.getAbsolutePath
+      val ex1 = intercept[UnknownConfigurationException] {
+        createUpdateTableWithProps(tablePath, createTable = true, Map("delta.unknown" -> "str"))
+      }
+      assert(ex1.getMessage.contains("Unknown configuration was specified: delta.unknown"))
+
+      // Try updating in an existing table
+      createUpdateTableWithProps(tablePath, createTable = true)
+      val ex2 = intercept[UnknownConfigurationException] {
+        createUpdateTableWithProps(tablePath, props = Map("Delta.unknown" -> "str"))
+      }
+      assert(ex2.getMessage.contains("Unknown configuration was specified: Delta.unknown"))
+    }
+  }
+
+  def createUpdateTableWithProps(
+      tablePath: String,
+      createTable: Boolean = false,
+      props: Map[String, String] = null): Unit = {
+    createTxn(defaultEngine, tablePath, createTable, testSchema, Seq.empty, props)
+      .commit(defaultEngine, emptyIterable())
+  }
+
+  // TODO: this will be replaced with get API in the next PRs.
+  def assertHasProp(tablePath: String, expProps: Map[String, String]): Unit = {
+    val snapshot = Table.forPath(defaultEngine, tablePath)
+      .getLatestSnapshot(defaultEngine).asInstanceOf[SnapshotImpl]
+    expProps.foreach { case (key, value) =>
+      assert(snapshot.getMetadata.getConfiguration.get(key) === value, key)
+    }
+  }
+}