[Kernel] Integrate TableFeature framework in read and write path (#…

…4159)

## Description
This integrates the end-to-end TableFeatures framework in read and write
path.

It also adds support for appending into a table with `deletionVectors`,
`v2Checkpoint` and 'timestampNtz` are supported.

## How was this patch tested?
Existing and new tests. We need a few more integration tests, which I will follow up on in a separate PR.

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

@@ -140,51 +140,52 @@ public static KernelException invalidVersionRange(long startVersion, long endVer
   }
 
   /* ------------------------ PROTOCOL EXCEPTIONS ----------------------------- */
-  public static KernelException unsupportedTableFeature(String feature) {
+  public static KernelException unsupportedReaderProtocol(
+      String tablePath, int tableReaderVersion) {
     String message =
         String.format(
-            "Unsupported Delta table feature: table requires feature \"%s\" "
+            "Unsupported Delta protocol reader version: table `%s` requires reader version %s "
                 + "which is unsupported by this version of Delta Kernel.",
-            feature);
+            tablePath, tableReaderVersion);
     return new KernelException(message);
   }
 
-  public static KernelException unsupportedReaderProtocol(
-      String tablePath, int tableReaderVersion) {
+  public static KernelException unsupportedWriterProtocol(
+      String tablePath, int tableWriterVersion) {
     String message =
         String.format(
-            "Unsupported Delta protocol reader version: table `%s` requires reader version %s "
+            "Unsupported Delta protocol writer version: table `%s` requires writer version %s "
                 + "which is unsupported by this version of Delta Kernel.",
-            tablePath, tableReaderVersion);
+            tablePath, tableWriterVersion);
     return new KernelException(message);
   }
 
-  public static KernelException unsupportedReaderFeature(
-      String tablePath, Set<String> unsupportedFeatures) {
+  public static KernelException unsupportedTableFeature(String feature) {
     String message =
         String.format(
-            "Unsupported Delta reader features: table `%s` requires reader table features [%s] "
+            "Unsupported Delta table feature: table requires feature \"%s\" "
                 + "which is unsupported by this version of Delta Kernel.",
-            tablePath, String.join(", ", unsupportedFeatures));
+            feature);
     return new KernelException(message);
   }
 
-  public static KernelException unsupportedWriterProtocol(
-      String tablePath, int tableWriterVersion) {
+  public static KernelException unsupportedReaderFeatures(
+      String tablePath, Set<String> readerFeatures) {
     String message =
         String.format(
-            "Unsupported Delta protocol writer version: table `%s` requires writer version %s "
+            "Unsupported Delta reader features: table `%s` requires reader table features [%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

@@ -19,13 +19,13 @@
 import static io.delta.kernel.internal.DeltaErrors.tableAlreadyExists;
 import static io.delta.kernel.internal.TransactionImpl.DEFAULT_READ_VERSION;
 import static io.delta.kernel.internal.TransactionImpl.DEFAULT_WRITE_VERSION;
-import static io.delta.kernel.internal.tablefeatures.TableFeatures.DOMAIN_METADATA_W_FEATURE;
 import static io.delta.kernel.internal.util.ColumnMapping.isColumnMappingModeEnabled;
 import static io.delta.kernel.internal.util.Preconditions.checkArgument;
 import static io.delta.kernel.internal.util.SchemaUtils.casePreservingPartitionColNames;
 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;
@@ -38,6 +38,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;
@@ -165,46 +166,28 @@ 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);
-
-      ColumnMapping.verifyColumnMappingChange(
-          metadata.getConfiguration(), newProperties, isNewTable);
-
-      if (!newProperties.isEmpty()) {
-        shouldUpdateMetadata = true;
-        metadata = metadata.withNewConfiguration(newProperties);
-      }
+    Map<String, String> validatedProperties =
+        TableConfig.validateDeltaProperties(tableProperties.orElse(Collections.emptyMap()));
+    Map<String, String> newProperties = metadata.filterOutUnchangedProperties(validatedProperties);
 
-      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));
-      }
+    ColumnMapping.verifyColumnMappingChange(metadata.getConfiguration(), newProperties, isNewTable);
+
+    if (!newProperties.isEmpty()) {
+      shouldUpdateMetadata = true;
+      metadata = metadata.withNewConfiguration(newProperties);
     }
 
-    /* --------------- Domain Metadata Protocol upgrade if necessary------------ */
-    if (!TableFeatures.isDomainMetadataSupported(protocol)) {
-      if (!domainMetadatasAdded.isEmpty()) {
-        // This txn is setting a domain metadata, enable the feature in the protocol
-        logger.info(
-            "Automatically enabling writer feature: {}", DOMAIN_METADATA_W_FEATURE.featureName());
-        protocol =
-            protocol.withNewWriterFeatures(
-                Collections.singleton(DOMAIN_METADATA_W_FEATURE.featureName()));
-        shouldUpdateProtocol = true;
-      }
-      // If domainMetadatasRemoved is non-empty we do nothing. A DomainDoesNotExistException will be
-      // thrown in `getDomainMetadatasToCommit` since the domain cannot exist in the readSnapshot.
+    Optional<Tuple2<Protocol, Set<TableFeature>>> newProtocolAndFeatures =
+        TableFeatures.autoUpgradeProtocolBasedOnMetadata(
+            metadata, !domainMetadatasAdded.isEmpty(), 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(
@@ -228,7 +211,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

@@ -15,13 +15,15 @@
  */
 package io.delta.kernel.internal.actions;
 
+import static io.delta.kernel.internal.tablefeatures.TableFeatures.TABLE_FEATURES;
 import static io.delta.kernel.internal.tablefeatures.TableFeatures.TABLE_FEATURES_MIN_WRITER_VERSION;
 import static io.delta.kernel.internal.util.Preconditions.checkArgument;
 import static io.delta.kernel.internal.util.VectorUtils.stringArrayValue;
 import static java.lang.String.format;
 import static java.util.Collections.emptySet;
 import static java.util.Collections.unmodifiableSet;
 import static java.util.Objects.requireNonNull;
+import static java.util.stream.Collectors.toSet;
 
 import io.delta.kernel.data.*;
 import io.delta.kernel.internal.data.GenericRow;
@@ -181,7 +183,7 @@ public Set<TableFeature> getImplicitlySupportedFeatures() {
     if (supportsReaderFeatures && supportsWriterFeatures) {
       return emptySet();
     } else {
-      return TableFeatures.TABLE_FEATURES.stream()
+      return TABLE_FEATURES.stream()
           .filter(f -> !supportsReaderFeatures && f.minReaderVersion() <= minReaderVersion)
           .filter(f -> !supportsWriterFeatures && f.minWriterVersion() <= minWriterVersion)
           .collect(Collectors.toSet());
@@ -221,6 +223,25 @@ public Set<TableFeature> getImplicitlyAndExplicitlySupportedFeatures() {
     return supportedFeatures;
   }
 
+  /**
+   * Get the set of reader writer features that are both implicitly and explicitly supported by the
+   * protocol. Usually, the protocol has either implicit or explicit features, but not both. This
+   * API provides a way to get all enabled reader writer features. It doesn't return any writer only
+   * features.
+   */
+  public Set<TableFeature> getImplicitlyAndExplicitlySupportedReaderWriterFeatures() {
+    return Stream.concat(
+            // implicit supported features
+            TABLE_FEATURES.stream()
+                .filter(
+                    f ->
+                        !supportsReaderFeatures
+                            && f.minReaderVersion() <= this.getMinReaderVersion()),
+            // explicitly supported features
+            readerFeatures.stream().map(TableFeatures::getTableFeature))
+        .collect(toSet());
+  }
+
   /** Create a new {@link Protocol} object with the given {@link TableFeature} supported. */
   public Protocol withFeatures(Iterable<TableFeature> newFeatures) {
     Protocol result = this;
@@ -431,23 +452,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");
@@ -496,7 +500,7 @@ protected void validate() {
       // ensure we don't get (minReaderVersion, minWriterVersion) that satisfy the readerWriter
       // feature version requirements. E.g. (1, 5) is invalid as writer version indicates
       // columnMapping supported but reader version does not support it (requires 2).
-      TableFeatures.TABLE_FEATURES.stream()
+      TABLE_FEATURES.stream()
           .filter(TableFeature::isReaderWriterFeature)
           .forEach(
               f -> {

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.
 }