This is an automated email from the ASF dual-hosted git repository.
bharos pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new dad9a08dcd [#11584] feat(iceberg-rest): Gate config endpoints by
backend capabilities (#11634)
dad9a08dcd is described below
commit dad9a08dcdf397ed9558e4c36d57bdd49b75f3bf
Author: Akshay Thorat <[email protected]>
AuthorDate: Thu Sep 3 11:19:25 2026 -0700
[#11584] feat(iceberg-rest): Gate config endpoints by backend capabilities
(#11634)
### What changes were proposed in this pull request?
Derive the endpoint set advertised by `/v1/config` from the backing
catalog's capabilities rather than a hardcoded static list.
`V1_SUBMIT_TABLE_SCAN_PLAN` is gated behind
`CatalogWrapperForREST.supportsScanPlanOperations()`:
- `CatalogWrapperForREST` returns `true`. Gravitino plans scans locally
on loaded table metadata, which works for every backend it handles
directly (Hive, JDBC, Memory, Custom).
- `FederatedCatalogWrapper` overrides it to report what the remote
catalog advertises in its own `/v1/config`, since it delegates
`planTableScan` upstream rather than planning locally. The result is
cached for the wrapper's lifetime, so the remote is queried at most once
instead of on every local `/v1/config` call.
A remote that omits `endpoints` is treated as not supporting scan
planning, matching the Iceberg client's fallback to an endpoint set that
predates scan planning. A failed lookup is not cached and does not
advertise the endpoint, so it can resolve once the remote recovers.
Also extracts the auth manager / HTTP client / auth session lifecycle
shared by the federated credential, scan-plan and config calls into
`callRemoteCatalog`, rather than adding a third copy of that
boilerplate.
### Why are the changes needed?
`IcebergConfigOperations` advertises a hardcoded `DEFAULT_ENDPOINTS`
list (including `V1_SUBMIT_TABLE_SCAN_PLAN`) for every catalog,
regardless of whether the configured backend actually supports those
operations. A client that trusts the advertised `endpoints` may call an
operation the backend cannot serve, producing confusing runtime errors.
Fix: #11584
### Does this PR introduce _any_ user-facing change?
A federated (REST backend) catalog advertises the scan plan endpoint
only when the remote catalog it federates advertises it. Clients that
rely on the advertised endpoints will see the operations their catalog
can actually serve. Non-REST backends are unaffected.
### How was this patch tested?
- `TestCatalogWrapperForREST` gains coverage against an embedded
`/v1/config` server: support reported when the remote advertises the
endpoint, not reported when it omits the endpoint or sends no
`endpoints` at all, the remote queried only once, `warehouse` forwarded
and URL-encoded, and a failed lookup not cached.
- `TestIcebergConfigEndpointGating` and `TestIcebergConfig` cover the
`/v1/config` wiring in both directions.
- `:iceberg:iceberg-rest-server:test` is green (404 tests, 0 failures).
---
.../iceberg/service/CatalogWrapperForREST.java | 14 ++
.../iceberg/service/FederatedCatalogWrapper.java | 216 +++++++++++++--------
.../service/rest/IcebergConfigOperations.java | 31 +--
.../iceberg/service/TestCatalogWrapperForREST.java | 195 +++++++++++++++++++
.../iceberg/service/rest/IcebergRestTestUtil.java | 21 +-
.../iceberg/service/rest/TestIcebergConfig.java | 23 +++
.../rest/TestIcebergConfigEndpointGating.java | 124 ++++++++++++
7 files changed, 530 insertions(+), 94 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 1af51f5295..c58547e4fc 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
@@ -181,6 +181,20 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
}
}
+ /**
+ * Whether this catalog wrapper supports server-side scan planning.
+ *
+ * <p>Gravitino plans scans locally on top of loaded table metadata, which
works for every backend
+ * this class handles directly (Hive, JDBC, Memory, Custom). {@link
FederatedCatalogWrapper}
+ * delegates planning to the remote catalog instead, so it overrides this to
report what that
+ * catalog actually advertises.
+ *
+ * @return {@code true} if the scan-plan endpoint should be advertised
+ */
+ public boolean supportsScanPlanOperations() {
+ return true;
+ }
+
@Override
public void close() throws Exception {
try {
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
index b4128ff724..698ae3220c 100644
---
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
@@ -30,6 +30,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.gravitino.credential.CredentialPrivilege;
import org.apache.gravitino.credential.CredentialPropertyUtils;
@@ -56,6 +57,7 @@ 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.Endpoint;
import org.apache.iceberg.rest.ErrorHandlers;
import org.apache.iceberg.rest.HTTPClient;
import org.apache.iceberg.rest.ParserContext;
@@ -71,6 +73,7 @@ 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.ConfigResponse;
import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
import org.apache.iceberg.rest.responses.PlanTableScanResponse;
@@ -94,6 +97,13 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
private static final String FORMAT_VERSION = "format-version";
private static final Schema EMPTY_SCHEMA = new Schema();
+ /**
+ * Caches whether the remote catalog advertises the scan-plan endpoint. Only
successful lookups
+ * are cached, so a failure to reach the remote can be retried. Races just
repeat an idempotent
+ * lookup.
+ */
+ private volatile Boolean remoteSupportsScanPlan;
+
/**
* Creates a federated wrapper.
*
@@ -188,12 +198,111 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
catalogCredentialManager.catalogName(), tableIdentifier, response);
}
+ /**
+ * Reports whether the remote catalog advertises the scan-plan endpoint,
since {@link
+ * #planTableScan} delegates planning to it rather than planning locally.
+ *
+ * <p>The answer comes from the remote catalog's own {@code /v1/config}
response and is cached for
+ * the lifetime of this wrapper, so the remote is queried at most once
rather than on every local
+ * {@code /v1/config} call.
+ *
+ * <p>A remote that omits {@code endpoints} is treated as not supporting
scan planning. That
+ * matches the Iceberg client, which falls back to a default endpoint set
that predates scan
+ * planning when the field is absent.
+ *
+ * <p>If the remote cannot be reached the result is not cached and the
endpoint is not advertised,
+ * so a later call can still resolve it once the remote recovers. Not
advertising is the safe
+ * direction here: the endpoint would fail anyway while the remote is
unreachable.
+ *
+ * @return {@code true} if the remote catalog advertises {@code
V1_SUBMIT_TABLE_SCAN_PLAN}.
+ */
+ @Override
+ public boolean supportsScanPlanOperations() {
+ Boolean cached = remoteSupportsScanPlan;
+ if (cached != null) {
+ return cached;
+ }
+
+ try {
+ boolean supported =
+
fetchRemoteConfig().endpoints().contains(Endpoint.V1_SUBMIT_TABLE_SCAN_PLAN);
+ remoteSupportsScanPlan = supported;
+ return supported;
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to read the endpoints advertised by the remote catalog of
{}; not advertising the"
+ + " scan plan endpoint",
+ catalogCredentialManager.catalogName(),
+ e);
+ return false;
+ }
+ }
+
+ /**
+ * Fetches the remote catalog's {@code /v1/config} response.
+ *
+ * <p>The {@code warehouse} query parameter is forwarded when configured, so
a remote serving
+ * several warehouses returns the endpoint set for the one this catalog
federates.
+ *
+ * <p>{@code RESTCatalog} already fetched this at init but keeps the
endpoint set private, so it
+ * has to be re-fetched here.
+ *
+ * @return the remote catalog's config response.
+ */
+ private ConfigResponse fetchRemoteConfig() {
+ RESTCatalog restCatalog = (RESTCatalog) getCatalog();
+ String warehouse =
restCatalog.properties().get(CatalogProperties.WAREHOUSE_LOCATION);
+ Map<String, String> queryParams =
+ warehouse == null || warehouse.isEmpty()
+ ? Collections.emptyMap()
+ : ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, warehouse);
+
+ return callRemoteCatalog(
+ restCatalog,
+ "reading the remote catalog config",
+ client ->
+ client.get(
+ ResourcePaths.config(),
+ queryParams,
+ ConfigResponse.class,
+ Collections.emptyMap(),
+ ErrorHandlers.configErrorHandler()));
+ }
+
private static LoadCredentialsResponse getRESTTableCredentials(
RESTCatalog restCatalog, TableIdentifier identifier) {
Map<String, String> properties = Maps.newHashMap(restCatalog.properties());
String credentialsPath =
ResourcePaths.forCatalogProperties(properties).table(identifier) +
"/credentials";
+ return callRemoteCatalog(
+ restCatalog,
+ String.format("loading credentials for table: %s", identifier),
+ client ->
+ client.get(
+ credentialsPath,
+ LoadCredentialsResponse.class,
+ Collections.emptyMap(),
+ ErrorHandlers.tableErrorHandler()));
+ }
+
+ /**
+ * Runs an action against the remote REST catalog through a short-lived
authenticated client.
+ *
+ * <p>Centralizes the auth manager, HTTP client and auth session lifecycle
shared by the federated
+ * credential, scan-plan and config requests. Resources are closed in
reverse order of creation,
+ * and a close failure on one does not prevent the others from being closed.
+ *
+ * @param restCatalog the underlying REST catalog whose properties supply
the URI and auth config.
+ * @param description what the action is doing, used in close-failure log
messages.
+ * @param action invoked with a client bound to an authenticated session.
+ * @param <T> the action's result type.
+ * @return the action's result.
+ */
+ private static <T> T callRemoteCatalog(
+ RESTCatalog restCatalog, String description, Function<RESTClient, T>
action) {
+ Map<String, String> properties = Maps.newHashMap(restCatalog.properties());
+
AuthManager authManager = null;
RESTClient client = null;
AuthSession authSession = null;
@@ -205,40 +314,24 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
.withHeaders(RESTUtil.configHeaders(properties))
.build();
authSession = authManager.catalogSession(client, properties);
- return client
- .withAuthSession(authSession)
- .get(
- credentialsPath,
- LoadCredentialsResponse.class,
- Collections.emptyMap(),
- ErrorHandlers.tableErrorHandler());
+ return action.apply(client.withAuthSession(authSession));
} finally {
- if (authSession != null) {
- try {
- authSession.close();
- } catch (Exception e) {
- LOG.warn(
- "Failed to close auth session when loading credentials for
table: {}", identifier, e);
- }
- }
+ closeQuietly(authSession, "auth session", description);
+ closeQuietly(client, "REST client", description);
+ closeQuietly(authManager, "auth manager", description);
+ }
+ }
- if (client != null) {
- try {
- client.close();
- } catch (Exception e) {
- LOG.warn(
- "Failed to close REST client when loading credentials for table:
{}", identifier, e);
- }
- }
+ private static void closeQuietly(
+ AutoCloseable closeable, String resourceName, String description) {
+ if (closeable == null) {
+ return;
+ }
- if (authManager != null) {
- try {
- authManager.close();
- } catch (Exception e) {
- LOG.warn(
- "Failed to close auth manager when loading credentials for
table: {}", identifier, e);
- }
- }
+ try {
+ closeable.close();
+ } catch (Exception e) {
+ LOG.warn("Failed to close {} when {}", resourceName, description, e);
}
}
@@ -280,55 +373,18 @@ public class FederatedCatalogWrapper extends
CatalogWrapperForREST {
.add("caseSensitive", scanRequest.caseSensitive())
.build();
- 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)
- .post(
- planPath,
- scanRequest,
- PlanTableScanResponse.class,
- headers,
- ErrorHandlers.planErrorHandler(),
- ignored -> {},
- parserContext);
- } finally {
- if (authSession != null) {
- try {
- authSession.close();
- } catch (Exception e) {
- LOG.warn(
- "Failed to close auth session when planning table scan for
table: {}", identifier, e);
- }
- }
-
- if (client != null) {
- try {
- client.close();
- } catch (Exception e) {
- LOG.warn(
- "Failed to close REST client when planning table scan for table:
{}", identifier, e);
- }
- }
-
- if (authManager != null) {
- try {
- authManager.close();
- } catch (Exception e) {
- LOG.warn(
- "Failed to close auth manager when planning table scan for
table: {}", identifier, e);
- }
- }
- }
+ return callRemoteCatalog(
+ restCatalog,
+ String.format("planning table scan for table: %s", identifier),
+ client ->
+ client.post(
+ planPath,
+ scanRequest,
+ PlanTableScanResponse.class,
+ headers,
+ ErrorHandlers.planErrorHandler(),
+ ignored -> {},
+ parserContext));
}
/**
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergConfigOperations.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergConfigOperations.java
index 0766b2f348..59e0f9fdde 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergConfigOperations.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergConfigOperations.java
@@ -75,9 +75,11 @@ public class IcebergConfigOperations {
.add(Endpoint.V1_REGISTER_TABLE)
.add(Endpoint.V1_REPORT_METRICS)
.add(Endpoint.V1_TABLE_CREDENTIALS)
- .add(Endpoint.V1_SUBMIT_TABLE_SCAN_PLAN)
.build();
+ private static final List<Endpoint> SCAN_PLAN_ENDPOINTS =
+ ImmutableList.of(Endpoint.V1_SUBMIT_TABLE_SCAN_PLAN);
+
private static final List<Endpoint> DEFAULT_VIEW_ENDPOINTS =
ImmutableList.<Endpoint>builder()
.add(Endpoint.V1_LIST_VIEWS)
@@ -101,21 +103,29 @@ public class IcebergConfigOperations {
@ResponseMetered(name = "config", absolute = true)
public Response getConfig(@DefaultValue("") @QueryParam("warehouse") String
warehouse) {
String catalogName = getCatalogName(warehouse);
- boolean supportsView = supportsViewOperations(catalogName);
+ CatalogWrapperForREST catalogWrapper = getCatalogWrapper(catalogName);
+ boolean supportsView = catalogWrapper.supportsViewOperations();
+ boolean supportsScanPlan = catalogWrapper.supportsScanPlanOperations();
ConfigResponse.Builder builder = ConfigResponse.builder();
-
builder.withDefaults(getDefaultConfig(catalogName)).withEndpoints(getEndpoints(supportsView));
+ builder
+ .withDefaults(getDefaultConfig(catalogName))
+ .withEndpoints(getEndpoints(supportsView, supportsScanPlan));
if (StringUtils.isNotBlank(warehouse)) {
builder.withDefault("prefix", warehouse);
}
return IcebergRESTUtils.ok(builder.build());
}
- private List<Endpoint> getEndpoints(boolean supportsViewOperations) {
- if (!supportsViewOperations) {
- return DEFAULT_ENDPOINTS;
+ private List<Endpoint> getEndpoints(
+ boolean supportsViewOperations, boolean supportsScanPlanOperations) {
+ Stream<Endpoint> endpoints = DEFAULT_ENDPOINTS.stream();
+ if (supportsScanPlanOperations) {
+ endpoints = Stream.concat(endpoints, SCAN_PLAN_ENDPOINTS.stream());
+ }
+ if (supportsViewOperations) {
+ endpoints = Stream.concat(endpoints, DEFAULT_VIEW_ENDPOINTS.stream());
}
- return Stream.concat(DEFAULT_ENDPOINTS.stream(),
DEFAULT_VIEW_ENDPOINTS.stream())
- .collect(Collectors.toList());
+ return endpoints.collect(Collectors.toList());
}
private Map<String, String> getCatalogConfig(String catalogName) {
@@ -133,11 +143,6 @@ public class IcebergConfigOperations {
}
}
- private boolean supportsViewOperations(String catalogName) {
- CatalogWrapperForREST catalogWrapperForREST =
getCatalogWrapper(catalogName);
- return catalogWrapperForREST.supportsViewOperations();
- }
-
private CatalogWrapperForREST getCatalogWrapper(String catalogName) {
return catalogWrapperManager.getCatalogWrapper(catalogName);
}
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 713091a9e4..50a3f684d1 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
@@ -36,6 +36,7 @@ import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -73,6 +74,7 @@ import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.ResolvingFileIO;
import org.apache.iceberg.io.StorageCredential;
import org.apache.iceberg.io.SupportsStorageCredentials;
+import org.apache.iceberg.rest.Endpoint;
import org.apache.iceberg.rest.PlanStatus;
import org.apache.iceberg.rest.RESTCatalog;
import org.apache.iceberg.rest.auth.AuthProperties;
@@ -82,6 +84,8 @@ import
org.apache.iceberg.rest.requests.ImmutableRegisterTableRequest;
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.ConfigResponse;
+import org.apache.iceberg.rest.responses.ConfigResponseParser;
import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
import org.apache.iceberg.rest.responses.PlanTableScanResponse;
@@ -1189,6 +1193,197 @@ public class TestCatalogWrapperForREST {
new MetadataUpdate.AssignUUID(UUID.randomUUID().toString())));
}
+ @Test
+ void testScanPlanSupportedForNonRESTBackend() {
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+
+ // Gravitino plans the scan locally for non-REST backends, so it always
supports the endpoint.
+ CatalogWrapperForREST wrapper = new CatalogWrapperForREST("local-catalog",
config);
+ Assertions.assertTrue(wrapper.supportsScanPlanOperations());
+ }
+
+ @Test
+ void testScanPlanSupportedWhenRemoteAdvertisesEndpoint() throws Exception {
+ ConfigResponse remoteConfig =
+ ConfigResponse.builder()
+ .withEndpoints(
+ Arrays.asList(Endpoint.V1_LOAD_TABLE,
Endpoint.V1_SUBMIT_TABLE_SCAN_PLAN))
+ .build();
+
+ withRemoteConfigServer(
+ remoteConfig,
+ (wrapper, requests) ->
Assertions.assertTrue(wrapper.supportsScanPlanOperations()));
+ }
+
+ @Test
+ void testScanPlanUnsupportedWhenRemoteOmitsEndpoint() throws Exception {
+ ConfigResponse remoteConfig =
+ ConfigResponse.builder()
+ .withEndpoints(Arrays.asList(Endpoint.V1_LOAD_TABLE,
Endpoint.V1_LIST_TABLES))
+ .build();
+
+ withRemoteConfigServer(
+ remoteConfig,
+ (wrapper, requests) ->
Assertions.assertFalse(wrapper.supportsScanPlanOperations()));
+ }
+
+ @Test
+ void testScanPlanUnsupportedWhenRemoteAdvertisesNoEndpoints() throws
Exception {
+ // A remote that omits "endpoints" is treated the same way the Iceberg
client treats it: as a
+ // catalog whose endpoint set predates scan planning.
+ withRemoteConfigServer(
+ ConfigResponse.builder().build(),
+ (wrapper, requests) ->
Assertions.assertFalse(wrapper.supportsScanPlanOperations()));
+ }
+
+ @Test
+ void testScanPlanSupportQueriesRemoteOnlyOnce() throws Exception {
+ ConfigResponse remoteConfig =
+ ConfigResponse.builder()
+
.withEndpoints(Collections.singletonList(Endpoint.V1_SUBMIT_TABLE_SCAN_PLAN))
+ .build();
+
+ withRemoteConfigServer(
+ remoteConfig,
+ (wrapper, requests) -> {
+ Assertions.assertTrue(wrapper.supportsScanPlanOperations());
+ Assertions.assertTrue(wrapper.supportsScanPlanOperations());
+ Assertions.assertTrue(wrapper.supportsScanPlanOperations());
+ Assertions.assertEquals(
+ 1, requests.size(), "The remote config should be fetched once
and then cached");
+ });
+ }
+
+ @Test
+ void testScanPlanSupportForwardsWarehouseToRemote() throws Exception {
+ ConfigResponse remoteConfig =
+ ConfigResponse.builder()
+
.withEndpoints(Collections.singletonList(Endpoint.V1_SUBMIT_TABLE_SCAN_PLAN))
+ .build();
+
+ withRemoteConfigServer(
+ remoteConfig,
+ ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION,
"s3://remote/warehouse"),
+ (wrapper, requests) -> {
+ Assertions.assertTrue(wrapper.supportsScanPlanOperations());
+ Assertions.assertEquals(1, requests.size());
+ Assertions.assertEquals("/v1/config", requests.get(0).path);
+ Assertions.assertEquals(
+ "warehouse=s3%3A%2F%2Fremote%2Fwarehouse",
requests.get(0).rawQuery);
+ });
+ }
+
+ @Test
+ void testScanPlanSupportLookupFailureIsNotCached() throws Exception {
+ List<RecordedRequest> requests = Collections.synchronizedList(new
ArrayList<>());
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ requests.add(
+ new RecordedRequest(
+ exchange.getRequestURI().getPath(),
exchange.getRequestURI().getRawQuery()));
+ // 400 rather than a retryable status, so each lookup is exactly one
request and the
+ // assertion below counts lookups rather than the REST client's
retry attempts.
+ exchange.sendResponseHeaders(400, -1);
+ exchange.close();
+ });
+ server.start();
+ try {
+ CatalogWrapperForREST wrapper = federatedWrapperFor(server,
Collections.emptyMap());
+
+ // A failed lookup must not be cached, so a later call can resolve it
once the remote is back.
+ Assertions.assertFalse(wrapper.supportsScanPlanOperations());
+ Assertions.assertFalse(wrapper.supportsScanPlanOperations());
+ Assertions.assertEquals(2, requests.size());
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ private void withRemoteConfigServer(ConfigResponse remoteConfig,
RemoteConfigAssertion assertion)
+ throws Exception {
+ withRemoteConfigServer(remoteConfig, Collections.emptyMap(), assertion);
+ }
+
+ /**
+ * Serves {@code remoteConfig} from an embedded {@code /v1/config} endpoint
and hands the test a
+ * federated wrapper pointed at it, along with the requests the wrapper made.
+ */
+ private void withRemoteConfigServer(
+ ConfigResponse remoteConfig,
+ Map<String, String> extraCatalogProperties,
+ RemoteConfigAssertion assertion)
+ throws Exception {
+ String responseJson = ConfigResponseParser.toJson(remoteConfig);
+ List<RecordedRequest> requests = Collections.synchronizedList(new
ArrayList<>());
+
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ requests.add(
+ new RecordedRequest(
+ exchange.getRequestURI().getPath(),
exchange.getRequestURI().getRawQuery()));
+ byte[] body = responseJson.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(200, body.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(body);
+ }
+ });
+ server.start();
+ try {
+ assertion.accept(federatedWrapperFor(server, extraCatalogProperties),
requests);
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ private CatalogWrapperForREST federatedWrapperFor(
+ HttpServer server, Map<String, String> extraCatalogProperties) {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.<String, String>builder()
+ .put(CatalogProperties.URI, uri)
+ .put(AuthProperties.AUTH_TYPE, AuthProperties.AUTH_TYPE_NONE)
+ .putAll(extraCatalogProperties)
+ .build());
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ return new StaticCatalogWrapperForREST("local", config, restCatalog);
+ }
+
+ @FunctionalInterface
+ private interface RemoteConfigAssertion {
+ void accept(CatalogWrapperForREST wrapper, List<RecordedRequest> requests);
+ }
+
+ private static class RecordedRequest {
+ private final String path;
+ private final String rawQuery;
+
+ RecordedRequest(String path, String rawQuery) {
+ this.path = path;
+ this.rawQuery = rawQuery;
+ }
+ }
+
@Test
void testStagedCreateBuilderUsesDerivedMetadataV3() {
RESTCatalog catalog = mock(RESTCatalog.class);
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/IcebergRestTestUtil.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/IcebergRestTestUtil.java
index 8c572999b6..c10a6140b3 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/IcebergRestTestUtil.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/IcebergRestTestUtil.java
@@ -87,6 +87,16 @@ public class IcebergRestTestUtil {
public static final String RENAME_VIEW_PATH = V_1 + "/views/rename";
public static final String REPORT_METRICS_POSTFIX = "metrics";
+ /** Factory for creating {@link IcebergCatalogWrapperManager} instances in
tests. */
+ @FunctionalInterface
+ public interface WrapperManagerFactory {
+ IcebergCatalogWrapperManager create(
+ Map<String, String> properties,
+ IcebergConfigProvider configProvider,
+ boolean auxMode,
+ String metalakeName);
+ }
+
public static final boolean DEBUG_SERVER_LOG_ENABLED = true;
public static ResourceConfig getIcebergResourceConfig(Class c) {
@@ -95,6 +105,15 @@ public class IcebergRestTestUtil {
public static ResourceConfig getIcebergResourceConfig(
Class c, boolean bindIcebergTableOps, List<EventListenerPlugin>
eventListenerPlugins) {
+ return getIcebergResourceConfig(
+ c, bindIcebergTableOps, eventListenerPlugins,
IcebergCatalogWrapperManagerForTest::new);
+ }
+
+ public static ResourceConfig getIcebergResourceConfig(
+ Class c,
+ boolean bindIcebergTableOps,
+ List<EventListenerPlugin> eventListenerPlugins,
+ WrapperManagerFactory wrapperManagerFactory) {
ResourceConfig resourceConfig = new ResourceConfig();
resourceConfig.register(c);
resourceConfig.register(IcebergObjectMapperProvider.class).register(JacksonFeature.class);
@@ -140,7 +159,7 @@ public class IcebergRestTestUtil {
configProvider.initialize(catalogConf);
// used to override register table interface
IcebergCatalogWrapperManager icebergCatalogWrapperManager =
- new IcebergCatalogWrapperManagerForTest(
+ wrapperManagerFactory.create(
catalogConf, configProvider, false,
configProvider.getMetalakeName());
IcebergRESTServerContext.create(
configProvider, false, false, true, icebergCatalogWrapperManager);
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergConfig.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergConfig.java
index f60eebf36f..b9192d31db 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergConfig.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergConfig.java
@@ -126,6 +126,29 @@ public class TestIcebergConfig extends IcebergTestBase {
"Config response should contain view list endpoint for catalog that
supports views");
}
+ @Test
+ public void testConfigEndpointsContainScanPlanForNonRESTBackend() {
+ // Gravitino plans scans locally for non-REST backends (memory, hive,
jdbc), so the scan plan
+ // endpoint must be advertised.
+ String warehouseName = IcebergRestTestUtil.PREFIX;
+ Map<String, String> queryParams = ImmutableMap.of("warehouse",
warehouseName);
+ Response resp =
+ getIcebergClientBuilder(IcebergRestTestUtil.CONFIG_PATH,
Optional.of(queryParams)).get();
+ Assertions.assertEquals(Response.Status.OK.getStatusCode(),
resp.getStatus());
+
+ ConfigResponse response = resp.readEntity(ConfigResponse.class);
+
+ boolean hasScanPlanEndpoint =
+ response.endpoints().stream()
+ .anyMatch(
+ endpoint ->
+ "POST".equals(endpoint.httpMethod())
+ &&
endpoint.path().contains("namespaces/{namespace}/tables/{table}/plan"));
+ Assertions.assertTrue(
+ hasScanPlanEndpoint,
+ "Config response must advertise scan plan endpoint for non-REST
backend catalogs");
+ }
+
@Test
public void testConfigScanPlanEndpointPathIsNamespaceScoped() {
// Iceberg 1.11+ advertises the namespace-scoped scan plan path via
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergConfigEndpointGating.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergConfigEndpointGating.java
new file mode 100644
index 0000000000..3fa6e38e7c
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergConfigEndpointGating.java
@@ -0,0 +1,124 @@
+/*
+ * 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.rest;
+
+import java.util.Collections;
+import java.util.Map;
+import javax.ws.rs.core.Application;
+import javax.ws.rs.core.Response;
+import org.apache.gravitino.iceberg.common.IcebergConfig;
+import org.apache.gravitino.iceberg.service.CatalogWrapperForREST;
+import org.apache.gravitino.iceberg.service.IcebergCatalogWrapperManager;
+import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider;
+import org.apache.iceberg.rest.responses.ConfigResponse;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests that the /v1/config endpoint gates optional endpoints (like scan
planning) based on the
+ * catalog backend's capabilities.
+ */
+public class TestIcebergConfigEndpointGating extends IcebergTestBase {
+
+ @Override
+ protected Application configure() {
+ return IcebergRestTestUtil.getIcebergResourceConfig(
+ IcebergConfigOperations.class,
+ true,
+ Collections.emptyList(),
+ NoScanPlanWrapperManager::new);
+ }
+
+ @Test
+ public void testScanPlanEndpointOmittedWhenNotSupported() {
+ Response resp = getConfigClientBuilder().get();
+ Assertions.assertEquals(Response.Status.OK.getStatusCode(),
resp.getStatus());
+
+ ConfigResponse response = resp.readEntity(ConfigResponse.class);
+
+ boolean hasScanPlanEndpoint =
+ response.endpoints().stream()
+ .anyMatch(
+ endpoint ->
+ "POST".equals(endpoint.httpMethod())
+ && endpoint.path().contains("tables/{table}/plan"));
+
+ Assertions.assertFalse(
+ hasScanPlanEndpoint,
+ "Config response must NOT advertise scan plan endpoint for catalogs
that do not support it");
+ }
+
+ @Test
+ public void testCoreEndpointsAlwaysPresent() {
+ Response resp = getConfigClientBuilder().get();
+ Assertions.assertEquals(Response.Status.OK.getStatusCode(),
resp.getStatus());
+
+ ConfigResponse response = resp.readEntity(ConfigResponse.class);
+
+ // Core table endpoints must always be present
+ boolean hasListTables =
+ response.endpoints().stream()
+ .anyMatch(
+ endpoint ->
+ "GET".equals(endpoint.httpMethod())
+ &&
endpoint.path().contains("namespaces/{namespace}/tables"));
+ Assertions.assertTrue(hasListTables, "Config must always advertise list
tables endpoint");
+
+ boolean hasLoadTable =
+ response.endpoints().stream()
+ .anyMatch(
+ endpoint ->
+ "GET".equals(endpoint.httpMethod())
+ && endpoint.path().contains("tables/{table}"));
+ Assertions.assertTrue(hasLoadTable, "Config must always advertise load
table endpoint");
+ }
+
+ /**
+ * Wrapper that reports no scan-plan support, standing in for a federated
catalog whose remote
+ * does not advertise {@code V1_SUBMIT_TABLE_SCAN_PLAN}. The remote lookup
itself is covered by
+ * {@code TestCatalogWrapperForREST}; this only pins the {@code /v1/config}
wiring.
+ */
+ static class NoScanPlanCatalogWrapper extends CatalogWrapperForTest {
+ public NoScanPlanCatalogWrapper(String catalogName, IcebergConfig
icebergConfig) {
+ super(catalogName, icebergConfig);
+ }
+
+ @Override
+ public boolean supportsScanPlanOperations() {
+ return false;
+ }
+ }
+
+ /** Manager that creates {@link NoScanPlanCatalogWrapper} instances. */
+ public static class NoScanPlanWrapperManager extends
IcebergCatalogWrapperManager {
+ public NoScanPlanWrapperManager(
+ Map<String, String> properties,
+ IcebergConfigProvider configProvider,
+ boolean auxMode,
+ String metalakeName) {
+ super(properties, configProvider, auxMode, metalakeName);
+ }
+
+ @Override
+ public CatalogWrapperForREST createCatalogWrapper(
+ String catalogName, IcebergConfig icebergConfig) {
+ return new NoScanPlanCatalogWrapper(catalogName, icebergConfig);
+ }
+ }
+}
