[SPARK-30183][SQL] Disallow to specify reserved properties in CREATE/ALTER NAMESPACE syntax

docs/sql-migration-guide.md

@@ -264,6 +264,8 @@ license: |
 
   - Since Spark 3.0, the function `percentile_approx` and its alias `approx_percentile` only accept integral value with range in `[1, 2147483647]` as its 3rd argument `accuracy`, fractional and string types are disallowed, e.g. `percentile_approx(10.0, 0.2, 1.8D)` will cause `AnalysisException`. In Spark version 2.4 and earlier, if `accuracy` is fractional or string value, it will be coerced to an int value, `percentile_approx(10.0, 0.2, 1.8D)` is operated as `percentile_approx(10.0, 0.2, 1)` which results in `10.0`.
 
+  - Since Spark 3.0, the namespace properties `location` and `comment` become reserved, it will fail with `ParseException` if we use them as members of `DBPROTERTIES` in `CREATE NAMESPACE` and `ALTER NAMESPACE ... SET PROPERTIES(...)`. We need their specific clauses to specify them, e.g. `CREATE NAMESPACE a.b.c COMMENT 'any comment' LOCATION 'some path'`. We can set `spark.sql.legacy.property.nonReserved` to `true` to ignore the `ParseException`, but notice that in this case, these properties will produce side effects, e.g `SET DBPROTERTIES('location'='/tmp')` might change the location of the database. In Spark version 2.4 and earlier, these properties are neither reserved nor have side effects, e.g.   `SET DBPROTERTIES('location'='/tmp')` will not change the location of the database but create a headless property just like `'a'='b'`.
+
 ## Upgrading from Spark SQL 2.4 to 2.4.1
 
   - The value of `spark.executor.heartbeatInterval`, when specified without units like "30" rather than "30s", was

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -2498,6 +2498,19 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
     }
   }
 
+  private def checkNamespaceProperties(
+      properties: Map[String, String], ctx: ParserRuleContext): Unit = withOrigin(ctx) {
+    import SupportsNamespaces._
+    if (!conf.getConf(SQLConf.LEGACY_PROPERTY_NON_RESERVED)) {
+      properties.foreach {
+        case (PROP_LOCATION, _) => throw new ParseException(s"$PROP_LOCATION is a reserved" +
+          s" namespace property, please use the LOCATION clause to specify it.", ctx)
+        case (PROP_COMMENT, _) => throw new ParseException(s"$PROP_COMMENT is a reserved" +
+          s" namespace property, please use the COMMENT clause to specify it.", ctx)
+        case _ =>
+      }
+    }
+  }
   /**
    * Create a [[CreateNamespaceStatement]] command.
    *
@@ -2513,6 +2526,7 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
    * }}}
    */
   override def visitCreateNamespace(ctx: CreateNamespaceContext): LogicalPlan = withOrigin(ctx) {
+    import SupportsNamespaces._
     checkDuplicateClauses(ctx.COMMENT, "COMMENT", ctx)
     checkDuplicateClauses(ctx.locationSpec, "LOCATION", ctx)
     checkDuplicateClauses(ctx.PROPERTIES, "WITH PROPERTIES", ctx)
@@ -2525,11 +2539,14 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
     var properties = ctx.tablePropertyList.asScala.headOption
       .map(visitPropertyKeyValues)
       .getOrElse(Map.empty)
-    Option(ctx.comment).map(string).map {
-      properties += SupportsNamespaces.PROP_COMMENT -> _
+
+    checkNamespaceProperties(properties, ctx)
+
+    Option(ctx.comment).map(string).foreach {
+      properties += PROP_COMMENT -> _
     }
-    ctx.locationSpec.asScala.headOption.map(visitLocationSpec).map {
-      properties += SupportsNamespaces.PROP_LOCATION -> _
+    ctx.locationSpec.asScala.headOption.map(visitLocationSpec).foreach {
+      properties += PROP_LOCATION -> _
     }
 
     CreateNamespaceStatement(
@@ -2564,9 +2581,11 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
    */
   override def visitSetNamespaceProperties(ctx: SetNamespacePropertiesContext): LogicalPlan = {
     withOrigin(ctx) {
+      val properties = visitPropertyKeyValues(ctx.tablePropertyList)
+      checkNamespaceProperties(properties, ctx)
       AlterNamespaceSetPropertiesStatement(
         visitMultipartIdentifier(ctx.multipartIdentifier),
-        visitPropertyKeyValues(ctx.tablePropertyList))
+        properties)
     }
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -2135,6 +2135,16 @@ object SQLConf {
         "defined by `from` and `to`.")
       .booleanConf
       .createWithDefault(false)
+
+  val LEGACY_PROPERTY_NON_RESERVED =
+    buildConf("spark.sql.legacy.property.nonReserved")
+      .internal()
+      .doc("When true, all namespace and table properties are not reserved and available for " +
+        "create/alter syntaxes. But please be aware that the reserved properties will still be " +
+        "used by Spark internally. It is highly discouraged to turn on this to avoid potential " +
+        "side-effects.")
+      .booleanConf
+      .createWithDefault(false)
 }
 
 /**

sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala

@@ -22,6 +22,7 @@ import scala.collection.JavaConverters._
 import org.apache.spark.SparkException
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.analysis.{CannotReplaceMissingTableException, NamespaceAlreadyExistsException, NoSuchDatabaseException, NoSuchNamespaceException, NoSuchTableException, TableAlreadyExistsException}
+import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.connector.catalog._
 import org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAME
 import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
@@ -864,6 +865,33 @@ class DataSourceV2SQLSuite
     }
   }
 
+  test("CreateNameSpace: reserved properties") {
+    withSQLConf((SQLConf.LEGACY_PROPERTY_NON_RESERVED.key, "false")) {
+      SupportsNamespaces.RESERVED_PROPERTIES.asScala.foreach { key =>
+        val exception = intercept[ParseException] {
+          sql(s"CREATE NAMESPACE testcat.reservedTest WITH DBPROPERTIES('$key'='dummyVal')")
+        }
+        assert(exception.getMessage.contains(s"$key is a reserved namespace property"))
+      }
+    }
+    withSQLConf((SQLConf.LEGACY_PROPERTY_NON_RESERVED.key, "true")) {
+      SupportsNamespaces.RESERVED_PROPERTIES.asScala.foreach { key =>
+        withNamespace("testcat.reservedTest") {
+          withTempDir { tmpDir =>
+            val sideEffectVal = tmpDir.getCanonicalPath
+            sql(s"CREATE NAMESPACE testcat.reservedTest WITH DBPROPERTIES('$key'='$sideEffectVal')")
+            assert(sql("DESC NAMESPACE EXTENDED testcat.reservedTest")
+              .toDF("k", "v")
+              .where("k='Properties'")
+              .isEmpty, s"$key is a reserved namespace property")
+            assert(catalog("testcat").asNamespaceCatalog
+              .loadNamespaceMetadata(Array("reservedTest")).get(key) === sideEffectVal)
+          }
+        }
+      }
+    }
+  }
+
   test("DropNamespace: basic tests") {
     // Session catalog is used.
     sql("CREATE NAMESPACE ns")
@@ -961,6 +989,37 @@ class DataSourceV2SQLSuite
     }
   }
 
+  test("AlterNamespaceSetProperties: reserved properties") {
+    withSQLConf((SQLConf.LEGACY_PROPERTY_NON_RESERVED.key, "false")) {
+      SupportsNamespaces.RESERVED_PROPERTIES.asScala.foreach { key =>
+        withNamespace("testcat.reservedTest") {
+          sql("CREATE NAMESPACE testcat.reservedTest")
+          val exception = intercept[ParseException] {
+            sql(s"ALTER NAMESPACE testcat.reservedTest SET PROPERTIES ('$key'='dummyVal')")
+          }
+          assert(exception.getMessage.contains(s"$key is a reserved namespace property"))
+        }
+      }
+    }
+    withSQLConf((SQLConf.LEGACY_PROPERTY_NON_RESERVED.key, "true")) {
+      SupportsNamespaces.RESERVED_PROPERTIES.asScala.foreach { key =>
+        withNamespace("testcat.reservedTest") {
+          withTempDir { tmpDir =>
+            val sideEffectVal = tmpDir.getCanonicalPath
+            sql(s"CREATE NAMESPACE testcat.reservedTest")
+            sql(s"ALTER NAMESPACE testcat.reservedTest SET PROPERTIES ('$key'='$sideEffectVal')")
+            assert(sql("DESC NAMESPACE EXTENDED testcat.reservedTest")
+              .toDF("k", "v")
+              .where("k='Properties'")
+              .isEmpty, s"$key is a reserved namespace property")
+            assert(catalog("testcat").asNamespaceCatalog
+              .loadNamespaceMetadata(Array("reservedTest")).get(key) === sideEffectVal)
+          }
+        }
+      }
+    }
+  }
+
   test("AlterNamespaceSetLocation using v2 catalog") {
     withNamespace("testcat.ns1.ns2") {
       sql("CREATE NAMESPACE IF NOT EXISTS testcat.ns1.ns2 COMMENT " +