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
Expand Up @@ -369,7 +369,8 @@ public Table getTable(Identifier identifier) throws TableNotExistException {
SnapshotCommit.Factory commitFactory =
new RenamingSnapshotCommit.Factory(
lockFactory().orElse(null), lockContext().orElse(null));
return CatalogUtils.loadTable(this, identifier, this::loadTableMetadata, commitFactory);
return CatalogUtils.loadTable(
this, identifier, fileIO(), this::loadTableMetadata, commitFactory);
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -171,6 +171,7 @@ public static List<Partition> listPartitionsFromFileSystem(Table table) {
public static Table loadTable(
Catalog catalog,
Identifier identifier,
FileIO fileIO,
TableMetadata.Loader metadataLoader,
SnapshotCommit.Factory commitFactory)
throws Catalog.TableNotExistException {
Expand All @@ -189,8 +190,7 @@ public static Table loadTable(
new CatalogEnvironment(
identifier, metadata.uuid(), catalog.catalogLoader(), commitFactory);
Path path = new Path(schema.options().get(PATH.key()));
FileStoreTable table =
FileStoreTableFactory.create(catalog.fileIO(), path, schema, catalogEnv);
FileStoreTable table = FileStoreTableFactory.create(fileIO, path, schema, catalogEnv);

if (options.type() == TableType.OBJECT_TABLE) {
table = toObjectTable(catalog, table);
Expand Down
40 changes: 33 additions & 7 deletions paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@ public class RESTCatalog implements Catalog {
private final ResourcePaths resourcePaths;
private final AuthSession catalogAuth;
private final Options options;
private final boolean fileIORefreshCredentialEnable;
private final FileIO fileIO;

private volatile ScheduledExecutorService refreshExecutor = null;
Expand All @@ -130,13 +131,19 @@ public RESTCatalog(CatalogContext context) {
.merge(context.options().toMap()));
this.resourcePaths = ResourcePaths.forCatalogProperties(options);

this.fileIORefreshCredentialEnable =
options.get(RESTCatalogOptions.FILE_IO_REFRESH_CREDENTIAL_ENABLE);
try {
String warehouseStr = options.get(CatalogOptions.WAREHOUSE);
this.fileIO =
FileIO.get(
new Path(warehouseStr),
CatalogContext.create(
options, context.preferIO(), context.fallbackIO()));
if (fileIORefreshCredentialEnable) {
this.fileIO = null;
} else {
String warehouseStr = options.get(CatalogOptions.WAREHOUSE);
this.fileIO =
FileIO.get(
new Path(warehouseStr),
CatalogContext.create(
options, context.preferIO(), context.fallbackIO()));
}
} catch (IOException e) {
LOG.warn("Can not get FileIO from options.");
throw new RuntimeException(e);
Expand All @@ -149,6 +156,8 @@ protected RESTCatalog(Options options, FileIO fileIO) {
this.options = options;
this.resourcePaths = ResourcePaths.forCatalogProperties(options);
this.fileIO = fileIO;
this.fileIORefreshCredentialEnable =
options.get(RESTCatalogOptions.FILE_IO_REFRESH_CREDENTIAL_ENABLE);
}

@Override
Expand All @@ -168,12 +177,20 @@ public RESTCatalogLoader catalogLoader() {

@Override
public FileIO fileIO() {
if (fileIORefreshCredentialEnable) {
throw new UnsupportedOperationException();
}
return fileIO;
}

@Override
public FileIO fileIO(Path path) {
return fileIO;
try {
return FileIO.get(path, CatalogContext.create(options));
} catch (IOException e) {
LOG.warn("Can not get FileIO from options.");
throw new RuntimeException(e);
}
}

@Override
Expand Down Expand Up @@ -289,6 +306,7 @@ public Table getTable(Identifier identifier) throws TableNotExistException {
return CatalogUtils.loadTable(
this,
identifier,
this.fileIO(identifier),
this::loadTableMetadata,
new RESTSnapshotCommitFactory(catalogLoader()));
}
Expand Down Expand Up @@ -645,4 +663,12 @@ private ScheduledExecutorService tokenRefreshExecutor() {

return refreshExecutor;
}

private FileIO fileIO(Identifier identifier) {
if (fileIORefreshCredentialEnable) {
return new RefreshCredentialFileIO(
resourcePaths, catalogAuth, options, client, identifier);
}
return fileIO;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -77,4 +77,10 @@ public class RESTCatalogOptions {
.stringType()
.noDefaultValue()
.withDescription("REST Catalog auth token provider path.");

public static final ConfigOption<Boolean> FILE_IO_REFRESH_CREDENTIAL_ENABLE =
ConfigOptions.key("file-io-refresh-credential.enabled")
.booleanType()
.defaultValue(false)
.withDescription("Whether to support file io refresh credential.");
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,148 @@
/*
* 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.paimon.rest;

import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.PositionOutputStream;
import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.rest.auth.AuthSession;
import org.apache.paimon.rest.responses.GetTableCredentialsResponse;

import java.io.IOException;
import java.util.Map;

/** A {@link FileIO} to support refresh credential. */
public class RefreshCredentialFileIO implements FileIO {
Copy link
Contributor

Choose a reason for hiding this comment

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

Add ser id


private static final long serialVersionUID = 1L;

private final ResourcePaths resourcePaths;
private final AuthSession catalogAuth;
protected Options options;
private final Identifier identifier;
private Long expireAtMillis;
private Map<String, String> credential;
private final transient RESTClient client;
private transient volatile FileIO lazyFileIO;

public RefreshCredentialFileIO(
ResourcePaths resourcePaths,
AuthSession catalogAuth,
Options options,
RESTClient client,
Identifier identifier) {
this.resourcePaths = resourcePaths;
this.catalogAuth = catalogAuth;
this.options = options;
this.identifier = identifier;
this.client = client;
}

@Override
public void configure(CatalogContext context) {
this.options = context.options();
}

@Override
public SeekableInputStream newInputStream(Path path) throws IOException {
return fileIO().newInputStream(path);
}

@Override
public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws IOException {
return fileIO().newOutputStream(path, overwrite);
}

@Override
public FileStatus getFileStatus(Path path) throws IOException {
return fileIO().getFileStatus(path);
}

@Override
public FileStatus[] listStatus(Path path) throws IOException {
return fileIO().listStatus(path);
}

@Override
public boolean exists(Path path) throws IOException {
return fileIO().exists(path);
}

@Override
public boolean delete(Path path, boolean recursive) throws IOException {
return fileIO().delete(path, recursive);
}

@Override
public boolean mkdirs(Path path) throws IOException {
return fileIO().mkdirs(path);
}

@Override
public boolean rename(Path src, Path dst) throws IOException {
return fileIO().rename(src, dst);
}

@Override
public boolean isObjectStore() {
try {
return fileIO().isObjectStore();
} catch (IOException e) {
throw new RuntimeException(e);
}
}

private FileIO fileIO() throws IOException {
if (lazyFileIO == null || shouldRefresh()) {
synchronized (this) {
if (lazyFileIO == null || shouldRefresh()) {
GetTableCredentialsResponse response = getCredential();
expireAtMillis = response.getExpiresAtMillis();
credential = response.getCredential();
Map<String, String> conf = RESTUtil.merge(options.toMap(), credential);
Options updateCredentialOption = new Options(conf);
lazyFileIO =
FileIO.get(
new Path(updateCredentialOption.get(CatalogOptions.WAREHOUSE)),
CatalogContext.create(updateCredentialOption));
}
}
}
return lazyFileIO;
}

// todo: handle exception
private GetTableCredentialsResponse getCredential() {
return client.get(
resourcePaths.tableCredentials(
identifier.getDatabaseName(), identifier.getObjectName()),
GetTableCredentialsResponse.class,
catalogAuth.getHeaders());
}

private boolean shouldRefresh() {
return expireAtMillis != null && expireAtMillis > System.currentTimeMillis();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,10 @@ public String commitTable(String databaseName) {
return SLASH.join(V1, prefix, DATABASES, databaseName, TABLES, "commit");
}

public String tableCredentials(String databaseName, String tableName) {
return SLASH.join(V1, prefix, DATABASES, databaseName, TABLES, tableName, "credentials");
}

public String partitions(String databaseName, String tableName) {
return SLASH.join(V1, prefix, DATABASES, databaseName, TABLES, tableName, "partitions");
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,60 @@
/*
* 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.paimon.rest.responses;

import org.apache.paimon.rest.RESTResponse;

import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;

import java.util.Map;

/** Response for table credentials. */
@JsonIgnoreProperties(ignoreUnknown = true)
public class GetTableCredentialsResponse implements RESTResponse {

private static final String FIELD_CREDENTIAL = "credential";
private static final String FIELD_EXPIREAT_MILLIS = "expiresAtMillis";

@JsonProperty(FIELD_CREDENTIAL)
private final Map<String, String> credential;

@JsonProperty(FIELD_EXPIREAT_MILLIS)
private long expiresAtMillis;

@JsonCreator
public GetTableCredentialsResponse(
@JsonProperty(FIELD_EXPIREAT_MILLIS) long expiresAtMillis,
@JsonProperty(FIELD_CREDENTIAL) Map<String, String> credential) {
this.expiresAtMillis = expiresAtMillis;
this.credential = credential;
}

@JsonGetter(FIELD_CREDENTIAL)
public Map<String, String> getCredential() {
return credential;
}

@JsonGetter(FIELD_EXPIREAT_MILLIS)
public long getExpiresAtMillis() {
return expiresAtMillis;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import org.apache.paimon.rest.responses.AlterDatabaseResponse;
import org.apache.paimon.rest.responses.CreateDatabaseResponse;
import org.apache.paimon.rest.responses.GetDatabaseResponse;
import org.apache.paimon.rest.responses.GetTableCredentialsResponse;
import org.apache.paimon.rest.responses.GetTableResponse;
import org.apache.paimon.rest.responses.GetViewResponse;
import org.apache.paimon.rest.responses.ListDatabasesResponse;
Expand All @@ -48,6 +49,7 @@
import org.apache.paimon.view.ViewSchema;

import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap;
import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;

import java.util.ArrayList;
Expand Down Expand Up @@ -248,6 +250,11 @@ public static ListViewsResponse listViewsResponse() {
return new ListViewsResponse(ImmutableList.of("view"));
}

public static GetTableCredentialsResponse getTableCredentialsResponse() {
return new GetTableCredentialsResponse(
System.currentTimeMillis(), ImmutableMap.of("key", "value"));
}

private static ViewSchema viewSchema() {
List<DataField> fields =
Arrays.asList(
Expand Down
Loading
Loading