[Kernel] Integrate TableFeature framework in read and write path

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

@@ -149,42 +149,23 @@ public static KernelException unsupportedTableFeature(String feature) {
     return new KernelException(message);
   }
 
-  public static KernelException unsupportedReaderProtocol(
-      String tablePath, int tableReaderVersion) {
-    String message =
-        String.format(
-            "Unsupported Delta protocol reader version: table `%s` requires reader version %s "
-                + "which is unsupported by this version of Delta Kernel.",
-            tablePath, tableReaderVersion);
-    return new KernelException(message);
-  }
-
-  public static KernelException unsupportedReaderFeature(
-      String tablePath, Set<String> unsupportedFeatures) {
+  public static KernelException unsupportedReaderFeatures(
+      String tablePath, Set<String> readerFeatures) {
     String message =
         String.format(
             "Unsupported Delta reader features: table `%s` requires reader table features [%s] "
                 + "which is unsupported by this version of Delta Kernel.",
-            tablePath, String.join(", ", unsupportedFeatures));
-    return new KernelException(message);
-  }
-
-  public static KernelException unsupportedWriterProtocol(
-      String tablePath, int tableWriterVersion) {
-    String message =
-        String.format(
-            "Unsupported Delta protocol writer version: table `%s` requires writer version %s "
-                + "which is unsupported by this version of Delta Kernel.",
-            tablePath, tableWriterVersion);
+            tablePath, String.join(", ", readerFeatures));
     return new KernelException(message);
   }
 
-  public static KernelException unsupportedWriterFeature(String tablePath, String writerFeature) {
+  public static KernelException unsupportedWriterFeatures(
+      String tablePath, Set<String> writerFeatures) {
     String message =
         String.format(
             "Unsupported Delta writer feature: table `%s` requires writer table feature \"%s\" "
                 + "which is unsupported by this version of Delta Kernel.",
-            tablePath, writerFeature);
+            tablePath, writerFeatures);
     return new KernelException(message);
   }
 

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

@@ -197,7 +197,7 @@ public CloseableIterator<ColumnarBatch> getChanges(
               for (int rowId = 0; rowId < protocolVector.getSize(); rowId++) {
                 if (!protocolVector.isNullAt(rowId)) {
                   Protocol protocol = Protocol.fromColumnVector(protocolVector, rowId);
-                  TableFeatures.validateReadSupportedTable(protocol, getDataPath().toString());
+                  TableFeatures.validateKernelCanReadTheTable(protocol, getDataPath().toString());
                 }
               }
               if (shouldDropProtocolColumn) {

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

@@ -25,6 +25,7 @@
 import static io.delta.kernel.internal.util.VectorUtils.stringArrayValue;
 import static io.delta.kernel.internal.util.VectorUtils.stringStringMapValue;
 import static java.util.Objects.requireNonNull;
+import static java.util.stream.Collectors.toSet;
 
 import io.delta.kernel.*;
 import io.delta.kernel.engine.Engine;
@@ -36,6 +37,7 @@
 import io.delta.kernel.internal.replay.LogReplay;
 import io.delta.kernel.internal.snapshot.LogSegment;
 import io.delta.kernel.internal.snapshot.SnapshotHint;
+import io.delta.kernel.internal.tablefeatures.TableFeature;
 import io.delta.kernel.internal.tablefeatures.TableFeatures;
 import io.delta.kernel.internal.util.ColumnMapping;
 import io.delta.kernel.internal.util.ColumnMapping.ColumnMappingMode;
@@ -136,31 +138,27 @@ public Transaction build(Engine engine) {
     boolean shouldUpdateProtocol = false;
     Metadata metadata = snapshot.getMetadata();
     Protocol protocol = snapshot.getProtocol();
-    if (tableProperties.isPresent()) {
-      Map<String, String> validatedProperties =
-          TableConfig.validateDeltaProperties(tableProperties.get());
-      Map<String, String> newProperties =
-          metadata.filterOutUnchangedProperties(validatedProperties);
+    Map<String, String> validatedProperties =
+        TableConfig.validateDeltaProperties(tableProperties.orElse(Collections.emptyMap()));
+    Map<String, String> newProperties = metadata.filterOutUnchangedProperties(validatedProperties);
 
-      ColumnMapping.verifyColumnMappingChange(
-          metadata.getConfiguration(), newProperties, isNewTable);
+    if (!newProperties.isEmpty()) {
+      shouldUpdateMetadata = true;
+      metadata = metadata.withNewConfiguration(newProperties);
+    }
 
-      if (!newProperties.isEmpty()) {
-        shouldUpdateMetadata = true;
-        metadata = metadata.withNewConfiguration(newProperties);
-      }
+    ColumnMapping.verifyColumnMappingChange(metadata.getConfiguration(), newProperties, isNewTable);
 
-      Set<String> newWriterFeatures =
-          TableFeatures.extractAutomaticallyEnabledWriterFeatures(metadata, protocol);
-      if (!newWriterFeatures.isEmpty()) {
-        logger.info("Automatically enabling writer features: {}", newWriterFeatures);
-        shouldUpdateProtocol = true;
-        Set<String> oldWriterFeatures = protocol.getWriterFeatures();
-        protocol = protocol.withNewWriterFeatures(newWriterFeatures);
-        Set<String> curWriterFeatures = protocol.getWriterFeatures();
-        checkArgument(!Objects.equals(oldWriterFeatures, curWriterFeatures));
-        TableFeatures.validateWriteSupportedTable(protocol, metadata, table.getPath(engine));
-      }
+    Optional<Tuple2<Protocol, Set<TableFeature>>> newProtocolAndFeatures =
+        TableFeatures.autoUpgradeProtocolBasedOnMetadata(metadata, protocol);
+    if (newProtocolAndFeatures.isPresent()) {
+      logger.info(
+          "Automatically enabling table features: {}",
+          newProtocolAndFeatures.get()._2.stream().map(TableFeature::featureName).collect(toSet()));
+
+      shouldUpdateProtocol = true;
+      protocol = newProtocolAndFeatures.get()._1;
+      TableFeatures.validateKernelCanWriteToTable(protocol, metadata, table.getPath(engine));
     }
 
     return new TransactionImpl(
@@ -183,7 +181,7 @@ public Transaction build(Engine engine) {
   private void validate(Engine engine, SnapshotImpl snapshot, boolean isNewTable) {
     String tablePath = table.getPath(engine);
     // Validate the table has no features that Kernel doesn't yet support writing into it.
-    TableFeatures.validateWriteSupportedTable(
+    TableFeatures.validateKernelCanWriteToTable(
         snapshot.getProtocol(), snapshot.getMetadata(), tablePath);
 
     if (!isNewTable) {

kernel/kernel-api/src/main/java/io/delta/kernel/internal/actions/Protocol.java

@@ -431,23 +431,6 @@ public Protocol merge(Protocol... others) {
     return mergedProtocol.denormalizedNormalized();
   }
 
-  /////////////////////////////////////////////////////////////////////////////////////////////////
-  /// Legacy method which will be removed after the table feature integration is done           ///
-  /////////////////////////////////////////////////////////////////////////////////////////////////
-  public Protocol withNewWriterFeatures(Set<String> writerFeatures) {
-    Tuple2<Integer, Integer> newProtocolVersions =
-        TableFeatures.minProtocolVersionFromAutomaticallyEnabledFeatures(writerFeatures);
-    Set<String> newWriterFeatures = new HashSet<>(writerFeatures);
-    if (this.writerFeatures != null) {
-      newWriterFeatures.addAll(this.writerFeatures);
-    }
-    return new Protocol(
-        newProtocolVersions._1,
-        newProtocolVersions._2,
-        this.readerFeatures == null ? null : new HashSet<>(this.readerFeatures),
-        newWriterFeatures);
-  }
-
   /** Validate the protocol contents represents a valid state */
   protected void validate() {
     checkArgument(minReaderVersion >= 1, "minReaderVersion should be at least 1");

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checkpoints/Checkpointer.java

@@ -19,7 +19,6 @@
 import static io.delta.kernel.internal.TableConfig.EXPIRED_LOG_CLEANUP_ENABLED;
 import static io.delta.kernel.internal.TableConfig.LOG_RETENTION;
 import static io.delta.kernel.internal.snapshot.MetadataCleanup.cleanupExpiredLogs;
-import static io.delta.kernel.internal.tablefeatures.TableFeatures.validateWriteSupportedTable;
 import static io.delta.kernel.internal.util.Utils.singletonCloseableIterator;
 
 import io.delta.kernel.data.ColumnarBatch;
@@ -32,6 +31,7 @@
 import io.delta.kernel.internal.actions.Metadata;
 import io.delta.kernel.internal.fs.Path;
 import io.delta.kernel.internal.replay.CreateCheckpointIterator;
+import io.delta.kernel.internal.tablefeatures.TableFeatures;
 import io.delta.kernel.internal.util.*;
 import io.delta.kernel.utils.CloseableIterator;
 import io.delta.kernel.utils.FileStatus;
@@ -65,7 +65,7 @@ public static void checkpoint(Engine engine, Clock clock, SnapshotImpl snapshot)
     logger.info("{}: Starting checkpoint for version: {}", tablePath, version);
 
     // Check if writing to the given table protocol version/features is supported in Kernel
-    validateWriteSupportedTable(
+    TableFeatures.validateKernelCanWriteToTable(
         snapshot.getProtocol(), snapshot.getMetadata(), snapshot.getDataPath().toString());
 
     final Path checkpointPath = FileNames.checkpointFileSingular(logPath, version);

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/LogReplay.java

@@ -288,7 +288,7 @@ protected Tuple2<Protocol, Metadata> loadTableProtocolAndMetadata(
 
               if (protocol != null) {
                 // Stop since we have found the latest Protocol and Metadata.
-                TableFeatures.validateReadSupportedTable(protocol, dataPath.toString());
+                TableFeatures.validateKernelCanReadTheTable(protocol, dataPath.toString());
                 return new Tuple2<>(protocol, metadata);
               }
 

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

@@ -43,6 +43,9 @@
  * protocol but its metadata requirements are not satisfied, then clients still have to understand
  * the feature (at least to the extent that they can read and preserve the existing data in the
  * table that uses the feature).
+ *
+ * <p>Important note: uses the default implementation of `equals` and `hashCode` methods. We expect
+ * that the feature instances are singletons, so we don't need to compare the fields.
  */
 public abstract class TableFeature {
 
@@ -220,7 +223,4 @@ private void validate() {
       checkArgument(minReaderVersion() == 0, "Writer-only feature must have minReaderVersion=0");
     }
   }
-
-  // Important note: uses the default implementation of `equals` and `hashCode` methods.
-  // We expect that the feature instances are singletons, so we don't need to compare the fields.
 }