[Kernel] Handle the preview and graduated table features when updating protocol through metadata (#4539)

## Description
Example preview and graduated features are:
* (typeWidening-preview -> typeWidening)
* (variantType-preview -> variant)

Currently we enable both the preview and non-preview feature names if
the feature enable criteria is true.

For example if the delta.enableTypeWidening is set to true, the table
will have both typeWidening-preview and typeWidening

Expected behavior:
* if the table has no typeWidening or typeWidening-preview
  * enable only the typeWidening feature
* if the table has typeWidening-preview
  * keep it as is and don't add typeWidening
* If the table has typeWidening feature
  * keep it as is

Resolves #4532


## How was this patch tested?
Unit and integration tests. Integration tests only for typeWiderning as
Kernel doesn't yet support writes with variant.

Author: vkorukanti

kernel/kernel-api/src/main/java/io/delta/kernel/internal/icebergcompat/IcebergWriterCompatV1MetadataValidatorAndUpdater.java

@@ -171,7 +171,7 @@ public static Optional<Metadata> validateAndUpdateIcebergWriterCompatV1Metadata(
               CLUSTERING_W_FEATURE,
               TIMESTAMP_NTZ_RW_FEATURE,
               TYPE_WIDENING_RW_FEATURE,
-              TYPE_WIDENING_PREVIEW_TABLE_FEATURE)
+              TYPE_WIDENING_RW_PREVIEW_FEATURE)
           .collect(toSet());
 
   /** Checks that all features supported in the protocol are in {@link #ALLOWED_TABLE_FEATURES} */

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

@@ -183,28 +183,46 @@ public boolean metadataRequiresFeatureToBeEnabled(Protocol protocol, Metadata me
     }
   }
 
-  public static final TableFeature VARIANT_RW_FEATURE = new VariantTypeTableFeature("variantType");
-  public static final TableFeature VARIANT_RW_PREVIEW_FEATURE =
-      new VariantTypeTableFeature("variantType-preview");
-
-  private static class VariantTypeTableFeature extends TableFeature.ReaderWriterFeature
-      implements FeatureAutoEnabledByMetadata {
-    VariantTypeTableFeature(String featureName) {
-      super(
-          /* featureName = */ featureName, /* minReaderVersion = */ 3, /* minWriterVersion = */ 7);
+  /* ---- Start: variantType ---- */
+  // Base class for variantType and variantType-preview features. Both features are same in terms
+  // of behavior and given the feature is graduated, we will enable the `variantType` by default
+  // if the metadata requirements are satisfied and the table doesn't already contain the
+  // `variantType-preview` feature. Also to note, with this version of Kernel, one can't
+  // auto upgrade to `variantType-preview` with metadata requirements. It can only be set
+  // manually using `delta.feature.variantType-preview=supported` property.
+  private static class VariantTypeTableFeatureBase extends TableFeature.ReaderWriterFeature {
+    VariantTypeTableFeatureBase(String featureName) {
+      super(featureName, /* minReaderVersion = */ 3, /* minWriterVersion = */ 7);
     }
 
     @Override
-    public boolean metadataRequiresFeatureToBeEnabled(Protocol protocol, Metadata metadata) {
-      return hasTypeColumn(metadata.getSchema(), VARIANT);
+    public boolean hasKernelWriteSupport(Metadata metadata) {
+      return false; // TODO: yet to be implemented in Kernel
+    }
+  }
+
+  private static class VariantTypeTableFeature extends VariantTypeTableFeatureBase
+      implements FeatureAutoEnabledByMetadata {
+    VariantTypeTableFeature() {
+      super("variantType");
     }
 
     @Override
-    public boolean hasKernelWriteSupport(Metadata metadata) {
-      return false; // TODO: yet to be implemented in Kernel
+    public boolean metadataRequiresFeatureToBeEnabled(Protocol protocol, Metadata metadata) {
+      return hasTypeColumn(metadata.getSchema(), VARIANT)
+          &&
+          // Don't automatically enable the stable feature if the preview feature is
+          // already supported, to avoid possibly breaking old clients that only
+          // support the preview feature.
+          !protocol.supportsFeature(VARIANT_RW_PREVIEW_FEATURE);
     }
   }
 
+  public static final TableFeature VARIANT_RW_FEATURE = new VariantTypeTableFeature();
+  public static final TableFeature VARIANT_RW_PREVIEW_FEATURE =
+      new VariantTypeTableFeatureBase("variantType-preview");
+  /* ---- End: variantType ---- */
+
   public static final TableFeature DOMAIN_METADATA_W_FEATURE = new DomainMetadataFeature();
 
   private static class DomainMetadataFeature extends TableFeature.WriterFeature {
@@ -286,28 +304,42 @@ public boolean metadataRequiresFeatureToBeEnabled(Protocol protocol, Metadata me
     }
   }
 
-  public static final TableFeature TYPE_WIDENING_RW_FEATURE =
-      new TypeWideningTableFeature("typeWidening");
-  public static final TableFeature TYPE_WIDENING_PREVIEW_TABLE_FEATURE =
-      new TypeWideningTableFeature("typeWidening-preview");
-
-  private static class TypeWideningTableFeature extends TableFeature.ReaderWriterFeature
-      implements FeatureAutoEnabledByMetadata {
-    TypeWideningTableFeature(String featureName) {
+  /* ---- Start: type widening ---- */
+  // Base class for typeWidening and typeWidening-preview features. Both features are same in terms
+  // of behavior and given the feature is graduated, we will enable the `typeWidening` by default
+  // if the metadata requirements are satisfied and the table doesn't already contain the
+  // `typeWidening-preview` feature. Also to note, with this version of Kernel, one can't
+  // auto upgrade to `typeWidening-preview` with metadata requirements. It can only be set
+  // manually using `delta.feature.typeWidening-preview=supported` property.
+  private static class TypeWideningTableFeatureBase extends TableFeature.ReaderWriterFeature {
+    TypeWideningTableFeatureBase(String featureName) {
       super(featureName, /* minReaderVersion = */ 3, /* minWriterVersion = */ 7);
     }
+  }
 
-    @Override
-    public boolean metadataRequiresFeatureToBeEnabled(Protocol protocol, Metadata metadata) {
-      return TableConfig.TYPE_WIDENING_ENABLED.fromMetadata(metadata);
+  private static class TypeWideningTableFeature extends TypeWideningTableFeatureBase
+      implements FeatureAutoEnabledByMetadata {
+    TypeWideningTableFeature() {
+      super("typeWidening");
     }
 
     @Override
-    public boolean hasKernelWriteSupport(Metadata metadata) {
-      return true;
+    public boolean metadataRequiresFeatureToBeEnabled(Protocol protocol, Metadata metadata) {
+      return TableConfig.TYPE_WIDENING_ENABLED.fromMetadata(metadata)
+          &&
+          // Don't automatically enable the stable feature if the preview feature is already
+          // supported, to
+          // avoid possibly breaking old clients that only support the preview feature.
+          !protocol.supportsFeature(TYPE_WIDENING_RW_PREVIEW_FEATURE);
     }
   }
 
+  public static final TableFeature TYPE_WIDENING_RW_FEATURE = new TypeWideningTableFeature();
+
+  public static final TableFeature TYPE_WIDENING_RW_PREVIEW_FEATURE =
+      new TypeWideningTableFeatureBase("typeWidening-preview");
+  /* ---- End: type widening ---- */
+
   public static final TableFeature IN_COMMIT_TIMESTAMP_W_FEATURE =
       new InCommitTimestampTableFeature();
 
@@ -414,7 +446,7 @@ public boolean metadataRequiresFeatureToBeEnabled(Protocol protocol, Metadata me
               INVARIANTS_W_FEATURE,
               ROW_TRACKING_W_FEATURE,
               TIMESTAMP_NTZ_RW_FEATURE,
-              TYPE_WIDENING_PREVIEW_TABLE_FEATURE,
+              TYPE_WIDENING_RW_PREVIEW_FEATURE,
               TYPE_WIDENING_RW_FEATURE,
               VACUUM_PROTOCOL_CHECK_RW_FEATURE,
               VARIANT_RW_FEATURE,

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/icebergcompat/IcebergCompatV2MetadataValidatorAndUpdaterSuite.scala

@@ -21,7 +21,7 @@ import io.delta.kernel.exceptions.KernelException
 import io.delta.kernel.internal.actions.{Metadata, Protocol}
 import io.delta.kernel.internal.icebergcompat.IcebergCompatV2MetadataValidatorAndUpdater.validateAndUpdateIcebergCompatV2Metadata
 import io.delta.kernel.internal.tablefeatures.TableFeature
-import io.delta.kernel.internal.tablefeatures.TableFeatures.{COLUMN_MAPPING_RW_FEATURE, DELETION_VECTORS_RW_FEATURE, ICEBERG_COMPAT_V2_W_FEATURE, TYPE_WIDENING_PREVIEW_TABLE_FEATURE, TYPE_WIDENING_RW_FEATURE}
+import io.delta.kernel.internal.tablefeatures.TableFeatures.{COLUMN_MAPPING_RW_FEATURE, DELETION_VECTORS_RW_FEATURE, ICEBERG_COMPAT_V2_W_FEATURE, TYPE_WIDENING_RW_FEATURE, TYPE_WIDENING_RW_PREVIEW_FEATURE}
 import io.delta.kernel.internal.util.ColumnMappingSuiteBase
 import io.delta.kernel.test.VectorTestUtils
 import io.delta.kernel.types._

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/tablefeatures/TableFeaturesSuite.scala

@@ -15,7 +15,6 @@
  */
 package io.delta.kernel.internal.tablefeatures
 
-import java.util
 import java.util.{Collections, Optional}
 import java.util.Collections.{emptySet, singleton}
 import java.util.stream.Collectors.toList
@@ -25,7 +24,7 @@ import scala.collection.JavaConverters._
 import io.delta.kernel.data.{ArrayValue, ColumnVector, MapValue}
 import io.delta.kernel.exceptions.KernelException
 import io.delta.kernel.internal.actions.{Format, Metadata, Protocol}
-import io.delta.kernel.internal.tablefeatures.TableFeatures.{validateKernelCanReadTheTable, validateKernelCanWriteToTable, CLUSTERING_W_FEATURE, DOMAIN_METADATA_W_FEATURE, TABLE_FEATURES}
+import io.delta.kernel.internal.tablefeatures.TableFeatures.{validateKernelCanReadTheTable, validateKernelCanWriteToTable, CLUSTERING_W_FEATURE, DOMAIN_METADATA_W_FEATURE, TABLE_FEATURES, TYPE_WIDENING_RW_FEATURE}
 import io.delta.kernel.internal.util.InternalUtils.singletonStringColumnVector
 import io.delta.kernel.internal.util.VectorUtils.buildColumnVector
 import io.delta.kernel.types._
@@ -119,16 +118,10 @@ class TableFeaturesSuite extends AnyFunSuite {
     ("identityColumns", testMetadata(includeIdentityColumn = false), false),
     ("columnMapping", testMetadata(tblProps = Map("delta.columnMapping.mode" -> "id")), true),
     ("columnMapping", testMetadata(tblProps = Map("delta.columnMapping.mode" -> "none")), false),
-    (
-      "typeWidening-preview",
-      testMetadata(tblProps = Map("delta.enableTypeWidening" -> "true")),
-      true),
-    (
-      "typeWidening-preview",
-      testMetadata(tblProps = Map("delta.enableTypeWidening" -> "false")),
-      false),
     ("typeWidening", testMetadata(tblProps = Map("delta.enableTypeWidening" -> "true")), true),
     ("typeWidening", testMetadata(tblProps = Map("delta.enableTypeWidening" -> "false")), false),
+    ("variantType", testMetadata(includeVariantTypeCol = true), true),
+    ("variantType", testMetadata(includeVariantTypeCol = false), false),
     // Disable this until we have support to enable row tracking through metadata
     // ("rowTracking", testMetadata(tblProps = Map("delta.enableRowTracking" -> "true")), true),
     ("rowTracking", testMetadata(tblProps = Map("delta.enableRowTracking" -> "false")), false),
@@ -183,7 +176,24 @@ class TableFeaturesSuite extends AnyFunSuite {
     }
   }
 
-  test("row tracking enable throguh metadata property is not supported") {
+  Seq(
+    ("variantType", testMetadata(includeVariantTypeCol = true)),
+    ("typeWidening", testMetadata(tblProps = Map("delta.enableTypeWidening" -> "true")))).foreach {
+    case (feature, metadataEnablingFeature) =>
+      test("special handling of tables containing preview features: " + feature) {
+        val protocolWithPreviewFeature = new Protocol(3, 7)
+          .withFeature(TableFeatures.getTableFeature(s"$feature-preview"))
+
+        val enable = TableFeatures.getTableFeature(feature)
+          .asInstanceOf[FeatureAutoEnabledByMetadata]
+          .metadataRequiresFeatureToBeEnabled(
+            protocolWithPreviewFeature,
+            metadataEnablingFeature)
+        assert(!enable, "shouldn't enable non-preview feature")
+      }
+  }
+
+  test("row tracking enable through metadata property is not supported") {
     val tableFeature = TableFeatures.getTableFeature("rowTracking")
     val ex = intercept[UnsupportedOperationException] {
       tableFeature.asInstanceOf[FeatureAutoEnabledByMetadata]

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableFeaturesSuite.scala

@@ -317,6 +317,68 @@ class DeltaTableFeaturesSuite extends DeltaTableWriteSuiteBase {
     }
   }
 
+  /* ---- Start: type widening tests ---- */
+  test("only typeWidening feature is enabled when metadata supports it: new table") {
+    withTempDirAndEngine { (tablePath, engine) =>
+      createEmptyTable(
+        engine,
+        tablePath = tablePath,
+        schema = testSchema,
+        tableProperties = Map("delta.enableTypeWidening" -> "true"))
+
+      val protocolV0 = getProtocol(engine, tablePath)
+      assert(!protocolV0.supportsFeature(TableFeatures.TYPE_WIDENING_RW_PREVIEW_FEATURE))
+      assert(protocolV0.supportsFeature(TableFeatures.TYPE_WIDENING_RW_FEATURE))
+
+      // try enabling type widening again and expect no change in protocol
+      updateTableMetadata(
+        engine = engine,
+        tablePath = tablePath,
+        tableProperties = Map("delta.enableTypeWidening" -> "true"))
+      val protocolV1 = getProtocol(engine, tablePath)
+      assert(protocolV1 === protocolV0)
+    }
+  }
+
+  test("only typeWidening feature is enabled when new metadata supports it: existing table") {
+    withTempDirAndEngine { (tablePath, engine) =>
+      createEmptyTable(engine, tablePath = tablePath, schema = testSchema)
+      val protocolV0 = getProtocol(engine, tablePath)
+      assert(!protocolV0.supportsFeature(TableFeatures.TYPE_WIDENING_RW_PREVIEW_FEATURE))
+      assert(!protocolV0.supportsFeature(TableFeatures.TYPE_WIDENING_RW_FEATURE))
+
+      // try enabling type widening  and expect change in protocol
+      updateTableMetadata(
+        engine = engine,
+        tablePath = tablePath,
+        tableProperties = Map("delta.enableTypeWidening" -> "true"))
+      val protocolV1 = getProtocol(engine, tablePath)
+      assert(!protocolV1.supportsFeature(TableFeatures.TYPE_WIDENING_RW_PREVIEW_FEATURE))
+      assert(protocolV1.supportsFeature(TableFeatures.TYPE_WIDENING_RW_FEATURE))
+    }
+  }
+
+  test("typeWidening-preview in existing table is respected") {
+    withTempDirAndEngine { (tablePath, engine) =>
+      spark.sql(s"CREATE TABLE delta.`$tablePath`(id INT) USING delta " +
+        s"TBLPROPERTIES ('delta.feature.typeWidening-preview' = 'supported')")
+
+      val protocolV0 = getProtocol(engine, tablePath)
+      require(protocolV0.supportsFeature(TableFeatures.TYPE_WIDENING_RW_PREVIEW_FEATURE))
+      require(!protocolV0.supportsFeature(TableFeatures.TYPE_WIDENING_RW_FEATURE))
+
+      // now through Kernel type enabling the type widening through table property
+      updateTableMetadata(
+        engine = engine,
+        tablePath = tablePath,
+        tableProperties = Map("delta.enableTypeWidening" -> "true"))
+      val protocolV1 = getProtocol(engine, tablePath)
+      assert(protocolV1.supportsFeature(TableFeatures.TYPE_WIDENING_RW_PREVIEW_FEATURE))
+      assert(!protocolV1.supportsFeature(TableFeatures.TYPE_WIDENING_RW_FEATURE))
+    }
+  }
+  /* ---- End: type widening tests ---- */
+
   ///////////////////////////////////////////////////////////////////////////
   // Helper methods
   ///////////////////////////////////////////////////////////////////////////