add more test cases; fix ICT disablement logic

Author: scottsand-db

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

@@ -419,10 +419,17 @@ private void handleInCommitTimestampDisablement() {
     final boolean txnExplicitlyDisablesICT =
         txnIctEnabledValue != null && txnIctEnabledValue.equalsIgnoreCase("false");
 
+    // Case 1: Txn is not explicitly disabling ICT. Exit.
     if (!txnExplicitlyDisablesICT) {
       return;
     }
 
+    // Case 2: Txn is explicitly disabling ICT on a catalogManaged table. Throw.
+    if (getEffectiveProtocol().supportsFeature(TableFeatures.CATALOG_MANAGED_R_W_FEATURE_PREVIEW)) {
+      throw new KernelException("Cannot disable inCommitTimestamp on a catalogManaged table");
+    }
+
+    // Case 3 (normal case): Txn is explicitly disabling ICT. Remove the ICT enablement properties.
     final Set<String> ictKeysToRemove =
         new HashSet<>(
             Arrays.asList(

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/catalogManaged/CatalogManagedEnablementSuite.scala

@@ -31,7 +31,7 @@ import org.scalatest.funsuite.AnyFunSuite
 
 class CatalogManagedEnablementSuite extends AnyFunSuite with TestUtils {
 
-  case class CatalogManagedEnablementTestCase(
+  case class CatalogManagedTestCase(
       testName: String,
       operationType: String, // "CREATE", "UPDATE", or "REPLACE"
       initialTableProperties: Map[String, String] = Map.empty,
@@ -42,42 +42,51 @@ class CatalogManagedEnablementSuite extends AnyFunSuite with TestUtils {
       expectedCatalogManagedSupported: Boolean = false)
 
   val catalogManagedTestCases = Seq(
-    CatalogManagedEnablementTestCase(
-      testName = "CREATE: catalogManaged enablement flag => enables catalogManaged and ICT",
+    // ===== CREATE cases =====
+    CatalogManagedTestCase(
+      testName = "CREATE: set catalogManaged=supported => enables catalogManaged and ICT",
       operationType = "CREATE",
       transactionProperties = Map("delta.feature.catalogOwned-preview" -> "supported"),
       expectedSuccess = true,
       expectedIctEnabled = true,
       expectedCatalogManagedSupported = true),
-    CatalogManagedEnablementTestCase(
-      testName = "UPDATE: catalogManaged enablement flag => enables catalogManaged and ICT",
+    CatalogManagedTestCase(
+      testName = "ILLEGAL CREATE: set catalogManaged=supported and explicitly disable ICT => THROW",
+      operationType = "CREATE",
+      transactionProperties = Map(
+        "delta.feature.catalogOwned-preview" -> "supported",
+        "delta.enableInCommitTimestamps" -> "false"),
+      expectedSuccess = false,
+      expectedExceptionMessage =
+        Some("Cannot disable inCommitTimestamp when enabling catalogManaged")),
+
+    // ===== UPDATE cases =====
+    CatalogManagedTestCase(
+      testName = "UPDATE: set catalogManaged=supported => enables catalogManaged and ICT",
       operationType = "UPDATE",
       initialTableProperties = Map.empty, // Start with basic table
       transactionProperties = Map("delta.feature.catalogOwned-preview" -> "supported"),
       expectedSuccess = true,
       expectedIctEnabled = true,
       expectedCatalogManagedSupported = true),
-    CatalogManagedEnablementTestCase(
-      testName = "UPDATE: catalogManaged enablement flag => enables ICT if previously disabled",
+    CatalogManagedTestCase(
+      testName = "UPDATE: set catalogManaged=supported => enables ICT if previously disabled",
       operationType = "UPDATE",
       initialTableProperties = Map("delta.enableInCommitTimestamps" -> "false"),
       transactionProperties = Map("delta.feature.catalogOwned-preview" -> "supported"),
       expectedSuccess = true,
       expectedIctEnabled = true,
       expectedCatalogManagedSupported = true),
-    CatalogManagedEnablementTestCase(
-      testName =
-        "ILLEGAL CREATE: catalogManaged enablement flag but ICT explicitly disabled too => THROW",
-      operationType = "CREATE",
-      transactionProperties = Map(
-        "delta.feature.catalogOwned-preview" -> "supported",
-        "delta.enableInCommitTimestamps" -> "false"),
-      expectedSuccess = false,
-      expectedExceptionMessage =
-        Some("Cannot disable inCommitTimestamp when enabling catalogManaged")),
-    CatalogManagedEnablementTestCase(
-      testName =
-        "ILLEGAL UPDATE: catalogManaged enablement flag but ICT explicitly disabled too => THROW",
+    CatalogManagedTestCase(
+      testName = "UPDATE: set catalogManaged=supported and ICT already enabled => Okay",
+      operationType = "UPDATE",
+      initialTableProperties = Map("delta.enableInCommitTimestamps" -> "true"),
+      transactionProperties = Map("delta.feature.catalogOwned-preview" -> "supported"),
+      expectedSuccess = true,
+      expectedIctEnabled = true,
+      expectedCatalogManagedSupported = true),
+    CatalogManagedTestCase(
+      testName = "ILLEGAL UPDATE: set catalogManaged=supported and disable ICT => THROW",
       operationType = "UPDATE",
       initialTableProperties = Map.empty,
       transactionProperties = Map(
@@ -86,38 +95,48 @@ class CatalogManagedEnablementSuite extends AnyFunSuite with TestUtils {
       expectedSuccess = false,
       expectedExceptionMessage =
         Some("Cannot disable inCommitTimestamp when enabling catalogManaged")),
-    CatalogManagedEnablementTestCase(
-      testName = "UPDATE: catalogManaged enablement flag => ICT already enabled",
+    CatalogManagedTestCase(
+      testName = "ILLEGAL UPDATE: catalogManaged already supported, then disable ICT => THROW",
       operationType = "UPDATE",
-      initialTableProperties = Map("delta.enableInCommitTimestamps" -> "true"),
-      transactionProperties = Map("delta.feature.catalogOwned-preview" -> "supported"),
-      expectedSuccess = true,
-      expectedIctEnabled = true,
-      expectedCatalogManagedSupported = true),
-    CatalogManagedEnablementTestCase(
-      testName = "No-op: catalogManaged not being enabled should not affect ICT",
+      initialTableProperties = Map("delta.feature.catalogOwned-preview" -> "supported"),
+      transactionProperties = Map("delta.enableInCommitTimestamps" -> "false"),
+      expectedSuccess = false,
+      expectedExceptionMessage =
+        Some("Cannot disable inCommitTimestamp on a catalogManaged table")),
+    CatalogManagedTestCase(
+      testName = "NO-OP UPDATE: catalogManaged not being enabled should not affect ICT",
       operationType = "UPDATE",
       initialTableProperties = Map.empty,
       transactionProperties = Map(),
       expectedSuccess = true,
       expectedIctEnabled = false,
       expectedCatalogManagedSupported = false),
-    CatalogManagedEnablementTestCase(
-      testName = "ILLEGAL REPLACE: catalogManaged enablement flag => THROW",
+
+    // ===== REPLACE cases =====
+    CatalogManagedTestCase(
+      testName = "REPLACE: normal replace should succeed on a catalogManaged table",
+      operationType = "REPLACE",
+      initialTableProperties = Map("delta.feature.catalogOwned-preview" -> "supported"),
+      transactionProperties = Map(),
+      expectedSuccess = true,
+      expectedIctEnabled = true,
+      expectedCatalogManagedSupported = true),
+    CatalogManagedTestCase(
+      testName = "ILLEGAL REPLACE: set catalogManaged=supported => THROW",
       operationType = "REPLACE",
       initialTableProperties = Map.empty,
       transactionProperties = Map("delta.feature.catalogOwned-preview" -> "supported"),
       expectedSuccess = false,
       expectedExceptionMessage =
         Some("Cannot enable the catalogManaged feature during a REPLACE command.")),
-    CatalogManagedEnablementTestCase(
-      testName = "REPLACE: should succeed on a catalogManaged table",
+    CatalogManagedTestCase(
+      testName = "ILLEGAL REPLACE: catalogManaged already supported, then disable ICT => THROW",
       operationType = "REPLACE",
       initialTableProperties = Map("delta.feature.catalogOwned-preview" -> "supported"),
-      transactionProperties = Map(),
-      expectedSuccess = true,
-      expectedIctEnabled = true,
-      expectedCatalogManagedSupported = true))
+      transactionProperties = Map("delta.enableInCommitTimestamps" -> "false"),
+      expectedSuccess = false,
+      expectedExceptionMessage =
+        Some("Cannot disable inCommitTimestamp on a catalogManaged table")))
 
   catalogManagedTestCases.foreach { testCase =>
     test(testCase.testName) {
@@ -199,7 +218,7 @@ class CatalogManagedEnablementSuite extends AnyFunSuite with TestUtils {
         } else {
           // Transaction building should fail
           val exception = intercept[Exception] {
-            txnBuilder.build(defaultEngine).commit(defaultEngine, emptyIterable[Row])
+            txnBuilder.build(defaultEngine)
           }
 
           testCase.expectedExceptionMessage.foreach { expectedMsg =>