[Kernel] [CatalogManaged] [UC] UCCatalogManagedClient loadTable tests (delta-io#4838)

## 🥞 Stacked PR
Use this [link](https://github.com/delta-io/delta/pull/4838/files) to
review incremental changes.
-
[**stack/uc_kernel_catalog_managed_3**](delta-io#4838)
[[Files changed](https://github.com/delta-io/delta/pull/4838/files)]

---------
#### 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

Use the UCCatalogManagedClient, instantiated with an InMemoryUCClient,
to load a ResolvedTable from a "real" UC catalog-managed table (a
published delta and two staged deltas that the InMemoryUCClient
acknowledges as ratified commits).

## How was this patch tested?

See above -- test only changes.

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

No.

Author: scottsand-db

unity/src/test/scala/io/delta/unity/InMemoryUCClient.scala

@@ -37,9 +37,9 @@ object InMemoryUCClient {
    * internal mutable state. This class is designed to be safely accessed by multiple threads
    * concurrently.
    */
-  class TableData {
-    private var maxRatifiedVersion = -1L
-    private val commits: ArrayBuffer[Commit] = ArrayBuffer.empty
+  class TableData(
+      private var maxRatifiedVersion: Long = -1L,
+      private val commits: ArrayBuffer[Commit] = ArrayBuffer.empty) {
 
     /** @return the maximum ratified version, or -1 if no commits have been made. */
     def getMaxRatifiedVersion: Long = synchronized {
@@ -152,6 +152,13 @@ class InMemoryUCClient(ucMetastoreId: String) extends UCClient {
 
   override def close(): Unit = {}
 
+  private[unity] def createTableIfNotExistsOrThrow(
+      ucTableId: String,
+      tableData: TableData): Unit = {
+    Option(tables.putIfAbsent(ucTableId, tableData))
+      .foreach(_ => throw new IllegalArgumentException(s"Table $ucTableId already exists"))
+  }
+
   private[unity] def getTablesCopy: Map[String, TableData] = {
     tables.asScala.toMap
   }

unity/src/test/scala/io/delta/unity/UCCatalogManagedClientSuite.scala

@@ -17,14 +17,56 @@
 package io.delta.unity
 
 import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
 
 import io.delta.kernel.internal.files.ParsedLogData.ParsedLogType
+import io.delta.kernel.internal.table.ResolvedTableInternal
+import io.delta.kernel.internal.tablefeatures.TableFeatures.{CATALOG_MANAGED_R_W_FEATURE_PREVIEW, TABLE_FEATURES_MIN_READER_VERSION, TABLE_FEATURES_MIN_WRITER_VERSION}
+import io.delta.kernel.internal.util.FileNames
+import io.delta.storage.commit.Commit
 
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileSystem, Path}
 import org.scalatest.funsuite.AnyFunSuite
 
 /** Unit tests for [[UCCatalogManagedClient]]. */
 class UCCatalogManagedClientSuite extends AnyFunSuite with UCCatalogManagedTestUtils {
 
+  private def testCatalogManagedTable(versionToLoad: Long): Unit = {
+    // Step 1: Create the in-memory table data (ratified commits v1, v2)
+    val tablePath = getTestResourceFilePath("catalog-owned-preview")
+    val ucClient = new InMemoryUCClient("ucMetastoreId")
+    val fs = FileSystem.get(new Configuration())
+    val catalogCommits = Seq(
+      // scalastyle:off line.size.limit
+      getTestResourceFilePath("catalog-owned-preview/_delta_log/_staged_commits/00000000000000000001.4cb9708e-b478-44de-b203-53f9ba9b2876.json"),
+      getTestResourceFilePath("catalog-owned-preview/_delta_log/_staged_commits/00000000000000000002.5b9bba4a-0085-430d-a65e-b0d38c1afbe9.json"))
+      // scalastyle:on line.size.limit
+      .map { path => fs.getFileStatus(new Path(path)) }
+      .map { fileStatus =>
+        new Commit(
+          FileNames.deltaVersion(fileStatus.getPath.toString),
+          fileStatus,
+          fileStatus.getModificationTime)
+      }
+    val tableData = new InMemoryUCClient.TableData(2, ArrayBuffer(catalogCommits: _*))
+    ucClient.createTableIfNotExistsOrThrow("ucTableId", tableData)
+
+    // Step 2: Load the table using UCCatalogManagedClient at the desired versionToLoad
+    val ucCatalogManagedClient = new UCCatalogManagedClient(ucClient)
+    val resolvedTable = ucCatalogManagedClient
+      .loadTable(defaultEngine, "ucTableId", tablePath, versionToLoad)
+      .asInstanceOf[ResolvedTableInternal]
+
+    // Step 3: Validate
+    val protocol = resolvedTable.getProtocol
+    assert(resolvedTable.getVersion == versionToLoad)
+    assert(protocol.getMinReaderVersion == TABLE_FEATURES_MIN_READER_VERSION)
+    assert(protocol.getMinWriterVersion == TABLE_FEATURES_MIN_WRITER_VERSION)
+    assert(protocol.getReaderFeatures.contains(CATALOG_MANAGED_R_W_FEATURE_PREVIEW.featureName()))
+    assert(protocol.getWriterFeatures.contains(CATALOG_MANAGED_R_W_FEATURE_PREVIEW.featureName()))
+  }
+
   test("constructor throws on invalid input") {
     assertThrows[NullPointerException] {
       new UCCatalogManagedClient(null)
@@ -49,6 +91,26 @@ class UCCatalogManagedClientSuite extends AnyFunSuite with UCCatalogManagedTestU
     }
   }
 
+  test("loadTable throws if version to load is greater than max ratified version") {
+    val exMsg = intercept[IllegalArgumentException] {
+      testCatalogManagedTable(versionToLoad = 9L)
+    }.getMessage
+
+    assert(exMsg.contains("Cannot load table version 9 as the latest version ratified by UC is 2"))
+  }
+
+  test("loadTable correctly loads a UC table -- versionToLoad is a ratified commit (the max)") {
+    testCatalogManagedTable(versionToLoad = 2L)
+  }
+
+  test("loadTable correctly loads a UC table -- versionToLoad is a ratified commit (not the max)") {
+    testCatalogManagedTable(versionToLoad = 1L)
+  }
+
+  test("loadTable correctly loads a UC table -- versionToLoad is a published commit") {
+    testCatalogManagedTable(versionToLoad = 0L)
+  }
+
   test("converts UC Commit into Kernel ParsedLogData.RATIFIED_STAGED_COMMIT") {
     val ucCommit = createCommit(1)
     val hadoopFS = ucCommit.getFileStatus