respond to pr comments

Author: scottsand-db

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

@@ -39,8 +39,12 @@ public class ReplaceTableTransactionBuilderV2Impl implements ReplaceTableTransac
       new HashSet<String>() {
         {
           add(TableConfig.COLUMN_MAPPING_MAX_COLUMN_ID.getKey());
+
+          // Must retail all ICT properties, else a client would not know when ICT was enabled,
+          // which could result in a failed query or incorrect results.
           add(TableConfig.IN_COMMIT_TIMESTAMPS_ENABLED.getKey());
-          // TODO are there any other table properties we should preserve?
+          add(TableConfig.IN_COMMIT_TIMESTAMP_ENABLEMENT_VERSION.getKey());
+          add(TableConfig.IN_COMMIT_TIMESTAMP_ENABLEMENT_TIMESTAMP.getKey());
         }
       };
 

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

@@ -661,11 +661,11 @@ private static void validateSchemaAndPartColsCreateOrReplace(
   }
 
   private static void validateNotEnablingCatalogManagedOnReplace(
-      Map<String, String> tableProperties) {
+      Map<String, String> userInputTableProperties) {
     if (TableFeatures.isPropertiesManuallySupportingTableFeature(
-        tableProperties, TableFeatures.CATALOG_MANAGED_R_W_FEATURE_PREVIEW)) {
+        userInputTableProperties, TableFeatures.CATALOG_MANAGED_R_W_FEATURE_PREVIEW)) {
       throw new UnsupportedOperationException(
-          "Cannot enable the catalogManaged feature during a REPLACE operation.");
+          "Cannot enable the catalogManaged feature during a REPLACE command.");
     }
   }
 }

kernel/kernel-api/src/main/java/io/delta/kernel/internal/tablefeatures/TableFeatures.java

@@ -50,6 +50,14 @@ public class TableFeatures {
    */
   public static String SET_TABLE_FEATURE_SUPPORTED_PREFIX = "delta.feature.";
 
+  /**
+   * Configuration value to turn on support for a table feature when used with {@link
+   * #SET_TABLE_FEATURE_SUPPORTED_PREFIX}.
+   *
+   * <p>Example: {@code "delta.feature.myFeature" -> "supported"}
+   */
+  public static String SET_TABLE_FEATURE_SUPPORTED_VALUE = "supported";
+
   /////////////////////////////////////////////////////////////////////////////////
   /// START: Define the {@link TableFeature}s                                   ///
   /// If feature instance variable ends with                                    ///
@@ -615,7 +623,7 @@ public static boolean isPropertiesManuallySupportingTableFeature(
       Map<String, String> userProperties, TableFeature tableFeature) {
     final String featurePropKey = SET_TABLE_FEATURE_SUPPORTED_PREFIX + tableFeature.featureName();
     final String propertyValue = userProperties.get(featurePropKey); // will be null if not found
-    return "supported".equals(propertyValue);
+    return SET_TABLE_FEATURE_SUPPORTED_VALUE.equals(propertyValue);
   }
 
   /**
@@ -642,7 +650,7 @@ public static Tuple2<Set<TableFeature>, Optional<Metadata>> extractFeatureProper
 
         TableFeature feature = getTableFeature(featureName);
         features.add(feature);
-        if (!entry.getValue().equals("supported")) {
+        if (!entry.getValue().equals(SET_TABLE_FEATURE_SUPPORTED_VALUE)) {
           throw DeltaErrors.invalidConfigurationValueException(
               entry.getKey(),
               entry.getValue(),

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

@@ -139,9 +139,9 @@ class CatalogManagedEnablementSuite extends AnyFunSuite with TestUtils {
             .commit(defaultEngine, emptyIterable[Row])
         }
 
-        // CreateTableTransactionBuilder and UpdateTableTransactionBuilder don't share a common
-        // parent interface. So, we treat the `txnBuilder` as a trait that has a `build(engine)`
-        // method. Scalastyle doesn't like this, but it's valid.
+        // CREATE, UPDATE, and REPLACE txnBuilders don't share a common parent interface. So, we
+        // treat the `txnBuilder` as a trait that has a `build(engine)` method. Scalastyle doesn't
+        // like this, but it's valid.
         //
         // scalastyle:off
         val txnBuilder: { def build(engine: Engine): Transaction } = testCase.operationType match {
@@ -164,7 +164,7 @@ class CatalogManagedEnablementSuite extends AnyFunSuite with TestUtils {
 
             TableManager
               .loadSnapshot(tablePath)
-              .withCommitter(committer)
+              .withCommitter(committerUsingPutIfAbsent)
               .build(defaultEngine)
               .asInstanceOf[SnapshotImpl]
               .buildReplaceTableTransaction(replaceSchema, "engineInfo")