[Kernel] Add JsonMetadataDomain and RowTrackingMetadataDomain

kernel/kernel-api/src/main/java/io/delta/kernel/internal/metadatadomain/JsonMetadataDomain.java

@@ -0,0 +1,100 @@
+/*
+ * 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.metadatadomain;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.*;
+import io.delta.kernel.exceptions.KernelException;
+import io.delta.kernel.internal.SnapshotImpl;
+import io.delta.kernel.internal.actions.DomainMetadata;
+import io.delta.kernel.internal.rowtracking.RowTrackingMetadataDomain;
+import java.util.Optional;
+
+/**
+ * Abstract class representing a JSON metadata domain, whose configuration string is a JSON
+ * serialization of a domain object. This class provides methods to serialize and deserialize a
+ * metadata domain to and from JSON. Concrete implementations, such as {@link
+ * RowTrackingMetadataDomain}, should extend this class to define a specific metadata domain.
+ */
+public abstract class JsonMetadataDomain {
+  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+  /**
+   * Deserializes a JSON string into an instance of the specified metadata domain.
+   *
+   * @param json the JSON string to deserialize
+   * @param clazz the concrete class of the metadata domain object to deserialize into
+   * @param <T> the type of the object
+   * @return the deserialized object
+   * @throws KernelException if the JSON string cannot be parsed
+   */
+  protected static <T> T fromJsonConfiguration(String json, Class<T> clazz) {
+    try {
+      return OBJECT_MAPPER.readValue(json, clazz);
+    } catch (JsonProcessingException e) {
+      throw new KernelException("Could not parse a JSON to a JsonMetadataDomain object", e);
+    }
+  }
+
+  /**
+   * Retrieves the domain metadata from a snapshot for a given domain, and deserializes it into an
+   * instance of the specified metadata domain class.
+   *
+   * @param snapshot the snapshot to read from
+   * @param clazz the metadata domain class of the object to deserialize into
+   * @param domainName the name of the domain
+   * @param <T> the type of the metadata domain object
+   * @return an Optional containing the deserialized object if the domain metadata is found,
+   *     otherwise an empty Optional
+   */
+  protected static <T> Optional<T> fromSnapshot(
+      SnapshotImpl snapshot, Class<T> clazz, String domainName) {
+    return Optional.ofNullable(snapshot.getDomainMetadataMap().get(domainName))
+        .map(domainMetadata -> fromJsonConfiguration(domainMetadata.getConfiguration(), clazz));
+  }
+
+  /**
+   * Returns the name of the domain.
+   *
+   * @return the domain name
+   */
+  @JsonIgnore
+  public abstract String getDomainName();
+
+  /**
+   * Serializes this object into a JSON string.
+   *
+   * @return the JSON string representation of this object
+   * @throws KernelException if the object cannot be serialized
+   */
+  public String toJsonConfiguration() {
+    try {
+      return OBJECT_MAPPER.writeValueAsString(this);
+    } catch (JsonProcessingException e) {
+      throw new KernelException("Could not serialize a JsonMetadataDomain object to JSON", e);
+    }
+  }
+
+  /**
+   * Generate a {@link DomainMetadata} action from this metadata domain.
+   *
+   * @return the DomainMetadata action instance
+   */
+  public DomainMetadata toDomainMetadata() {
+    return new DomainMetadata(getDomainName(), toJsonConfiguration(), false);
+  }
+}

kernel/kernel-api/src/main/java/io/delta/kernel/internal/rowtracking/RowTrackingMetadataDomain.java

@@ -0,0 +1,77 @@
+/*
+ * 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.rowtracking;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.delta.kernel.internal.SnapshotImpl;
+import io.delta.kernel.internal.metadatadomain.JsonMetadataDomain;
+import java.util.Optional;
+
+/** Represents the metadata domain for row tracking. */
+public class RowTrackingMetadataDomain extends JsonMetadataDomain {
+
+  /**
+   * Creates an instance of {@link RowTrackingMetadataDomain} from a JSON configuration string.
+   *
+   * @param json the JSON configuration string
+   * @return an instance of {@link RowTrackingMetadataDomain}
+   */
+  public static RowTrackingMetadataDomain fromJsonConfiguration(String json) {
+    return JsonMetadataDomain.fromJsonConfiguration(json, RowTrackingMetadataDomain.class);
+  }
+
+  /**
+   * Creates an instance of {@link RowTrackingMetadataDomain} from a {@link SnapshotImpl}.
+   *
+   * @param snapshot the snapshot instance
+   * @return an {@link Optional} containing the {@link RowTrackingMetadataDomain} if present
+   */
+  public static Optional<RowTrackingMetadataDomain> fromSnapshot(SnapshotImpl snapshot) {
+    return JsonMetadataDomain.fromSnapshot(snapshot, RowTrackingMetadataDomain.class, DOMAIN_NAME);
+  }
+
+  public static final String DOMAIN_NAME = "delta.rowTracking";
+
+  /** Default value for row ID high watermark when it is missing in the table */
+  public static final long MISSING_ROW_ID_HIGH_WATERMARK = -1L;
+
+  /** The highest assigned fresh row id for the table */
+  private long rowIdHighWaterMark;
+
+  /**
+   * Constructs a RowTrackingMetadataDomain with the specified row ID high water mark.
+   *
+   * @param rowIdHighWaterMark the row ID high water mark
+   */
+  @JsonCreator
+  public RowTrackingMetadataDomain(@JsonProperty("rowIdHighWaterMark") long rowIdHighWaterMark) {
+    this.rowIdHighWaterMark = rowIdHighWaterMark;
+  }
+
+  @Override
+  public String getDomainName() {
+    return DOMAIN_NAME;
+  }
+
+  public long getRowIdHighWaterMark() {
+    return rowIdHighWaterMark;
+  }
+
+  public void setRowIdHighWaterMark(long rowIdHighWaterMark) {
+    this.rowIdHighWaterMark = rowIdHighWaterMark;
+  }
+}

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DomainMetadataSuite.scala

@@ -22,6 +22,7 @@ import io.delta.kernel.exceptions._
 import io.delta.kernel.internal.{SnapshotImpl, TableImpl, TransactionBuilderImpl, TransactionImpl}
 import io.delta.kernel.internal.actions.{DomainMetadata, Protocol, SingleAction}
 import io.delta.kernel.internal.util.Utils.toCloseableIterator
+import io.delta.kernel.internal.rowtracking.RowTrackingMetadataDomain
 import io.delta.kernel.utils.CloseableIterable.{emptyIterable, inMemoryIterable}
 import org.apache.hadoop.fs.Path
 import org.apache.spark.sql.delta.DeltaLog
@@ -472,4 +473,51 @@ class DomainMetadataSuite extends DeltaTableWriteSuiteBase with ParquetSuiteBase
       }
     })
   }
+
+  test("RowTrackingMetadataDomain is serializable and deserializable") {
+    withTempDirAndEngine((tablePath, engine) => {
+      // Create a RowTrackingMetadataDomain
+      val rowTrackingMetadataDomain = new RowTrackingMetadataDomain(10)
+
+      // Generate a DomainMetadata action from it and verify. Its configuration should be
+      // a JSON serialization of the rowTrackingMetadataDomain
+      val dm = rowTrackingMetadataDomain.toDomainMetadata
+      assert(dm.getDomain === rowTrackingMetadataDomain.getDomainName)
+      assert(dm.getConfiguration === """{"rowIdHighWaterMark":10}""")
+
+      // Verify the deserialization from DomainMetadata action into concrete domain object
+      val deserializedDomain = RowTrackingMetadataDomain.fromJsonConfiguration(dm.getConfiguration)
+      assert(deserializedDomain.getDomainName === rowTrackingMetadataDomain.getDomainName)
+      assert(
+        rowTrackingMetadataDomain.getRowIdHighWaterMark
+        === deserializedDomain.getRowIdHighWaterMark
+      )
+
+      // Verify the domainMetadata can be committed and read back
+      createTableWithDomainMetadataSupported(engine, tablePath)
+      // Commit the domain metadata and verify
+      commitDomainMetadataAndVerify(
+        engine,
+        tablePath,
+        domainMetadatas = Seq(dm),
+        expectedValue = Map(rowTrackingMetadataDomain.getDomainName -> dm)
+      )
+
+      // Read the domain metadata back from the table snapshot
+      val table = Table.forPath(engine, tablePath)
+      val snapshot = table.getLatestSnapshot(engine).asInstanceOf[SnapshotImpl]
+      val rowTrackingMetadataDomainFromSnapshot =
+        RowTrackingMetadataDomain.fromSnapshot(snapshot).get
+
+      // Verify the domain metadata read back from the snapshot
+      assert(
+        rowTrackingMetadataDomain.getDomainName ===
+        rowTrackingMetadataDomainFromSnapshot.getDomainName
+      )
+      assert(
+        rowTrackingMetadataDomain.getRowIdHighWaterMark ===
+        rowTrackingMetadataDomainFromSnapshot.getRowIdHighWaterMark
+      )
+    })
+  }
 }