nerf

common/utils/src/main/resources/error/error-classes.json

@@ -3241,6 +3241,12 @@
     ],
     "sqlState" : "0A000"
   },
+  "STATE_STORE_CANNOT_REMOVE_DEFAULT_COLUMN_FAMILY" : {
+    "message" : [
+      "Failed to remove default column family with reserved name=<colFamilyName>."
+    ],
+    "sqlState" : "42802"
+  },
   "STATE_STORE_MULTIPLE_VALUES_PER_KEY" : {
     "message" : [
       "Store does not support multiple values per key"
@@ -3950,6 +3956,11 @@
           "Creating multiple column families with <stateStoreProvider> is not supported."
         ]
       },
+      "STATE_STORE_REMOVING_COLUMN_FAMILIES" : {
+        "message" : [
+          "Removing column families with <stateStoreProvider> is not supported."
+        ]
+      },
       "TABLE_OPERATION" : {
         "message" : [
           "Table <tableName> does not support <operation>. Please check the current catalog and namespace to make sure the qualified table name is expected, and also check the catalog implementation which is configured by \"spark.sql.catalog\"."

docs/sql-error-conditions-unsupported-feature-error-class.md

@@ -194,6 +194,10 @@ set PROPERTIES and DBPROPERTIES at the same time.
 
 Creating multiple column families with `<stateStoreProvider>` is not supported.
 
+## STATE_STORE_REMOVING_COLUMN_FAMILIES
+
+Removing column families with `<stateStoreProvider>` is not supported.
+
 ## TABLE_OPERATION
 
 Table `<tableName>` does not support `<operation>`. Please check the current catalog and namespace to make sure the qualified table name is expected, and also check the catalog implementation which is configured by "spark.sql.catalog".

docs/sql-error-conditions.md

@@ -2025,6 +2025,12 @@ The SQL config `<sqlConf>` cannot be found. Please verify that the config exists
 
 Star (*) is not allowed in a select list when GROUP BY an ordinal position is used.
 
+### STATE_STORE_CANNOT_REMOVE_DEFAULT_COLUMN_FAMILY
+
+[SQLSTATE: 42802](sql-error-conditions-sqlstates.html#class-42-syntax-error-or-access-rule-violation)
+
+Failed to remove default column family with reserved name=`<colFamilyName>`.
+
 ### STATE_STORE_MULTIPLE_VALUES_PER_KEY
 
 [SQLSTATE: 42802](sql-error-conditions-sqlstates.html#class-42-syntax-error-or-access-rule-violation)

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/HDFSBackedStateStoreProvider.scala

@@ -203,8 +203,9 @@ private[sql] class HDFSBackedStateStoreProvider extends StateStoreProvider with
     }
 
     override def removeColFamilyIfExists(colFamilyName: String): Unit = {
-      throw new UnsupportedOperationException("Removing columnFamily with " +
-        "HDFSBackedStateStoreProvider is not supported")
+      throw StateStoreErrors.removingColumnFamiliesNotSupported(
+        "HDFSBackedStateStoreProvider")
+
     }
   }
 

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

@@ -269,7 +269,7 @@ class RocksDB(
    */
   def removeColFamilyIfExists(colFamilyName: String): Unit = {
     if (colFamilyName == StateStore.DEFAULT_COL_FAMILY_NAME) {
-      throw new UnsupportedOperationException("Removing default column family is not allowed")
+      throw StateStoreErrors.cannotRemoveDefaultColumnFamily(colFamilyName)
     }
 
     if (checkColFamilyExists(colFamilyName)) {

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreErrors.scala

@@ -37,6 +37,16 @@ object StateStoreErrors {
       new StateStoreMultipleColumnFamiliesNotSupportedException(stateStoreProvider)
     }
 
+  def removingColumnFamiliesNotSupported(stateStoreProvider: String):
+    StateStoreRemovingColumnFamiliesNotSupportedException = {
+        new StateStoreRemovingColumnFamiliesNotSupportedException(stateStoreProvider)
+    }
+
+  def cannotRemoveDefaultColumnFamily(colFamilyName: String):
+    StateStoreCannotRemoveDefaultColumnFamily = {
+        new StateStoreCannotRemoveDefaultColumnFamily(colFamilyName)
+    }
+
   def unsupportedOperationException(operationName: String, entity: String):
     StateStoreUnsupportedOperationException = {
       new StateStoreUnsupportedOperationException(operationName, entity)
@@ -48,6 +58,18 @@ class StateStoreMultipleColumnFamiliesNotSupportedException(stateStoreProvider:
     errorClass = "UNSUPPORTED_FEATURE.STATE_STORE_MULTIPLE_COLUMN_FAMILIES",
     messageParameters = Map("stateStoreProvider" -> stateStoreProvider)
   )
+class StateStoreRemovingColumnFamiliesNotSupportedException(stateStoreProvider: String)
+  extends SparkUnsupportedOperationException(
+    errorClass = "UNSUPPORTED_FEATURE.STATE_STORE_REMOVING_COLUMN_FAMILIES",
+    messageParameters = Map("stateStoreProvider" -> stateStoreProvider)
+  )
+
+  class StateStoreCannotRemoveDefaultColumnFamily(colFamilyName: String)
+  extends SparkUnsupportedOperationException(
+    errorClass = "STATE_STORE_CANNOT_REMOVE_DEFAULT_COLUMN_FAMILY",
+    messageParameters = Map("colFamilyName" -> colFamilyName)
+  )
+
 
 class StateStoreUnsupportedOperationException(operationType: String, entity: String)
   extends SparkUnsupportedOperationException(

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/MemoryStateStore.scala

@@ -34,7 +34,7 @@ class MemoryStateStore extends StateStore() {
   }
 
   override def removeColFamilyIfExists(colFamilyName: String): Unit = {
-    throw new UnsupportedOperationException("Doesn't support removing column families!")
+    throw StateStoreErrors.removingColumnFamiliesNotSupported("MemoryStateStoreProvider")
   }
 
   override def get(key: UnsafeRow, colFamilyName: String): UnsafeRow = map.get(key)