[SPARK-48720][SQL] Align the command ALTER TABLE ... UNSET TBLPROPERTIES ... in v1 and v2

docs/sql-ref-syntax-ddl-alter-table.md

@@ -236,20 +236,30 @@ ALTER TABLE table_identifier DROP [ IF EXISTS ] partition_spec [PURGE]
 
 ### SET AND UNSET
 
-#### SET TABLE PROPERTIES
+#### SET PROPERTIES
 
 `ALTER TABLE SET` command is used for setting the table properties. If a particular property was already set, 
 this overrides the old value with the new one.
 
-`ALTER TABLE UNSET` is used to drop the table property. 
-
 ##### Syntax
 
 ```sql
--- Set Table Properties 
+-- Set Properties
 ALTER TABLE table_identifier SET TBLPROPERTIES ( key1 = val1, key2 = val2, ... )
+```
+
+#### UNSET PROPERTIES
+
+`ALTER TABLE UNSET` command is used to drop the table property.
 
--- Unset Table Properties
+**Note:** If the specified property key does not exist, when you use the v1 command and do not specify `IF EXISTS`,
+it will throw the error-condition `UNSET_NONEXISTENT_PROPERTIES` and finally `failed`,
+however regardless of whether `IF EXISTS` is set or not, the v2 command will ignore it and finally `succeed`.
+
+##### Syntax
+
+```sql
+-- Unset Properties
 ALTER TABLE table_identifier UNSET TBLPROPERTIES [ IF EXISTS ] ( key1, key2, ... )
 ```
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala

@@ -1075,25 +1075,6 @@ class DDLParserSuite extends AnalysisTest {
         ifExists = true))
   }
 
-  // ALTER TABLE table_name UNSET TBLPROPERTIES [IF EXISTS] ('comment', 'key');
-  test("alter table: alter table properties") {
-    val sql2_table = "ALTER TABLE table_name UNSET TBLPROPERTIES ('comment', 'test')"
-    val sql3_table = "ALTER TABLE table_name UNSET TBLPROPERTIES IF EXISTS ('comment', 'test')"
-
-    comparePlans(
-      parsePlan(sql2_table),
-      UnsetTableProperties(
-        UnresolvedTable(Seq("table_name"), "ALTER TABLE ... UNSET TBLPROPERTIES", true),
-        Seq("comment", "test"),
-        ifExists = false))
-    comparePlans(
-      parsePlan(sql3_table),
-      UnsetTableProperties(
-        UnresolvedTable(Seq("table_name"), "ALTER TABLE ... UNSET TBLPROPERTIES", true),
-        Seq("comment", "test"),
-        ifExists = true))
-  }
-
   test("alter table: add column") {
     comparePlans(
       parsePlan("ALTER TABLE table_name ADD COLUMN x int"),

sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableSetTblPropertiesSuiteBase.scala

@@ -19,6 +19,10 @@ package org.apache.spark.sql.execution.command
 
 import org.apache.spark.sql.{AnalysisException, QueryTest}
 import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, TableCatalog}
+import org.apache.spark.sql.errors.DataTypeErrors.toSQLId
+import org.apache.spark.sql.internal.SQLConf
 
 /**
  * This base suite contains unified tests for the `ALTER TABLE .. SET TBLPROPERTIES`
@@ -39,7 +43,26 @@ trait AlterTableSetTblPropertiesSuiteBase extends QueryTest with DDLCommandTestU
 
   def checkTblProps(tableIdent: TableIdentifier, expectedTblProps: Map[String, String]): Unit
 
-  test("alter table set tblproperties") {
+  def getTblPropertyValue(tableIdent: TableIdentifier, key: String): String
+
+  test("table to alter does not exist") {
+    withNamespaceAndTable("ns", "does_not_exist") { t =>
+      val sqlText = s"ALTER TABLE $t SET TBLPROPERTIES ('k1' = 'v1')"
+      checkError(
+        exception = intercept[AnalysisException] {
+          sql(sqlText)
+        },
+        errorClass = "TABLE_OR_VIEW_NOT_FOUND",
+        parameters = Map("relationName" -> toSQLId(t)),
+      context = ExpectedContext(
+        fragment = t,
+        start = 12,
+        stop = 11 + t.length)
+      )
+    }
+  }
+
+  test("alter table set properties") {
     withNamespaceAndTable("ns", "tbl") { t =>
       sql(s"CREATE TABLE $t (col1 int, col2 string, a int, b int) $defaultUsing")
       val tableIdent = TableIdentifier("tbl", Some("ns"), Some(catalog))
@@ -54,16 +77,53 @@ trait AlterTableSetTblPropertiesSuiteBase extends QueryTest with DDLCommandTestU
 
       sql(s"ALTER TABLE $t SET TBLPROPERTIES ('k1' = 'v1', 'k2' = 'v8')")
       checkTblProps(tableIdent, Map("k1" -> "v1", "k2" -> "v8", "k3" -> "v3"))
+    }
+  }
 
-      // table to alter does not exist
-      checkError(
-        exception = intercept[AnalysisException] {
-          sql("ALTER TABLE does_not_exist SET TBLPROPERTIES ('winner' = 'loser')")
-        },
-        errorClass = "TABLE_OR_VIEW_NOT_FOUND",
-        parameters = Map("relationName" -> "`does_not_exist`"),
-        context = ExpectedContext(fragment = "does_not_exist", start = 12, stop = 25)
-      )
+  test("alter table set reserved properties") {
+    import TableCatalog._
+    val keyParameters = Map[String, String](
+      PROP_PROVIDER -> "please use the USING clause to specify it",
+      PROP_LOCATION -> "please use the LOCATION clause to specify it",
+      PROP_OWNER -> "it will be set to the current user",
+      PROP_EXTERNAL -> "please use CREATE EXTERNAL TABLE"
+    )
+    withSQLConf((SQLConf.LEGACY_PROPERTY_NON_RESERVED.key, "false")) {
+      CatalogV2Util.TABLE_RESERVED_PROPERTIES.filterNot(_ == PROP_COMMENT).foreach { key =>
+        withNamespaceAndTable("ns", "tbl") { t =>
+          val sqlText = s"ALTER TABLE $t SET TBLPROPERTIES ('$key'='bar')"
+          checkError(
+            exception = intercept[ParseException] {
+              sql(sqlText)
+            },
+            errorClass = "UNSUPPORTED_FEATURE.SET_TABLE_PROPERTY",
+            parameters = Map(
+              "property" -> key,
+              "msg" -> keyParameters.getOrElse(
+                key, "please remove it from the TBLPROPERTIES list.")),
+            context = ExpectedContext(
+              fragment = sqlText,
+              start = 0,
+              stop = 40 + t.length + key.length))
+        }
+      }
+    }
+    withSQLConf((SQLConf.LEGACY_PROPERTY_NON_RESERVED.key, "true")) {
+      CatalogV2Util.TABLE_RESERVED_PROPERTIES.filterNot(_ == PROP_COMMENT).foreach { key =>
+        Seq("OPTIONS", "TBLPROPERTIES").foreach { clause =>
+          withNamespaceAndTable("ns", "tbl") { t =>
+            sql(s"CREATE TABLE $t (key int) USING parquet $clause ('$key'='bar')")
+            val tableIdent = TableIdentifier("tbl", Some("ns"), Some(catalog))
+
+            val originValue = getTblPropertyValue(tableIdent, key)
+            assert(originValue != "bar", "reserved properties should not have side effects")
+
+            sql(s"ALTER TABLE $t SET TBLPROPERTIES ('$key'='newValue')")
+            assert(getTblPropertyValue(tableIdent, key) == originValue,
+              "reserved properties should not have side effects")
+          }
+        }
+      }
     }
   }
 }

sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableUnsetTblPropertiesParserSuite.scala

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.execution.command
+
+import org.apache.spark.SparkThrowable
+import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, UnresolvedTable}
+import org.apache.spark.sql.catalyst.parser.CatalystSqlParser.parsePlan
+import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.catalyst.plans.logical.UnsetTableProperties
+import org.apache.spark.sql.test.SharedSparkSession
+
+class AlterTableUnsetTblPropertiesParserSuite extends AnalysisTest with SharedSparkSession {
+
+  private def parseException(sqlText: String): SparkThrowable = {
+    intercept[ParseException](sql(sqlText).collect())
+  }
+
+  // ALTER TABLE table_name UNSET TBLPROPERTIES [IF EXISTS] ('comment', 'key');
+  test("alter table: alter table properties") {
+    val sql1 = "ALTER TABLE table_name UNSET TBLPROPERTIES ('comment', 'test')"
+    val sql2 = "ALTER TABLE table_name UNSET TBLPROPERTIES IF EXISTS ('comment', 'test')"
+
+    comparePlans(
+      parsePlan(sql1),
+      UnsetTableProperties(
+        UnresolvedTable(Seq("table_name"), "ALTER TABLE ... UNSET TBLPROPERTIES",
+          suggestAlternative = true),
+        Seq("comment", "test"),
+        ifExists = false))
+    comparePlans(
+      parsePlan(sql2),
+      UnsetTableProperties(
+        UnresolvedTable(Seq("table_name"), "ALTER TABLE ... UNSET TBLPROPERTIES",
+          suggestAlternative = true),
+        Seq("comment", "test"),
+        ifExists = true))
+  }
+
+  test("alter table unset properties - property values must NOT be set") {
+    val sql = "ALTER TABLE my_tab UNSET TBLPROPERTIES('key_without_value', 'key_with_value'='x')"
+    checkError(
+      exception = parseException(sql),
+      errorClass = "_LEGACY_ERROR_TEMP_0035",
+      parameters = Map("message" -> "Values should not be specified for key(s): [key_with_value]"),
+      context = ExpectedContext(
+        fragment = sql,
+        start = 0,
+        stop = 80))
+  }
+}

sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlterTableUnsetTblPropertiesSuiteBase.scala

@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.execution.command
+
+import org.apache.spark.sql.{AnalysisException, QueryTest}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.parser.ParseException
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, TableCatalog}
+import org.apache.spark.sql.errors.DataTypeErrors.toSQLId
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * This base suite contains unified tests for the `ALTER TABLE .. UNSET TBLPROPERTIES`
+ * command that check V1 and V2 table catalogs. The tests that cannot run for all supported
+ * catalogs are located in more specific test suites:
+ *
+ *   - V2 table catalog tests:
+ *     `org.apache.spark.sql.execution.command.v2.AlterTableUnsetTblPropertiesSuite`
+ *   - V1 table catalog tests:
+ *     `org.apache.spark.sql.execution.command.v1.AlterTableUnsetTblPropertiesSuiteBase`
+ *     - V1 In-Memory catalog:
+ *       `org.apache.spark.sql.execution.command.v1.AlterTableUnsetTblPropertiesSuite`
+ *     - V1 Hive External catalog:
+ *        `org.apache.spark.sql.hive.execution.command.AlterTableUnsetTblPropertiesSuite`
+ */
+trait AlterTableUnsetTblPropertiesSuiteBase extends QueryTest with DDLCommandTestUtils {
+  override val command = "ALTER TABLE .. UNSET TBLPROPERTIES"
+
+  def checkTblProps(tableIdent: TableIdentifier, expectedTblProps: Map[String, String]): Unit
+
+  def getTblPropertyValue(tableIdent: TableIdentifier, key: String): String
+
+  test("table to alter does not exist") {
+    withNamespaceAndTable("ns", "does_not_exist") { t =>
+      val sqlText = s"ALTER TABLE $t UNSET TBLPROPERTIES ('k1')"
+      checkError(
+        exception = intercept[AnalysisException] {
+          sql(sqlText)
+        },
+        errorClass = "TABLE_OR_VIEW_NOT_FOUND",
+        parameters = Map("relationName" -> toSQLId(t)),
+        context = ExpectedContext(
+          fragment = t,
+          start = 12,
+          stop = 11 + t.length)
+      )
+    }
+  }
+
+  test("alter table unset properties") {
+    withNamespaceAndTable("ns", "tbl") { t =>
+      sql(s"CREATE TABLE $t (col1 int, col2 string, a int, b int) $defaultUsing")
+      val tableIdent = TableIdentifier("tbl", Some("ns"), Some(catalog))
+      checkTblProps(tableIdent, Map.empty[String, String])
+
+      sql(s"ALTER TABLE $t SET TBLPROPERTIES ('k1' = 'v1', 'k2' = 'v2', 'k3' = 'v3', 'k4' = 'v4')")
+      checkTblProps(tableIdent, Map("k1" -> "v1", "k2" -> "v2", "k3" -> "v3", "k4" -> "v4"))
+
+      // unset table properties
+      sql(s"ALTER TABLE $t UNSET TBLPROPERTIES ('k1')")
+      checkTblProps(tableIdent, Map("k2" -> "v2", "k3" -> "v3", "k4" -> "v4"))
+
+      // unset table properties without explicitly specifying database
+      sql(s"USE $catalog.ns")
+      sql(s"ALTER TABLE tbl UNSET TBLPROPERTIES ('k2')")
+      checkTblProps(tableIdent, Map("k3" -> "v3", "k4" -> "v4"))
+    }
+  }
+
+  test("alter table unset reserved properties") {
+    import TableCatalog._
+    val keyParameters = Map[String, String](
+      PROP_PROVIDER -> "please use the USING clause to specify it",
+      PROP_LOCATION -> "please use the LOCATION clause to specify it",
+      PROP_OWNER -> "it will be set to the current user",
+      PROP_EXTERNAL -> "please use CREATE EXTERNAL TABLE"
+    )
+    withSQLConf((SQLConf.LEGACY_PROPERTY_NON_RESERVED.key, "false")) {
+      CatalogV2Util.TABLE_RESERVED_PROPERTIES.filterNot(_ == PROP_COMMENT).foreach { key =>
+        withNamespaceAndTable("ns", "tbl") { t =>
+          val sqlText = s"ALTER TABLE $t UNSET TBLPROPERTIES ('$key')"
+          checkError(
+            exception = intercept[ParseException] {
+              sql(sqlText)
+            },
+            errorClass = "UNSUPPORTED_FEATURE.SET_TABLE_PROPERTY",
+            parameters = Map(
+              "property" -> key,
+              "msg" -> keyParameters.getOrElse(
+                key, "please remove it from the TBLPROPERTIES list.")),
+            context = ExpectedContext(
+              fragment = sqlText,
+              start = 0,
+              stop = 36 + t.length + key.length))
+        }
+      }
+    }
+    withSQLConf((SQLConf.LEGACY_PROPERTY_NON_RESERVED.key, "true")) {
+      CatalogV2Util.TABLE_RESERVED_PROPERTIES.filterNot(_ == PROP_COMMENT).foreach { key =>
+        Seq("OPTIONS", "TBLPROPERTIES").foreach { clause =>
+          withNamespaceAndTable("ns", "tbl") { t =>
+            sql(s"CREATE TABLE $t (key int) USING parquet $clause ('$key'='bar')")
+            val tableIdent = TableIdentifier("tbl", Some("ns"), Some(catalog))
+
+            val originValue = getTblPropertyValue(tableIdent, key)
+            assert(originValue != "bar", "reserved properties should not have side effects")
+
+            sql(s"ALTER TABLE $t UNSET TBLPROPERTIES ('$key')")
+            assert(getTblPropertyValue(tableIdent, key) == originValue,
+              "reserved properties should not have side effects")
+          }
+        }
+      }
+    }
+  }
+}

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLParserSuite.scala

@@ -107,18 +107,6 @@ class DDLParserSuite extends AnalysisTest with SharedSparkSession {
         stop = 98))
   }
 
-  test("alter table unset properties - property values must NOT be set") {
-    val sql = "ALTER TABLE my_tab UNSET TBLPROPERTIES('key_without_value', 'key_with_value'='x')"
-    checkError(
-      exception = parseException(sql),
-      errorClass = "_LEGACY_ERROR_TEMP_0035",
-      parameters = Map("message" -> "Values should not be specified for key(s): [key_with_value]"),
-      context = ExpectedContext(
-        fragment = sql,
-        start = 0,
-        stop = 80))
-  }
-
   test("alter table: exchange partition (not supported)") {
     val sql =
       """ALTER TABLE table_name_1 EXCHANGE PARTITION