[Kernel] Added Domain Metadata support to Delta Kernel

connectors/golden-tables/src/main/resources/golden/kernel-domain-metadata/_delta_log/00000000000000000000.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1730671956424,"operation":"CREATE TABLE","operationParameters":{"partitionBy":"[]","clusterBy":"[]","description":null,"isManaged":"false","properties":"{\"delta.checkpointInterval\":\"3\"}"},"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{},"engineInfo":"Apache-Spark/3.5.3 Delta-Lake/3.3.0-SNAPSHOT","txnId":"90158bef-da23-4500-aafc-d2932e80f8cb"}}
+{"metaData":{"id":"04e4bf27-b577-4f7d-b002-08b3bbc00ce5","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{"delta.checkpointInterval":"3"},"createdTime":1730671956256}}
+{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}

connectors/golden-tables/src/main/resources/golden/kernel-domain-metadata/_delta_log/00000000000000000001.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1730671958509,"operation":"SET TBLPROPERTIES","operationParameters":{"properties":"{\"delta.feature.domainmetadata\":\"enabled\"}"},"readVersion":0,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{},"engineInfo":"Apache-Spark/3.5.3 Delta-Lake/3.3.0-SNAPSHOT","txnId":"4b1dc72c-432c-4753-b9d3-68ab89f3cb91"}}
+{"metaData":{"id":"04e4bf27-b577-4f7d-b002-08b3bbc00ce5","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{"delta.checkpointInterval":"3"},"createdTime":1730671956256}}
+{"protocol":{"minReaderVersion":1,"minWriterVersion":7,"writerFeatures":["domainMetadata","appendOnly","invariants"]}}

connectors/golden-tables/src/main/resources/golden/kernel-domain-metadata/_delta_log/00000000000000000002.json

@@ -0,0 +1,4 @@
+{"commitInfo":{"timestamp":1730671958797,"operation":"Manual Update","operationParameters":{},"readVersion":1,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{},"engineInfo":"Apache-Spark/3.5.3 Delta-Lake/3.3.0-SNAPSHOT","txnId":"9a6324de-800e-4c8f-9ce3-7766d0462474"}}
+{"domainMetadata":{"domain":"testDomain1","configuration":"{\"key1\":\"1\"}","removed":false}}
+{"domainMetadata":{"domain":"testDomain2","configuration":"","removed":false}}
+{"domainMetadata":{"domain":"testDomain3","configuration":"","removed":false}}

connectors/golden-tables/src/main/resources/golden/kernel-domain-metadata/_delta_log/00000000000000000003.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1730671959801,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[]"},"readVersion":2,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"2","numOutputRows":"2","numOutputBytes":"956"},"engineInfo":"Apache-Spark/3.5.3 Delta-Lake/3.3.0-SNAPSHOT","txnId":"54aecdaa-e039-40c8-ac9b-bcd7f30183fa"}}
+{"add":{"path":"test%25file%25prefix-part-00000-48cf7913-43ae-45bf-ab2c-94eb2fe77358-c000.snappy.parquet","partitionValues":{},"size":478,"modificationTime":1730671959767,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":0},\"maxValues\":{\"id\":0},\"nullCount\":{\"id\":0}}"}}
+{"add":{"path":"test%25file%25prefix-part-00001-071539c0-ef9e-478c-b550-035c6b5a31c2-c000.snappy.parquet","partitionValues":{},"size":478,"modificationTime":1730671959767,"dataChange":true,"stats":"{\"numRecords\":1,\"minValues\":{\"id\":1},\"maxValues\":{\"id\":1},\"nullCount\":{\"id\":0}}"}}

connectors/golden-tables/src/main/resources/golden/kernel-domain-metadata/_delta_log/00000000000000000004.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1730671962209,"operation":"Manual Update","operationParameters":{},"readVersion":3,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{},"engineInfo":"Apache-Spark/3.5.3 Delta-Lake/3.3.0-SNAPSHOT","txnId":"063e4cd4-4091-4a30-a089-4e8a0fc5c0ac"}}
+{"domainMetadata":{"domain":"testDomain1","configuration":"{\"key1\":\"10\"}","removed":false}}
+{"domainMetadata":{"domain":"testDomain2","configuration":"","removed":true}}

connectors/golden-tables/src/main/resources/golden/kernel-domain-metadata/_delta_log/_last_checkpoint

@@ -0,0 +1 @@
+{"version":3,"size":7,"sizeInBytes":16337,"numOfAddFiles":2,"checkpointSchema":{"type":"struct","fields":[{"name":"txn","type":{"type":"struct","fields":[{"name":"appId","type":"string","nullable":true,"metadata":{}},{"name":"version","type":"long","nullable":true,"metadata":{}},{"name":"lastUpdated","type":"long","nullable":true,"metadata":{}}]},"nullable":true,"metadata":{}},{"name":"add","type":{"type":"struct","fields":[{"name":"path","type":"string","nullable":true,"metadata":{}},{"name":"partitionValues","type":{"type":"map","keyType":"string","valueType":"string","valueContainsNull":true},"nullable":true,"metadata":{}},{"name":"size","type":"long","nullable":true,"metadata":{}},{"name":"modificationTime","type":"long","nullable":true,"metadata":{}},{"name":"dataChange","type":"boolean","nullable":true,"metadata":{}},{"name":"tags","type":{"type":"map","keyType":"string","valueType":"string","valueContainsNull":true},"nullable":true,"metadata":{}},{"name":"deletionVector","type":{"type":"struct","fields":[{"name":"storageType","type":"string","nullable":true,"metadata":{}},{"name":"pathOrInlineDv","type":"string","nullable":true,"metadata":{}},{"name":"offset","type":"integer","nullable":true,"metadata":{}},{"name":"sizeInBytes","type":"integer","nullable":true,"metadata":{}},{"name":"cardinality","type":"long","nullable":true,"metadata":{}},{"name":"maxRowIndex","type":"long","nullable":true,"metadata":{}}]},"nullable":true,"metadata":{}},{"name":"baseRowId","type":"long","nullable":true,"metadata":{}},{"name":"defaultRowCommitVersion","type":"long","nullable":true,"metadata":{}},{"name":"clusteringProvider","type":"string","nullable":true,"metadata":{}},{"name":"stats","type":"string","nullable":true,"metadata":{}}]},"nullable":true,"metadata":{}},{"name":"remove","type":{"type":"struct","fields":[{"name":"path","type":"string","nullable":true,"metadata":{}},{"name":"deletionTimestamp","type":"long","nullable":true,"metadata":{}},{"name":"dataChange","type":"boolean","nullable":true,"metadata":{}},{"name":"extendedFileMetadata","type":"boolean","nullable":true,"metadata":{}},{"name":"partitionValues","type":{"type":"map","keyType":"string","valueType":"string","valueContainsNull":true},"nullable":true,"metadata":{}},{"name":"size","type":"long","nullable":true,"metadata":{}},{"name":"deletionVector","type":{"type":"struct","fields":[{"name":"storageType","type":"string","nullable":true,"metadata":{}},{"name":"pathOrInlineDv","type":"string","nullable":true,"metadata":{}},{"name":"offset","type":"integer","nullable":true,"metadata":{}},{"name":"sizeInBytes","type":"integer","nullable":true,"metadata":{}},{"name":"cardinality","type":"long","nullable":true,"metadata":{}},{"name":"maxRowIndex","type":"long","nullable":true,"metadata":{}}]},"nullable":true,"metadata":{}},{"name":"baseRowId","type":"long","nullable":true,"metadata":{}},{"name":"defaultRowCommitVersion","type":"long","nullable":true,"metadata":{}}]},"nullable":true,"metadata":{}},{"name":"metaData","type":{"type":"struct","fields":[{"name":"id","type":"string","nullable":true,"metadata":{}},{"name":"name","type":"string","nullable":true,"metadata":{}},{"name":"description","type":"string","nullable":true,"metadata":{}},{"name":"format","type":{"type":"struct","fields":[{"name":"provider","type":"string","nullable":true,"metadata":{}},{"name":"options","type":{"type":"map","keyType":"string","valueType":"string","valueContainsNull":true},"nullable":true,"metadata":{}}]},"nullable":true,"metadata":{}},{"name":"schemaString","type":"string","nullable":true,"metadata":{}},{"name":"partitionColumns","type":{"type":"array","elementType":"string","containsNull":true},"nullable":true,"metadata":{}},{"name":"configuration","type":{"type":"map","keyType":"string","valueType":"string","valueContainsNull":true},"nullable":true,"metadata":{}},{"name":"createdTime","type":"long","nullable":true,"metadata":{}}]},"nullable":true,"metadata":{}},{"name":"protocol","type":{"type":"struct","fields":[{"name":"minReaderVersion","type":"integer","nullable":true,"metadata":{}},{"name":"minWriterVersion","type":"integer","nullable":true,"metadata":{}},{"name":"readerFeatures","type":{"type":"array","elementType":"string","containsNull":true},"nullable":true,"metadata":{}},{"name":"writerFeatures","type":{"type":"array","elementType":"string","containsNull":true},"nullable":true,"metadata":{}}]},"nullable":true,"metadata":{}},{"name":"domainMetadata","type":{"type":"struct","fields":[{"name":"domain","type":"string","nullable":true,"metadata":{}},{"name":"configuration","type":"string","nullable":true,"metadata":{}},{"name":"removed","type":"boolean","nullable":true,"metadata":{}}]},"nullable":true,"metadata":{}}]},"checksum":"7a97cc187ffb0604cc50e51bddb3cbfa"}

connectors/golden-tables/src/test/scala/io/delta/golden/GoldenTables.scala

@@ -1663,6 +1663,46 @@ class GoldenTables extends QueryTest with SharedSparkSession {
       Row(0, 0) :: Nil,
       schema)
   }
+
+  generateGoldenTable("kernel-domain-metadata") { tablePath =>
+    withSQLConf(
+      ("spark.databricks.delta.properties.defaults.checkpointInterval", "3")
+    ) {
+      val tbl = "tbl"
+
+      sql(s"CREATE TABLE $tbl (id LONG) USING delta LOCATION '$tablePath'")
+      sql(s"ALTER TABLE $tbl SET TBLPROPERTIES('delta.feature.domainMetadata' = 'enabled')")
+
+      val deltaLog = DeltaLog.forTable(spark, new Path(tablePath))
+
+      deltaLog
+        .startTransaction()
+        .commitManually(
+          List(
+            DomainMetadata("testDomain1", "{\"key1\":\"1\"}", removed = false),
+            DomainMetadata("testDomain2", "", removed = false),
+            DomainMetadata("testDomain3", "", removed = false)
+          ): _*
+        )
+
+      spark.range(0, 2).write.format("delta").mode("append").save(tablePath) // Checkpoint created
+
+      deltaLog
+        .startTransaction()
+        .commitManually(
+          List(
+            DomainMetadata("testDomain1", "{\"key1\":\"10\"}".stripMargin, removed = false),
+            DomainMetadata("testDomain2", "", removed = true)
+          ): _*
+        )
+
+      // In the end, we need to read 1 checkpoint file and 1 log file to replay the golden table
+      // The state of the domain metadata should be:
+      // testDomain1: "\"key1\":\"10\"", removed = false
+      // testDomain2: "", removed = true
+      // testDomain3: "", removed = false
+    }
+  }
 }
 
 case class TestStruct(f1: String, f2: Long)

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, String action1, String 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, action2);
+    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 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,10 @@ public Protocol getProtocol() {
     return protocol;
   }
 
+  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");
             }
           });
 
@@ -93,7 +94,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
@@ -137,6 +138,8 @@ public static void validateWriteSupportedTable(
               break;
             case "typeWidening":
               break;
+            case "domainMetadata":
+              break;
             default:
               throw unsupportedWriterFeature(tablePath, writerFeature);
           }

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;
@@ -75,6 +71,7 @@ public class TransactionImpl implements Transaction {
   private final Clock clock;
   private Metadata metadata;
   private boolean shouldUpdateMetadata;
+  private final List<DomainMetadata> domainMetadatas = new ArrayList<>();
 
   private boolean closed; // To avoid trying to commit the same transaction again.
 
@@ -221,11 +218,22 @@ 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);
+
+    // Create domain metadata action rows
+    List<Row> domainMetadataActions = new ArrayList<>();
+    for (DomainMetadata domainMetadata : domainMetadatas) {
+      domainMetadataActions.add(createDomainMetadataSingleAction(domainMetadata.toRow()));
+    }
+
     try (CloseableIterator<Row> stageDataIter = dataActions.iterator()) {
       // Create a new CloseableIterator that will return the metadata actions followed by the
       // data actions.
       CloseableIterator<Row> dataAndMetadataActions =
-          toCloseableIterator(metadataActions.iterator()).combine(stageDataIter);
+          toCloseableIterator(metadataActions.iterator())
+              .combine(toCloseableIterator(domainMetadataActions.iterator()))
+              .combine(stageDataIter);
 
       if (commitAsVersion == 0) {
         // New table, create a delta log directory
@@ -269,6 +277,19 @@ public Optional<SetTransaction> getSetTxnOpt() {
     return setTxnOpt;
   }
 
+  /**
+   * Add domain metadata to the transaction to be committed.
+   *
+   * @param domainMetadatas List of domain metadata to be added.
+   */
+  public void addDomainMetadata(List<DomainMetadata> domainMetadatas) {
+    this.domainMetadatas.addAll(domainMetadatas);
+  }
+
+  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.