Skip to content

Commit a1edad4

Browse files
timothyw553timothyw553
committed
[Storage] Add tableIdentifier to UCClient getCommits
Adds `TableIdentifier` to `UCClient#getCommits` so UC clients can receive the catalog/schema/table name when fetching commits. The identifier is forwarded from `UCCommitCoordinatorClient` when available in the `TableDescriptor`. Kernel catalog-managed snapshot loading also has an overload that forwards the identifier, and Spark v2 UC snapshot metadata now carries the identifier from `CatalogTable` into that Kernel path. The legacy `UCTokenBasedRestClient` accepts the new argument but keeps sending the existing legacy request fields. `tableIdentifier` is the three-part `catalog.schema.table` name (not the UC UUID `tableId`); callers pass it when they have catalog context and null otherwise, and receivers either require it (rejecting null) or ignore it. Resolves #6784.
1 parent b5e5aec commit a1edad4

30 files changed

Lines changed: 371 additions & 52 deletions

File tree

‎flink/src/main/java/io/delta/flink/table/CatalogManagedTable.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import io.delta.kernel.transaction.CreateTableTransactionBuilder;
2525
import io.delta.kernel.types.StructType;
2626
import io.delta.kernel.unitycatalog.UCCatalogManagedClient;
27+
import io.delta.kernel.unitycatalog.UCTableIdentifier;
2728
import io.delta.kernel.utils.CloseableIterable;
2829
import io.delta.storage.commit.uccommitcoordinator.UCClient;
2930
import io.delta.storage.commit.uccommitcoordinator.UCTokenBasedRestClient;
@@ -72,6 +73,12 @@ public class CatalogManagedTable extends AbstractKernelTable {
7273
protected transient UCClient ucClient;
7374
protected transient UCCatalogManagedClient catalogManagedClient;
7475

76+
private UCTableIdentifier getUcTableIdentifier() {
77+
Preconditions.checkArgument(
78+
catalog instanceof UnityCatalog, "Catalog-managed tables require a UnityCatalog catalog");
79+
return ((UnityCatalog) catalog).toUcTableIdentifier(tableId);
80+
}
81+
7582
public CatalogManagedTable(
7683
DeltaCatalog catalog, String tableId, Map<String, String> conf, URI endpoint, String token) {
7784
this(catalog, tableId, conf, null, null, endpoint, token);
@@ -141,6 +148,7 @@ protected Snapshot loadLatestSnapshot() {
141148
/* engine */ getEngine(),
142149
/* ucTableId */ tableUUID,
143150
/* tablePath */ tablePath.toString(),
151+
/* ucTableIdentifier */ getUcTableIdentifier(),
144152
/* versionOpt */ Optional.empty(),
145153
/* timestampOpt */ Optional.empty()));
146154
}
@@ -182,6 +190,7 @@ protected boolean versionExists(Long version) {
182190
/* engine */ getEngine(),
183191
/* ucTableId */ getTableUUID(),
184192
/* tablePath */ getTablePath().toString(),
193+
/* ucTableIdentifier */ getUcTableIdentifier(),
185194
/* startVersionOpt */ Optional.of(version),
186195
/* startTimestampOpt */ Optional.empty(),
187196
/* endVersionOpt */ Optional.empty(),

‎flink/src/main/java/io/delta/flink/table/UnityCatalog.java‎

Lines changed: 31 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import dev.failsafe.function.CheckedSupplier;
2222
import io.delta.kernel.internal.types.DataTypeJsonSerDe;
2323
import io.delta.kernel.types.*;
24+
import io.delta.kernel.unitycatalog.UCTableIdentifier;
2425
import io.unitycatalog.client.ApiClient;
2526
import io.unitycatalog.client.ApiClientBuilder;
2627
import io.unitycatalog.client.ApiException;
@@ -231,6 +232,33 @@ public ApiClient getApiClient() {
231232
return apiClient;
232233
}
233234

235+
/**
236+
* Parses {@code schema.table} or {@code catalog.schema.table} into a {@link UCTableIdentifier}.
237+
* In the 2-part form the catalog defaults to this catalog's name; in the 3-part form the leading
238+
* segment must equal this catalog's name.
239+
*/
240+
UCTableIdentifier toUcTableIdentifier(String qualifiedTableName) {
241+
String[] namespaces = qualifiedTableName.split("\\.");
242+
Preconditions.checkArgument(namespaces.length == 2 || namespaces.length == 3);
243+
String catalogName;
244+
String schemaName;
245+
String tableName;
246+
if (namespaces.length == 3) {
247+
Preconditions.checkArgument(
248+
namespaces[0].equals(getName()),
249+
String.format(
250+
"table's catalog name %s must match catalog's name %s", namespaces[0], getName()));
251+
catalogName = namespaces[0];
252+
schemaName = namespaces[1];
253+
tableName = namespaces[2];
254+
} else {
255+
catalogName = getName();
256+
schemaName = namespaces[0];
257+
tableName = namespaces[1];
258+
}
259+
return new UCTableIdentifier(catalogName, schemaName, tableName);
260+
}
261+
234262
@Override
235263
public void open() {
236264
if (apiClient == null) {
@@ -329,22 +357,9 @@ public void createTable(
329357
() -> {
330358
TablesApi tablesApi = new TablesApi(apiClient);
331359
// Obtain names
332-
String[] namespaces = tableId.split("\\.");
333-
Preconditions.checkArgument(namespaces.length == 2 || namespaces.length == 3);
334-
String schemaName;
335-
String tableName;
336-
if (namespaces.length == 3) {
337-
Preconditions.checkArgument(
338-
namespaces[0].equals(getName()),
339-
String.format(
340-
"table's catalog name %s must match catalog's name %s",
341-
namespaces[0], getName()));
342-
schemaName = namespaces[1];
343-
tableName = namespaces[2];
344-
} else {
345-
schemaName = namespaces[0];
346-
tableName = namespaces[1];
347-
}
360+
UCTableIdentifier tableIdentifier = toUcTableIdentifier(tableId);
361+
String schemaName = tableIdentifier.getSchemaName();
362+
String tableName = tableIdentifier.getTableName();
348363
// Column Info
349364
List<ColumnInfo> columnInfos =
350365
IntStream.range(0, schema.fields().size())

‎flink/src/test/java/io/delta/flink/table/UnityCatalogTest.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,16 +17,42 @@
1717
package io.delta.flink.table;
1818

1919
import static org.junit.jupiter.api.Assertions.assertEquals;
20+
import static org.junit.jupiter.api.Assertions.assertThrows;
2021

2122
import io.delta.flink.MockHttp;
2223
import io.delta.flink.TestHelper;
2324
import io.delta.kernel.types.*;
25+
import io.delta.kernel.unitycatalog.UCTableIdentifier;
2426
import java.net.URI;
2527
import org.junit.jupiter.api.Test;
2628

2729
/** JUnit 6 test suite for UnityCatalog. */
2830
class UnityCatalogTest extends TestHelper {
2931

32+
@Test
33+
void testToUcTableIdentifier() {
34+
UnityCatalog catalog = new UnityCatalog("main", URI.create("http://localhost"), "");
35+
36+
UCTableIdentifier twoPartIdentifier = catalog.toUcTableIdentifier("default.tbl");
37+
assertEquals("main", twoPartIdentifier.getCatalogName());
38+
assertEquals("default", twoPartIdentifier.getSchemaName());
39+
assertEquals("tbl", twoPartIdentifier.getTableName());
40+
41+
UCTableIdentifier threePartIdentifier = catalog.toUcTableIdentifier("main.default.tbl");
42+
assertEquals("main", threePartIdentifier.getCatalogName());
43+
assertEquals("default", threePartIdentifier.getSchemaName());
44+
assertEquals("tbl", threePartIdentifier.getTableName());
45+
}
46+
47+
@Test
48+
void testToUcTableIdentifierRejectsInvalidNames() {
49+
UnityCatalog catalog = new UnityCatalog("main", URI.create("http://localhost"), "");
50+
51+
assertThrows(IllegalArgumentException.class, () -> catalog.toUcTableIdentifier("tbl"));
52+
assertThrows(
53+
IllegalArgumentException.class, () -> catalog.toUcTableIdentifier("other.default.tbl"));
54+
}
55+
3056
@Test
3157
void testGetTable() {
3258
withTempDir(

‎kernel/kernel-benchmarks/src/test/java/io/delta/kernel/benchmarks/models/UcCatalogInfo.java‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import io.delta.kernel.internal.util.FileNames;
2525
import io.delta.kernel.unitycatalog.InMemoryUCClient;
2626
import io.delta.kernel.unitycatalog.UCCatalogManagedCommitter;
27+
import io.delta.kernel.unitycatalog.UCTableIdentifier;
2728
import io.delta.kernel.utils.FileStatus;
2829
import io.delta.storage.commit.Commit;
2930
import java.io.File;
@@ -43,6 +44,9 @@
4344
* <pre>{@code
4445
* {
4546
* "uc_table_id": "12345678-1234-1234-1234-123456789abc",
47+
* "catalog_name": "benchmark_catalog",
48+
* "schema_name": "benchmark_schema",
49+
* "table_name": "benchmark_table",
4650
* "max_ratified_version": 2,
4751
* "log_tail": [
4852
* {
@@ -141,6 +145,18 @@ public Commit toCommit(Engine engine, String tableRoot) throws IOException {
141145
@JsonProperty(value = "uc_table_id", required = true)
142146
private String ucTableId;
143147

148+
/** The Unity Catalog catalog name. */
149+
@JsonProperty(value = "catalog_name", required = true)
150+
private String catalogName;
151+
152+
/** The Unity Catalog schema name. */
153+
@JsonProperty(value = "schema_name", required = true)
154+
private String schemaName;
155+
156+
/** The Unity Catalog table name. */
157+
@JsonProperty(value = "table_name", required = true)
158+
private String tableName;
159+
144160
/** The maximum ratified version for this table in Unity Catalog. */
145161
@JsonProperty(value = "max_ratified_version", required = true)
146162
private long maxRatifiedVersion;
@@ -167,6 +183,11 @@ public String getUcTableId() {
167183
return ucTableId;
168184
}
169185

186+
/** @return the three-part Unity Catalog table identifier */
187+
public UCTableIdentifier getUcTableIdentifier() {
188+
return new UCTableIdentifier(catalogName, schemaName, tableName);
189+
}
190+
170191
/**
171192
* Creates an InMemoryUCClient for this table with the staged commits pre-loaded.
172193
*

‎kernel/kernel-benchmarks/src/test/java/io/delta/kernel/benchmarks/workloadrunners/WorkloadRunner.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,7 @@ protected Snapshot loadSnapshot(Engine engine, TableInfo tableInfo, Optional<Lon
111111
engine,
112112
ucCatalogInfo.getUcTableId(),
113113
tableUri.toString(),
114+
ucCatalogInfo.getUcTableIdentifier(),
114115
versionOpt,
115116
Optional.empty() /* timestampOpt */);
116117
} else {

‎kernel/kernel-benchmarks/src/test/resources/workload_specs/basic_catalog_managed/catalog_managed_info.json‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,8 @@
11
{
22
"uc_table_id": "12345678-1234-1234-1234-123456789abc",
3+
"catalog_name": "benchmark_catalog",
4+
"schema_name": "benchmark_schema",
5+
"table_name": "benchmark_table",
36
"max_ratified_version": 3,
47
"log_tail": [
58
{

‎kernel/unitycatalog/src/main/java/io/delta/kernel/unitycatalog/UCCatalogManagedClient.java‎

Lines changed: 25 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import io.delta.kernel.unitycatalog.metrics.UcLoadSnapshotTelemetry;
3838
import io.delta.storage.commit.Commit;
3939
import io.delta.storage.commit.GetCommitsResponse;
40+
import io.delta.storage.commit.TableIdentifier;
4041
import io.delta.storage.commit.uccommitcoordinator.UCClient;
4142
import io.delta.storage.commit.uccommitcoordinator.UCCommitCoordinatorException;
4243
import java.io.IOException;
@@ -82,6 +83,7 @@ public UCCatalogManagedClient(UCClient ucClient) {
8283
* @param engine The Delta Kernel {@link Engine} to use for loading the table.
8384
* @param ucTableId The Unity Catalog table ID, which is a unique identifier for the table in UC.
8485
* @param tablePath The path to the Delta table in the underlying storage system.
86+
* @param ucTableIdentifier The three-part Unity Catalog table identifier.
8587
* @param versionOpt The optional version to time-travel to when loading the table. This must be
8688
* mutually exclusive with timestampOpt.
8789
* @param timestampOpt The optional timestamp to time-travel to when loading the table. This must
@@ -93,11 +95,13 @@ public Snapshot loadSnapshot(
9395
Engine engine,
9496
String ucTableId,
9597
String tablePath,
98+
UCTableIdentifier ucTableIdentifier,
9699
Optional<Long> versionOpt,
97100
Optional<Long> timestampOpt) {
98101
Objects.requireNonNull(engine, "engine is null");
99102
Objects.requireNonNull(ucTableId, "ucTableId is null");
100103
Objects.requireNonNull(tablePath, "tablePath is null");
104+
Objects.requireNonNull(ucTableIdentifier, "ucTableIdentifier is null");
101105
Objects.requireNonNull(versionOpt, "versionOpt is null");
102106
Objects.requireNonNull(timestampOpt, "timestampOpt is null");
103107
versionOpt.ifPresent(version -> checkArgument(version >= 0, "version must be non-negative"));
@@ -122,7 +126,13 @@ public Snapshot loadSnapshot(
122126
() -> {
123127
final GetCommitsResponse response =
124128
metricsCollector.getCommitsTimer.timeChecked(
125-
() -> getRatifiedCommitsFromUC(ucTableId, tablePath, versionOpt));
129+
() ->
130+
getRatifiedCommitsFromUC(
131+
ucTableId,
132+
tablePath,
133+
versionOpt,
134+
UCCatalogManagedCommitter.toStorageTableIdentifier(
135+
ucTableIdentifier)));
126136

127137
metricsCollector.setNumCatalogCommits(response.getCommits().size());
128138

@@ -249,6 +259,7 @@ public CreateTableTransactionBuilder buildCreateTableTransaction(
249259
* @param engine The Delta Kernel {@link Engine} to use for loading the table.
250260
* @param ucTableId The Unity Catalog table ID, which is a unique identifier for the table in UC.
251261
* @param tablePath The path to the Delta table in the underlying storage system.
262+
* @param ucTableIdentifier The three-part Unity Catalog table identifier.
252263
* @param startVersionOpt The optional start version boundary. This must be mutually exclusive
253264
* with startTimestampOpt. Either this or startTimestampOpt must be provided.
254265
* @param startTimestampOpt The optional start timestamp boundary. This must be mutually exclusive
@@ -267,13 +278,15 @@ public CommitRange loadCommitRange(
267278
Engine engine,
268279
String ucTableId,
269280
String tablePath,
281+
UCTableIdentifier ucTableIdentifier,
270282
Optional<Long> startVersionOpt,
271283
Optional<Long> startTimestampOpt,
272284
Optional<Long> endVersionOpt,
273285
Optional<Long> endTimestampOpt) {
274286
Objects.requireNonNull(engine, "engine is null");
275287
Objects.requireNonNull(ucTableId, "ucTableId is null");
276288
Objects.requireNonNull(tablePath, "tablePath is null");
289+
Objects.requireNonNull(ucTableIdentifier, "ucTableIdentifier is null");
277290
Objects.requireNonNull(startVersionOpt, "startVersionOpt is null");
278291
Objects.requireNonNull(startTimestampOpt, "startTimestampOpt is null");
279292
Objects.requireNonNull(endVersionOpt, "endVersionOpt is null");
@@ -308,7 +321,11 @@ public CommitRange loadCommitRange(
308321
Optional<Long> endVersionOptForCommitQuery =
309322
endVersionOpt.filter(v -> !startTimestampOpt.isPresent());
310323
final GetCommitsResponse response =
311-
getRatifiedCommitsFromUC(ucTableId, tablePath, endVersionOptForCommitQuery);
324+
getRatifiedCommitsFromUC(
325+
ucTableId,
326+
tablePath,
327+
endVersionOptForCommitQuery,
328+
UCCatalogManagedCommitter.toStorageTableIdentifier(ucTableIdentifier));
312329
final long ucTableVersion = response.getLatestTableVersion();
313330
validateVersionBoundariesExist(ucTableId, startVersionOpt, endVersionOpt, ucTableVersion);
314331
final List<ParsedLogData> logData =
@@ -413,7 +430,11 @@ private String getCommitRangeBoundariesString(
413430
}
414431

415432
private GetCommitsResponse getRatifiedCommitsFromUC(
416-
String ucTableId, String tablePath, Optional<Long> versionOpt) {
433+
String ucTableId,
434+
String tablePath,
435+
Optional<Long> versionOpt,
436+
TableIdentifier tableIdentifier) {
437+
Objects.requireNonNull(tableIdentifier, "tableIdentifier is null");
417438
logger.info(
418439
"[{}] Invoking the UCClient to get ratified commits at version {}",
419440
ucTableId,
@@ -430,6 +451,7 @@ private GetCommitsResponse getRatifiedCommitsFromUC(
430451
return ucClient.getCommits(
431452
ucTableId,
432453
new Path(tablePath).toUri(),
454+
tableIdentifier,
433455
Optional.empty() /* startVersion */,
434456
versionOpt /* endVersion */);
435457
} catch (IOException ex) {

‎kernel/unitycatalog/src/main/java/io/delta/kernel/unitycatalog/UCTableIdentifier.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818

1919
import static java.util.Objects.requireNonNull;
2020

21-
/** Logical Unity Catalog table identifier used for create-time finalization. */
21+
/** Logical Unity Catalog table identifier used for table lifecycle and read operations. */
2222
public final class UCTableIdentifier {
2323
private final String catalogName;
2424
private final String schemaName;

‎kernel/unitycatalog/src/test/scala/io/delta/kernel/unitycatalog/InMemoryUCClient.scala‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -206,6 +206,7 @@ class InMemoryUCClient(ucMetastoreId: String) extends UCClient {
206206
override def getCommits(
207207
tableId: String,
208208
tableUri: URI,
209+
tableIdentifier: TableIdentifier,
209210
startVersion: Optional[JLong],
210211
endVersion: Optional[JLong]): GetCommitsResponse = {
211212
val tableData = getTableDataElseThrow(tableId)

‎kernel/unitycatalog/src/test/scala/io/delta/kernel/unitycatalog/InMemoryUCClientSuite.scala‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,12 @@ class InMemoryUCClientSuite extends AnyFunSuite with UCCatalogManagedTestUtils {
3737
endVersionOpt: Optional[JLong],
3838
expectedVersions: Seq[Long]): Unit = {
3939
val client = getInMemoryUCClientWithCommitsForTableId("tableId", allVersions)
40-
val response = client.getCommits("tableId", fakeURI, startVersionOpt, endVersionOpt)
40+
val response = client.getCommits(
41+
"tableId",
42+
fakeURI,
43+
/* tableIdentifier = */ null,
44+
startVersionOpt,
45+
endVersionOpt)
4146
val actualVersions = response.getCommits.asScala.map(_.getVersion)
4247

4348
assert(actualVersions == expectedVersions)
@@ -82,7 +87,12 @@ class InMemoryUCClientSuite extends AnyFunSuite with UCCatalogManagedTestUtils {
8287
test("getCommits throws InvalidTargetTableException for non-existent table") {
8388
val client = new InMemoryUCClient("ucMetastoreId")
8489
val exception = intercept[InvalidTargetTableException] {
85-
client.getCommits("abcd", new URI("s3://bucket/table"), Optional.empty(), Optional.empty())
90+
client.getCommits(
91+
"abcd",
92+
new URI("s3://bucket/table"),
93+
/* tableIdentifier = */ null,
94+
Optional.empty(),
95+
Optional.empty())
8696
}
8797
assert(exception.getMessage.contains(s"Table not found: abcd"))
8898
}

0 commit comments

Comments
 (0)