[Kernel] [CC Refactor #1] Add TableIdentifier API (delta-io#3795)

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [ ] Spark
- [ ] Standalone
- [ ] Flink
- [X] Kernel
- [ ] Other (fill in here)

## Description

- adds a new `TableIdentifier` class, that kernel will pass on to Commit
Coordinator Client
- adds a new `Table::forPathWithTableId(engine, path, tableId)`
interface
- the tableId is stored as an `Optional` in the `Table`, and this PR
does **not** propagate that value into SnapshotManager, Snapshot, etc.
Future PRs can take care of that.

## How was this patch tested?

TableIdentifier UTs

## Does this PR introduce _any_ user-facing changes?

Yes. See the above.

Author: scottsand-db

kernel/kernel-api/src/main/java/io/delta/kernel/Table.java

@@ -22,6 +22,7 @@
 import io.delta.kernel.exceptions.TableNotFoundException;
 import io.delta.kernel.internal.TableImpl;
 import java.io.IOException;
+import java.util.Optional;
 
 /**
  * Represents the Delta Lake table for a given path.
@@ -57,6 +58,24 @@ static Table forPath(Engine engine, String path) {
     return TableImpl.forPath(engine, path);
   }
 
+  /**
+   * Instantiate a table object for the Delta Lake table at the given path and associate it with the
+   * given {@link TableIdentifier}.
+   *
+   * <p>See {@link #forPath(Engine, String)} for more details on behavior when the table path does
+   * or does not exist.
+   *
+   * @param engine the {@link Engine} instance to use in Delta Kernel.
+   * @param path location of the table. Path is resolved to fully qualified path using the given
+   *     {@code engine}.
+   * @param tableId the {@link TableIdentifier} to associate with the {@link Table}
+   * @return an instance of {@link Table} representing the Delta table at the given path and
+   *     associated with the given {@link TableIdentifier}
+   */
+  static Table forPathWithTableId(Engine engine, String path, TableIdentifier tableId) {
+    return TableImpl.forPathWithTableId(engine, path, tableId);
+  }
+
   /**
    * The fully qualified path of this {@link Table} instance.
    *
@@ -66,6 +85,14 @@ static Table forPath(Engine engine, String path) {
    */
   String getPath(Engine engine);
 
+  /**
+   * The table identifier of this {@link Table} instance.
+   *
+   * @return the table identifier, or {@link Optional#empty()} if none is set.
+   * @since 3.3.0
+   */
+  Optional<TableIdentifier> getTableIdentifier();
+
   /**
    * Get the latest snapshot of the table.
    *

kernel/kernel-api/src/main/java/io/delta/kernel/TableIdentifier.java

@@ -0,0 +1,84 @@
+/*
+ * 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;
+
+import static io.delta.kernel.internal.util.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+
+import io.delta.kernel.annotation.Evolving;
+import java.util.Arrays;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * Identifier for a table. e.g. $catalog / $schema / $table
+ *
+ * @since 3.3
+ */
+@Evolving
+public class TableIdentifier {
+  /** The namespace of the table. */
+  private final String[] namespace;
+
+  /** The name of the table. */
+  private final String name;
+
+  public TableIdentifier(String[] namespace, String name) {
+    checkArgument(namespace != null && namespace.length > 0, "namespace cannot be null or empty");
+    this.namespace = namespace;
+    this.name = requireNonNull(name, "name is null");
+  }
+
+  /** @return The namespace of the table. e.g. $catalog / $schema */
+  public String[] getNamespace() {
+    return namespace;
+  }
+
+  /** @return The name of the table. */
+  public String getName() {
+    return name;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    TableIdentifier that = (TableIdentifier) o;
+    return Arrays.equals(getNamespace(), that.getNamespace()) && getName().equals(that.getName());
+  }
+
+  @Override
+  public int hashCode() {
+    return 31 * Objects.hash(getName()) + Arrays.hashCode(getNamespace());
+  }
+
+  @Override
+  public String toString() {
+    final String quotedNamespace =
+        Arrays.stream(namespace).map(this::quoteIdentifier).collect(Collectors.joining("."));
+    return "TableIdentifier{" + quotedNamespace + "." + quoteIdentifier(name) + "}";
+  }
+
+  /** Escapes back-ticks within the identifier name with double-back-ticks. */
+  private String quoteIdentifier(String identifier) {
+    return identifier.replace("`", "``");
+  }
+}

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

@@ -44,6 +44,10 @@
 
 public class TableImpl implements Table {
 
+  //////////////////////////////
+  // Static Methods / Members //
+  //////////////////////////////
+
   private static final Logger logger = LoggerFactory.getLogger(TableImpl.class);
 
   public static Table forPath(Engine engine, String path) {
@@ -60,34 +64,60 @@ public static Table forPath(Engine engine, String path) {
    * @return an instance of {@link Table} representing the Delta table at the given path
    */
   public static Table forPath(Engine engine, String path, Clock clock) {
-    String resolvedPath;
+    return forPathWithTableId(engine, path, Optional.empty(), clock);
+  }
+
+  public static Table forPathWithTableId(Engine engine, String path, TableIdentifier tableId) {
+    return forPathWithTableId(engine, path, Optional.of(tableId), System::currentTimeMillis);
+  }
+
+  public static Table forPathWithTableId(
+      Engine engine, String path, Optional<TableIdentifier> tableIdOpt, Clock clock) {
     try {
-      resolvedPath =
+      final String resolvedPath =
           wrapEngineExceptionThrowsIO(
               () -> engine.getFileSystemClient().resolvePath(path), "Resolving path %s", path);
+      return new TableImpl(resolvedPath, tableIdOpt, clock);
     } catch (IOException io) {
       throw new UncheckedIOException(io);
     }
-    return new TableImpl(resolvedPath, clock);
   }
 
-  private final SnapshotManager snapshotManager;
+  ////////////////////////////////
+  // Instance Methods / Members //
+  ////////////////////////////////
+
   private final String tablePath;
+  private final Optional<TableIdentifier> tableIdOpt;
   private final Clock clock;
 
-  public TableImpl(String tablePath, Clock clock) {
+  private final Path dataPath;
+  private final Path logPath;
+  private final SnapshotManager snapshotManager;
+
+  private TableImpl(String tablePath, Optional<TableIdentifier> tableIdOpt, Clock clock) {
     this.tablePath = tablePath;
-    final Path dataPath = new Path(tablePath);
-    final Path logPath = new Path(dataPath, "_delta_log");
-    this.snapshotManager = new SnapshotManager(logPath, dataPath);
+    this.tableIdOpt = tableIdOpt;
     this.clock = clock;
+    this.dataPath = new Path(tablePath);
+    this.logPath = new Path(dataPath, "_delta_log");
+    this.snapshotManager = new SnapshotManager(logPath, dataPath);
   }
 
+  /////////////////
+  // Public APIs //
+  /////////////////
+
   @Override
   public String getPath(Engine engine) {
     return tablePath;
   }
 
+  @Override
+  public Optional<TableIdentifier> getTableIdentifier() {
+    return tableIdOpt;
+  }
+
   @Override
   public Snapshot getLatestSnapshot(Engine engine) throws TableNotFoundException {
     return snapshotManager.buildLatestSnapshot(engine);
@@ -169,8 +199,7 @@ public CloseableIterator<ColumnarBatch> getChanges(
               for (int rowId = 0; rowId < protocolVector.getSize(); rowId++) {
                 if (!protocolVector.isNullAt(rowId)) {
                   Protocol protocol = Protocol.fromColumnVector(protocolVector, rowId);
-                  TableFeatures.validateReadSupportedTable(
-                      protocol, getDataPath().toString(), Optional.empty());
+                  TableFeatures.validateReadSupportedTable(protocol, tablePath, Optional.empty());
                 }
               }
               if (shouldDropProtocolColumn) {
@@ -181,14 +210,6 @@ public CloseableIterator<ColumnarBatch> getChanges(
             });
   }
 
-  protected Path getDataPath() {
-    return new Path(tablePath);
-  }
-
-  protected Path getLogPath() {
-    return new Path(tablePath, "_delta_log");
-  }
-
   /**
    * Returns the latest version that was committed before or at {@code millisSinceEpochUTC}. If no
    * version exists, throws a {@link KernelException}
@@ -269,6 +290,22 @@ public long getVersionAtOrAfterTimestamp(Engine engine, long millisSinceEpochUTC
     }
   }
 
+  ////////////////////
+  // Protected APIs //
+  ////////////////////
+
+  protected Path getDataPath() {
+    return dataPath;
+  }
+
+  protected Path getLogPath() {
+    return logPath;
+  }
+
+  ////////////////////////////
+  // Private Helper Methods //
+  ////////////////////////////
+
   /**
    * Returns the raw delta actions for each version between startVersion and endVersion. Only reads
    * the actions requested in actionSet from the JSON log files.

kernel/kernel-api/src/test/scala/io/delta/kernel/TableIdentifierSuite.scala

@@ -0,0 +1,75 @@
+/*
+ * 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
+
+import org.scalatest.funsuite.AnyFunSuite
+
+class TableIdentifierSuite extends AnyFunSuite {
+
+  test("TableIdentifier should throw IllegalArgumentException for null or empty namespace") {
+    assertThrows[IllegalArgumentException] {
+      new TableIdentifier(null, "table")
+    }
+    assertThrows[IllegalArgumentException] {
+      new TableIdentifier(Array(), "table")
+    }
+  }
+
+  test("TableIdentifier should throw NullPointerException for null table name") {
+    assertThrows[NullPointerException] {
+      new TableIdentifier(Array("catalog", "schema"), null)
+    }
+  }
+
+  test("TableIdentifier should return the correct namespace and name") {
+    val namespace = Array("catalog", "schema")
+    val name = "testTable"
+    val tid = new TableIdentifier(namespace, name)
+
+    assert(tid.getNamespace.sameElements(namespace))
+    assert(tid.getName == name)
+  }
+
+  test("TableIdentifiers with same namespace and name should be equal") {
+    val tid1 = new TableIdentifier(Array("catalog", "schema"), "table")
+    val tid2 = new TableIdentifier(Array("catalog", "schema"), "table")
+
+    assert(tid1 == tid2)
+    assert(tid1.hashCode == tid2.hashCode)
+  }
+
+  test("TableIdentifiers with different namespace or name should not be equal") {
+    val tid1 = new TableIdentifier(Array("catalog", "schema1"), "table1")
+    val tid2 = new TableIdentifier(Array("catalog", "schema2"), "table1")
+    val tid3 = new TableIdentifier(Array("catalog", "schema1"), "table2")
+
+    assert(tid1 != tid2)
+    assert(tid1 != tid3)
+  }
+
+  test("TableIdentifier toString") {
+    // Normal case
+    val tidNormal = new TableIdentifier(Array("catalog", "schema"), "table")
+    val expectedNormal = "TableIdentifier{catalog.schema.table}"
+    assert(tidNormal.toString == expectedNormal)
+
+    // Special case: should escape backticks
+    val tidSpecial = new TableIdentifier(Array("catalog", "sche`ma"), "tab`le")
+    val expectedSpecial = "TableIdentifier{catalog.sche``ma.tab``le}"
+    assert(tidSpecial.toString == expectedSpecial)
+  }
+}

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

@@ -236,7 +236,7 @@ class InCommitTimestampSuite extends DeltaTableWriteSuiteBase {
 
   test("Enablement tracking works when ICT is enabled post commit 0") {
     withTempDirAndEngine { (tablePath, engine) =>
-      val table = TableImpl.forPath(engine, tablePath)
+      val table = Table.forPath(engine, tablePath)
       val txnBuilder = table.createTransactionBuilder(engine, testEngineInfo, CREATE_TABLE)
 
       val txn = txnBuilder
@@ -319,7 +319,7 @@ class InCommitTimestampSuite extends DeltaTableWriteSuiteBase {
 
   test("Metadata toString should work with ICT enabled") {
     withTempDirAndEngine { (tablePath, engine) =>
-      val table = TableImpl.forPath(engine, tablePath)
+      val table = Table.forPath(engine, tablePath)
       val txnBuilder = table.createTransactionBuilder(engine, testEngineInfo, CREATE_TABLE)
 
       val txn = txnBuilder
@@ -356,7 +356,7 @@ class InCommitTimestampSuite extends DeltaTableWriteSuiteBase {
 
   test("Table with ICT enabled is readable") {
     withTempDirAndEngine { (tablePath, engine) =>
-      val table = TableImpl.forPath(engine, tablePath)
+      val table = Table.forPath(engine, tablePath)
       val txnBuilder = table.createTransactionBuilder(engine, testEngineInfo, CREATE_TABLE)
 
       val txn = txnBuilder