[Kernel] [CC Refactor] Make Protocol implement AbstractProtocol

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

@@ -18,12 +18,14 @@
 import static java.lang.String.format;
 
 import io.delta.kernel.exceptions.*;
+import io.delta.kernel.internal.TableFeatures.TableFeatureType;
 import io.delta.kernel.types.DataType;
 import io.delta.kernel.types.StructType;
 import io.delta.kernel.utils.DataFileStatus;
 import java.io.IOException;
 import java.sql.Timestamp;
 import java.util.List;
+import java.util.Locale;
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Supplier;
@@ -138,7 +140,39 @@ public static KernelException invalidVersionRange(long startVersion, long endVer
     return new KernelException(message);
   }
 
-  /* ------------------------ PROTOCOL EXCEPTIONS ----------------------------- */
+  /* ------------------------ PROTOCOL EXCEPTIONS - START ----------------------------- */
+
+  public static KernelException mismatchedProtocolVersionFeatureSet(
+      TableFeatureType featureType,
+      int tableFeatureVersion,
+      int minRequiredVersion,
+      Set<String> tableFeatures) {
+    final String featureTypeStr = featureType.toString().toLowerCase(Locale.ROOT);
+    final String message =
+        String.format(
+            "Found %s features %s in %s protocol version %s but these are only supported in "
+                + "%s protocol version %s and above.",
+            featureTypeStr,
+            tableFeatures,
+            featureTypeStr,
+            tableFeatureVersion,
+            featureTypeStr,
+            minRequiredVersion);
+    return new KernelException(message);
+  }
+
+  public static KernelException tableFeatureReadRequiresWrite(
+      int tableReaderVersion, int tableWriterVersion) {
+    final String message =
+        String.format(
+            "Table reader features are supported (current reader protocol version is %s) yet "
+                + "table writer features are not (current writer protocol version is %s). Writer "
+                + "protocol version must be at least %s to proceed.",
+            tableReaderVersion,
+            tableWriterVersion,
+            TableFeatures.TABLE_FEATURES_MIN_WRITER_VERSION);
+    return new KernelException(message);
+  }
 
   public static KernelException unsupportedReaderProtocol(
       String tablePath, int tableReaderVersion) {
@@ -186,6 +220,8 @@ public static KernelException columnInvariantsNotSupported() {
     return new KernelException(message);
   }
 
+  /* ------------------------ PROTOCOL EXCEPTIONS - END ----------------------------- */
+
   public static KernelException unsupportedDataType(DataType dataType) {
     return new KernelException("Kernel doesn't support writing data of type: " + dataType);
   }

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

@@ -30,6 +30,17 @@
 /** Contains utility methods related to the Delta table feature support in protocol. */
 public class TableFeatures {
 
+  public enum TableFeatureType {
+    READER,
+    WRITER;
+  }
+
+  /** Min reader version that supports reader features. */
+  public static final int TABLE_FEATURES_MIN_READER_VERSION = 3;
+
+  /** Min writer version that supports writer features. */
+  public static final int TABLE_FEATURES_MIN_WRITER_VERSION = 7;
+
   private static final Set<String> SUPPORTED_WRITER_FEATURES =
       Collections.unmodifiableSet(
           new HashSet<String>() {
@@ -58,7 +69,7 @@ public class TableFeatures {
           });
 
   ////////////////////
-  // Helper Methods //
+  // Public Methods //
   ////////////////////
 
   public static void validateReadSupportedTable(
@@ -70,7 +81,7 @@ public static void validateReadSupportedTable(
         metadata.ifPresent(ColumnMapping::throwOnUnsupportedColumnMappingMode);
         break;
       case 3:
-        List<String> readerFeatures = protocol.getReaderFeatures();
+        final Set<String> readerFeatures = protocol.getReaderFeatures();
         if (!SUPPORTED_READER_FEATURES.containsAll(readerFeatures)) {
           Set<String> unsupportedFeatures = new HashSet<>(readerFeatures);
           unsupportedFeatures.removeAll(SUPPORTED_READER_FEATURES);
@@ -187,6 +198,10 @@ public static Set<String> extractAutomaticallyEnabledWriterFeatures(
         .collect(Collectors.toSet());
   }
 
+  ////////////////////
+  // Helper Methods //
+  ////////////////////
+
   /**
    * Get the minimum reader version required for a feature.
    *

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

@@ -135,9 +135,9 @@ public Transaction build(Engine engine) {
       if (!newWriterFeatures.isEmpty()) {
         logger.info("Automatically enabling writer features: {}", newWriterFeatures);
         shouldUpdateProtocol = true;
-        List<String> oldWriterFeatures = protocol.getWriterFeatures();
-        protocol = protocol.withNewWriterFeatures(newWriterFeatures);
-        List<String> curWriterFeatures = protocol.getWriterFeatures();
+        final Set<String> oldWriterFeatures = protocol.getWriterFeatures();
+        protocol = protocol.withAdditionalWriterFeatures(newWriterFeatures);
+        final Set<String> curWriterFeatures = protocol.getWriterFeatures();
         checkArgument(!Objects.equals(oldWriterFeatures, curWriterFeatures));
         TableFeatures.validateWriteSupportedTable(
             protocol, metadata, metadata.getSchema(), table.getPath(engine));
@@ -257,7 +257,7 @@ private Protocol getInitialProtocol() {
     return new Protocol(
         DEFAULT_READ_VERSION,
         DEFAULT_WRITE_VERSION,
-        null /* readerFeatures */,
-        null /* writerFeatures */);
+        Collections.emptySet() /* readerFeatures */,
+        Collections.emptySet() /* writerFeatures */);
   }
 }

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

@@ -16,9 +16,13 @@
 package io.delta.kernel.internal.actions;
 
 import static io.delta.kernel.internal.util.VectorUtils.stringArrayValue;
+import static java.util.Objects.requireNonNull;
 
 import io.delta.kernel.data.*;
+import io.delta.kernel.engine.coordinatedcommits.actions.AbstractProtocol;
+import io.delta.kernel.internal.DeltaErrors;
 import io.delta.kernel.internal.TableFeatures;
+import io.delta.kernel.internal.TableFeatures.TableFeatureType;
 import io.delta.kernel.internal.data.GenericRow;
 import io.delta.kernel.internal.util.Tuple2;
 import io.delta.kernel.internal.util.VectorUtils;
@@ -28,7 +32,22 @@
 import io.delta.kernel.types.StructType;
 import java.util.*;
 
-public class Protocol {
+public class Protocol implements AbstractProtocol {
+
+  ///////////////////
+  // Static Fields //
+  ///////////////////
+
+  public static final StructType FULL_SCHEMA =
+      new StructType()
+          .add("minReaderVersion", IntegerType.INTEGER, false /* nullable */)
+          .add("minWriterVersion", IntegerType.INTEGER, false /* nullable */)
+          .add("readerFeatures", new ArrayType(StringType.STRING, false /* contains null */))
+          .add("writerFeatures", new ArrayType(StringType.STRING, false /* contains null */));
+
+  ////////////////////
+  // Static Methods //
+  ////////////////////
 
   public static Protocol fromColumnVector(ColumnVector vector, int rowId) {
     if (vector.isNullAt(rowId)) {
@@ -39,34 +58,58 @@ public static Protocol fromColumnVector(ColumnVector vector, int rowId) {
         vector.getChild(0).getInt(rowId),
         vector.getChild(1).getInt(rowId),
         vector.getChild(2).isNullAt(rowId)
-            ? Collections.emptyList()
-            : VectorUtils.toJavaList(vector.getChild(2).getArray(rowId)),
+            ? Collections.emptySet()
+            : new HashSet<>(VectorUtils.toJavaList(vector.getChild(2).getArray(rowId))),
         vector.getChild(3).isNullAt(rowId)
-            ? Collections.emptyList()
-            : VectorUtils.toJavaList(vector.getChild(3).getArray(rowId)));
+            ? Collections.emptySet()
+            : new HashSet<>(VectorUtils.toJavaList(vector.getChild(3).getArray(rowId))));
   }
 
-  public static final StructType FULL_SCHEMA =
-      new StructType()
-          .add("minReaderVersion", IntegerType.INTEGER, false /* nullable */)
-          .add("minWriterVersion", IntegerType.INTEGER, false /* nullable */)
-          .add("readerFeatures", new ArrayType(StringType.STRING, false /* contains null */))
-          .add("writerFeatures", new ArrayType(StringType.STRING, false /* contains null */));
+  /////////////////////////////
+  // Member Fields / Methods //
+  /////////////////////////////
 
   private final int minReaderVersion;
   private final int minWriterVersion;
-  private final List<String> readerFeatures;
-  private final List<String> writerFeatures;
+  private final Set<String> readerFeatures;
+  private final Set<String> writerFeatures;
 
   public Protocol(
       int minReaderVersion,
       int minWriterVersion,
-      List<String> readerFeatures,
-      List<String> writerFeatures) {
+      Set<String> readerFeatures,
+      Set<String> writerFeatures) {
     this.minReaderVersion = minReaderVersion;
     this.minWriterVersion = minWriterVersion;
-    this.readerFeatures = readerFeatures;
-    this.writerFeatures = writerFeatures;
+    this.readerFeatures =
+        Collections.unmodifiableSet(requireNonNull(readerFeatures, "readerFeatures is null"));
+    this.writerFeatures =
+        Collections.unmodifiableSet(requireNonNull(writerFeatures, "writerFeatures is null"));
+
+    final boolean supportsReaderFeatures =
+        minReaderVersion >= TableFeatures.TABLE_FEATURES_MIN_READER_VERSION;
+    final boolean supportsWriterFeatures =
+        minWriterVersion >= TableFeatures.TABLE_FEATURES_MIN_WRITER_VERSION;
+
+    if (!supportsReaderFeatures && !readerFeatures.isEmpty()) {
+      throw DeltaErrors.mismatchedProtocolVersionFeatureSet(
+          TableFeatureType.READER,
+          minReaderVersion /* tableFeatureVersion */,
+          TableFeatures.TABLE_FEATURES_MIN_READER_VERSION /* minRequiredVersion */,
+          readerFeatures /* tableFeatures */);
+    }
+
+    if (!supportsWriterFeatures && !writerFeatures.isEmpty()) {
+      throw DeltaErrors.mismatchedProtocolVersionFeatureSet(
+          TableFeatureType.WRITER,
+          minWriterVersion /* tableFeatureVersion */,
+          TableFeatures.TABLE_FEATURES_MIN_WRITER_VERSION /* minRequiredVersion */,
+          writerFeatures /* tableFeatures */);
+    }
+
+    if (supportsReaderFeatures && !supportsWriterFeatures) {
+      throw DeltaErrors.tableFeatureReadRequiresWrite(minReaderVersion, minWriterVersion);
+    }
   }
 
   public int getMinReaderVersion() {
@@ -77,11 +120,11 @@ public int getMinWriterVersion() {
     return minWriterVersion;
   }
 
-  public List<String> getReaderFeatures() {
+  public Set<String> getReaderFeatures() {
     return readerFeatures;
   }
 
-  public List<String> getWriterFeatures() {
+  public Set<String> getWriterFeatures() {
     return writerFeatures;
   }
 
@@ -105,23 +148,42 @@ public Row toRow() {
     Map<Integer, Object> protocolMap = new HashMap<>();
     protocolMap.put(0, minReaderVersion);
     protocolMap.put(1, minWriterVersion);
-    protocolMap.put(2, stringArrayValue(readerFeatures));
-    protocolMap.put(3, stringArrayValue(writerFeatures));
+
+    // Note that we only write readerFeatures when the minReaderVersion is at least
+    // TableFeatures.TABLE_FEATURES_MIN_READER_VERSION, else we write null.
+    protocolMap.put(
+        2,
+        minReaderVersion < TableFeatures.TABLE_FEATURES_MIN_READER_VERSION
+            ? null
+            : stringArrayValue(new ArrayList<>(readerFeatures)));
+
+    // Note that we only write writerFeatures when the minWriterVersion is at least
+    // TableFeatures.TABLE_FEATURES_MIN_WRITER_VERSION, else we write null.
+    protocolMap.put(
+        3,
+        minWriterVersion < TableFeatures.TABLE_FEATURES_MIN_WRITER_VERSION
+            ? null
+            : stringArrayValue(new ArrayList<>(writerFeatures)));
 
     return new GenericRow(Protocol.FULL_SCHEMA, protocolMap);
   }
 
-  public Protocol withNewWriterFeatures(Set<String> writerFeatures) {
-    Tuple2<Integer, Integer> newProtocolVersions =
-        TableFeatures.minProtocolVersionFromAutomaticallyEnabledFeatures(writerFeatures);
-    List<String> newWriterFeatures = new ArrayList<>(writerFeatures);
-    if (this.writerFeatures != null) {
-      newWriterFeatures.addAll(this.writerFeatures);
+  public Protocol withAdditionalWriterFeatures(Set<String> additionalWriterFeatures) {
+    requireNonNull(additionalWriterFeatures, "additionalWriterFeatures is null");
+
+    Tuple2<Integer, Integer> protocolVersions =
+        TableFeatures.minProtocolVersionFromAutomaticallyEnabledFeatures(additionalWriterFeatures);
+
+    Set<String> combinedWriterFeatures = new HashSet<>(additionalWriterFeatures);
+
+    if (!writerFeatures.isEmpty()) {
+      combinedWriterFeatures.addAll(writerFeatures);
     }
+
     return new Protocol(
-        newProtocolVersions._1,
-        newProtocolVersions._2,
-        this.readerFeatures == null ? null : new ArrayList<>(this.readerFeatures),
-        newWriterFeatures);
+        protocolVersions._1,
+        protocolVersions._2,
+        readerFeatures.isEmpty() ? Collections.emptySet() : new HashSet<>(this.readerFeatures),
+        combinedWriterFeatures);
   }
 }