[Kernel] [CC Refactor #1] TableIdentifier API

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.4.0
+   */
+  Optional<TableIdentifier> getTableId();
+
   /**
    * Get the latest snapshot of the table.
    *

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

@@ -47,39 +47,58 @@
 
 public class TableImpl implements Table {
 
+  //////////////////////////////////
+  // Static variables and methods //
+  //////////////////////////////////
+
   private static final Logger logger = LoggerFactory.getLogger(TableImpl.class);
 
   public static Table forPath(Engine engine, String path) {
     return forPath(engine, path, System::currentTimeMillis);
   }
 
+  public static Table forPathWithTableId(Engine engine, String path, TableIdentifier tableId) {
+    return create(engine, path, Optional.of(tableId), System::currentTimeMillis);
+  }
+
+  public static Table forPath(Engine engine, String path, Clock clock) {
+    return create(engine, path, Optional.empty() /* tableId */, clock);
+  }
+
   /**
    * Instantiate a table object for the Delta Lake table at the given path. It takes an additional
    * parameter called {@link Clock} which helps in testing.
    *
    * @param engine {@link Engine} instance to use in Delta Kernel.
    * @param path location of the table.
+   * @param tableId the {@link TableIdentifier} to associate with the table.
    * @param clock {@link Clock} instance to use for time-related operations.
    * @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;
+  private static Table create(
+      Engine engine, String path, Optional<TableIdentifier> tableId, Clock clock) {
     try {
-      resolvedPath =
+      final String resolvedPath =
           wrapEngineExceptionThrowsIO(
               () -> engine.getFileSystemClient().resolvePath(path), "Resolving path %s", path);
+      return new TableImpl(resolvedPath, tableId, clock);
     } catch (IOException io) {
       throw new UncheckedIOException(io);
     }
-    return new TableImpl(resolvedPath, clock);
   }
 
+  //////////////////////////////////
+  // Member variables and methods //
+  //////////////////////////////////
+
   private final SnapshotManager snapshotManager;
   private final String tablePath;
+  private final Optional<TableIdentifier> tableId;
   private final Clock clock;
 
-  public TableImpl(String tablePath, Clock clock) {
+  public TableImpl(String tablePath, Optional<TableIdentifier> tableId, Clock clock) {
     this.tablePath = tablePath;
+    this.tableId = tableId;
     final Path dataPath = new Path(tablePath);
     final Path logPath = new Path(dataPath, "_delta_log");
     this.snapshotManager = new SnapshotManager(logPath, dataPath);
@@ -91,6 +110,11 @@ public String getPath(Engine engine) {
     return tablePath;
   }
 
+  @Override
+  public Optional<TableIdentifier> getTableId() {
+    return tableId;
+  }
+
   @Override
   public Snapshot getLatestSnapshot(Engine engine) throws TableNotFoundException {
     SnapshotQueryContext snapshotContext = SnapshotQueryContext.forLatestSnapshot(tablePath);

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)
+  }
+}