[Kernel] Assign base row ID and default row commit version to AddFile

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

@@ -293,6 +293,18 @@ public static ConcurrentWriteException concurrentDomainMetadataAction(
     return new ConcurrentWriteException(message);
   }
 
+  public static KernelException rowIDAssignmentWithoutStats() {
+    return new KernelException(
+        "All AddFile actions must have statistics that include the number of records "
+            + "when writing to a Delta table with the 'rowTracking' table feature supported");
+  }
+
+  public static KernelException rowTrackingSupportedWithDomainMetadataUnsupported() {
+    return new KernelException(
+        "Feature 'rowTracking' is supported and depends on feature 'domainMetadata',"
+            + " but 'domainMetadata' is unsupported");
+  }
+
   /* ------------------------ HELPER METHODS ----------------------------- */
   private static String formatTimestamp(long millisSinceEpochUTC) {
     return new Timestamp(millisSinceEpochUTC).toInstant().toString();

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

@@ -32,6 +32,7 @@
 import java.net.URI;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 
 /**
  * Utilities to extract information out of the scan file rows returned by {@link
@@ -87,6 +88,11 @@ private InternalScanFileUtils() {}
 
   private static final int ADD_FILE_DV_ORDINAL = ADD_FILE_SCHEMA.indexOf("deletionVector");
 
+  private static final int ADD_FILE_BASE_ROW_ID_ORDINAL = ADD_FILE_SCHEMA.indexOf("baseRowId");
+
+  private static final int ADD_FILE_DEFAULT_ROW_COMMIT_VERSION =
+      ADD_FILE_SCHEMA.indexOf("defaultRowCommitVersion");
+
   private static final int TABLE_ROOT_ORDINAL = SCAN_FILE_SCHEMA.indexOf(TABLE_ROOT_COL_NAME);
 
   public static final int ADD_FILE_STATS_ORDINAL = AddFile.SCHEMA_WITH_STATS.indexOf("stats");
@@ -190,4 +196,30 @@ public static DeletionVectorDescriptor getDeletionVectorDescriptorFromRow(Row sc
   public static Column getPartitionValuesParsedRefInAddFile(String partitionColName) {
     return new Column(new String[] {"add", "partitionValues_parsed", partitionColName});
   }
+
+  /**
+   * Get the base row id from the given scan file row.
+   *
+   * @param scanFile {@link Row} representing one scan file.
+   * @return base row id if present, otherwise empty.
+   */
+  public static Optional<Long> getBaseRowId(Row scanFile) {
+    Row addFile = getAddFileEntry(scanFile);
+    return addFile.isNullAt(ADD_FILE_BASE_ROW_ID_ORDINAL)
+        ? Optional.empty()
+        : Optional.of(addFile.getLong(ADD_FILE_BASE_ROW_ID_ORDINAL));
+  }
+
+  /**
+   * Get the default row commit version from the given scan file row.
+   *
+   * @param scanFile {@link Row} representing one scan file.
+   * @return default row commit version if present, otherwise empty.
+   */
+  public static Optional<Long> getDefaultRowCommitVersion(Row scanFile) {
+    Row addFile = getAddFileEntry(scanFile);
+    return addFile.isNullAt(ADD_FILE_DEFAULT_ROW_COMMIT_VERSION)
+        ? Optional.empty()
+        : Optional.of(addFile.getLong(ADD_FILE_DEFAULT_ROW_COMMIT_VERSION));
+  }
 }

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

@@ -40,6 +40,7 @@ public class TableFeatures {
               add("typeWidening-preview");
               add("typeWidening");
               add(DOMAIN_METADATA_FEATURE_NAME);
+              add(ROW_TRACKING_FEATURE_NAME);
             }
           });
 
@@ -61,6 +62,9 @@ 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 table features. */
   public static final int TABLE_FEATURES_MIN_WRITER_VERSION = 7;
 
@@ -100,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}, {@code domainMetadata} feature enabled.
+   *       columnMapping}, {@code typeWidening}, {@code domainMetadata}, {@code rowTracking} feature
+   *       enabled.
    * </ul>
    *
    * @param protocol Table protocol
@@ -136,6 +141,12 @@ public static void validateWriteSupportedTable(
             throw unsupportedWriterFeature(tablePath, writerFeature);
           }
         }
+        // Eventually we may have a way to declare and enforce dependencies between features.
+        // By putting this check for row tracking here, it makes it easier to spot that row
+        // tracking defines such a dependency that can be implicitly checked.
+        if (isRowTrackingSupported(protocol) && !isDomainMetadataSupported(protocol)) {
+          throw DeltaErrors.rowTrackingSupportedWithDomainMetadataUnsupported();
+        }
         break;
       default:
         throw unsupportedWriterProtocol(tablePath, minWriterVersion);
@@ -189,12 +200,17 @@ public static Set<String> extractAutomaticallyEnabledWriterFeatures(
    * @return true if the "domainMetadata" feature is supported, false otherwise
    */
   public static boolean isDomainMetadataSupported(Protocol protocol) {
-    List<String> writerFeatures = protocol.getWriterFeatures();
-    if (writerFeatures == null) {
-      return false;
-    }
-    return writerFeatures.contains(DOMAIN_METADATA_FEATURE_NAME)
-        && protocol.getMinWriterVersion() >= TABLE_FEATURES_MIN_WRITER_VERSION;
+    return isWriterFeatureSupported(protocol, DOMAIN_METADATA_FEATURE_NAME);
+  }
+
+  /**
+   * Check if the table protocol supports the "rowTracking" writer feature.
+   *
+   * @param protocol the protocol to check
+   * @return true if the protocol supports row tracking, false otherwise
+   */
+  public static boolean isRowTrackingSupported(Protocol protocol) {
+    return isWriterFeatureSupported(protocol, ROW_TRACKING_FEATURE_NAME);
   }
 
   /**
@@ -253,4 +269,13 @@ private static void validateNoInvariants(StructType tableSchema) {
       throw columnInvariantsNotSupported();
     }
   }
+
+  private static boolean isWriterFeatureSupported(Protocol protocol, String featureName) {
+    List<String> writerFeatures = protocol.getWriterFeatures();
+    if (writerFeatures == null) {
+      return false;
+    }
+    return writerFeatures.contains(featureName)
+        && protocol.getMinWriterVersion() >= TABLE_FEATURES_MIN_WRITER_VERSION;
+  }
 }

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

@@ -32,6 +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.rowtracking.RowTracking;
 import io.delta.kernel.internal.util.*;
 import io.delta.kernel.types.StructType;
 import io.delta.kernel.utils.CloseableIterable;
@@ -145,6 +146,18 @@ public TransactionCommitResult commit(Engine engine, CloseableIterable<Row> data
       CommitInfo attemptCommitInfo = generateCommitAction(engine);
       updateMetadataWithICTIfRequired(
           engine, attemptCommitInfo.getInCommitTimestamp(), readSnapshot.getVersion(engine));
+
+      // If row tracking is supported, assign base row IDs and default row commit versions to any
+      // AddFile actions that do not yet have them. If the row ID high watermark changes, emit a
+      // DomainMetadata action to update it.
+      if (TableFeatures.isRowTrackingSupported(protocol)) {
+        RowTracking.createNewHighWaterMarkIfNeeded(readSnapshot, dataActions)
+            .ifPresent(domainMetadatas::add);
+        dataActions =
+            RowTracking.assignBaseRowIdAndDefaultRowCommitVersion(
+                readSnapshot, commitAsVersion, dataActions);
+      }
+
       int numRetries = 0;
       do {
         logger.info("Committing transaction as version = {}.", commitAsVersion);