[Kernel] Assign base row ID to AddFile actions

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,40 @@ public static KernelException invalidConfigurationValueException(
     return new InvalidConfigurationValueException(key, value, helpMessage);
   }
 
+  public static KernelException domainMetadataUnsupported() {
+    String message =
+        "Cannot commit DomainMetadata action(s) because the 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);
+  }
+
+  public static KernelException rowIDAssignmentWithoutStats() {
+    return new KernelException(
+        "Cannot assign baseRowId to add action. "
+            + "The number of records in this data file is missing.");
+  }
+
   /* ------------------------ 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);
@@ -115,6 +128,15 @@ public Path getDataPath() {
     return dataPath;
   }
 
+  /**
+   * Returns the log replay object for this snapshot. Visible for testing.
+   *
+   * @return the {@link LogReplay} object
+   */
+  public LogReplay getLogReplay() {
+    return logReplay;
+  }
+
   /**
    * Returns the timestamp of the latest commit of this snapshot. For an uninitialized snapshot,
    * this returns -1.

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

@@ -39,6 +39,8 @@ public class TableFeatures {
               add("columnMapping");
               add("typeWidening-preview");
               add("typeWidening");
+              add(DOMAIN_METADATA_FEATURE_NAME);
+              add(ROW_TRACKING_FEATURE_NAME);
             }
           });
 
@@ -57,6 +59,15 @@ public class TableFeatures {
             }
           });
 
+  /** The feature name for domain metadata. */
+  public static final String DOMAIN_METADATA_FEATURE_NAME = "domainMetadata";
+
+  /** The feature name for row tracking. */
+  public static final String ROW_TRACKING_FEATURE_NAME = "rowTracking";
+
+  /** The minimum writer version required to support domain metadata. */
+  public static final int TABLE_FEATURES_MIN_WRITER_VERSION = 7;
+
   ////////////////////
   // Helper Methods //
   ////////////////////
@@ -93,7 +104,8 @@ 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}, {@code rowTracking} feature
+   *       enabled.
    * </ul>
    *
    * @param protocol Table protocol
@@ -125,20 +137,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/TransactionImpl.java

@@ -32,11 +32,9 @@
 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.rowtracking.RowIDAssignmentResult;
+import io.delta.kernel.internal.rowtracking.RowTracking;
+import io.delta.kernel.internal.util.*;
 import io.delta.kernel.types.StructType;
 import io.delta.kernel.utils.CloseableIterable;
 import io.delta.kernel.utils.CloseableIterator;
@@ -76,6 +74,12 @@ public class TransactionImpl implements Transaction {
   private Metadata metadata;
   private boolean shouldUpdateMetadata;
 
+  // Contains domain metadata actions known prior to iterating and writing the data actions
+  private final List<DomainMetadata> domainMetadatas = new ArrayList<>();
+  // Contains domain metadata actions generated on the fly while writing the data actions
+  private CloseableIterator<DomainMetadata> domainMetadataIter =
+      toCloseableIterator(Collections.emptyIterator());
+
   private boolean closed; // To avoid trying to commit the same transaction again.
 
   public TransactionImpl(
@@ -120,6 +124,23 @@ public StructType getSchema(Engine engine) {
     return readSnapshot.getSchema(engine);
   }
 
+  public Optional<SetTransaction> getSetTxnOpt() {
+    return setTxnOpt;
+  }
+
+  /**
+   * Internal API to add domain metadata actions for this transaction. Visible for testing.
+   *
+   * @param domainMetadatas List of domain metadata to be added to the transaction.
+   */
+  public void addDomainMetadatas(List<DomainMetadata> domainMetadatas) {
+    this.domainMetadatas.addAll(domainMetadatas);
+  }
+
+  public List<DomainMetadata> getDomainMetadatas() {
+    return domainMetadatas;
+  }
+
   @Override
   public TransactionCommitResult commit(Engine engine, CloseableIterable<Row> dataActions)
       throws ConcurrentWriteException {
@@ -131,6 +152,16 @@ public TransactionCommitResult commit(Engine engine, CloseableIterable<Row> data
       CommitInfo attemptCommitInfo = generateCommitAction(engine);
       updateMetadataWithICTIfRequired(
           engine, attemptCommitInfo.getInCommitTimestamp(), readSnapshot.getVersion(engine));
+
+      // For Row Tracking, we assign base row IDs to all AddFiles inside dataActions that
+      // do not have it yet. If the high water mark has changed, we also emit a
+      // DomainMetadata action with the new high water mark.
+      RowIDAssignmentResult rowIDAssignmentResult =
+          RowTracking.assignBaseRowId(protocol, readSnapshot, FULL_SCHEMA, dataActions);
+      dataActions = rowIDAssignmentResult.getDataActions();
+      domainMetadataIter =
+          domainMetadataIter.combine(rowIDAssignmentResult.getDomainMetadatasIter());
+
       int numRetries = 0;
       do {
         logger.info("Committing transaction as version = {}.", commitAsVersion);
@@ -223,9 +254,12 @@ private TransactionCommitResult doCommit(
 
     try (CloseableIterator<Row> stageDataIter = dataActions.iterator()) {
       // Create a new CloseableIterator that will return the metadata actions followed by the
-      // data actions.
+      // data actions, and then DomainMetadata actions. The order is crucial as DomainMetadata
+      // actions may depend on the data actions.
       CloseableIterator<Row> dataAndMetadataActions =
-          toCloseableIterator(metadataActions.iterator()).combine(stageDataIter);
+          toCloseableIterator(metadataActions.iterator())
+              .combine(stageDataIter)
+              .combine(getDomainMetadataActions());
 
       if (commitAsVersion == 0) {
         // New table, create a delta log directory
@@ -265,10 +299,6 @@ public boolean isBlindAppend() {
     return true;
   }
 
-  public Optional<SetTransaction> getSetTxnOpt() {
-    return setTxnOpt;
-  }
-
   /**
    * 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.
@@ -313,6 +343,29 @@ private Map<String, String> getOperationParameters() {
     return Collections.emptyMap();
   }
 
+  /**
+   * Returns an iterator of all domain metadata actions to be committed. Domain metadata actions
+   * exist in two places:
+   *
+   * <ol>
+   *   <li>{@code List<DomainMetadata> domainMetadatas}: Contains domain metadata actions known
+   *       prior to iterating and writing the data actions.
+   *   <li>{@code CloseableIterator<DomainMetadata> domainMetadataIter}`: Contains domain metadata
+   *       actions generated on the fly while writing the data actions.
+   * </ol>
+   *
+   * This method combines both sources of domain metadata actions and returns an iterator of all
+   * domain metadata action rows to be committed.
+   *
+   * @return a {@link CloseableIterator} of domain metadata action rows
+   */
+  private CloseableIterator<Row> getDomainMetadataActions() {
+    // TODO: Implement the check for duplicate domain metadata and if the protocol supports here
+    return toCloseableIterator(domainMetadatas.listIterator())
+        .combine(domainMetadataIter)
+        .map(domainMetadata -> createDomainMetadataSingleAction(domainMetadata.toRow()));
+  }
+
   /**
    * Get the part of the schema of the table that needs the statistics to be collected per file.
    *