[Kernel] [Refactor] Refactor TestUtils to provide reading APIs using both legacy (Snapshot) and new (ResolvedTable) APIs (delta-io#4676)

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

---------
#### Which Delta project/connector is this regarding?

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

## Description

As part of the catalogManaged initiative, we are introducing new APIs to
read and write delta tables: `TableManager` and `ResolvedTable` instead
of `Table` and `Snapshot`

As this initiative is ongoing, we want to test both APIs until we are
ready and confident in deprecating the legacy API.

This PR refactors `TestUtils` so that: we can have both `New` and
`Legacy` implementations of it.

We also update `DeltaTableReadsSuite` to show an example of it working.

## How was this patch tested?

Existing UTs.

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

No.

Author: scottsand-db

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

@@ -24,7 +24,7 @@ import scala.collection.JavaConverters._
 
 import io.delta.golden.GoldenTableUtils.goldenTablePath
 import io.delta.kernel.Table
-import io.delta.kernel.defaults.utils.{TestRow, TestUtils}
+import io.delta.kernel.defaults.utils.{AbstractTestUtils, TestRow, TestUtils, TestUtilsWithLegacyKernelAPIs}
 import io.delta.kernel.exceptions.{InvalidTableException, KernelException, TableNotFoundException}
 import io.delta.kernel.internal.TableImpl
 import io.delta.kernel.internal.fs.Path
@@ -33,15 +33,22 @@ import io.delta.kernel.internal.util.InternalUtils.daysSinceEpoch
 import io.delta.kernel.types.{LongType, StructType}
 
 import org.apache.spark.sql.delta.{DeltaLog, DeltaOperations}
-import org.apache.spark.sql.delta.actions.{AddFile, Metadata}
-import org.apache.spark.sql.delta.implicits.stringEncoder
+import org.apache.spark.sql.delta.actions.AddFile
 
 import org.apache.hadoop.shaded.org.apache.commons.io.FileUtils
-import org.apache.spark.sql.functions.{col, current_timestamp, to_timestamp}
 import org.apache.spark.sql.functions.col
 import org.scalatest.funsuite.AnyFunSuite
 
-class DeltaTableReadsSuite extends AnyFunSuite with TestUtils {
+class LegacyDeltaTableReadsSuite
+    extends AbstractDeltaTableReadsSuite
+    with TestUtilsWithLegacyKernelAPIs
+
+// TODO: uncomment when new Kernel APIs supports ScanBuilder
+// class DeltaTableReadsSuite extends
+//     AbstractDeltaTableReadsSuite
+//     with TestUtilsWithTableManagerAPIs
+
+trait AbstractDeltaTableReadsSuite extends AnyFunSuite { self: AbstractTestUtils =>
 
   //////////////////////////////////////////////////////////////////////////////////
   // Timestamp type tests

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

@@ -0,0 +1,66 @@
+/*
+ * Copyright (2025) 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.defaults.test
+
+import io.delta.kernel.{ResolvedTable, ScanBuilder, Snapshot}
+import io.delta.kernel.internal.SnapshotImpl
+import io.delta.kernel.types.StructType
+
+/** Implicit conversions that enable easy and succinct adapter creation in tests. */
+object ResolvedTableAdapterImplicits {
+
+  /** Converts a [[Snapshot]] to a [[LegacyResolvedTableAdapter]]. */
+  implicit class AdapterFromSnapshot(private val snapshot: Snapshot) extends AnyVal {
+    def toTestAdapter: LegacyResolvedTableAdapter = snapshot match {
+      case impl: SnapshotImpl => new LegacyResolvedTableAdapter(impl)
+      case _ => throw new IllegalArgumentException("Snapshot must be an instance of SnapshotImpl")
+    }
+  }
+
+  /** Converts a [[ResolvedTable]] to a [[ResolvedTableAdapter]]. */
+  implicit class AdapterFromResolvedTable(private val resolvedTable: ResolvedTable) extends AnyVal {
+    def toTestAdapter: ResolvedTableAdapter = new ResolvedTableAdapter(resolvedTable)
+  }
+}
+
+/**
+ * Test framework adapter that provides a unified interface for **accessing** Delta tables.
+ *
+ * This trait abstracts over the differences between [[Snapshot]] (legacy API) and [[ResolvedTable]]
+ * (current API) via the [[LegacyResolvedTableAdapter]] and [[ResolvedTableAdapter]] child classes.
+ */
+trait AbstractResolvedTableAdapter {
+  def getPath(): String
+  def getVersion(): Long
+  def getSchema(): StructType
+  def getScanBuilder(): ScanBuilder
+}
+
+class LegacyResolvedTableAdapter(snapshot: SnapshotImpl) extends AbstractResolvedTableAdapter {
+  override def getPath(): String = snapshot.getDataPath.toString
+  override def getVersion(): Long = snapshot.getVersion
+  override def getSchema(): StructType = snapshot.getSchema
+  override def getScanBuilder(): ScanBuilder = snapshot.getScanBuilder
+}
+
+// TODO: Use ResolvedTableInternal
+class ResolvedTableAdapter(rt: ResolvedTable) extends AbstractResolvedTableAdapter {
+  override def getPath(): String = rt.getPath
+  override def getVersion(): Long = rt.getVersion
+  override def getSchema(): StructType = rt.getSchema
+  override def getScanBuilder(): ScanBuilder = rt.getScanBuilder
+}

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

@@ -0,0 +1,99 @@
+/*
+ * Copyright (2025) 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.defaults.test
+
+import io.delta.kernel.{Table, TableManager}
+import io.delta.kernel.defaults.test.ResolvedTableAdapterImplicits._
+import io.delta.kernel.engine.Engine
+
+/**
+ * Test framework adapter that provides a unified interface for **loading** Delta tables.
+ *
+ * This trait enables test suites to be parameterized over different Kernel APIs via the
+ * [[LegacyTableManagerAdapter]] and [[TableManagerAdapter]] child classes.
+ *
+ * By using this adapter pattern, the same test suite can verify both APIs work correctly, without
+ * duplicating test logic.
+ *
+ * Tests can switch between implementations by mixing in either
+ * [[io.delta.kernel.defaults.utils.TestUtilsWithLegacyKernelAPIs]] or
+ * [[io.delta.kernel.defaults.utils.TestUtilsWithTableManagerAPIs]].
+ */
+trait AbstractTableManagerAdapter {
+  def getResolvedTableAdapterAtLatest(engine: Engine, path: String): AbstractResolvedTableAdapter
+
+  def getResolvedTableAdapterAtVersion(
+      engine: Engine,
+      path: String,
+      version: Long): AbstractResolvedTableAdapter
+
+  def getResolvedTableAdapterAtTimestamp(
+      engine: Engine,
+      path: String,
+      timestamp: Long): AbstractResolvedTableAdapter
+}
+
+/**
+ * Legacy implementation using the [[Table.forPath]] API which then wraps a [[Snapshot]].
+ */
+class LegacyTableManagerAdapter extends AbstractTableManagerAdapter {
+  override def getResolvedTableAdapterAtLatest(
+      engine: Engine,
+      path: String): AbstractResolvedTableAdapter = {
+    Table.forPath(engine, path).getLatestSnapshot(engine).toTestAdapter
+  }
+
+  override def getResolvedTableAdapterAtVersion(
+      engine: Engine,
+      path: String,
+      version: Long): AbstractResolvedTableAdapter = {
+    Table.forPath(engine, path).getSnapshotAsOfVersion(engine, version).toTestAdapter
+  }
+
+  override def getResolvedTableAdapterAtTimestamp(
+      engine: Engine,
+      path: String,
+      timestamp: Long): AbstractResolvedTableAdapter = {
+    Table.forPath(engine, path).getSnapshotAsOfTimestamp(engine, timestamp).toTestAdapter
+  }
+}
+
+/**
+ * Current implementation using the [[TableManager.loadTable]] API, which then wraps a
+ * [[ResolvedTable]].
+ */
+class TableManagerAdapter extends AbstractTableManagerAdapter {
+  override def getResolvedTableAdapterAtLatest(
+      engine: Engine,
+      path: String): AbstractResolvedTableAdapter = {
+    TableManager.loadTable(path).build(engine).toTestAdapter
+  }
+
+  override def getResolvedTableAdapterAtVersion(
+      engine: Engine,
+      path: String,
+      version: Long): AbstractResolvedTableAdapter = {
+    TableManager.loadTable(path).atVersion(version).build(engine).toTestAdapter
+  }
+
+  override def getResolvedTableAdapterAtTimestamp(
+      engine: Engine,
+      path: String,
+      timestamp: Long): AbstractResolvedTableAdapter = {
+    throw new UnsupportedOperationException("not implemented")
+  }
+}

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/utils/TestUtils.scala

@@ -28,10 +28,11 @@ import io.delta.kernel.{Scan, Snapshot, Table, TransactionCommitResult}
 import io.delta.kernel.data.{ColumnarBatch, ColumnVector, FilteredColumnarBatch, MapValue, Row}
 import io.delta.kernel.defaults.engine.DefaultEngine
 import io.delta.kernel.defaults.internal.data.vector.{DefaultGenericVector, DefaultStructVector}
+import io.delta.kernel.defaults.test.{AbstractResolvedTableAdapter, AbstractTableManagerAdapter, LegacyTableManagerAdapter, TableManagerAdapter}
 import io.delta.kernel.engine.Engine
 import io.delta.kernel.expressions.{Column, Predicate}
 import io.delta.kernel.hook.PostCommitHook.PostCommitHookType
-import io.delta.kernel.internal.{InternalScanFileUtils, SnapshotImpl, TableImpl}
+import io.delta.kernel.internal.{InternalScanFileUtils, SnapshotImpl}
 import io.delta.kernel.internal.actions.DomainMetadata
 import io.delta.kernel.internal.checksum.{ChecksumReader, ChecksumWriter, CRCInfo}
 import io.delta.kernel.internal.clustering.ClusteringMetadataDomain
@@ -55,7 +56,25 @@ import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.plans.SQLHelper
 import org.scalatest.Assertions
 
-trait TestUtils extends Assertions with SQLHelper {
+trait TestUtils extends AbstractTestUtils {
+  override def getTableManagerAdapter: AbstractTableManagerAdapter = new LegacyTableManagerAdapter()
+}
+
+/**
+ * DO NOT MODIFY this trait -- this is just syntactic sugar to clearly indicate we are extending the
+ * "default" TestUtils which happens to use the legacy Kernel APIs
+ */
+trait TestUtilsWithLegacyKernelAPIs extends TestUtils
+
+trait TestUtilsWithTableManagerAPIs extends AbstractTestUtils {
+  override def getTableManagerAdapter: AbstractTableManagerAdapter = new TableManagerAdapter()
+}
+
+trait AbstractTestUtils extends Assertions with SQLHelper {
+
+  import io.delta.kernel.defaults.test.ResolvedTableAdapterImplicits._
+
+  def getTableManagerAdapter: AbstractTableManagerAdapter
 
   lazy val configuration = new Configuration()
   lazy val defaultEngine = DefaultEngine.create(configuration)
@@ -210,10 +229,24 @@ trait TestUtils extends Assertions with SQLHelper {
       filter: Predicate = null,
       expectedRemainingFilter: Predicate = null,
       engine: Engine = defaultEngine): Seq[Row] = {
+    readResolvedTableAdapter(
+      snapshot.toTestAdapter,
+      readSchema,
+      filter,
+      expectedRemainingFilter,
+      engine)
+  }
+
+  def readResolvedTableAdapter(
+      resolvedTableAdapter: AbstractResolvedTableAdapter,
+      readSchema: StructType = null,
+      filter: Predicate = null,
+      expectedRemainingFilter: Predicate = null,
+      engine: Engine = defaultEngine): Seq[Row] = {
 
     val result = ArrayBuffer[Row]()
 
-    var scanBuilder = snapshot.getScanBuilder()
+    var scanBuilder = resolvedTableAdapter.getScanBuilder()
 
     if (readSchema != null) {
       scanBuilder = scanBuilder.withReadSchema(readSchema)
@@ -375,40 +408,46 @@ trait TestUtils extends Assertions with SQLHelper {
       expectedVersion: Option[Long] = None): Unit = {
     assert(version.isEmpty || timestamp.isEmpty, "Cannot provide both a version and timestamp")
 
-    val snapshot = if (version.isDefined) {
-      Table.forPath(engine, path)
-        .getSnapshotAsOfVersion(engine, version.get)
+    val resolvedTableAdapter = if (version.isDefined) {
+      getTableManagerAdapter.getResolvedTableAdapterAtVersion(engine, path, version.get)
     } else if (timestamp.isDefined) {
-      Table.forPath(engine, path)
-        .getSnapshotAsOfTimestamp(engine, timestamp.get)
+      getTableManagerAdapter.getResolvedTableAdapterAtTimestamp(engine, path, timestamp.get)
     } else {
-      latestSnapshot(path, engine)
+      getTableManagerAdapter.getResolvedTableAdapterAtLatest(engine, path)
     }
 
     val readSchema = if (readCols == null) {
       null
     } else {
-      val schema = snapshot.getSchema()
+      val schema = resolvedTableAdapter.getSchema()
       new StructType(readCols.map(schema.get(_)).asJava)
     }
 
     if (expectedSchema != null) {
       assert(
-        expectedSchema == snapshot.getSchema(),
+        expectedSchema == resolvedTableAdapter.getSchema(),
         s"""
            |Expected schema does not match actual schema:
            |Expected schema: $expectedSchema
-           |Actual schema: ${snapshot.getSchema()}
+           |Actual schema: ${resolvedTableAdapter.getSchema()}
            |""".stripMargin)
     }
 
+    val actualVersion = resolvedTableAdapter.getVersion()
+
     expectedVersion.foreach { version =>
       assert(
-        version == snapshot.getVersion(),
-        s"Expected version $version does not match actual version ${snapshot.getVersion()}")
+        version == actualVersion,
+        s"Expected version $version does not match actual version $actualVersion}")
     }
 
-    val result = readSnapshot(snapshot, readSchema, filter, expectedRemainingFilter, engine)
+    val result =
+      readResolvedTableAdapter(
+        resolvedTableAdapter,
+        readSchema,
+        filter,
+        expectedRemainingFilter,
+        engine)
     checkAnswer(result, expectedAnswer)
   }