Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
/*
* 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.iceberg.rest;

public final class RESTCatalogProperties {

private RESTCatalogProperties() {}

public static final String SNAPSHOT_LOADING_MODE = "snapshot-loading-mode";
public static final String SNAPSHOT_LOADING_MODE_DEFAULT = SnapshotMode.ALL.name();

public static final String METRICS_REPORTING_ENABLED = "rest-metrics-reporting-enabled";
public static final boolean METRICS_REPORTING_ENABLED_DEFAULT = true;

// for backwards compatibility with older REST servers where it can be assumed that a particular
// server supports view endpoints but doesn't send the "endpoints" field in the ConfigResponse
public static final String VIEW_ENDPOINTS_SUPPORTED = "view-endpoints-supported";
public static final boolean VIEW_ENDPOINTS_SUPPORTED_DEFAULT = false;

public static final String PAGE_SIZE = "rest-page-size";

public enum SnapshotMode {
ALL,
REFS
}
}
49 changes: 28 additions & 21 deletions core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.rest.RESTCatalogProperties.SnapshotMode;
import org.apache.iceberg.rest.auth.AuthManager;
import org.apache.iceberg.rest.auth.AuthManagers;
import org.apache.iceberg.rest.auth.AuthSession;
Expand Down Expand Up @@ -104,12 +105,12 @@ public class RESTSessionCatalog extends BaseViewSessionCatalog
implements Configurable<Object>, Closeable {
private static final Logger LOG = LoggerFactory.getLogger(RESTSessionCatalog.class);
private static final String DEFAULT_FILE_IO_IMPL = "org.apache.iceberg.io.ResolvingFileIO";
private static final String REST_METRICS_REPORTING_ENABLED = "rest-metrics-reporting-enabled";
private static final String REST_SNAPSHOT_LOADING_MODE = "snapshot-loading-mode";
// for backwards compatibility with older REST servers where it can be assumed that a particular
// server supports view endpoints but doesn't send the "endpoints" field in the ConfigResponse
static final String VIEW_ENDPOINTS_SUPPORTED = "view-endpoints-supported";
public static final String REST_PAGE_SIZE = "rest-page-size";

/**
* @deprecated will be removed in 2.0.0. Use {@link
* org.apache.iceberg.rest.RESTCatalogProperties#PAGE_SIZE} instead.
*/
@Deprecated public static final String REST_PAGE_SIZE = "rest-page-size";

// these default endpoints must not be updated in order to maintain backwards compatibility with
// legacy servers
Expand Down Expand Up @@ -159,15 +160,6 @@ public class RESTSessionCatalog extends BaseViewSessionCatalog
private CloseableGroup closeables = null;
private Set<Endpoint> endpoints;

enum SnapshotMode {
ALL,
REFS;

Map<String, String> params() {
return ImmutableMap.of("snapshots", this.name().toLowerCase(Locale.US));
}
}

public RESTSessionCatalog() {
this(
config ->
Expand Down Expand Up @@ -212,7 +204,10 @@ public void initialize(String name, Map<String, String> unresolved) {

if (config.endpoints().isEmpty()) {
this.endpoints =
PropertyUtil.propertyAsBoolean(mergedProps, VIEW_ENDPOINTS_SUPPORTED, false)
PropertyUtil.propertyAsBoolean(
mergedProps,
RESTCatalogProperties.VIEW_ENDPOINTS_SUPPORTED,
RESTCatalogProperties.VIEW_ENDPOINTS_SUPPORTED_DEFAULT)
? ImmutableSet.<Endpoint>builder()
.addAll(DEFAULT_ENDPOINTS)
.addAll(VIEW_ENDPOINTS)
Expand All @@ -230,10 +225,13 @@ public void initialize(String name, Map<String, String> unresolved) {
this.catalogAuth = authManager.catalogSession(client, mergedProps);
this.closeables.addCloseable(this.catalogAuth);

this.pageSize = PropertyUtil.propertyAsNullableInt(mergedProps, REST_PAGE_SIZE);
this.pageSize =
PropertyUtil.propertyAsNullableInt(mergedProps, RESTCatalogProperties.PAGE_SIZE);
if (pageSize != null) {
Preconditions.checkArgument(
pageSize > 0, "Invalid value for %s, must be a positive integer", REST_PAGE_SIZE);
pageSize > 0,
"Invalid value for %s, must be a positive integer",
RESTCatalogProperties.PAGE_SIZE);
}

this.io = newFileIO(SessionContext.createEmpty(), mergedProps);
Expand All @@ -246,13 +244,18 @@ public void initialize(String name, Map<String, String> unresolved) {
this.snapshotMode =
SnapshotMode.valueOf(
PropertyUtil.propertyAsString(
mergedProps, REST_SNAPSHOT_LOADING_MODE, SnapshotMode.ALL.name())
mergedProps,
RESTCatalogProperties.SNAPSHOT_LOADING_MODE,
RESTCatalogProperties.SNAPSHOT_LOADING_MODE_DEFAULT)
.toUpperCase(Locale.US));

this.reporter = CatalogUtil.loadMetricsReporter(mergedProps);

this.reportingViaRestEnabled =
PropertyUtil.propertyAsBoolean(mergedProps, REST_METRICS_REPORTING_ENABLED, true);
PropertyUtil.propertyAsBoolean(
mergedProps,
RESTCatalogProperties.METRICS_REPORTING_ENABLED,
RESTCatalogProperties.METRICS_REPORTING_ENABLED_DEFAULT);
super.initialize(name, mergedProps);
}

Expand Down Expand Up @@ -366,6 +369,10 @@ public boolean tableExists(SessionContext context, TableIdentifier identifier) {
}
}

private static Map<String, String> snapshotModeToParam(SnapshotMode mode) {
return ImmutableMap.of("snapshots", mode.name().toLowerCase(Locale.US));
}

private LoadTableResponse loadInternal(
SessionContext context, TableIdentifier identifier, SnapshotMode mode) {
Endpoint.check(endpoints, Endpoint.V1_LOAD_TABLE);
Expand All @@ -374,7 +381,7 @@ private LoadTableResponse loadInternal(
.withAuthSession(contextualSession)
.get(
paths.table(identifier),
mode.params(),
snapshotModeToParam(mode),
LoadTableResponse.class,
Map.of(),
ErrorHandlers.tableErrorHandler());
Expand Down
30 changes: 11 additions & 19 deletions core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.rest.HTTPRequest.HTTPMethod;
import org.apache.iceberg.rest.RESTSessionCatalog.SnapshotMode;
import org.apache.iceberg.rest.RESTCatalogProperties.SnapshotMode;
import org.apache.iceberg.rest.auth.AuthSessionUtil;
import org.apache.iceberg.rest.auth.OAuth2Properties;
import org.apache.iceberg.rest.auth.OAuth2Util;
Expand Down Expand Up @@ -887,13 +887,6 @@ public void testTableCredential(String oauth2ServerUri) {
oauth2ServerUri);
}

@Test
public void testSnapshotParams() {
assertThat(SnapshotMode.ALL.params()).isEqualTo(ImmutableMap.of("snapshots", "all"));

assertThat(SnapshotMode.REFS.params()).isEqualTo(ImmutableMap.of("snapshots", "refs"));
}

@Test
public void testTableSnapshotLoading() {
RESTCatalogAdapter adapter = Mockito.spy(new RESTCatalogAdapter(backendCatalog));
Expand All @@ -908,8 +901,8 @@ public void testTableSnapshotLoading() {
CatalogProperties.FILE_IO_IMPL,
"org.apache.iceberg.inmemory.InMemoryFileIO",
// default loading to refs only
"snapshot-loading-mode",
"refs"));
RESTCatalogProperties.SNAPSHOT_LOADING_MODE,
SnapshotMode.REFS.name()));

if (requiresNamespaceCreate()) {
catalog.createNamespace(TABLE.namespace());
Expand Down Expand Up @@ -1004,8 +997,8 @@ public void testTableSnapshotLoadingWithDivergedBranches(String formatVersion) {
"ignored",
CatalogProperties.FILE_IO_IMPL,
"org.apache.iceberg.inmemory.InMemoryFileIO",
"snapshot-loading-mode",
"refs"));
RESTCatalogProperties.SNAPSHOT_LOADING_MODE,
SnapshotMode.REFS.name()));

if (requiresNamespaceCreate()) {
catalog.createNamespace(TABLE.namespace());
Expand Down Expand Up @@ -1137,8 +1130,8 @@ public void lazySnapshotLoadingWithDivergedHistory() {
"ignored",
CatalogProperties.FILE_IO_IMPL,
"org.apache.iceberg.inmemory.InMemoryFileIO",
"snapshot-loading-mode",
"refs"));
RESTCatalogProperties.SNAPSHOT_LOADING_MODE,
SnapshotMode.REFS.name()));

if (requiresNamespaceCreate()) {
catalog.createNamespace(TABLE.namespace());
Expand Down Expand Up @@ -2307,13 +2300,12 @@ public void testInvalidPageSize() {
new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter);
assertThatThrownBy(
() ->
catalog.initialize(
"test", ImmutableMap.of(RESTSessionCatalog.REST_PAGE_SIZE, "-1")))
catalog.initialize("test", ImmutableMap.of(RESTCatalogProperties.PAGE_SIZE, "-1")))
.isInstanceOf(IllegalArgumentException.class)
.hasMessage(
String.format(
"Invalid value for %s, must be a positive integer",
RESTSessionCatalog.REST_PAGE_SIZE));
RESTCatalogProperties.PAGE_SIZE));
}

@ParameterizedTest
Expand All @@ -2322,7 +2314,7 @@ public void testPaginationForListNamespaces(int numberOfItems) {
RESTCatalogAdapter adapter = Mockito.spy(new RESTCatalogAdapter(backendCatalog));
RESTCatalog catalog =
new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter);
catalog.initialize("test", ImmutableMap.of(RESTSessionCatalog.REST_PAGE_SIZE, "10"));
catalog.initialize("test", ImmutableMap.of(RESTCatalogProperties.PAGE_SIZE, "10"));
String namespaceName = "newdb";

// create several namespaces for listing and verify
Expand Down Expand Up @@ -2381,7 +2373,7 @@ public void testPaginationForListTables(int numberOfItems) {
RESTCatalogAdapter adapter = Mockito.spy(new RESTCatalogAdapter(backendCatalog));
RESTCatalog catalog =
new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter);
catalog.initialize("test", ImmutableMap.of(RESTSessionCatalog.REST_PAGE_SIZE, "10"));
catalog.initialize("test", ImmutableMap.of(RESTCatalogProperties.PAGE_SIZE, "10"));
String namespaceName = "newdb";
String tableName = "newtable";
Namespace namespace = Namespace.of(namespaceName);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -168,7 +168,7 @@ public void testPaginationForListViews(int numberOfItems) {
RESTCatalogAdapter adapter = Mockito.spy(new RESTCatalogAdapter(backendCatalog));
RESTCatalog catalog =
new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter);
catalog.initialize("test", ImmutableMap.of(RESTSessionCatalog.REST_PAGE_SIZE, "10"));
catalog.initialize("test", ImmutableMap.of(RESTCatalogProperties.PAGE_SIZE, "10"));

String namespaceName = "newdb";
String viewName = "newview";
Expand Down Expand Up @@ -305,7 +305,7 @@ public void viewExistsFallbackToGETRequestWithLegacyServer() {
// client relies on the default endpoints
verifyViewExistsFallbackToGETRequest(
ConfigResponse.builder().build(),
ImmutableMap.of(RESTSessionCatalog.VIEW_ENDPOINTS_SUPPORTED, "true"));
ImmutableMap.of(RESTCatalogProperties.VIEW_ENDPOINTS_SUPPORTED, "true"));
}

@Override
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ public <T extends RESTResponse> T handleRequest(
"credential",
"catalog:12345",
// assume that the server supports view endpoints
RESTSessionCatalog.VIEW_ENDPOINTS_SUPPORTED,
RESTCatalogProperties.VIEW_ENDPOINTS_SUPPORTED,
"true",
CatalogProperties.VIEW_DEFAULT_PREFIX + "key1",
"catalog-default-key1",
Expand Down