This is an automated email from the ASF dual-hosted git repository. github-actions[bot] pushed a commit to branch cherry-pick-9419878b-to-branch-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 9cef93ec5e3c1de8cec8b5d0c04d8e653a99e98e Author: roryqi <[email protected]> AuthorDate: Fri Jun 5 20:58:54 2026 +0800 [#11138][#11137] refactor(iceberg-rest): federation handling via FederatedCatalogWrapper subclass (#11367) ### What changes were proposed in this pull request? Federation logic in `CatalogWrapperForREST` was implemented as scattered `instanceof RESTCatalog` / `isRESTCatalog()` short-circuits across the class. This PR refactors them into polymorphic dispatch and fixes the missing `registerTable` federation path: - Add `FederatedCatalogWrapper extends CatalogWrapperForREST` that overrides the operations needing federation-aware behavior: `createTable`, `loadTable`, `updateTable`, `registerTable`, and `buildCatalogConfigToClients`. - Add a `CatalogWrapperForREST.create(...)` factory that returns the federated subclass when the backend is `rest`, and wire `IcebergCatalogWrapperManager` to it. - Add `registerTableInternal`, mirroring `loadTableInternal`, so federated `registerTable` extracts FileIO-derived client config and credential fields from the remote catalog's `table.io()` (previously dropped). (#11137) - Remove the now-dead `instanceof`/`isRESTCatalog()` branches from the base class (`createTable`, `loadTable`, `updateTable`, `shouldGenerateCredential`), and make `buildCatalogConfigToClients` an overridable instance method backed by a shared `filterCatalogConfigForClients` helper. ### Why are the changes needed? Adding a new federated operation required remembering to add a corresponding `instanceof` check; nothing in the type system enforced it. The missing `registerTable` federation handling was a direct bug from this pattern: a federated `registerTable` returned a `LoadTableResponse` without the FileIO-derived properties and credential fields that `createTable`/`loadTable` include. Polymorphic dispatch removes the foot-gun and fixes the bug. Fix: #11138 Fix: #11137 ### Does this PR introduce _any_ user-facing change? No. Federated `registerTable` responses now include the FileIO/credential client config they should have included before; no API or property-key changes. ### How was this patch tested? `./gradlew :iceberg:iceberg-rest-server:test -PskipITs` (full module unit suite green), plus new unit tests: - `testCreateReturnsFederatedWrapperForRestBackend` - `testCreateReturnsBaseWrapperForNonRestBackend` - `testFederatedRegisterTableIncludesFileIo` Existing `buildCatalogConfigToClients` tests were updated to exercise the polymorphic source selection and the shared filter helper. --- .../iceberg/service/CatalogWrapperForREST.java | 474 ++----------------- .../iceberg/service/FederatedCatalogWrapper.java | 506 +++++++++++++++++++++ .../service/IcebergCatalogWrapperManager.java | 14 +- .../iceberg/service/IcebergRESTUtils.java | 52 ++- .../iceberg/service/TestCatalogWrapperForREST.java | 115 +++-- .../TestIcebergCatalogWrapperManagerForREST.java | 42 ++ .../iceberg/service/TestIcebergRESTUtils.java | 32 ++ 7 files changed, 766 insertions(+), 469 deletions(-) diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java index 7e1aac0c0d..e02c6d4f55 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java @@ -20,14 +20,10 @@ package org.apache.gravitino.iceberg.service; import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; -import com.google.common.collect.Maps; import java.io.IOException; import java.net.URI; -import java.time.OffsetDateTime; -import java.time.ZoneOffset; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -36,7 +32,6 @@ import java.util.Locale; import java.util.Map; import java.util.Optional; import java.util.Set; -import java.util.stream.Collectors; import java.util.stream.Stream; import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; @@ -44,7 +39,6 @@ import org.apache.gravitino.credential.CatalogCredentialManager; import org.apache.gravitino.credential.Credential; import org.apache.gravitino.credential.CredentialConstants; import org.apache.gravitino.credential.CredentialPrivilege; -import org.apache.gravitino.credential.CredentialPropertyUtils; import org.apache.gravitino.credential.PathBasedCredentialContext; import org.apache.gravitino.iceberg.common.IcebergConfig; import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper; @@ -54,48 +48,22 @@ import org.apache.gravitino.storage.GCSProperties; import org.apache.gravitino.utils.ClassUtils; import org.apache.gravitino.utils.MapUtils; import org.apache.gravitino.utils.PrincipalUtils; -import org.apache.iceberg.BaseMetadataTable; -import org.apache.iceberg.BaseTable; -import org.apache.iceberg.BaseTransaction; -import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.FileScanTask; import org.apache.iceberg.IncrementalAppendScan; -import org.apache.iceberg.MetadataUpdate; -import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Scan; -import org.apache.iceberg.Schema; -import org.apache.iceberg.SortOrder; import org.apache.iceberg.Table; import org.apache.iceberg.TableMetadata; -import org.apache.iceberg.TableOperations; import org.apache.iceberg.TableProperties; import org.apache.iceberg.TableScan; -import org.apache.iceberg.Transaction; -import org.apache.iceberg.UpdateRequirement; -import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; -import org.apache.iceberg.exceptions.AlreadyExistsException; -import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.exceptions.ServiceUnavailableException; -import org.apache.iceberg.inmemory.InMemoryFileIO; import org.apache.iceberg.io.CloseableIterable; -import org.apache.iceberg.io.FileIO; import org.apache.iceberg.rest.CatalogHandlers; -import org.apache.iceberg.rest.ErrorHandlers; -import org.apache.iceberg.rest.HTTPClient; import org.apache.iceberg.rest.PlanStatus; -import org.apache.iceberg.rest.RESTCatalog; -import org.apache.iceberg.rest.RESTClient; -import org.apache.iceberg.rest.RESTUtil; -import org.apache.iceberg.rest.ResourcePaths; -import org.apache.iceberg.rest.auth.AuthManager; -import org.apache.iceberg.rest.auth.AuthManagers; -import org.apache.iceberg.rest.auth.AuthSession; import org.apache.iceberg.rest.requests.CreateTableRequest; import org.apache.iceberg.rest.requests.PlanTableScanRequest; import org.apache.iceberg.rest.requests.RegisterTableRequest; -import org.apache.iceberg.rest.requests.UpdateTableRequest; import org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse; import org.apache.iceberg.rest.responses.LoadCredentialsResponse; import org.apache.iceberg.rest.responses.LoadTableResponse; @@ -104,8 +72,8 @@ import org.apache.iceberg.rest.responses.PlanTableScanResponse; /** Process Iceberg REST specific operations, like credential vending. */ public class CatalogWrapperForREST extends IcebergCatalogWrapper { - private static final String FORMAT_VERSION = "format-version"; - private final CatalogCredentialManager catalogCredentialManager; + /** Vends Gravitino-managed credentials and exposes the catalog name for credential refresh. */ + protected final CatalogCredentialManager catalogCredentialManager; private volatile Map<String, String> catalogConfigToClients; private final Object catalogConfigToClientsLock = new Object(); @@ -114,9 +82,12 @@ public class CatalogWrapperForREST extends IcebergCatalogWrapper { private static final String DATA_ACCESS_VENDED_CREDENTIALS = "vended-credentials"; private static final String DATA_ACCESS_REMOTE_SIGNING = "remote-signing"; - private static final Schema EMPTY_SCHEMA = new Schema(); - private static final Set<String> catalogPropertiesToClientKeys = + /** + * Client-facing catalog property keys retained when building the IRC {@code /v1/config} defaults + * and when extracting FileIO-derived config in {@link FederatedCatalogWrapper}. + */ + protected static final Set<String> catalogPropertiesToClientKeys = ImmutableSet.of( IcebergConstants.IO_IMPL, IcebergConstants.AWS_S3_REGION, @@ -144,12 +115,7 @@ public class CatalogWrapperForREST extends IcebergCatalogWrapper { public LoadTableResponse createTable( Namespace namespace, CreateTableRequest request, boolean requestCredential) { - LoadTableResponse loadTableResponse; - if (isRESTCatalog()) { - loadTableResponse = createTableInternal(namespace, request); - } else { - loadTableResponse = super.createTable(namespace, request); - } + LoadTableResponse loadTableResponse = super.createTable(namespace, request); if (shouldGenerateCredential(loadTableResponse, requestCredential)) { return injectCredentialConfig( TableIdentifier.of(namespace, request.name()), @@ -161,12 +127,7 @@ public class CatalogWrapperForREST extends IcebergCatalogWrapper { public LoadTableResponse loadTable( TableIdentifier identifier, boolean requestCredential, CredentialPrivilege privilege) { - LoadTableResponse loadTableResponse; - if (isRESTCatalog()) { - loadTableResponse = loadTableInternal(identifier); - } else { - loadTableResponse = super.loadTable(identifier); - } + LoadTableResponse loadTableResponse = super.loadTable(identifier); if (shouldGenerateCredential(loadTableResponse, requestCredential)) { return injectCredentialConfig(identifier, loadTableResponse, privilege); } @@ -188,34 +149,19 @@ public class CatalogWrapperForREST extends IcebergCatalogWrapper { return loadTableResponse; } - @Override - public LoadTableResponse updateTable( - TableIdentifier tableIdentifier, UpdateTableRequest updateTableRequest) { - if (isRESTCatalog()) { - return tableUpdateInternal(tableIdentifier, updateTableRequest); - } else { - return super.updateTable(tableIdentifier, updateTableRequest); - } - } - /** - * Get table credentials. + * Get table credentials from the local (Gravitino-managed) catalog backend. + * + * <p>{@link FederatedCatalogWrapper} overrides this to fetch credentials from the remote REST + * catalog instead, so the base implementation never needs an {@code instanceof RESTCatalog} + * check. * * @param identifier table identifier - * @param privilege used for local credential vending; ignored for REST catalog backends + * @param privilege used for local credential vending * @return table credentials response */ public LoadCredentialsResponse getTableCredentials( TableIdentifier identifier, CredentialPrivilege privilege) { - if (isRESTCatalog()) { - return getRESTTableCredentials((RESTCatalog) getCatalog(), identifier); - } else { - return getLocalTableCredentials(identifier, privilege); - } - } - - private LoadCredentialsResponse getLocalTableCredentials( - TableIdentifier identifier, CredentialPrivilege privilege) { try { LoadTableResponse loadTableResponse = super.loadTable(identifier); Credential credential = getCredential(loadTableResponse.tableMetadata(), privilege); @@ -233,60 +179,6 @@ public class CatalogWrapperForREST extends IcebergCatalogWrapper { } } - private static LoadCredentialsResponse getRESTTableCredentials( - RESTCatalog restCatalog, TableIdentifier identifier) { - Map<String, String> properties = Maps.newHashMap(restCatalog.properties()); - String credentialsPath = - ResourcePaths.forCatalogProperties(properties).table(identifier) + "/credentials"; - - AuthManager authManager = null; - RESTClient client = null; - AuthSession authSession = null; - try { - authManager = AuthManagers.loadAuthManager(restCatalog.name(), properties); - client = - HTTPClient.builder(properties) - .uri(properties.get(CatalogProperties.URI)) - .withHeaders(RESTUtil.configHeaders(properties)) - .build(); - authSession = authManager.catalogSession(client, properties); - return client - .withAuthSession(authSession) - .get( - credentialsPath, - LoadCredentialsResponse.class, - Collections.emptyMap(), - ErrorHandlers.tableErrorHandler()); - } finally { - if (authSession != null) { - try { - authSession.close(); - } catch (Exception e) { - LOG.warn( - "Failed to close auth session when loading credentials for table: {}", identifier, e); - } - } - - if (client != null) { - try { - client.close(); - } catch (Exception e) { - LOG.warn( - "Failed to close REST client when loading credentials for table: {}", identifier, e); - } - } - - if (authManager != null) { - try { - authManager.close(); - } catch (Exception e) { - LOG.warn( - "Failed to close auth manager when loading credentials for table: {}", identifier, e); - } - } - } - } - @Override public void close() throws Exception { try { @@ -313,7 +205,7 @@ public class CatalogWrapperForREST extends IcebergCatalogWrapper { synchronized (catalogConfigToClientsLock) { if (catalogConfigToClients == null) { - catalogConfigToClients = buildCatalogConfigToClients(getIcebergConfig(), getCatalog()); + catalogConfigToClients = buildCatalogConfigToClients(); } return catalogConfigToClients; } @@ -322,29 +214,35 @@ public class CatalogWrapperForREST extends IcebergCatalogWrapper { /** * Builds properties exposed to Iceberg clients via the IRC {@code /v1/config} defaults. * - * <p>For {@link RESTCatalog}, uses {@link RESTCatalog#properties()} so defaults reflect the - * remote catalog's config response merged with client properties (after REST handshake), not only - * static Gravitino catalog configuration. + * <p>The base implementation uses the static Gravitino catalog configuration as the property + * source. {@link FederatedCatalogWrapper} overrides this to use {@code RESTCatalog.properties()} + * so defaults reflect the remote catalog's config response merged with client properties (after + * REST handshake). * * <p>{@link IcebergConstants#IO_IMPL} is passed through when present (e.g. Iceberg {@link * org.apache.iceberg.io.ResolvingFileIO}), so clients multiplex by URI scheme without server-side * rewriting per table. + * + * @return the immutable, filtered properties exposed to Iceberg clients. */ @VisibleForTesting - static Map<String, String> buildCatalogConfigToClients(IcebergConfig config, Catalog catalog) { - Map<String, String> sourceProps; - if (catalog instanceof RESTCatalog) { - Map<String, String> merged = ((RESTCatalog) catalog).properties(); - sourceProps = merged != null ? new HashMap<>(merged) : new HashMap<>(); - } else { - sourceProps = new HashMap<>(config.getIcebergCatalogProperties()); - } + Map<String, String> buildCatalogConfigToClients() { + return filterCatalogConfigForClients(getIcebergConfig().getIcebergCatalogProperties()); + } + /** + * Keeps only the client-facing catalog properties and validates the data-access property. + * + * @param sourceProps the candidate properties to filter. + * @return the immutable, filtered properties exposed to Iceberg clients. + */ + protected static Map<String, String> filterCatalogConfigForClients( + Map<String, String> sourceProps) { Map<String, String> filtered = - MapUtils.getFilteredMap(sourceProps, key -> catalogPropertiesToClientKeys.contains(key)); - filtered = new HashMap<>(filtered); + new HashMap<>( + MapUtils.getFilteredMap( + sourceProps, key -> catalogPropertiesToClientKeys.contains(key))); validateAndNormalizeDataAccessProperty(filtered); - return Collections.unmodifiableMap(filtered); } @@ -388,20 +286,21 @@ public class CatalogWrapperForREST extends IcebergCatalogWrapper { credential.credentialType(), tableIdentifier); - // Merge temporary credential fields as Iceberg client config entries in the load-table - // response. - Map<String, String> credentialConfig = + // Vend the temporary credential both as a first-class REST credential (storage-prefix scoped, + // for Iceberg 1.7+ clients) and, for backward compatibility, flattened into the load-table + // response config. This mirrors FederatedCatalogWrapper, which also populates both fields. + org.apache.iceberg.rest.credentials.Credential restCredential = IcebergRESTUtils.toRESTCredential( - catalogCredentialManager.catalogName(), - tableIdentifier, - credential, - loadTableResponse.tableMetadata()) - .config(); + catalogCredentialManager.catalogName(), + tableIdentifier, + credential, + loadTableResponse.tableMetadata()); return LoadTableResponse.builder() .withTableMetadata(loadTableResponse.tableMetadata()) .addAllConfig(loadTableResponse.config()) .addAllConfig(getCatalogConfigToClient()) - .addAllConfig(credentialConfig) + .addAllConfig(restCredential.config()) + .addCredential(restCredential) .build(); } @@ -437,11 +336,6 @@ public class CatalogWrapperForREST extends IcebergCatalogWrapper { return false; } - // RESTCatalog will fetch credential from the remote catalog instead of generating credential - if (getCatalog() instanceof RESTCatalog) { - return false; - } - validateCredentialLocation(loadTableResponse.tableMetadata().location()); return !isLocalOrHdfsTable(loadTableResponse.tableMetadata()); } @@ -730,280 +624,4 @@ public class CatalogWrapperForREST extends IcebergCatalogWrapper { properties.put(newProperty, deprecatedValue); } } - - private LoadTableResponse createTableInternal(Namespace namespace, CreateTableRequest request) { - Catalog loadedCatalog = getCatalog(); - - request.validate(); - - if (request.stageCreate()) { - return stageTableCreateInternal(namespace, request); - } - - TableIdentifier ident = TableIdentifier.of(namespace, request.name()); - Table table = - loadedCatalog - .buildTable(ident, request.schema()) - .withLocation(request.location()) - .withPartitionSpec(request.spec()) - .withSortOrder(request.writeOrder()) - .withProperties(request.properties()) - .create(); - - if (table instanceof BaseTable) { - Map<String, String> properties = retrieveFileIOProperties(table.io()); - Map<String, String> filteredCredentialProperties = - CredentialPropertyUtils.filterCredentialProperties(properties); - return LoadTableResponse.builder() - .withTableMetadata(((BaseTable) table).operations().current()) - .addAllConfig( - MapUtils.getFilteredMap( - properties, key -> catalogPropertiesToClientKeys.contains(key))) - // Keep only credential fields from FileIO properties before returning them to the - // client. - .addAllConfig(filteredCredentialProperties) - .addAllConfig( - IcebergRESTUtils.buildRefreshProps( - catalogCredentialManager.catalogName(), ident, filteredCredentialProperties)) - .addAllCredentials( - IcebergRESTUtils.buildStorageCreds( - catalogCredentialManager.catalogName(), ident, table.io())) - .build(); - } - - throw new IllegalStateException("Cannot wrap catalog that does not produce BaseTable"); - } - - private LoadTableResponse stageTableCreateInternal( - Namespace namespace, CreateTableRequest request) { - Catalog loadedCatalog = getCatalog(); - TableIdentifier ident = TableIdentifier.of(namespace, request.name()); - if (loadedCatalog.tableExists(ident)) { - throw new AlreadyExistsException("Table already exists: %s", ident); - } - - Map<String, String> properties = Maps.newHashMap(); - properties.put("created-at", OffsetDateTime.now(ZoneOffset.UTC).toString()); - properties.putAll(request.properties()); - - Map<String, String> config = Maps.newHashMap(); - Catalog.TableBuilder tableBuilder = - loadedCatalog - .buildTable(ident, request.schema()) - .withPartitionSpec(request.spec()) - .withSortOrder(request.writeOrder()) - .withProperties(properties); - - Table table; - if (request.location() != null) { - table = tableBuilder.withLocation(request.location()).createTransaction().table(); - } else { - table = tableBuilder.createTransaction().table(); - } - - Map<String, String> tableProperties = retrieveFileIOProperties(table.io()); - Map<String, String> filteredCredentialProperties = - CredentialPropertyUtils.filterCredentialProperties(tableProperties); - config.putAll( - MapUtils.getFilteredMap( - tableProperties, key -> catalogPropertiesToClientKeys.contains(key))); - config.putAll(filteredCredentialProperties); - config.putAll( - IcebergRESTUtils.buildRefreshProps( - catalogCredentialManager.catalogName(), ident, filteredCredentialProperties)); - - List<org.apache.iceberg.rest.credentials.Credential> credentials = - IcebergRESTUtils.buildStorageCreds( - catalogCredentialManager.catalogName(), ident, table.io()); - - TableMetadata metadata = - TableMetadata.newTableMetadata( - request.schema(), - request.spec() != null ? request.spec() : PartitionSpec.unpartitioned(), - request.writeOrder() != null ? request.writeOrder() : SortOrder.unsorted(), - table.location(), - properties); - - return LoadTableResponse.builder() - .withTableMetadata(metadata) - .addAllConfig(config) - .addAllCredentials(credentials) - .build(); - } - - private LoadTableResponse tableUpdateInternal(TableIdentifier ident, UpdateTableRequest request) { - if (isCreate(request)) { - // this is a hacky way to get TableOperations for an uncommitted table - Optional<Integer> formatVersion = - request.updates().stream() - .filter(update -> update instanceof MetadataUpdate.UpgradeFormatVersion) - .map(update -> ((MetadataUpdate.UpgradeFormatVersion) update).formatVersion()) - .findFirst(); - - Schema schema = - request.updates().stream() - .filter(update -> update instanceof MetadataUpdate.AddSchema) - .map(update -> ((MetadataUpdate.AddSchema) update).schema()) - .findFirst() - .orElse(EMPTY_SCHEMA); - - Catalog.TableBuilder tableBuilder = getCatalog().buildTable(ident, schema); - - TableMetadata.Builder changedMetadata = - formatVersion.map(TableMetadata::buildFromEmpty).orElse(TableMetadata.buildFromEmpty()); - request.updates().forEach(update -> update.applyTo(changedMetadata)); - - TableMetadata changedTableMeta = changedMetadata.build(); - tableBuilder.withPartitionSpec(changedTableMeta.spec()); - tableBuilder.withSortOrder(changedTableMeta.sortOrder()); - tableBuilder.withLocation(changedTableMeta.location()); - tableBuilder.withProperty(FORMAT_VERSION, String.valueOf(changedTableMeta.formatVersion())); - tableBuilder.withProperties(changedTableMeta.properties()); - - Transaction transaction = tableBuilder.createOrReplaceTransaction(); - if (transaction instanceof BaseTransaction) { - BaseTransaction baseTransaction = (BaseTransaction) transaction; - - return LoadTableResponse.builder() - .withTableMetadata(create(baseTransaction, request)) - .build(); - } else { - throw new IllegalStateException( - "Cannot wrap catalog that does not produce BaseTransaction"); - } - - } else { - return CatalogHandlers.updateTable(getCatalog(), ident, request); - } - } - - private LoadTableResponse loadTableInternal(TableIdentifier ident) { - Table table = getCatalog().loadTable(ident); - - if (table instanceof BaseTable) { - Map<String, String> properties = retrieveFileIOProperties(table.io()); - Map<String, String> filteredCredentialProperties = - CredentialPropertyUtils.filterCredentialProperties(properties); - return LoadTableResponse.builder() - .withTableMetadata(((BaseTable) table).operations().current()) - .addAllConfig( - MapUtils.getFilteredMap( - properties, key -> catalogPropertiesToClientKeys.contains(key))) - // Keep only credential fields from FileIO properties before returning them to the client. - .addAllConfig(filteredCredentialProperties) - .addAllConfig( - IcebergRESTUtils.buildRefreshProps( - catalogCredentialManager.catalogName(), ident, filteredCredentialProperties)) - .addAllCredentials( - IcebergRESTUtils.buildStorageCreds( - catalogCredentialManager.catalogName(), ident, table.io())) - .build(); - } else if (table instanceof BaseMetadataTable) { - // metadata tables are loaded on the client side, return NoSuchTableException for now - throw new NoSuchTableException("Table does not exist: %s", ident.toString()); - } - - throw new IllegalStateException("Cannot wrap catalog that does not produce BaseTable"); - } - - private static boolean isCreate(UpdateTableRequest request) { - boolean isCreate = - request.requirements().stream() - .anyMatch(UpdateRequirement.AssertTableDoesNotExist.class::isInstance); - - if (isCreate) { - List<UpdateRequirement> invalidRequirements = - request.requirements().stream() - .filter(req -> !(req instanceof UpdateRequirement.AssertTableDoesNotExist)) - .collect(Collectors.toList()); - Preconditions.checkArgument( - invalidRequirements.isEmpty(), "Invalid create requirements: %s", invalidRequirements); - } - - return isCreate; - } - - private static TableMetadata create(BaseTransaction baseTransaction, UpdateTableRequest request) { - // the only valid requirement is that the table will be created - TableOperations ops = baseTransaction.underlyingOps(); - request.requirements().forEach(requirement -> requirement.validate(ops.current())); - - TableMetadata.Builder builder = TableMetadata.buildFrom(baseTransaction.currentMetadata()); - request - .updates() - .forEach( - update -> { - if (shouldApplyMetadataUpdateAfterBuilder(update)) { - update.applyTo(builder); - } - }); - - // create transactions do not retry. if the table exists, retrying is not a solution - ops.commit(null, builder.build()); - - return ops.current(); - } - - /** - * Returns {@code false} for updates already reflected through {@link Catalog.TableBuilder} during - * staged create; those must not be applied again on {@link TableMetadata.Builder}. - */ - @VisibleForTesting - static boolean shouldApplyMetadataUpdateAfterBuilder(MetadataUpdate update) { - if (update instanceof MetadataUpdate.UpgradeFormatVersion) { - return false; - } - - if (update instanceof MetadataUpdate.AddSchema) { - return false; - } - - if (update instanceof MetadataUpdate.SetCurrentSchema) { - return false; - } - - if (update instanceof MetadataUpdate.RemoveSchemas) { - return false; - } - - if (update instanceof MetadataUpdate.SetLocation) { - return false; - } - - if (update instanceof MetadataUpdate.SetProperties) { - return false; - } - - if (update instanceof MetadataUpdate.RemoveProperties) { - return false; - } - - if (update instanceof MetadataUpdate.AddSortOrder) { - return false; - } - - if (update instanceof MetadataUpdate.SetDefaultSortOrder) { - return false; - } - - if (update instanceof MetadataUpdate.AddPartitionSpec) { - return false; - } - - if (update instanceof MetadataUpdate.SetDefaultPartitionSpec) { - return false; - } - - if (update instanceof MetadataUpdate.RemovePartitionSpecs) { - return false; - } - - return true; - } - - private static Map<String, String> retrieveFileIOProperties(FileIO fileIO) { - return fileIO instanceof InMemoryFileIO - ? Maps.newHashMap() - : new HashMap<>(fileIO.properties()); - } } diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java new file mode 100644 index 0000000000..e726ca5617 --- /dev/null +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java @@ -0,0 +1,506 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.gravitino.iceberg.service; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import com.google.common.collect.Maps; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.stream.Collectors; +import org.apache.gravitino.credential.CredentialPrivilege; +import org.apache.gravitino.credential.CredentialPropertyUtils; +import org.apache.gravitino.iceberg.common.IcebergConfig; +import org.apache.gravitino.utils.MapUtils; +import org.apache.iceberg.BaseMetadataTable; +import org.apache.iceberg.BaseTable; +import org.apache.iceberg.BaseTransaction; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.MetadataUpdate; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableOperations; +import org.apache.iceberg.Transaction; +import org.apache.iceberg.UpdateRequirement; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.exceptions.NoSuchTableException; +import org.apache.iceberg.inmemory.InMemoryFileIO; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.rest.CatalogHandlers; +import org.apache.iceberg.rest.ErrorHandlers; +import org.apache.iceberg.rest.HTTPClient; +import org.apache.iceberg.rest.RESTCatalog; +import org.apache.iceberg.rest.RESTClient; +import org.apache.iceberg.rest.RESTUtil; +import org.apache.iceberg.rest.ResourcePaths; +import org.apache.iceberg.rest.auth.AuthManager; +import org.apache.iceberg.rest.auth.AuthManagers; +import org.apache.iceberg.rest.auth.AuthSession; +import org.apache.iceberg.rest.credentials.Credential; +import org.apache.iceberg.rest.requests.CreateTableRequest; +import org.apache.iceberg.rest.requests.RegisterTableRequest; +import org.apache.iceberg.rest.requests.UpdateTableRequest; +import org.apache.iceberg.rest.responses.LoadCredentialsResponse; +import org.apache.iceberg.rest.responses.LoadTableResponse; + +/** + * A {@link CatalogWrapperForREST} for a federated Iceberg REST catalog (the underlying catalog is a + * {@link RESTCatalog}). + * + * <p>Federation-specific behavior is expressed through polymorphic overrides instead of {@code + * instanceof RESTCatalog} checks scattered across the base class. Table operations are routed to + * federation-aware {@code *Internal} methods so client-facing FileIO and credential properties are + * extracted from the remote catalog's {@code table.io()}. Credentials are vended by the remote + * catalog, so this wrapper never injects Gravitino-generated credentials. + */ +public class FederatedCatalogWrapper extends CatalogWrapperForREST { + + private static final String FORMAT_VERSION = "format-version"; + private static final Schema EMPTY_SCHEMA = new Schema(); + + /** + * Creates a federated wrapper. + * + * @param catalogName the catalog name. + * @param config the Iceberg catalog configuration (backend {@code rest}). + */ + public FederatedCatalogWrapper(String catalogName, IcebergConfig config) { + super(catalogName, config); + } + + @Override + public LoadTableResponse createTable( + Namespace namespace, CreateTableRequest request, boolean requestCredential) { + // The remote REST catalog vends its own credentials, so the requestCredential flag is not used + // here; FileIO-derived client config is extracted by createTableInternal. + return createTableInternal(namespace, request); + } + + @Override + public LoadTableResponse loadTable( + TableIdentifier identifier, boolean requestCredential, CredentialPrivilege privilege) { + return loadTableInternal(identifier); + } + + @Override + public LoadTableResponse registerTable( + Namespace namespace, RegisterTableRequest request, boolean requestCredential) { + return registerTableInternal(namespace, request); + } + + @Override + public LoadTableResponse updateTable( + TableIdentifier tableIdentifier, UpdateTableRequest updateTableRequest) { + return tableUpdateInternal(tableIdentifier, updateTableRequest); + } + + @Override + Map<String, String> buildCatalogConfigToClients() { + Map<String, String> merged = ((RESTCatalog) getCatalog()).properties(); + return filterCatalogConfigForClients(merged != null ? merged : Collections.emptyMap()); + } + + /** + * Fetches table credentials from the remote REST catalog instead of vending Gravitino-managed + * credentials. The {@code privilege} is ignored because the remote catalog decides what to vend. + * + * <p>The upstream credentials are rewritten via {@link IcebergRESTUtils#rewriteTableCredentials} + * so their {@code refresh-credentials-endpoint} points at this IRC catalog rather than the remote + * catalog, consistent with the {@link #loadTable}/{@code createTable} federation paths. + * + * @param identifier the table identifier. + * @param privilege ignored; the remote REST catalog vends its own credentials. + * @return the remote credentials with IRC-local refresh endpoints. + */ + @Override + public LoadCredentialsResponse getTableCredentials( + TableIdentifier identifier, CredentialPrivilege privilege) { + LoadCredentialsResponse upstream = + getRESTTableCredentials((RESTCatalog) getCatalog(), identifier); + return IcebergRESTUtils.rewriteTableCredentials( + catalogCredentialManager.catalogName(), identifier, upstream); + } + + private static LoadCredentialsResponse getRESTTableCredentials( + RESTCatalog restCatalog, TableIdentifier identifier) { + Map<String, String> properties = Maps.newHashMap(restCatalog.properties()); + String credentialsPath = + ResourcePaths.forCatalogProperties(properties).table(identifier) + "/credentials"; + + AuthManager authManager = null; + RESTClient client = null; + AuthSession authSession = null; + try { + authManager = AuthManagers.loadAuthManager(restCatalog.name(), properties); + client = + HTTPClient.builder(properties) + .uri(properties.get(CatalogProperties.URI)) + .withHeaders(RESTUtil.configHeaders(properties)) + .build(); + authSession = authManager.catalogSession(client, properties); + return client + .withAuthSession(authSession) + .get( + credentialsPath, + LoadCredentialsResponse.class, + Collections.emptyMap(), + ErrorHandlers.tableErrorHandler()); + } finally { + if (authSession != null) { + try { + authSession.close(); + } catch (Exception e) { + LOG.warn( + "Failed to close auth session when loading credentials for table: {}", identifier, e); + } + } + + if (client != null) { + try { + client.close(); + } catch (Exception e) { + LOG.warn( + "Failed to close REST client when loading credentials for table: {}", identifier, e); + } + } + + if (authManager != null) { + try { + authManager.close(); + } catch (Exception e) { + LOG.warn( + "Failed to close auth manager when loading credentials for table: {}", identifier, e); + } + } + } + } + + /** + * Federation-aware {@code createTable}: creates the table on the underlying (remote) catalog and + * extracts client-facing FileIO/credential properties from {@code table.io()}. + */ + private LoadTableResponse createTableInternal(Namespace namespace, CreateTableRequest request) { + Catalog loadedCatalog = getCatalog(); + + request.validate(); + + if (request.stageCreate()) { + return stageTableCreateInternal(namespace, request); + } + + TableIdentifier ident = TableIdentifier.of(namespace, request.name()); + Table table = + loadedCatalog + .buildTable(ident, request.schema()) + .withLocation(request.location()) + .withPartitionSpec(request.spec()) + .withSortOrder(request.writeOrder()) + .withProperties(request.properties()) + .create(); + + if (table instanceof BaseTable) { + return buildLoadTableResponseFromFileIo(ident, (BaseTable) table); + } + + throw new IllegalStateException("Cannot wrap catalog that does not produce BaseTable"); + } + + private LoadTableResponse stageTableCreateInternal( + Namespace namespace, CreateTableRequest request) { + Catalog loadedCatalog = getCatalog(); + TableIdentifier ident = TableIdentifier.of(namespace, request.name()); + if (loadedCatalog.tableExists(ident)) { + throw new AlreadyExistsException("Table already exists: %s", ident); + } + + Map<String, String> properties = Maps.newHashMap(); + properties.put("created-at", OffsetDateTime.now(ZoneOffset.UTC).toString()); + properties.putAll(request.properties()); + + Map<String, String> config = Maps.newHashMap(); + Catalog.TableBuilder tableBuilder = + loadedCatalog + .buildTable(ident, request.schema()) + .withPartitionSpec(request.spec()) + .withSortOrder(request.writeOrder()) + .withProperties(properties); + + Table table; + if (request.location() != null) { + table = tableBuilder.withLocation(request.location()).createTransaction().table(); + } else { + table = tableBuilder.createTransaction().table(); + } + + Map<String, String> tableProperties = retrieveFileIOProperties(table.io()); + Map<String, String> filteredCredentialProperties = + CredentialPropertyUtils.filterCredentialProperties(tableProperties); + config.putAll( + MapUtils.getFilteredMap( + tableProperties, key -> catalogPropertiesToClientKeys.contains(key))); + config.putAll(filteredCredentialProperties); + config.putAll( + IcebergRESTUtils.buildRefreshProps( + catalogCredentialManager.catalogName(), ident, filteredCredentialProperties)); + + List<Credential> credentials = + IcebergRESTUtils.buildStorageCreds( + catalogCredentialManager.catalogName(), ident, table.io()); + + TableMetadata metadata = + TableMetadata.newTableMetadata( + request.schema(), + request.spec() != null ? request.spec() : PartitionSpec.unpartitioned(), + request.writeOrder() != null ? request.writeOrder() : SortOrder.unsorted(), + table.location(), + properties); + + return LoadTableResponse.builder() + .withTableMetadata(metadata) + .addAllConfig(config) + .addAllCredentials(credentials) + .build(); + } + + /** + * Federation-aware {@code registerTable}: registers the existing table metadata on the underlying + * (remote) catalog and extracts client-facing FileIO/credential properties from {@code + * table.io()}, mirroring {@link #loadTableInternal(TableIdentifier)}. + */ + private LoadTableResponse registerTableInternal( + Namespace namespace, RegisterTableRequest request) { + TableIdentifier ident = TableIdentifier.of(namespace, request.name()); + Table table = getCatalog().registerTable(ident, request.metadataLocation()); + + if (table instanceof BaseTable) { + return buildLoadTableResponseFromFileIo(ident, (BaseTable) table); + } + + throw new IllegalStateException("Cannot wrap catalog that does not produce BaseTable"); + } + + /** + * Federation-aware {@code updateTable}: applies the update against the underlying (remote) + * catalog, including the staged-create path used by federated table creation. + */ + private LoadTableResponse tableUpdateInternal(TableIdentifier ident, UpdateTableRequest request) { + if (isCreate(request)) { + // this is a hacky way to get TableOperations for an uncommitted table + Optional<Integer> formatVersion = + request.updates().stream() + .filter(update -> update instanceof MetadataUpdate.UpgradeFormatVersion) + .map(update -> ((MetadataUpdate.UpgradeFormatVersion) update).formatVersion()) + .findFirst(); + + Schema schema = + request.updates().stream() + .filter(update -> update instanceof MetadataUpdate.AddSchema) + .map(update -> ((MetadataUpdate.AddSchema) update).schema()) + .findFirst() + .orElse(EMPTY_SCHEMA); + + Catalog.TableBuilder tableBuilder = getCatalog().buildTable(ident, schema); + + TableMetadata.Builder changedMetadata = + formatVersion.map(TableMetadata::buildFromEmpty).orElse(TableMetadata.buildFromEmpty()); + request.updates().forEach(update -> update.applyTo(changedMetadata)); + + TableMetadata changedTableMeta = changedMetadata.build(); + tableBuilder.withPartitionSpec(changedTableMeta.spec()); + tableBuilder.withSortOrder(changedTableMeta.sortOrder()); + tableBuilder.withLocation(changedTableMeta.location()); + tableBuilder.withProperty(FORMAT_VERSION, String.valueOf(changedTableMeta.formatVersion())); + tableBuilder.withProperties(changedTableMeta.properties()); + + Transaction transaction = tableBuilder.createOrReplaceTransaction(); + if (transaction instanceof BaseTransaction) { + BaseTransaction baseTransaction = (BaseTransaction) transaction; + + return LoadTableResponse.builder() + .withTableMetadata(create(baseTransaction, request)) + .build(); + } else { + throw new IllegalStateException( + "Cannot wrap catalog that does not produce BaseTransaction"); + } + + } else { + return CatalogHandlers.updateTable(getCatalog(), ident, request); + } + } + + /** + * Federation-aware {@code loadTable}: loads the table from the underlying (remote) catalog and + * extracts client-facing FileIO/credential properties from {@code table.io()}. + */ + private LoadTableResponse loadTableInternal(TableIdentifier ident) { + Table table = getCatalog().loadTable(ident); + + if (table instanceof BaseTable) { + return buildLoadTableResponseFromFileIo(ident, (BaseTable) table); + } else if (table instanceof BaseMetadataTable) { + // metadata tables are loaded on the client side, return NoSuchTableException for now + throw new NoSuchTableException("Table does not exist: %s", ident.toString()); + } + + throw new IllegalStateException("Cannot wrap catalog that does not produce BaseTable"); + } + + /** + * Builds a {@link LoadTableResponse} from a remote {@link BaseTable}, exposing the client-facing + * FileIO and credential properties extracted from {@code table.io()}, including the refreshable + * vended credentials and refresh properties for the remote storage. + * + * @param ident the table identifier, used to build the credential-refresh endpoint. + * @param table the remote base table whose {@code io()} carries the storage credentials. + * @return the load-table response including FileIO-derived client config and vended credentials. + */ + private LoadTableResponse buildLoadTableResponseFromFileIo( + TableIdentifier ident, BaseTable table) { + Map<String, String> properties = retrieveFileIOProperties(table.io()); + Map<String, String> filteredCredentialProperties = + CredentialPropertyUtils.filterCredentialProperties(properties); + return LoadTableResponse.builder() + .withTableMetadata(table.operations().current()) + .addAllConfig( + MapUtils.getFilteredMap(properties, key -> catalogPropertiesToClientKeys.contains(key))) + // Keep only credential fields from FileIO properties before returning them to the client. + .addAllConfig(filteredCredentialProperties) + .addAllConfig( + IcebergRESTUtils.buildRefreshProps( + catalogCredentialManager.catalogName(), ident, filteredCredentialProperties)) + .addAllCredentials( + IcebergRESTUtils.buildStorageCreds( + catalogCredentialManager.catalogName(), ident, table.io())) + .build(); + } + + private static boolean isCreate(UpdateTableRequest request) { + boolean isCreate = + request.requirements().stream() + .anyMatch(UpdateRequirement.AssertTableDoesNotExist.class::isInstance); + + if (isCreate) { + List<UpdateRequirement> invalidRequirements = + request.requirements().stream() + .filter(req -> !(req instanceof UpdateRequirement.AssertTableDoesNotExist)) + .collect(Collectors.toList()); + Preconditions.checkArgument( + invalidRequirements.isEmpty(), "Invalid create requirements: %s", invalidRequirements); + } + + return isCreate; + } + + private static TableMetadata create(BaseTransaction baseTransaction, UpdateTableRequest request) { + // the only valid requirement is that the table will be created + TableOperations ops = baseTransaction.underlyingOps(); + request.requirements().forEach(requirement -> requirement.validate(ops.current())); + + TableMetadata.Builder builder = TableMetadata.buildFrom(baseTransaction.currentMetadata()); + request + .updates() + .forEach( + update -> { + if (shouldApplyMetadataUpdateAfterBuilder(update)) { + update.applyTo(builder); + } + }); + + // create transactions do not retry. if the table exists, retrying is not a solution + ops.commit(null, builder.build()); + + return ops.current(); + } + + /** + * Returns {@code false} for updates already reflected through {@link Catalog.TableBuilder} during + * staged create; those must not be applied again on {@link TableMetadata.Builder}. + */ + @VisibleForTesting + static boolean shouldApplyMetadataUpdateAfterBuilder(MetadataUpdate update) { + if (update instanceof MetadataUpdate.UpgradeFormatVersion) { + return false; + } + + if (update instanceof MetadataUpdate.AddSchema) { + return false; + } + + if (update instanceof MetadataUpdate.SetCurrentSchema) { + return false; + } + + if (update instanceof MetadataUpdate.RemoveSchemas) { + return false; + } + + if (update instanceof MetadataUpdate.SetLocation) { + return false; + } + + if (update instanceof MetadataUpdate.SetProperties) { + return false; + } + + if (update instanceof MetadataUpdate.RemoveProperties) { + return false; + } + + if (update instanceof MetadataUpdate.AddSortOrder) { + return false; + } + + if (update instanceof MetadataUpdate.SetDefaultSortOrder) { + return false; + } + + if (update instanceof MetadataUpdate.AddPartitionSpec) { + return false; + } + + if (update instanceof MetadataUpdate.SetDefaultPartitionSpec) { + return false; + } + + if (update instanceof MetadataUpdate.RemovePartitionSpecs) { + return false; + } + + return true; + } + + private static Map<String, String> retrieveFileIOProperties(FileIO fileIO) { + return fileIO instanceof InMemoryFileIO + ? Maps.newHashMap() + : new HashMap<>(fileIO.properties()); + } +} diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java index 8d8edcf185..f9561e3f8f 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java @@ -23,11 +23,13 @@ import com.github.benmanes.caffeine.cache.Caffeine; import com.github.benmanes.caffeine.cache.Scheduler; import com.google.common.annotations.VisibleForTesting; import com.google.common.util.concurrent.ThreadFactoryBuilder; +import java.util.Locale; import java.util.Map; import java.util.Optional; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; import org.apache.gravitino.GravitinoEnv; +import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend; import org.apache.gravitino.exceptions.NoSuchCatalogException; import org.apache.gravitino.iceberg.common.IcebergConfig; import org.apache.gravitino.iceberg.common.authentication.AuthenticationConfig; @@ -127,7 +129,17 @@ public class IcebergCatalogWrapperManager implements AutoCloseable { @VisibleForTesting protected CatalogWrapperForREST createCatalogWrapper( String catalogName, IcebergConfig icebergConfig) { - CatalogWrapperForREST rest = new CatalogWrapperForREST(catalogName, icebergConfig); + // When the backend is a federated Iceberg REST catalog, use FederatedCatalogWrapper so + // federation-aware behavior (FileIO property extraction, remote credential vending, remote + // /v1/config defaults) is applied through polymorphic dispatch rather than scattered + // instanceof checks. All other backends use the base CatalogWrapperForREST. + IcebergCatalogBackend backend = + IcebergCatalogBackend.valueOf( + icebergConfig.get(IcebergConfig.CATALOG_BACKEND).toUpperCase(Locale.ROOT)); + CatalogWrapperForREST rest = + backend == IcebergCatalogBackend.REST + ? new FederatedCatalogWrapper(catalogName, icebergConfig) + : new CatalogWrapperForREST(catalogName, icebergConfig); AuthenticationConfig authenticationConfig = new AuthenticationConfig(icebergConfig.getAllConfig()); if (authenticationConfig.isKerberosAuth() && rest.getCatalog() instanceof SupportsKerberos) { diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java index 465ebfc138..9a4d74091d 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java @@ -55,6 +55,8 @@ import org.apache.iceberg.io.StorageCredential; import org.apache.iceberg.io.SupportsStorageCredentials; import org.apache.iceberg.rest.RESTUtil; import org.apache.iceberg.rest.responses.ErrorResponse; +import org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse; +import org.apache.iceberg.rest.responses.LoadCredentialsResponse; import org.apache.iceberg.rest.responses.LoadTableResponse; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -178,16 +180,58 @@ public class IcebergRESTUtils { List<org.apache.iceberg.rest.credentials.Credential> restCredentials = new ArrayList<>(); for (StorageCredential credential : credentials) { - Map<String, String> filteredConfig = - CredentialPropertyUtils.filterCredentialProperties(credential.config()); - filteredConfig.putAll(buildRefreshProps(catalogName, tableIdentifier, filteredConfig)); restCredentials.add( - toRESTCredential(credential.prefix(), ImmutableMap.copyOf(filteredConfig))); + rewriteCredential( + catalogName, tableIdentifier, credential.prefix(), credential.config())); } return List.copyOf(restCredentials); } + /** + * Rewrites credentials returned by an upstream REST catalog so their {@code + * *.refresh-credentials-endpoint} entries point at this IRC instance instead of the upstream + * catalog. Without this, a federated {@code getTableCredentials} would return refresh endpoints + * scoped to the remote catalog's prefix, inconsistent with the loadTable/createTable paths that + * already rewrite via {@link #buildStorageCreds}. + * + * @param catalogName IRC catalog name used to build refresh paths + * @param tableIdentifier table receiving the credentials + * @param upstream the credentials response returned by the upstream REST catalog + * @return a credentials response with IRC-local refresh endpoints + */ + public static LoadCredentialsResponse rewriteTableCredentials( + String catalogName, TableIdentifier tableIdentifier, LoadCredentialsResponse upstream) { + ImmutableLoadCredentialsResponse.Builder builder = ImmutableLoadCredentialsResponse.builder(); + for (org.apache.iceberg.rest.credentials.Credential credential : upstream.credentials()) { + builder.addCredentials( + rewriteCredential( + catalogName, tableIdentifier, credential.prefix(), credential.config())); + } + return builder.build(); + } + + /** + * Filters an upstream credential's config to recognized credential properties and re-applies + * IRC-local refresh-endpoint properties, dropping any refresh endpoint scoped to the upstream + * catalog. + * + * @param catalogName IRC catalog name used to build refresh paths + * @param tableIdentifier table receiving the credential + * @param prefix storage location prefix for the credential + * @param config upstream credential config + * @return an Iceberg REST credential with IRC-local refresh endpoints + */ + private static org.apache.iceberg.rest.credentials.Credential rewriteCredential( + String catalogName, + TableIdentifier tableIdentifier, + String prefix, + Map<String, String> config) { + Map<String, String> filteredConfig = CredentialPropertyUtils.filterCredentialProperties(config); + filteredConfig.putAll(buildRefreshProps(catalogName, tableIdentifier, filteredConfig)); + return toRESTCredential(prefix, ImmutableMap.copyOf(filteredConfig)); + } + public static <T> Response ok(T t) { return Response.status(Response.Status.OK).entity(t).type(MediaType.APPLICATION_JSON).build(); } diff --git a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java index 55d30018a0..ce845ce020 100644 --- a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java +++ b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java @@ -75,6 +75,8 @@ import org.apache.iceberg.rest.RESTCatalog; import org.apache.iceberg.rest.auth.AuthProperties; import org.apache.iceberg.rest.credentials.Credential; import org.apache.iceberg.rest.requests.CreateTableRequest; +import org.apache.iceberg.rest.requests.ImmutableRegisterTableRequest; +import org.apache.iceberg.rest.requests.RegisterTableRequest; import org.apache.iceberg.rest.requests.UpdateTableRequest; import org.apache.iceberg.rest.responses.LoadCredentialsResponse; import org.apache.iceberg.rest.responses.LoadTableResponse; @@ -129,7 +131,9 @@ public class TestCatalogWrapperForREST { "{\"storage-credentials\":[{\"prefix\":\"s3://upstream/db/tbl/\",\"config\":{" + "\"s3.access-key-id\":\"upstream-key\"," + "\"s3.secret-access-key\":\"upstream-secret\"," - + "\"s3.session-token\":\"upstream-token\"}}]}"; + + "\"s3.session-token\":\"upstream-token\"," + + "\"client.refresh-credentials-endpoint\":" + + "\"v1/upstream/namespaces/db/tables/tbl/credentials\"}}]}"; AtomicReference<String> requestPath = new AtomicReference<>(); HttpServer server = HttpServer.create(new InetSocketAddress(0), 0); @@ -178,6 +182,11 @@ public class TestCatalogWrapperForREST { Assertions.assertEquals("upstream-key", credential.config().get("s3.access-key-id")); Assertions.assertEquals("upstream-secret", credential.config().get("s3.secret-access-key")); Assertions.assertEquals("upstream-token", credential.config().get("s3.session-token")); + // The upstream refresh endpoint is rewritten to this IRC catalog's prefix ("local"), not the + // remote catalog's ("upstream"), matching the loadTable/createTable federation paths. + Assertions.assertEquals( + "v1/local/namespaces/db/tables/tbl/credentials", + credential.config().get("client.refresh-credentials-endpoint")); } finally { server.stop(0); } @@ -552,8 +561,9 @@ public class TestCatalogWrapperForREST { IcebergConstants.WAREHOUSE, "s3://remote/warehouse")); - Map<String, String> configToClients = - CatalogWrapperForREST.buildCatalogConfigToClients(config, restCatalog); + // FederatedCatalogWrapper sources the client config from the remote RESTCatalog's properties(). + CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("test", config, restCatalog); + Map<String, String> configToClients = wrapper.buildCatalogConfigToClients(); Assertions.assertEquals( "org.apache.iceberg.aws.s3.S3FileIO", configToClients.get(IcebergConstants.IO_IMPL)); @@ -570,33 +580,30 @@ public class TestCatalogWrapperForREST { ImmutableMap.of( IcebergConstants.CATALOG_BACKEND, "hive", + IcebergConstants.URI, + "thrift://hive-metastore:9083", IcebergConstants.IO_IMPL, ResolvingFileIO.class.getName(), IcebergConstants.WAREHOUSE, "s3://bucket/warehouse")); - Catalog catalog = mock(Catalog.class); - Map<String, String> configToClients = - CatalogWrapperForREST.buildCatalogConfigToClients(config, catalog); + // Base CatalogWrapperForREST sources the client config from static catalog configuration. + CatalogWrapperForREST wrapper = new CatalogWrapperForREST("test", config); + Map<String, String> configToClients = wrapper.buildCatalogConfigToClients(); Assertions.assertEquals( ResolvingFileIO.class.getName(), configToClients.get(IcebergConstants.IO_IMPL)); } @Test - void testNonRESTCatalogClientConfig() { - Catalog catalog = mock(Catalog.class); - IcebergConfig config = - new IcebergConfig( + void testNonRestCatalogClientConfig() { + Map<String, String> configToClients = + CatalogWrapperForREST.filterCatalogConfigForClients( ImmutableMap.of( - IcebergConstants.CATALOG_BACKEND, - "hive", IcebergConstants.URI, "thrift://hive-metastore:9083", IcebergConstants.IO_IMPL, "org.apache.iceberg.aws.s3.S3FileIO")); - Map<String, String> configToClients = - CatalogWrapperForREST.buildCatalogConfigToClients(config, catalog); Assertions.assertFalse(configToClients.containsKey(IcebergConstants.URI)); Assertions.assertEquals( "org.apache.iceberg.aws.s3.S3FileIO", configToClients.get(IcebergConstants.IO_IMPL)); @@ -605,18 +612,52 @@ public class TestCatalogWrapperForREST { @Test void testCatalogClientConfigRejectsBadDataAccess() { - Catalog catalog = mock(Catalog.class); + Map<String, String> source = + ImmutableMap.of(IcebergConstants.ICEBERG_ACCESS_DELEGATION, "invalid-mode"); + Assertions.assertThrows( + IllegalArgumentException.class, + () -> CatalogWrapperForREST.filterCatalogConfigForClients(source)); + } + + @Test + void testFederatedRegisterTableIncludesFileIo() { + RESTCatalog catalog = mock(RESTCatalog.class); + BaseTable table = mock(BaseTable.class); + TableOperations ops = mock(TableOperations.class); + FileIO fileIO = mock(FileIO.class); + when(catalog.registerTable(any(TableIdentifier.class), anyString())).thenReturn(table); + when(table.operations()).thenReturn(ops); + when(ops.current()).thenReturn(minimalTableMetadataForStagedCreateTest()); + when(table.io()).thenReturn(fileIO); + when(fileIO.properties()) + .thenReturn( + ImmutableMap.of( + IcebergConstants.IO_IMPL, + "org.apache.iceberg.aws.s3.S3FileIO", + IcebergConstants.ICEBERG_S3_ENDPOINT, + "http://localhost:9000";)); + IcebergConfig config = new IcebergConfig( ImmutableMap.of( IcebergConstants.CATALOG_BACKEND, - "hive", - IcebergConstants.ICEBERG_ACCESS_DELEGATION, - "invalid-mode")); + "memory", + IcebergConstants.WAREHOUSE, + "/tmp/warehouse")); + CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("test", config, catalog); - Assertions.assertThrows( - IllegalArgumentException.class, - () -> CatalogWrapperForREST.buildCatalogConfigToClients(config, catalog)); + RegisterTableRequest request = + ImmutableRegisterTableRequest.builder() + .name("tbl") + .metadataLocation("s3://bucket/warehouse/tbl/metadata/v1.metadata.json") + .build(); + + LoadTableResponse response = wrapper.registerTable(Namespace.of("db"), request, false); + + Assertions.assertEquals( + "org.apache.iceberg.aws.s3.S3FileIO", response.config().get(IcebergConstants.IO_IMPL)); + Assertions.assertEquals( + "http://localhost:9000";, response.config().get(IcebergConstants.ICEBERG_S3_ENDPOINT)); } @Test @@ -760,41 +801,41 @@ public class TestCatalogWrapperForREST { void testPostBuilderMetadataSkipsHandledKinds() { Schema schema = new Schema(Types.NestedField.required(1, "c", Types.LongType.get())); Assertions.assertFalse( - CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder( + FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder( new MetadataUpdate.AddSchema(schema))); Assertions.assertFalse( - CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder( + FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder( new MetadataUpdate.UpgradeFormatVersion(2))); Assertions.assertFalse( - CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder( + FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder( new MetadataUpdate.SetCurrentSchema(-1))); Assertions.assertFalse( - CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder( + FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder( new MetadataUpdate.SetLocation("file:///tmp/loc"))); Assertions.assertFalse( - CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder( + FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder( new MetadataUpdate.SetProperties(ImmutableMap.of("k", "v")))); Assertions.assertFalse( - CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder( + FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder( new MetadataUpdate.RemoveProperties(Collections.singleton("k")))); Assertions.assertFalse( - CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder( + FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder( new MetadataUpdate.AddPartitionSpec(PartitionSpec.unpartitioned()))); Assertions.assertFalse( - CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder( + FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder( new MetadataUpdate.SetDefaultPartitionSpec(PartitionSpec.unpartitioned().specId()))); Assertions.assertFalse( - CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder( + FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder( new MetadataUpdate.AddSortOrder(SortOrder.unsorted()))); Assertions.assertFalse( - CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder( + FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder( new MetadataUpdate.SetDefaultSortOrder(SortOrder.unsorted().orderId()))); } @Test void testPostBuilderMetadataAllowsAssignUuid() { Assertions.assertTrue( - CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder( + FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder( new MetadataUpdate.AssignUUID(UUID.randomUUID().toString()))); } @@ -906,8 +947,8 @@ public class TestCatalogWrapperForREST { } /** - * Same derivation as {@link CatalogWrapperForREST#tableUpdateInternal} for staged create: replay - * metadata updates and read {@link TableMetadata#formatVersion()}. + * Same derivation as {@link FederatedCatalogWrapper#tableUpdateInternal} for staged create: + * replay metadata updates and read {@link TableMetadata#formatVersion()}. */ private static String expectedFormatVersionStringAfterStagedUpdates( Schema schema, Optional<Integer> formatVersionForUpgrade) { @@ -925,7 +966,7 @@ public class TestCatalogWrapperForREST { /** * Minimal valid Iceberg staged-create update sequence so {@link TableMetadata.Builder#build()} - * succeeds in {@link CatalogWrapperForREST#tableUpdateInternal}. + * succeeds in {@link FederatedCatalogWrapper#tableUpdateInternal}. */ private static List<MetadataUpdate> stagedCreateMetadataUpdates( Schema schema, Optional<Integer> formatVersion) { @@ -969,7 +1010,9 @@ public class TestCatalogWrapperForREST { } } - private static class StaticCatalogWrapperForREST extends CatalogWrapperForREST { + // Extends FederatedCatalogWrapper so table operations route through the federation-aware + // *Internal paths (FileIO extraction) against the injected catalog. + private static class StaticCatalogWrapperForREST extends FederatedCatalogWrapper { private final Catalog catalog; StaticCatalogWrapperForREST(String catalogName, IcebergConfig config, Catalog catalog) { diff --git a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java index f81ea32633..2d4386b730 100644 --- a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java +++ b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java @@ -18,12 +18,15 @@ */ package org.apache.gravitino.iceberg.service; +import com.google.common.collect.ImmutableMap; import com.google.common.collect.Maps; import java.util.Map; import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.gravitino.GravitinoEnv; import org.apache.gravitino.catalog.CatalogManager; +import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; +import org.apache.gravitino.iceberg.common.IcebergConfig; import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper; import org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext; import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider; @@ -110,4 +113,43 @@ public class TestIcebergCatalogWrapperManagerForREST { .getMessage() .contains("gravitino.iceberg-rest.catalog-config-provider=dynamic-config-provider")); } + + @Test + public void testCreateFederatedWrapperForRestBackend() { + IcebergConfig icebergConfig = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "rest", + IcebergConstants.URI, + "http://localhost:8181";)); + + CatalogWrapperForREST wrapper = newManager().createCatalogWrapper("test", icebergConfig); + + Assertions.assertInstanceOf(FederatedCatalogWrapper.class, wrapper); + } + + @Test + public void testCreateBaseWrapperForNonRestBackend() { + IcebergConfig icebergConfig = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.WAREHOUSE, + "/tmp/warehouse")); + + CatalogWrapperForREST wrapper = newManager().createCatalogWrapper("test", icebergConfig); + + Assertions.assertFalse(wrapper instanceof FederatedCatalogWrapper); + Assertions.assertEquals(CatalogWrapperForREST.class, wrapper.getClass()); + } + + private static IcebergCatalogWrapperManager newManager() { + Map<String, String> config = Maps.newHashMap(); + IcebergConfigProvider configProvider = IcebergConfigProviderFactory.create(config); + configProvider.initialize(config); + return new IcebergCatalogWrapperManager( + config, configProvider, false, configProvider.getMetalakeName()); + } } diff --git a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java index f7dd7cb2f0..2ad1f85c3d 100644 --- a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java +++ b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java @@ -45,6 +45,8 @@ import org.apache.iceberg.io.StorageCredential; import org.apache.iceberg.io.SupportsStorageCredentials; import org.apache.iceberg.rest.credentials.Credential; import org.apache.iceberg.rest.requests.CreateTableRequest; +import org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse; +import org.apache.iceberg.rest.responses.LoadCredentialsResponse; import org.apache.iceberg.types.Types.IntegerType; import org.apache.iceberg.types.Types.NestedField; import org.apache.iceberg.types.Types.StringType; @@ -318,4 +320,34 @@ public class TestIcebergRESTUtils { Assertions.assertFalse(config.containsKey("client.refresh-credentials-endpoint")); } + + @Test + void testRewriteTableCredentials() { + TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl"); + LoadCredentialsResponse upstream = + ImmutableLoadCredentialsResponse.builder() + .addCredentials( + IcebergRESTUtils.toRESTCredential( + "s3://bucket/db/tbl/", + ImmutableMap.of( + "s3.session-token", + "upstream-token", + "s3.session-token-expires-at-ms", + "123", + "client.refresh-credentials-endpoint", + "v1/upstream/namespaces/db/tables/tbl/credentials"))) + .build(); + + LoadCredentialsResponse rewritten = + IcebergRESTUtils.rewriteTableCredentials("irc1", table, upstream); + + Assertions.assertEquals(1, rewritten.credentials().size()); + Credential credential = rewritten.credentials().get(0); + Assertions.assertEquals("s3://bucket/db/tbl/", credential.prefix()); + Assertions.assertEquals("upstream-token", credential.config().get("s3.session-token")); + // The refresh endpoint is re-scoped to the IRC catalog, dropping the upstream-scoped one. + Assertions.assertEquals( + "v1/irc1/namespaces/db/tables/tbl/credentials", + credential.config().get("client.refresh-credentials-endpoint")); + } }
