Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
openinx merged 6 commits into from
May 20, 2026
+234 −30
Merged

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

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
153b88a
[Storage] Implement UC token-based getCommits
TimothyW553 May 19, 2026
473ef06
[Storage] Address review feedback for UC token-based getCommits
TimothyW553 May 19, 2026
9b9b931
[Storage] Address openinx review comments
TimothyW553 May 19, 2026
d6167e4
[Storage] Address review nits
TimothyW553 May 19, 2026
ee014c8
[Storage] Compare tableId as String, not UUID
TimothyW553 May 19, 2026
d246102
Merge branch 'master' into ucdelta-getcommits-pr6788
TimothyW553 May 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 124 additions & 26 deletions ...rc/main/java/io/delta/storage/commit/uccommitcoordinator/UCDeltaTokenBasedRestClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -233,7 +236,7 @@ public void commit(
throws IOException, CommitFailedException, UCCommitCoordinatorException {
ensureOpen();
Objects.requireNonNull(tableId, "tableId must not be null");
Objects.requireNonNull(tableIdentifier, "tableIdentifier must not be null");
ResolvedTableName name = resolveThreePartName(tableIdentifier);

UpdateTableRequest request = new UpdateTableRequest();
request.addRequirementsItem(new AssertTableUUID()
Expand Down Expand Up @@ -266,14 +269,10 @@ public void commit(
.protocol(toSDKDeltaProtocol(newProtocol.get())));
}

String catalog = tableIdentifier.getNamespace()[0];
String schema = tableIdentifier.getNamespace()[1];
String table = tableIdentifier.getName();

try {
deltaTablesApi.updateTable(catalog, schema, table, request);
deltaTablesApi.updateTable(name.catalog, name.schema, name.table, request);
} catch (ApiException e) {
handleUpdateTableException(e, catalog, schema, table);
handleUpdateTableException(e, name.catalog, name.schema, name.table);
}
}

Expand All @@ -284,9 +283,82 @@ 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(startVersion, "startVersion must not be null");
Objects.requireNonNull(endVersion, "endVersion must not be null");

UUID expectedTableUuid = UUID.fromString(tableId);
Copy link
Copy Markdown
Collaborator

@openinx openinx May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

wait, does all the tableId is a unique string that can be serialized to be strickly UUID ? I think OSS UC server should have the implication, but pls take a look for the delta uc spec, and see if it clearly says it's a string that can be deserialized to a UUID ?

Otherwise, if the table id is a unique one, but it will not be able to deserialize to a UUID, then that's a serious bug.

Copy link
Copy Markdown
Collaborator

@openinx openinx May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I will more suggest to use a String, rather than explicitly deserializing to a UUID instance.

Copy link
Copy Markdown
Collaborator

@openinx openinx May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

okay, the spec clearly says it's UUID. https://github.com/unitycatalog/unitycatalog/blob/main/api/delta-docs/Models/TableMetadata.md

But I will still suggest to use a String.

Copy link
Copy Markdown
Collaborator Author

@TimothyW553 TimothyW553 May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good catch, switched to String.

Copy link
Copy Markdown
Collaborator

@yili-db yili-db May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is fine to be a String here, only because the function signature says "String".
In this case, aString.equals(anotherUUID.toString()) is better than UUID.fromString(...).equals(anotherUUID)

ResolvedTableName name = resolveThreePartName(tableIdentifier);

// 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(name.catalog, name.schema, name.table);
} catch (ApiException e) {
if (e.getCode() == HTTP_NOT_FOUND) {
throw new NoSuchTableException(
String.format("Table %s not found in Unity Catalog", name.fullName), e);
}
throw new IOException(
String.format("Failed to load commits for table %s (HTTP %s): %s",
name.fullName, e.getCode(), e.getResponseBody()),
e);
}

TableMetadata metadata = response.getMetadata();
UUID actualTableUuid = metadata != null ? metadata.getTableUuid() : null;
if (!expectedTableUuid.equals(actualTableUuid)) {
Copy link
Copy Markdown
Collaborator

@openinx openinx May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So this will happen in the following case:

  • t1: create a table cat.db.table, with tableId = 111.
  • t2: drop the table .
  • t3: construct the getCommits (cat.db.table, tableId=111).
  • t4: re-create the table, tableId=222.
  • t5: send the getCommits(cat.db.table, tableId=111).

Copy link
Copy Markdown
Collaborator Author

@TimothyW553 TimothyW553 May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes — this is exactly the case the UUID check protects against.

throw new InvalidTargetTableException(
String.format(
"Table UUID mismatch for %s: expected %s but got %s",
name.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;
Copy link
Copy Markdown
Collaborator

@openinx openinx May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

will it be the case if the table does not even have a single table version ?

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"),
Copy link
Copy Markdown
Collaborator

@openinx openinx May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

will it possible for this getFileModificationTimestamp to be nullable ? the annotation already says is not nullable.

@jakarta.annotation.Nonnull
  @JsonProperty(JSON_PROPERTY_FILE_MODIFICATION_TIMESTAMP)
  @JsonInclude(value = JsonInclude.Include.ALWAYS)
  public Long getFileModificationTimestamp() {
    return fileModificationTimestamp;
  }

Copy link
Copy Markdown
Collaborator Author

@TimothyW553 TimothyW553 May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, it's @Nonnull in the SDK — removed the requireNonNull.

new Path(basePath, Objects.requireNonNull(
deltaCommit.getFileName(), "commit fileName must not be null")));
Copy link
Copy Markdown
Collaborator

@openinx openinx May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same comment for getFileName.

Copy link
Copy Markdown
Collaborator Author

@TimothyW553 TimothyW553 May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same here — removed the requireNonNull on getFileName too.

return new Commit(
Objects.requireNonNull(deltaCommit.getVersion(), "commit version must not be null"),
fileStatus,
Objects.requireNonNull(deltaCommit.getTimestamp(), "commit timestamp must not be null"));
Copy link
Copy Markdown
Collaborator

@openinx openinx May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

those nullable checks may not necessary.

Copy link
Copy Markdown
Collaborator Author

@TimothyW553 TimothyW553 May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

you're right, removed all the requireNonNull from fromDeltaCommit.

}

@Override
Expand Down Expand Up @@ -339,38 +411,30 @@ public void close() throws IOException {
@Override
public TableInfo loadTable(TableIdentifier tableIdentifier) throws IOException {
ensureOpen();
Objects.requireNonNull(tableIdentifier, "tableIdentifier must not be null");
String[] namespace = tableIdentifier.getNamespace();
if (namespace == null || namespace.length != 2) {
throw new IllegalArgumentException(
"UC tableIdentifier must have a 2-component namespace [catalog, schema]; got " +
(namespace == null ? "null" : java.util.Arrays.toString(namespace)));
}
String catalog = namespace[0];
String schema = namespace[1];
String table = tableIdentifier.getName();
ResolvedTableName name = resolveThreePartName(tableIdentifier);

try {
return toTableInfo(
deltaTablesApi.loadTable(catalog, schema, table), catalog, schema, table);
deltaTablesApi.loadTable(name.catalog, name.schema, name.table),
name.catalog, name.schema, name.table);
} catch (ApiException e) {
if (e.getCode() == HTTP_NOT_FOUND) {
throw new NoSuchTableException(
String.format("Table %s.%s.%s not found in Unity Catalog", catalog, schema, table), e);
String.format("Table %s not found in Unity Catalog", name.fullName), e);
}
// UC encodes non-Delta-format errors as HTTP 400 with error.type =
// "UnsupportedTableFormatException"; substring-match the body to avoid coupling to an
// ErrorResponse parser.
String body = e.getResponseBody();
if (body != null && body.contains("UnsupportedTableFormatException")) {
throw new UnsupportedTableFormatException(
String.format("Table %s.%s.%s is not in Delta format; the Delta REST API cannot "
+ "serve it. Body: %s", catalog, schema, table, body),
String.format("Table %s is not in Delta format; the Delta REST API cannot "
+ "serve it. Body: %s", name.fullName, body),
e);
}
throw new IOException(
String.format("Failed to load table %s.%s.%s (HTTP %s): %s",
catalog, schema, table, e.getCode(), e.getResponseBody()), e);
String.format("Failed to load table %s (HTTP %s): %s",
name.fullName, e.getCode(), e.getResponseBody()), e);
}
}

Expand Down Expand Up @@ -640,6 +704,25 @@ private void addMetadataUpdates(
}
}

// ===========================
// Table Identifier Helpers
// ===========================

/**
* Validates that the given {@code tableIdentifier} is a Unity Catalog three-part
* (catalog.schema.table) name and returns its resolved parts.
*/
private static ResolvedTableName resolveThreePartName(TableIdentifier tableIdentifier) {
Objects.requireNonNull(tableIdentifier, "tableIdentifier must not be null");
String[] namespace = tableIdentifier.getNamespace();
if (namespace == null || namespace.length != 2) {
throw new IllegalArgumentException(
"UC tableIdentifier must have a 2-component namespace [catalog, schema]; got " +
(namespace == null ? "null" : java.util.Arrays.toString(namespace)));
}
return new ResolvedTableName(namespace[0], namespace[1], tableIdentifier.getName());
}

// ===========================
// Exception Handling
// ===========================
Expand Down Expand Up @@ -673,6 +756,21 @@ private void handleUpdateTableException(
// Inner Classes
// ===========================

/** A Unity Catalog three-part table name resolved from a {@link TableIdentifier}. */
Copy link
Copy Markdown
Collaborator Author

@TimothyW553 TimothyW553 May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is intentional kept inner - its a private impl detail

private static final class ResolvedTableName {
final String catalog;
final String schema;
final String table;
final String fullName;

ResolvedTableName(String catalog, String schema, String table) {
this.catalog = catalog;
this.schema = schema;
this.table = table;
this.fullName = catalog + "." + schema + "." + table;
}
}

/**
* Adapts a UC SDK {@link TableMetadata} to {@link AbstractMetadata}.
*/
Expand Down
108 changes: 104 additions & 4 deletions .../scala/io/delta/storage/commit/uccommitcoordinator/UCDeltaTokenBasedRestClientSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -585,9 +598,96 @@ 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 =>
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 NoSuchTableException on 404") {
deltaHandler = (exchange, _) =>
sendJson(exchange, HttpStatus.SC_NOT_FOUND, """{"error":"not found"}""")
withClient { c =>
val e = intercept[NoSuchTableException] {
c.getCommits(testTableId.toString, new URI("s3://b/t"), testIdentifier,
Optional.empty(), Optional.empty())
}
assert(e.getMessage.contains(s"$testCatalog.$testSchema.$testTable"))
assert(e.getMessage.contains("not found"))
}
}

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[UnsupportedOperationException] {
intercept[InvalidTargetTableException] {
c.getCommits(testTableId.toString, new URI("s3://b/t"), testIdentifier,
Optional.empty(), Optional.empty())
}
Expand Down
Loading