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

This reverts commit 024dadb.

We seem to be rethinking our Coordinated Commits CUJ / APIs, and we
don't want these APIs leaked in Delta 3.3.

Author: scottsand-db

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

@@ -22,7 +22,6 @@
 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.
@@ -58,24 +57,6 @@ 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.
    *
@@ -85,14 +66,6 @@ static Table forPathWithTableId(Engine engine, String path, TableIdentifier tabl
    */
   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/internal/TableImpl.java

@@ -44,10 +44,6 @@
 
 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) {
@@ -64,60 +60,34 @@ 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) {
-    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) {
+    String resolvedPath;
     try {
-      final String resolvedPath =
+      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);
   }
 
-  ////////////////////////////////
-  // Instance Methods / Members //
-  ////////////////////////////////
-
+  private final SnapshotManager snapshotManager;
   private final String tablePath;
-  private final Optional<TableIdentifier> tableIdOpt;
   private final Clock clock;
 
-  private final Path dataPath;
-  private final Path logPath;
-  private final SnapshotManager snapshotManager;
-
-  private TableImpl(String tablePath, Optional<TableIdentifier> tableIdOpt, Clock clock) {
+  public TableImpl(String tablePath, Clock clock) {
     this.tablePath = tablePath;
-    this.tableIdOpt = tableIdOpt;
-    this.clock = clock;
-    this.dataPath = new Path(tablePath);
-    this.logPath = new Path(dataPath, "_delta_log");
+    final Path dataPath = new Path(tablePath);
+    final Path logPath = new Path(dataPath, "_delta_log");
     this.snapshotManager = new SnapshotManager(logPath, dataPath);
+    this.clock = clock;
   }
 
-  /////////////////
-  // 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);
@@ -199,7 +169,8 @@ 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, tablePath, Optional.empty());
+                  TableFeatures.validateReadSupportedTable(
+                      protocol, getDataPath().toString(), Optional.empty());
                 }
               }
               if (shouldDropProtocolColumn) {
@@ -210,6 +181,14 @@ 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}
@@ -290,22 +269,6 @@ 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

@@ -236,7 +236,7 @@ class InCommitTimestampSuite extends DeltaTableWriteSuiteBase {
 
   test("Enablement tracking works when ICT is enabled post commit 0") {
     withTempDirAndEngine { (tablePath, engine) =>
-      val table = Table.forPath(engine, tablePath)
+      val table = TableImpl.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 = Table.forPath(engine, tablePath)
+      val table = TableImpl.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 = Table.forPath(engine, tablePath)
+      val table = TableImpl.forPath(engine, tablePath)
       val txnBuilder = table.createTransactionBuilder(engine, testEngineInfo, CREATE_TABLE)
 
       val txn = txnBuilder