[Kernel] Added Domain Metadata support to Delta Kernel

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

@@ -18,6 +18,7 @@
 import static java.lang.String.format;
 
 import io.delta.kernel.exceptions.*;
+import io.delta.kernel.internal.actions.DomainMetadata;
 import io.delta.kernel.types.DataType;
 import io.delta.kernel.types.StructType;
 import io.delta.kernel.utils.DataFileStatus;
@@ -274,6 +275,34 @@ public static KernelException invalidConfigurationValueException(
     return new InvalidConfigurationValueException(key, value, helpMessage);
   }
 
+  public static KernelException domainMetadataUnsupported() {
+    String message =
+        "Found DomainMetadata action(s) but table feature 'domainMetadata' "
+            + "is not supported on this table.";
+    return new KernelException(message);
+  }
+
+  public static KernelException duplicateDomainMetadataAction(
+      String domain, DomainMetadata action1, DomainMetadata action2) {
+    String message =
+        String.format(
+            "Multiple actions detected for domain '%s' in single transaction: '%s' and '%s'. "
+                + "Only one action per domain is allowed.",
+            domain, action1.toString(), action2.toString());
+    return new KernelException(message);
+  }
+
+  public static ConcurrentWriteException concurrentDomainMetadataAction(
+      DomainMetadata domainMetadataAttempt, DomainMetadata winningDomainMetadata) {
+    String message =
+        String.format(
+            "A concurrent writer added a domainMetadata action for the same domain: %s. "
+                + "No domain-specific conflict resolution is available for this domain. "
+                + "Attempted domainMetadata: %s. Winning domainMetadata: %s",
+            domainMetadataAttempt.getDomain(), domainMetadataAttempt, winningDomainMetadata);
+    return new ConcurrentWriteException(message);
+  }
+
   /* ------------------------ HELPER METHODS ----------------------------- */
   private static String formatTimestamp(long millisSinceEpochUTC) {
     return new Timestamp(millisSinceEpochUTC).toInstant().toString();

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

@@ -23,6 +23,7 @@
 import io.delta.kernel.engine.CommitCoordinatorClientHandler;
 import io.delta.kernel.engine.Engine;
 import io.delta.kernel.internal.actions.CommitInfo;
+import io.delta.kernel.internal.actions.DomainMetadata;
 import io.delta.kernel.internal.actions.Metadata;
 import io.delta.kernel.internal.actions.Protocol;
 import io.delta.kernel.internal.fs.Path;
@@ -31,6 +32,7 @@
 import io.delta.kernel.internal.snapshot.LogSegment;
 import io.delta.kernel.internal.snapshot.TableCommitCoordinatorClientHandler;
 import io.delta.kernel.types.StructType;
+import java.util.Map;
 import java.util.Optional;
 
 /** Implementation of {@link Snapshot}. */
@@ -83,6 +85,17 @@ public Protocol getProtocol() {
     return protocol;
   }
 
+  /**
+   * Get the domain metadata map from the log replay, which lazily loads and replays a history of
+   * domain metadata actions, resolving them to produce the current state of the domain metadata.
+   *
+   * @return A map where the keys are domain names and the values are {@link DomainMetadata}
+   *     objects.
+   */
+  public Map<String, DomainMetadata> getDomainMetadataMap() {
+    return logReplay.getDomainMetadataMap();
+  }
+
   public CreateCheckpointIterator getCreateCheckpointIterator(Engine engine) {
     long minFileRetentionTimestampMillis =
         System.currentTimeMillis() - TOMBSTONE_RETENTION.fromMetadata(metadata);

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

@@ -39,6 +39,7 @@ public class TableFeatures {
               add("columnMapping");
               add("typeWidening-preview");
               add("typeWidening");
+              add("domainMetadata");
             }
           });
 
@@ -57,6 +58,12 @@ public class TableFeatures {
             }
           });
 
+  /** The feature name for domain metadata. */
+  public static final String DOMAIN_METADATA_FEATURE_NAME = "domainMetadata";
+
+  /** The minimum writer version required to support domain metadata. */
+  public static final int DOMAIN_METADATA_MIN_WRITER_VERSION_REQUIRED = 7;
+
   ////////////////////
   // Helper Methods //
   ////////////////////
@@ -93,7 +100,7 @@ public static void validateReadSupportedTable(
    *   <li>protocol writer version 1.
    *   <li>protocol writer version 2 only with appendOnly feature enabled.
    *   <li>protocol writer version 7 with {@code appendOnly}, {@code inCommitTimestamp}, {@code
-   *       columnMapping}, {@code typeWidening} feature enabled.
+   *       columnMapping}, {@code typeWidening}, {@code domainMetadata} feature enabled.
    * </ul>
    *
    * @param protocol Table protocol
@@ -125,20 +132,8 @@ public static void validateWriteSupportedTable(
         throw unsupportedWriterProtocol(tablePath, minWriterVersion);
       case 7:
         for (String writerFeature : protocol.getWriterFeatures()) {
-          switch (writerFeature) {
-              // Only supported writer features as of today in Kernel
-            case "appendOnly":
-              break;
-            case "inCommitTimestamp":
-              break;
-            case "columnMapping":
-              break;
-            case "typeWidening-preview":
-              break;
-            case "typeWidening":
-              break;
-            default:
-              throw unsupportedWriterFeature(tablePath, writerFeature);
+          if (!SUPPORTED_WRITER_FEATURES.contains(writerFeature)) {
+            throw unsupportedWriterFeature(tablePath, writerFeature);
           }
         }
         break;

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

@@ -54,6 +54,7 @@ public class TransactionBuilderImpl implements TransactionBuilder {
   private Optional<List<String>> partitionColumns = Optional.empty();
   private Optional<SetTransaction> setTxnOpt = Optional.empty();
   private Optional<Map<String, String>> tableProperties = Optional.empty();
+  private List<DomainMetadata> domainMetadatas = new ArrayList<>();
 
   public TransactionBuilderImpl(TableImpl table, String engineInfo, Operation operation) {
     this.table = table;
@@ -93,6 +94,16 @@ public TransactionBuilder withTableProperties(Engine engine, Map<String, String>
     return this;
   }
 
+  /**
+   * Internal API to set the domain metadata for the transaction. Visible for testing.
+   *
+   * @param domainMetadatas List of domain metadata to be added to the transaction.
+   */
+  public TransactionBuilder withDomainMetadatas(List<DomainMetadata> domainMetadatas) {
+    this.domainMetadatas = new ArrayList<>(domainMetadatas);
+    return this;
+  }
+
   @Override
   public Transaction build(Engine engine) {
     SnapshotImpl snapshot;
@@ -156,7 +167,8 @@ public Transaction build(Engine engine) {
         setTxnOpt,
         shouldUpdateMetadata,
         shouldUpdateProtocol,
-        table.getClock());
+        table.getClock(),
+        domainMetadatas);
   }
 
   /** Validate the given parameters for the transaction. */

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

@@ -32,11 +32,7 @@
 import io.delta.kernel.internal.fs.Path;
 import io.delta.kernel.internal.replay.ConflictChecker;
 import io.delta.kernel.internal.replay.ConflictChecker.TransactionRebaseState;
-import io.delta.kernel.internal.util.Clock;
-import io.delta.kernel.internal.util.ColumnMapping;
-import io.delta.kernel.internal.util.FileNames;
-import io.delta.kernel.internal.util.InCommitTimestampUtils;
-import io.delta.kernel.internal.util.VectorUtils;
+import io.delta.kernel.internal.util.*;
 import io.delta.kernel.types.StructType;
 import io.delta.kernel.utils.CloseableIterable;
 import io.delta.kernel.utils.CloseableIterator;
@@ -73,6 +69,7 @@ public class TransactionImpl implements Transaction {
   private final Optional<SetTransaction> setTxnOpt;
   private final boolean shouldUpdateProtocol;
   private final Clock clock;
+  private final List<DomainMetadata> domainMetadatas;
   private Metadata metadata;
   private boolean shouldUpdateMetadata;
 
@@ -90,7 +87,8 @@ public TransactionImpl(
       Optional<SetTransaction> setTxnOpt,
       boolean shouldUpdateMetadata,
       boolean shouldUpdateProtocol,
-      Clock clock) {
+      Clock clock,
+      List<DomainMetadata> domainMetadatas) {
     this.isNewTable = isNewTable;
     this.dataPath = dataPath;
     this.logPath = logPath;
@@ -103,6 +101,7 @@ public TransactionImpl(
     this.shouldUpdateMetadata = shouldUpdateMetadata;
     this.shouldUpdateProtocol = shouldUpdateProtocol;
     this.clock = clock;
+    this.domainMetadatas = domainMetadatas;
   }
 
   @Override
@@ -221,6 +220,12 @@ private TransactionCommitResult doCommit(
     }
     setTxnOpt.ifPresent(setTxn -> metadataActions.add(createTxnSingleAction(setTxn.toRow())));
 
+    // Check for duplicate domain metadata and if the protocol supports
+    DomainMetadataUtils.validateDomainMetadatas(domainMetadatas, protocol);
+
+    domainMetadatas.forEach(
+        dm -> metadataActions.add(createDomainMetadataSingleAction(dm.toRow())));
+
     try (CloseableIterator<Row> stageDataIter = dataActions.iterator()) {
       // Create a new CloseableIterator that will return the metadata actions followed by the
       // data actions.
@@ -269,6 +274,10 @@ public Optional<SetTransaction> getSetTxnOpt() {
     return setTxnOpt;
   }
 
+  public List<DomainMetadata> getDomainMetadatas() {
+    return domainMetadatas;
+  }
+
   /**
    * Generates a timestamp which is greater than the commit timestamp of the readSnapshot. This can
    * result in an additional file read and that this will only happen if ICT is enabled.

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

@@ -0,0 +1,127 @@
+/*
+ * Copyright (2024) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.delta.kernel.internal.actions;
+
+import static io.delta.kernel.internal.util.InternalUtils.requireNonNull;
+import static java.util.Objects.requireNonNull;
+
+import io.delta.kernel.data.ColumnVector;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.internal.data.GenericRow;
+import io.delta.kernel.types.BooleanType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructType;
+import java.util.HashMap;
+import java.util.Map;
+
+/** Delta log action representing an `DomainMetadata` action */
+public class DomainMetadata {
+  /** Full schema of the {@link DomainMetadata} action in the Delta Log. */
+  public static final StructType FULL_SCHEMA =
+      new StructType()
+          .add("domain", StringType.STRING, false /* nullable */)
+          .add("configuration", StringType.STRING, false /* nullable */)
+          .add("removed", BooleanType.BOOLEAN, false /* nullable */);
+
+  public static DomainMetadata fromColumnVector(ColumnVector vector, int rowId) {
+    if (vector.isNullAt(rowId)) {
+      return null;
+    }
+    return new DomainMetadata(
+        requireNonNull(vector.getChild(0), rowId, "domain").getString(rowId),
+        requireNonNull(vector.getChild(1), rowId, "configuration").getString(rowId),
+        requireNonNull(vector.getChild(2), rowId, "removed").getBoolean(rowId));
+  }
+
+  public static DomainMetadata fromRow(Row row) {
+    if (row == null) {
+      return null;
+    }
+    assert (row.getSchema().equals(FULL_SCHEMA));
+    return new DomainMetadata(
+        requireNonNull(row, 0, "domain").getString(0),
+        requireNonNull(row, 1, "configuration").getString(1),
+        requireNonNull(row, 2, "removed").getBoolean(2));
+  }
+
+  private final String domain;
+  private final String configuration;
+  private final boolean removed;
+
+  /**
+   * The domain metadata action contains a configuration string for a named metadata domain. Two
+   * overlapping transactions conflict if they both contain a domain metadata action for the same
+   * metadata domain. Per-domain conflict resolution logic can be implemented.
+   *
+   * @param domain A string used to identify a specific domain.
+   * @param configuration A string containing configuration for the metadata domain.
+   * @param removed If it is true it serves as a tombstone to logically delete a {@link
+   *     DomainMetadata} action.
+   */
+  public DomainMetadata(String domain, String configuration, boolean removed) {
+    this.domain = requireNonNull(domain, "domain is null");
+    this.configuration = requireNonNull(configuration, "configuration is null");
+    this.removed = removed;
+  }
+
+  public String getDomain() {
+    return domain;
+  }
+
+  public String getConfiguration() {
+    return configuration;
+  }
+
+  public boolean isRemoved() {
+    return removed;
+  }
+
+  /**
+   * Encode as a {@link Row} object with the schema {@link DomainMetadata#FULL_SCHEMA}.
+   *
+   * @return {@link Row} object with the schema {@link DomainMetadata#FULL_SCHEMA}
+   */
+  public Row toRow() {
+    Map<Integer, Object> domainMetadataMap = new HashMap<>();
+    domainMetadataMap.put(0, domain);
+    domainMetadataMap.put(1, configuration);
+    domainMetadataMap.put(2, removed);
+
+    return new GenericRow(DomainMetadata.FULL_SCHEMA, domainMetadataMap);
+  }
+
+  @Override
+  public String toString() {
+    return String.format(
+        "DomainMetadata{domain='%s', configuration='%s', removed='%s'}",
+        domain, configuration, removed);
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (this == obj) return true;
+    if (obj == null || getClass() != obj.getClass()) return false;
+    DomainMetadata that = (DomainMetadata) obj;
+    return removed == that.removed
+        && domain.equals(that.domain)
+        && configuration.equals(that.configuration);
+  }
+
+  @Override
+  public int hashCode() {
+    return java.util.Objects.hash(domain, configuration, removed);
+  }
+}