[Storage] Implement getCommits in UC Delta token-based client

storage/src/main/java/io/delta/storage/commit/uccommitcoordinator/UCDeltaTokenBasedRestClient.java

@@ -18,6 +18,7 @@
 
 import io.delta.storage.commit.Commit;
 import io.delta.storage.commit.CommitFailedException;
+import io.delta.storage.commit.CoordinatedCommitsUtils;
 import io.delta.storage.commit.GetCommitsResponse;
 import io.delta.storage.commit.TableIdentifier;
 import io.delta.storage.commit.actions.AbstractMetadata;
@@ -70,6 +71,8 @@
 import java.util.function.Supplier;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
 
 /**
  * A REST client implementation of {@link UCDeltaClient} that uses the UC Delta REST API for
@@ -284,9 +287,92 @@ public GetCommitsResponse getCommits(
       TableIdentifier tableIdentifier,
       Optional<Long> startVersion,
       Optional<Long> endVersion) throws IOException, UCCommitCoordinatorException {
-    throw new UnsupportedOperationException(
-        "getCommits is not yet supported by UCDeltaTokenBasedRestClient. " +
-            "A separate PR will add this once the tableIdentifier mapping is available.");
+    ensureOpen();
+    Objects.requireNonNull(tableId, "tableId must not be null");
+    Objects.requireNonNull(tableUri, "tableUri must not be null");
+    Objects.requireNonNull(tableIdentifier, "tableIdentifier must not be null");
+    Objects.requireNonNull(startVersion, "startVersion must not be null");
+    Objects.requireNonNull(endVersion, "endVersion must not be null");
+
+    UUID expectedTableUuid = UUID.fromString(tableId);
+    String[] namespace = Objects.requireNonNull(
+        tableIdentifier.getNamespace(), "tableIdentifier namespace must not be null");
+    if (namespace.length != 2) {
+      throw new IllegalArgumentException(
+          "tableIdentifier must be a three-part Unity Catalog table name");
+    }
+    String catalog = Objects.requireNonNull(namespace[0], "catalog name must not be null");
+    String schema = Objects.requireNonNull(namespace[1], "schema name must not be null");
+    String table = Objects.requireNonNull(tableIdentifier.getName(), "table name must not be null");
+    String fullName = catalog + "." + schema + "." + table;
+
+    // The UC loadTable endpoint does not support server-side filtering by version range, so
+    // we fetch the full unbackfilled commit window and filter client-side below. The server
+    // bounds the window size, so this list is not unbounded in practice.
+    LoadTableResponse response;
+    try {
+      response = deltaTablesApi.loadTable(catalog, schema, table);
+    } catch (ApiException e) {
+      if (e.getCode() == HTTP_NOT_FOUND) {
+        throw new InvalidTargetTableException(
+            String.format("Table not found %s: %s", fullName, e.getResponseBody()));
+      }
+      throw new IOException(
+          String.format("Failed to load commits for table %s (HTTP %s): %s",
+              fullName, e.getCode(), e.getResponseBody()),
+          e);
+    }
+
+    TableMetadata metadata = response.getMetadata();
+    UUID actualTableUuid = metadata != null ? metadata.getTableUuid() : null;
+    if (!expectedTableUuid.equals(actualTableUuid)) {
+      throw new InvalidTargetTableException(
+          String.format(
+              "Table UUID mismatch for %s: expected %s but got %s",
+              fullName,
+              expectedTableUuid,
+              actualTableUuid));
+    }
+
+    Path basePath = CoordinatedCommitsUtils.commitDirPath(
+        CoordinatedCommitsUtils.logDirPath(new Path(tableUri)));
+    List<Commit> commits = new ArrayList<>();
+    if (response.getCommits() != null) {
+      for (DeltaCommit deltaCommit : response.getCommits()) {
+        long version = Objects.requireNonNull(
+            deltaCommit.getVersion(), "commit version must not be null");
+        if (startVersion.isPresent() && version < startVersion.get()) {
+          continue;
+        }
+        if (endVersion.isPresent() && version > endVersion.get()) {
+          continue;
+        }
+        commits.add(fromDeltaCommit(deltaCommit, basePath));
+      }
+    }
+
+    long latestTableVersion = response.getLatestTableVersion() != null
+        ? response.getLatestTableVersion() : -1L;
+    return new GetCommitsResponse(commits, latestTableVersion);
+  }
+
+  /** Converts a UC SDK {@link DeltaCommit} to a Delta {@link Commit}. */
+  private Commit fromDeltaCommit(DeltaCommit deltaCommit, Path basePath) {
+    FileStatus fileStatus = new FileStatus(
+        Objects.requireNonNull(
+            deltaCommit.getFileSize(), "commit fileSize must not be null"),
+        false /* isdir */,
+        0 /* block_replication */,
+        0 /* blocksize */,
+        Objects.requireNonNull(
+            deltaCommit.getFileModificationTimestamp(),
+            "commit fileModificationTimestamp must not be null"),
+        new Path(basePath, Objects.requireNonNull(
+            deltaCommit.getFileName(), "commit fileName must not be null")));
+    return new Commit(
+        Objects.requireNonNull(deltaCommit.getVersion(), "commit version must not be null"),
+        fileStatus,
+        Objects.requireNonNull(deltaCommit.getTimestamp(), "commit timestamp must not be null"));
   }
 
   @Override

storage/src/test/scala/io/delta/storage/commit/uccommitcoordinator/UCDeltaTokenBasedRestClientSuite.scala

@@ -32,6 +32,7 @@ import org.apache.hadoop.fs.{FileStatus, Path}
 import org.apache.http.HttpStatus
 import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
 import org.scalatest.funsuite.AnyFunSuite
+import scala.jdk.CollectionConverters._
 
 class UCDeltaTokenBasedRestClientSuite
     extends AnyFunSuite
@@ -81,15 +82,27 @@ class UCDeltaTokenBasedRestClientSuite
       tableUuid: UUID = testTableId,
       format: String = "DELTA",
       location: String = "s3://bucket/table",
-      tableType: String = "MANAGED"): String =
+      tableType: String = "MANAGED",
+      commitsJson: String = "[]",
+      latestTableVersion: Long = -1L): String =
     s"""{"metadata":{"table-uuid":"$tableUuid","data-source-format":"$format",""" +
     s""""table-type":"$tableType",""" +
     s""""location":"$location",""" +
     s""""columns":{"type":"struct","fields":[""" +
     s"""{"name":"date","type":"string","nullable":true,"metadata":{}},""" +
     s"""{"name":"value","type":"integer","nullable":true,"metadata":{}}""" +
     s"""]},""" +
-    s""""properties":{"key1":"val1"},"partition-columns":["date"],"created-time":1000}}"""
+    s""""properties":{"key1":"val1"},"partition-columns":["date"],"created-time":1000},""" +
+    s""""commits":$commitsJson,"latest-table-version":$latestTableVersion}"""
+
+  private def deltaCommitJson(
+      version: Long,
+      fileName: String,
+      fileSize: Long,
+      timestamp: Long,
+      fileModificationTimestamp: Long): String =
+    s"""{"version":$version,"file-name":"$fileName","file-size":$fileSize,""" +
+    s""""timestamp":$timestamp,"file-modification-timestamp":$fileModificationTimestamp}"""
 
   private def readBody(exchange: HttpExchange): String = {
     val is = exchange.getRequestBody
@@ -585,9 +598,93 @@ class UCDeltaTokenBasedRestClientSuite
 
   // --------------- getCommits ---------------
 
-  test("getCommits throws UnsupportedOperationException") {
+  test("getCommits loads table by identifier and returns commits") {
+    var capturedMethod: String = null
+    var capturedPath: String = null
+    val commitsJson = "[" +
+      deltaCommitJson(2L, "00000000000000000002.uuid.json", 200L, 2000L, 2001L) +
+      "]"
+    deltaHandler = (exchange, _) => {
+      capturedMethod = exchange.getRequestMethod
+      capturedPath = exchange.getRequestURI.getPath
+      sendJson(exchange, HttpStatus.SC_OK,
+        loadTableJson(commitsJson = commitsJson, latestTableVersion = 2L))
+    }
+
+    withClient { c =>
+      val response = c.getCommits(testTableId.toString, new URI("s3://b/t"), testIdentifier,
+        Optional.empty(), Optional.empty())
+
+      assert(capturedMethod === "GET")
+      assert(capturedPath ===
+        "/api/2.1/unity-catalog/delta/v1/catalogs/cat/schemas/sch/tables/tbl")
+      assert(response.getLatestTableVersion === 2L)
+      assert(response.getCommits.size() === 1)
+
+      val commit = response.getCommits.get(0)
+      assert(commit.getVersion === 2L)
+      assert(commit.getCommitTimestamp === 2000L)
+      assert(commit.getFileStatus.getLen === 200L)
+      assert(commit.getFileStatus.getModificationTime === 2001L)
+      assert(commit.getFileStatus.getPath.toString ===
+        "s3://b/t/_delta_log/_staged_commits/00000000000000000002.uuid.json")
+    }
+  }
+
+  test("getCommits filters loaded commits by requested version range") {
+    val commitsJson = Seq(
+      deltaCommitJson(1L, "1.uuid.json", 100L, 1000L, 1001L),
+      deltaCommitJson(2L, "2.uuid.json", 200L, 2000L, 2001L),
+      deltaCommitJson(3L, "3.uuid.json", 300L, 3000L, 3001L),
+      deltaCommitJson(4L, "4.uuid.json", 400L, 4000L, 4001L)
+    ).mkString("[", ",", "]")
+    deltaHandler = (exchange, _) =>
+      sendJson(exchange, HttpStatus.SC_OK,
+        loadTableJson(commitsJson = commitsJson, latestTableVersion = 4L))
+
     withClient { c =>
-      intercept[UnsupportedOperationException] {
+      val response = c.getCommits(testTableId.toString, new URI("s3://b/t"), testIdentifier,
+        Optional.of(java.lang.Long.valueOf(2L)),
+        Optional.of(java.lang.Long.valueOf(3L)))
+
+      assert(response.getLatestTableVersion === 4L)
+      assert(response.getCommits.asScala.map(_.getVersion).toSeq === Seq(2L, 3L))
+    }
+  }
+
+  test("getCommits validates required parameters") {
+    withClient { c =>
+      intercept[NullPointerException] {
+        c.getCommits(null, new URI("s3://b/t"), testIdentifier,
+          Optional.empty(), Optional.empty())
+      }
+      intercept[NullPointerException] {
+        c.getCommits(testTableId.toString, null, testIdentifier,
+          Optional.empty(), Optional.empty())
+      }
+      intercept[NullPointerException] {
+        c.getCommits(testTableId.toString, new URI("s3://b/t"), null,
+          Optional.empty(), Optional.empty())
+      }
+    }
+  }
+
+  test("getCommits throws InvalidTargetTableException on 404") {
+    deltaHandler = (exchange, _) => sendJson(exchange, HttpStatus.SC_NOT_FOUND, "{}")
+    withClient { c =>
+      intercept[InvalidTargetTableException] {
+        c.getCommits(testTableId.toString, new URI("s3://b/t"), testIdentifier,
+          Optional.empty(), Optional.empty())
+      }
+    }
+  }
+
+  test("getCommits throws InvalidTargetTableException when table UUID does not match") {
+    deltaHandler = (exchange, _) =>
+      sendJson(exchange, HttpStatus.SC_OK,
+        loadTableJson(tableUuid = UUID.fromString("550e8400-e29b-41d4-a716-446655440001")))
+    withClient { c =>
+      intercept[InvalidTargetTableException] {
         c.getCommits(testTableId.toString, new URI("s3://b/t"), testIdentifier,
           Optional.empty(), Optional.empty())
       }