diff --git a/core/src/main/scala/org/apache/spark/sql/delta/DeltaSharedExceptions.scala b/core/src/main/scala/org/apache/spark/sql/delta/DeltaSharedExceptions.scala
index da62e756ae..1af300abf3 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/DeltaSharedExceptions.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/DeltaSharedExceptions.scala
@@ -16,6 +16,8 @@
 
 package org.apache.spark.sql.delta
 
+import scala.collection.JavaConverters._
+
 import org.antlr.v4.runtime.ParserRuleContext
 
 import org.apache.spark.sql.AnalysisException
@@ -47,6 +49,11 @@ class DeltaIllegalArgumentException(
     with DeltaThrowable {
     override def getErrorClass: String = errorClass
   def getMessageParametersArray: Array[String] = messageParameters
+
+  override def getMessageParameters: java.util.Map[String, String] = {
+    DeltaThrowableHelper.getParameterNames(errorClass, errorSubClass = null)
+      .zip(messageParameters).toMap.asJava
+  }
 }
 
 class DeltaUnsupportedOperationException(
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/OptimisticTransaction.scala b/core/src/main/scala/org/apache/spark/sql/delta/OptimisticTransaction.scala
index 195bd1fbf4..d3f4a94e47 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/OptimisticTransaction.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/OptimisticTransaction.scala
@@ -561,7 +561,7 @@ trait OptimisticTransactionImpl extends TransactionalWrite
 
 
     RowId.verifyMetadata(
-      spark, protocol, snapshot.metadata, newMetadataTmp, isCreatingNewTable)
+      spark, snapshot.protocol, protocol, snapshot.metadata, newMetadataTmp, isCreatingNewTable)
 
     assertMetadata(newMetadataTmp)
     logInfo(s"Updated metadata from ${newMetadata.getOrElse("-")} to $newMetadataTmp")
diff --git a/core/src/main/scala/org/apache/spark/sql/delta/RowId.scala b/core/src/main/scala/org/apache/spark/sql/delta/RowId.scala
index ea04394e94..995be80620 100644
--- a/core/src/main/scala/org/apache/spark/sql/delta/RowId.scala
+++ b/core/src/main/scala/org/apache/spark/sql/delta/RowId.scala
@@ -64,14 +64,15 @@ object RowId {
    */
   private[delta] def verifyMetadata(
       spark: SparkSession,
-      protocol: Protocol,
+      oldProtocol: Protocol,
+      newProtocol: Protocol,
       oldMetadata: Metadata,
       newMetadata: Metadata,
       isCreatingNewTable: Boolean): Unit = {
     if (!isAllowed(spark)) return
 
-    val rowIdsEnabledBefore = isEnabled(protocol, oldMetadata)
-    val rowIdsEnabledAfter = isEnabled(protocol, newMetadata)
+    val rowIdsEnabledBefore = isEnabled(oldProtocol, oldMetadata)
+    val rowIdsEnabledAfter = isEnabled(newProtocol, newMetadata)
 
     if (rowIdsEnabledAfter && !rowIdsEnabledBefore && !isCreatingNewTable) {
       throw new UnsupportedOperationException(