You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2024/04/24 16:07:26 UTC
Change in asterixdb[master]: Add GCS Cloud Client
From Savyasach Reddy <es...@gmail.com>:
Savyasach Reddy has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/18253 )
Change subject: Add GCS Cloud Client
......................................................................
Add GCS Cloud Client
Change-Id: I0b9cad99de2d32d1e672a3d396897807687685e8
---
M asterixdb/asterix-app/pom.xml
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/IParallelCacher.java
A asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/GCSExternalFileWriterFactory.java
A asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSCloudClient.java
M asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/accessor/ReplaceableCloudAccessor.java
A asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/GCSExternalFileWriter.java
A asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSParallelDownloader.java
A asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSClientConfig.java
M asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/LocalCloudUtil.java
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudClient.java
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/CloudClientProvider.java
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/IParallelDownloader.java
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/ParallelCacher.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/bulk/IBulkOperationCallBack.java
A asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSBufferedWriter.java
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudBufferedWriter.java
18 files changed, 853 insertions(+), 11 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/53/18253/1
diff --git a/asterixdb/asterix-app/pom.xml b/asterixdb/asterix-app/pom.xml
index dbca45f..8017a1b 100644
--- a/asterixdb/asterix-app/pom.xml
+++ b/asterixdb/asterix-app/pom.xml
@@ -1009,5 +1009,17 @@
<artifactId>converter</artifactId>
<version>0.2.15</version>
</dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
+ <version>1.19.7</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <version>1.19.7</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/LocalCloudUtil.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/LocalCloudUtil.java
index 9e3fdc4..a5b71a8 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/LocalCloudUtil.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/LocalCloudUtil.java
@@ -21,11 +21,25 @@
import static org.apache.hyracks.util.file.FileUtil.joinPath;
import java.io.File;
+import java.io.IOException;
import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.util.ArrayList;
+import java.util.List;
import org.apache.commons.io.FileUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import org.testcontainers.containers.GenericContainer;
+
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Bucket;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageClass;
+import com.google.cloud.storage.StorageOptions;
import io.findify.s3mock.S3Mock;
import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
@@ -44,6 +58,7 @@
public static final String MOCK_SERVER_REGION = "us-west-2";
private static final String MOCK_FILE_BACKEND = joinPath("target", "s3mock");
private static S3Mock s3MockServer;
+ private static GenericContainer<?> fakeGcs;
private LocalCloudUtil() {
throw new AssertionError("Do not instantiate");
@@ -53,7 +68,8 @@
String cleanStartString = System.getProperty("cleanup.start", "true");
boolean cleanStart = Boolean.parseBoolean(cleanStartString);
// Change to 'true' if you want to delete "s3mock" folder on start
- startS3CloudEnvironment(cleanStart);
+ // startS3CloudEnvironment(cleanStart);
+ startGCSCloudEnvironment();
}
public static S3Mock startS3CloudEnvironment(boolean cleanStart) {
@@ -83,6 +99,52 @@
return s3MockServer;
}
+ public static void startGCSCloudEnvironment() {
+ // Starting GCS mock server to be used instead of real GCS server
+ LOGGER.info("Starting GCS mock server");
+
+ fakeGcs = new GenericContainer<>("fsouza/fake-gcs-server:1.48").withExposedPorts(8001, 8002)
+ .withCreateContainerCmdModifier(cmd -> cmd.withEntrypoint("/bin/fake-gcs-server", "-scheme", "both",
+ "-port-http", "8001", "-port", "8002"));
+ List<String> portBindings = new ArrayList<>();
+ portBindings.add("8001:8001"); // hostPort:containerPort
+ portBindings.add("8002:8002");
+ fakeGcs.setPortBindings(portBindings);
+
+ fakeGcs.start();
+ String fakeGcsExternalUrl = MOCK_SERVER_HOSTNAME;
+ String modifyExternalUrlRequestUri = fakeGcsExternalUrl + "/_internal/config";
+ String updateExternalUrlJson = "{" + "\"externalUrl\": \"" + fakeGcsExternalUrl + "\"" + "}";
+
+ HttpRequest req = HttpRequest.newBuilder().uri(URI.create(modifyExternalUrlRequestUri))
+ .header("Content-Type", "application/json")
+ .PUT(HttpRequest.BodyPublishers.ofString(updateExternalUrlJson)).build();
+ HttpResponse<Void> response = null;
+ try {
+ response = HttpClient.newBuilder().build().send(req, HttpResponse.BodyHandlers.discarding());
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+
+ if (response.statusCode() != 200) {
+ throw new RuntimeException("error updating fake-gcs-server with external url, response status code "
+ + response.statusCode() + " != 200");
+ }
+
+ Storage storage = StorageOptions.newBuilder().setHost(MOCK_SERVER_HOSTNAME)
+ .setCredentials(NoCredentials.getInstance()).build().getService();
+
+ Bucket bucket = storage.create(BucketInfo.newBuilder(CLOUD_STORAGE_BUCKET)
+ .setStorageClass(StorageClass.STANDARD).setLocation("US").build());
+ LOGGER.info("Created bucket {} for cloud storage", CLOUD_STORAGE_BUCKET);
+
+ try {
+ storage.close();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
private static void shutdownSilently() {
if (s3MockServer != null) {
try {
@@ -91,5 +153,8 @@
// do nothing
}
}
+ if (fakeGcs != null) {
+ fakeGcs.stop();
+ }
}
}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/bulk/IBulkOperationCallBack.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/bulk/IBulkOperationCallBack.java
index f0b336d..a4223eb 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/bulk/IBulkOperationCallBack.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/bulk/IBulkOperationCallBack.java
@@ -20,8 +20,9 @@
import java.util.List;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
public interface IBulkOperationCallBack {
- void call(List<FileReference> fileReferences);
+ void call(List<FileReference> fileReferences) throws HyracksDataException;
}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/CloudClientProvider.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/CloudClientProvider.java
index 21450c4..cc511c7 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/CloudClientProvider.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/CloudClientProvider.java
@@ -20,7 +20,10 @@
import org.apache.asterix.cloud.clients.aws.s3.S3ClientConfig;
import org.apache.asterix.cloud.clients.aws.s3.S3CloudClient;
+import org.apache.asterix.cloud.clients.google.gcs.GCSClientConfig;
+import org.apache.asterix.cloud.clients.google.gcs.GCSCloudClient;
import org.apache.asterix.common.config.CloudProperties;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
public class CloudClientProvider {
@@ -28,11 +31,14 @@
throw new AssertionError("do not instantiate");
}
- public static ICloudClient getClient(CloudProperties cloudProperties) {
+ public static ICloudClient getClient(CloudProperties cloudProperties) throws HyracksDataException {
String storageScheme = cloudProperties.getStorageScheme();
if ("s3".equalsIgnoreCase(storageScheme)) {
S3ClientConfig config = S3ClientConfig.of(cloudProperties);
return new S3CloudClient(config);
+ } else if ("gcs".equalsIgnoreCase(storageScheme)) {
+ GCSClientConfig config = GCSClientConfig.of(cloudProperties);
+ return new GCSCloudClient(config);
}
throw new IllegalStateException("unsupported cloud storage scheme: " + storageScheme);
}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudBufferedWriter.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudBufferedWriter.java
index 16e79f9..6b51964 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudBufferedWriter.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudBufferedWriter.java
@@ -31,7 +31,7 @@
* @param length length
* @return amount uploaded
*/
- int upload(InputStream stream, int length);
+ int upload(InputStream stream, int length) throws HyracksDataException;
/**
* Checks whether the writer has not written anything
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudClient.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudClient.java
index 7941ada..2bd0802 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudClient.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/ICloudClient.java
@@ -139,7 +139,7 @@
* @param ioManager local {@link IOManager}
* @return an instance of a new parallel downloader
*/
- IParallelDownloader createParallelDownloader(String bucket, IOManager ioManager);
+ IParallelDownloader createParallelDownloader(String bucket, IOManager ioManager) throws HyracksDataException;
/**
* Produces a {@link JsonNode} that contains information about the stored objects in the cloud
@@ -153,5 +153,5 @@
/**
* Performs any necessary closing and cleaning up
*/
- void close();
+ void close() throws HyracksDataException;
}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/IParallelDownloader.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/IParallelDownloader.java
index 32d0a74..184e015 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/IParallelDownloader.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/IParallelDownloader.java
@@ -43,5 +43,5 @@
/**
* Close the downloader and release all of its resources
*/
- void close();
+ void close() throws HyracksDataException;
}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSBufferedWriter.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSBufferedWriter.java
new file mode 100644
index 0000000..f58bef6
--- /dev/null
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSBufferedWriter.java
@@ -0,0 +1,96 @@
+package org.apache.asterix.cloud.clients.google.gcs;
+
+import static org.apache.asterix.cloud.CloudResettableInputStream.MIN_BUFFER_SIZE;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+
+import org.apache.asterix.cloud.clients.ICloudBufferedWriter;
+import org.apache.asterix.cloud.clients.profiler.IRequestProfiler;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import com.google.cloud.WriteChannel;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Storage;
+
+public class GCSBufferedWriter implements ICloudBufferedWriter {
+ private static final Logger LOGGER = LogManager.getLogger();
+ private final String bucket;
+ private final String path;
+ private final IRequestProfiler profiler;
+ private final Storage gcsClient;
+ private boolean uploadStarted = false;
+ private int partNumber;
+ private WriteChannel writer = null;
+
+ public GCSBufferedWriter(String bucket, String path, Storage gcsClient, IRequestProfiler profiler) {
+ this.bucket = bucket;
+ this.path = path;
+ this.profiler = profiler;
+ this.gcsClient = gcsClient;
+ }
+
+ @Override
+ public int upload(InputStream stream, int length) throws HyracksDataException {
+ profiler.objectMultipartUpload();
+ setUploadId();
+ try {
+ ByteBuffer buffer = ByteBuffer.wrap(stream.readNBytes(length));
+ while (buffer.hasRemaining()) {
+ writer.write(buffer);
+ }
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ }
+ return partNumber++;
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return !uploadStarted;
+ }
+
+ @Override
+ public void finish() throws HyracksDataException {
+ if (!uploadStarted) {
+ throw new IllegalStateException("Cannot finish without writing any bytes");
+ }
+ profiler.objectMultipartUpload();
+ try {
+ writer.close();
+ writer = null;
+ uploadStarted = false;
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ }
+ log("FINISHED");
+ }
+
+ @Override
+ public void abort() {
+ // https://github.com/googleapis/java-storage/issues/202
+ // Cannot abort. Upload Ids and data are discarded after a week
+ writer = null;
+ LOGGER.warn("Multipart upload for {} was aborted", path);
+ }
+
+ private void setUploadId() {
+ if (!uploadStarted) {
+ uploadStarted = true;
+ partNumber = 1;
+ writer = gcsClient.writer(BlobInfo.newBuilder(BlobId.of(bucket, path)).build());
+ writer.setChunkSize(MIN_BUFFER_SIZE);
+ log("STARTED");
+ }
+ }
+
+ private void log(String op) {
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("{} multipart upload for {}", op, path);
+ }
+ }
+}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSClientConfig.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSClientConfig.java
new file mode 100644
index 0000000..57ee2c5
--- /dev/null
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSClientConfig.java
@@ -0,0 +1,77 @@
+package org.apache.asterix.cloud.clients.google.gcs;
+
+import static org.apache.asterix.external.util.google.gcs.GCSConstants.ENDPOINT_FIELD_NAME;
+
+import java.io.IOException;
+import java.util.Map;
+
+import org.apache.asterix.common.config.CloudProperties;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.auth.oauth2.OAuth2Credentials;
+import com.google.cloud.NoCredentials;
+
+public class GCSClientConfig {
+ // The maximum number of file that can be deleted (GCS restriction)
+ static final int DELETE_BATCH_SIZE = 100;
+ private final String region;
+ private final String endpoint;
+ private final String prefix;
+ private final boolean anonymousAuth;
+ private final long profilerLogInterval;
+
+ public GCSClientConfig(String region, String endpoint, String prefix, boolean anonymousAuth,
+ long profilerLogInterval) {
+ this.region = region;
+ this.endpoint = endpoint;
+ this.prefix = prefix;
+ this.anonymousAuth = anonymousAuth;
+ this.profilerLogInterval = profilerLogInterval;
+ }
+
+ public static GCSClientConfig of(CloudProperties cloudProperties) {
+ return new GCSClientConfig(cloudProperties.getStorageRegion(), cloudProperties.getStorageEndpoint(),
+ cloudProperties.getStoragePrefix(), cloudProperties.isStorageAnonymousAuth(),
+ cloudProperties.getProfilerLogInterval());
+ }
+
+ public static GCSClientConfig of(Map<String, String> configuration) {
+ String endPoint = configuration.getOrDefault(ENDPOINT_FIELD_NAME, "");
+ long profilerLogInterval = 0;
+
+ String region = "";
+ String prefix = "";
+ boolean anonymousAuth = false;
+
+ return new GCSClientConfig(region, endPoint, prefix, anonymousAuth, profilerLogInterval);
+ }
+
+ public String getRegion() {
+ return region;
+ }
+
+ public String getEndpoint() {
+ return endpoint;
+ }
+
+ public String getPrefix() {
+ return prefix;
+ }
+
+ public long getProfilerLogInterval() {
+ return profilerLogInterval;
+ }
+
+ public boolean isAnonymousAuth() {
+ return anonymousAuth;
+ }
+
+ public OAuth2Credentials createCredentialsProvider() throws HyracksDataException {
+ try {
+ return anonymousAuth ? NoCredentials.getInstance() : GoogleCredentials.getApplicationDefault();
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ }
+ }
+}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSCloudClient.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSCloudClient.java
new file mode 100644
index 0000000..9ccf380
--- /dev/null
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSCloudClient.java
@@ -0,0 +1,233 @@
+package org.apache.asterix.cloud.clients.google.gcs;
+
+import static org.apache.asterix.cloud.clients.google.gcs.GCSClientConfig.DELETE_BATCH_SIZE;
+
+import java.io.FilenameFilter;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.asterix.cloud.clients.ICloudBufferedWriter;
+import org.apache.asterix.cloud.clients.ICloudClient;
+import org.apache.asterix.cloud.clients.IParallelDownloader;
+import org.apache.asterix.cloud.clients.profiler.CountRequestProfiler;
+import org.apache.asterix.cloud.clients.profiler.IRequestProfiler;
+import org.apache.asterix.cloud.clients.profiler.NoOpRequestProfiler;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.api.util.IoUtil;
+import org.apache.hyracks.control.nc.io.IOManager;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.api.gax.paging.Page;
+import com.google.cloud.ReadChannel;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.Storage.BlobListOption;
+import com.google.cloud.storage.Storage.CopyRequest;
+import com.google.cloud.storage.StorageBatch;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.StorageOptions;
+
+public class GCSCloudClient implements ICloudClient {
+
+ private final Storage gcsClient;
+ private final GCSClientConfig config;
+ private final IRequestProfiler profiler;
+
+ public GCSCloudClient(GCSClientConfig config, Storage gcsClient) {
+ this.gcsClient = gcsClient;
+ this.config = config;
+ long profilerInterval = config.getProfilerLogInterval();
+ if (profilerInterval > 0) {
+ profiler = new CountRequestProfiler(profilerInterval);
+ } else {
+ profiler = NoOpRequestProfiler.INSTANCE;
+ }
+ }
+
+ public GCSCloudClient(GCSClientConfig config) throws HyracksDataException {
+ this(config, buildClient(config));
+ }
+
+ private static Storage buildClient(GCSClientConfig config) throws HyracksDataException {
+ StorageOptions.Builder builder = StorageOptions.newBuilder().setCredentials(config.createCredentialsProvider());
+
+ if (config.getEndpoint() != null && !config.getEndpoint().isEmpty()) {
+ builder.setHost(config.getEndpoint());
+ }
+ return builder.build().getService();
+ }
+
+ @Override
+ public ICloudBufferedWriter createBufferedWriter(String bucket, String path) {
+ return new GCSBufferedWriter(bucket, path, gcsClient, profiler);
+ }
+
+ @Override
+ public Set<String> listObjects(String bucket, String path, FilenameFilter filter) {
+ profiler.objectsList();
+ Page<Blob> blobs = gcsClient.list(bucket, BlobListOption.prefix(path));
+
+ Set<String> files = new HashSet<>();
+ for (Blob blob : blobs.iterateAll()) {
+ if (filter.accept(null, IoUtil.getFileNameFromPath(blob.getName()))) {
+ files.add(blob.getName());
+ }
+ }
+ return files;
+ }
+
+ @Override
+ public int read(String bucket, String path, long offset, ByteBuffer buffer) throws HyracksDataException {
+ profiler.objectGet();
+ BlobId blobId = BlobId.of(bucket, path);
+ long readTo = offset + buffer.remaining();
+ int totalRead = 0;
+ int read = 0;
+ try (ReadChannel from = gcsClient.reader(blobId)) {
+ from.limit(readTo);
+ while (buffer.remaining() > 0) {
+ from.seek(offset + totalRead);
+ read = from.read(buffer);
+ totalRead += read;
+ }
+ } catch (IOException | StorageException ex) {
+ throw HyracksDataException.create(ex);
+ }
+
+ if (buffer.remaining() != 0) {
+ throw new IllegalStateException("Expected buffer remaining = 0, found: " + buffer.remaining());
+ }
+ return totalRead;
+ }
+
+ @Override
+ public byte[] readAllBytes(String bucket, String path) {
+ profiler.objectGet();
+ BlobId blobId = BlobId.of(bucket, path);
+ try {
+ return gcsClient.readAllBytes(blobId);
+ } catch (StorageException e) {
+ return null;
+ }
+ }
+
+ @Override
+ public InputStream getObjectStream(String bucket, String path) {
+ profiler.objectGet();
+ try (ReadChannel reader = gcsClient.reader(bucket, path)) {
+ return Channels.newInputStream(reader);
+ } catch (StorageException ex) {
+ throw new IllegalStateException(ex);
+ }
+ }
+
+ @Override
+ public void write(String bucket, String path, byte[] data) {
+ profiler.objectWrite();
+ BlobInfo blobInfo = BlobInfo.newBuilder(bucket, path).build();
+ gcsClient.create(blobInfo, data);
+ }
+
+ @Override
+ public void copy(String bucket, String srcPath, FileReference destPath) {
+ Page<Blob> blobs = gcsClient.list(bucket, BlobListOption.prefix(srcPath));
+ profiler.objectsList();
+ for (Blob blob : blobs.iterateAll()) {
+ profiler.objectCopy();
+ BlobId source = blob.getBlobId();
+ String targetName = destPath.getChildPath(IoUtil.getFileNameFromPath(source.getName()));
+ BlobId target = BlobId.of(bucket, targetName);
+ CopyRequest copyReq = CopyRequest.newBuilder().setSource(source).setTarget(target).build();
+ gcsClient.copy(copyReq);
+ }
+ }
+
+ @Override
+ public void deleteObjects(String bucket, Collection<String> paths) {
+ if (paths.isEmpty()) {
+ return;
+ }
+
+ StorageBatch batchRequest;
+ Iterator<String> pathIter = paths.iterator();
+ while (pathIter.hasNext()) {
+ batchRequest = gcsClient.batch();
+ for (int i = 0; pathIter.hasNext() && i < DELETE_BATCH_SIZE; i++) {
+ BlobId blobId = BlobId.of(bucket, pathIter.next());
+ batchRequest.delete(blobId);
+ }
+
+ batchRequest.submit();
+ profiler.objectDelete();
+ }
+ }
+
+ @Override
+ public long getObjectSize(String bucket, String path) {
+ profiler.objectGet();
+ Blob blob = gcsClient.get(bucket, path, Storage.BlobGetOption.fields(Storage.BlobField.SIZE));
+ if (blob == null) {
+ return 0;
+ }
+ return blob.getSize();
+ }
+
+ @Override
+ public boolean exists(String bucket, String path) {
+ profiler.objectGet();
+ Blob blob = gcsClient.get(bucket, path, Storage.BlobGetOption.fields(Storage.BlobField.values()));
+ return blob != null;
+ }
+
+ @Override
+ public boolean isEmptyPrefix(String bucket, String path) {
+ profiler.objectsList();
+ Page<Blob> blobs = gcsClient.list(bucket, BlobListOption.prefix(path));
+ return !blobs.hasNextPage();
+ }
+
+ @Override
+ public IParallelDownloader createParallelDownloader(String bucket, IOManager ioManager)
+ throws HyracksDataException {
+ return new GCSParallelDownloader(bucket, ioManager, config, profiler);
+ }
+
+ @Override
+ public JsonNode listAsJson(ObjectMapper objectMapper, String bucket) {
+ Page<Blob> blobs = gcsClient.list(bucket, BlobListOption.fields(Storage.BlobField.SIZE));
+ ArrayNode objectsInfo = objectMapper.createArrayNode();
+
+ List<Blob> objects = new ArrayList<>();
+ blobs.iterateAll().forEach(objects::add);
+ objects.sort((x, y) -> String.CASE_INSENSITIVE_ORDER.compare(x.getName(), y.getName()));
+ for (Blob blob : objects) {
+ ObjectNode objectInfo = objectsInfo.addObject();
+ objectInfo.put("path", blob.getName());
+ objectInfo.put("size", blob.getSize());
+ }
+ return objectsInfo;
+ }
+
+ @Override
+ public void close() throws HyracksDataException {
+ try {
+ gcsClient.close();
+ } catch (Exception ex) {
+ throw HyracksDataException.create(ex);
+ }
+ }
+}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSParallelDownloader.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSParallelDownloader.java
new file mode 100644
index 0000000..e8ab562
--- /dev/null
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSParallelDownloader.java
@@ -0,0 +1,124 @@
+package org.apache.asterix.cloud.clients.google.gcs;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.asterix.cloud.clients.IParallelDownloader;
+import org.apache.asterix.cloud.clients.profiler.IRequestProfiler;
+import org.apache.commons.io.FileUtils;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.control.nc.io.IOManager;
+
+import com.google.api.gax.paging.Page;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import com.google.cloud.storage.transfermanager.DownloadJob;
+import com.google.cloud.storage.transfermanager.DownloadResult;
+import com.google.cloud.storage.transfermanager.ParallelDownloadConfig;
+import com.google.cloud.storage.transfermanager.TransferManager;
+import com.google.cloud.storage.transfermanager.TransferManagerConfig;
+import com.google.cloud.storage.transfermanager.TransferStatus;
+
+public class GCSParallelDownloader implements IParallelDownloader {
+
+ // private static final Logger LOGGER = LogManager.getLogger();
+ private final String bucket;
+ private final IOManager ioManager;
+ private final Storage gcsClient;
+ private final TransferManager transferManager;
+ private final IRequestProfiler profiler;
+
+ public GCSParallelDownloader(String bucket, IOManager ioManager, GCSClientConfig config, IRequestProfiler profiler)
+ throws HyracksDataException {
+ this.bucket = bucket;
+ this.ioManager = ioManager;
+ this.profiler = profiler;
+ StorageOptions.Builder builder = StorageOptions.newBuilder();
+ if (config.getEndpoint() != null && !config.getEndpoint().isEmpty()) {
+ builder.setHost(config.getEndpoint());
+ }
+ builder.setCredentials(config.createCredentialsProvider());
+ this.gcsClient = builder.build().getService();
+ this.transferManager =
+ TransferManagerConfig.newBuilder().setStorageOptions(builder.build()).build().getService();
+ }
+
+ @Override
+ public void downloadFiles(Collection<FileReference> toDownload) throws HyracksDataException {
+ ParallelDownloadConfig.Builder config = ParallelDownloadConfig.newBuilder().setBucketName(bucket);
+ Map<Path, List<BlobInfo>> pathListMap = new HashMap<>();
+ try {
+ for (FileReference fileReference : toDownload) {
+ profiler.objectGet();
+ FileUtils.createParentDirectories(fileReference.getFile());
+ addToMap(pathListMap, fileReference.getDeviceHandle().getMount().toPath(),
+ BlobInfo.newBuilder(BlobId.of(bucket, fileReference.getRelativePath())).build());
+ }
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ }
+ List<DownloadJob> downloadJobs = new ArrayList<>(pathListMap.size());
+ for (Map.Entry<Path, List<BlobInfo>> entry : pathListMap.entrySet()) {
+ downloadJobs.add(transferManager.downloadBlobs(entry.getValue(),
+ config.setDownloadDirectory(entry.getKey()).build()));
+ }
+ downloadJobs.forEach(DownloadJob::getDownloadResults);
+ }
+
+ @Override
+ public Collection<FileReference> downloadDirectories(Collection<FileReference> toDownload)
+ throws HyracksDataException {
+ Set<FileReference> failedFiles = new HashSet<>();
+ ParallelDownloadConfig.Builder config = ParallelDownloadConfig.newBuilder().setBucketName(bucket);
+
+ Map<Path, List<BlobInfo>> pathListMap = new HashMap<>();
+ for (FileReference fileReference : toDownload) {
+ profiler.objectMultipartDownload();
+ Page<Blob> blobs = gcsClient.list(bucket, Storage.BlobListOption.prefix(fileReference.getRelativePath()));
+ for (Blob blob : blobs.iterateAll()) {
+ addToMap(pathListMap, fileReference.getDeviceHandle().getMount().toPath(), blob.asBlobInfo());
+ }
+ }
+ List<DownloadJob> downloadJobs = new ArrayList<>(pathListMap.size());
+ for (Map.Entry<Path, List<BlobInfo>> entry : pathListMap.entrySet()) {
+ downloadJobs.add(transferManager.downloadBlobs(entry.getValue(),
+ config.setDownloadDirectory(entry.getKey()).build()));
+ }
+ List<DownloadResult> results;
+ for (DownloadJob job : downloadJobs) {
+ results = job.getDownloadResults();
+ for (DownloadResult result : results) {
+ if (result.getStatus() != TransferStatus.SUCCESS) {
+ FileReference failedFile = ioManager.resolve(result.getInput().getName());
+ failedFiles.add(failedFile);
+ }
+ }
+ }
+ return failedFiles;
+ }
+
+ @Override
+ public void close() throws HyracksDataException {
+ try {
+ transferManager.close();
+ gcsClient.close();
+ } catch (Exception e) {
+ throw HyracksDataException.create(e);
+ }
+ }
+
+ private <K, V> void addToMap(Map<K, List<V>> map, K key, V value) {
+ map.computeIfAbsent(key, k -> new ArrayList<>()).add(value);
+ }
+}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/IParallelCacher.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/IParallelCacher.java
index 6600356..e3d978d 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/IParallelCacher.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/IParallelCacher.java
@@ -73,5 +73,5 @@
/**
* Close cacher resources
*/
- void close();
+ void close() throws HyracksDataException;
}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/ParallelCacher.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/ParallelCacher.java
index 3cc481e..95330d9 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/ParallelCacher.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/ParallelCacher.java
@@ -157,7 +157,7 @@
}
@Override
- public void close() {
+ public void close() throws HyracksDataException {
downloader.close();
LOGGER.info("Parallel cacher was closed");
}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/accessor/ReplaceableCloudAccessor.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/accessor/ReplaceableCloudAccessor.java
index bfa353a..277d425 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/accessor/ReplaceableCloudAccessor.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/lazy/accessor/ReplaceableCloudAccessor.java
@@ -175,7 +175,7 @@
return partitions.contains(StoragePathUtil.getPartitionNumFromRelativePath(path));
}
- private void replace() {
+ private void replace() throws HyracksDataException {
cacher.close();
replacer.replace();
}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/GCSExternalFileWriter.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/GCSExternalFileWriter.java
new file mode 100644
index 0000000..fab4e28
--- /dev/null
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/GCSExternalFileWriter.java
@@ -0,0 +1,33 @@
+package org.apache.asterix.cloud.writer;
+
+import org.apache.asterix.cloud.clients.ICloudClient;
+import org.apache.asterix.external.util.ExternalDataConstants;
+import org.apache.asterix.runtime.writer.IExternalPrinter;
+import org.apache.hyracks.api.exceptions.IWarningCollector;
+import org.apache.hyracks.api.exceptions.SourceLocation;
+
+import com.google.cloud.BaseServiceException;
+
+final class GCSExternalFileWriter extends AbstractCloudExternalFileWriter {
+ static int MAX_LENGTH_IN_BYTES = 1024;
+
+ GCSExternalFileWriter(IExternalPrinter printer, ICloudClient cloudClient, String bucket, boolean partitionedPath,
+ IWarningCollector warningCollector, SourceLocation pathSourceLocation) {
+ super(printer, cloudClient, bucket, partitionedPath, warningCollector, pathSourceLocation);
+ }
+
+ @Override
+ String getAdapterName() {
+ return ExternalDataConstants.KEY_ADAPTER_NAME_GCS;
+ }
+
+ @Override
+ int getPathMaxLengthInBytes() {
+ return MAX_LENGTH_IN_BYTES;
+ }
+
+ @Override
+ boolean isSdkException(Exception e) {
+ return e instanceof BaseServiceException;
+ }
+}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/GCSExternalFileWriterFactory.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/GCSExternalFileWriterFactory.java
new file mode 100644
index 0000000..ca0a057
--- /dev/null
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/GCSExternalFileWriterFactory.java
@@ -0,0 +1,183 @@
+package org.apache.asterix.cloud.writer;
+
+import static org.apache.asterix.cloud.writer.AbstractCloudExternalFileWriter.isExceedingMaxLength;
+import static org.apache.hyracks.api.util.ExceptionUtils.getMessageOrToString;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Random;
+
+import org.apache.asterix.cloud.CloudResettableInputStream;
+import org.apache.asterix.cloud.WriterSingleBufferProvider;
+import org.apache.asterix.cloud.clients.ICloudBufferedWriter;
+import org.apache.asterix.cloud.clients.ICloudClient;
+import org.apache.asterix.cloud.clients.google.gcs.GCSClientConfig;
+import org.apache.asterix.cloud.clients.google.gcs.GCSCloudClient;
+import org.apache.asterix.common.exceptions.CompilationException;
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.external.util.ExternalDataConstants;
+import org.apache.asterix.external.util.google.gcs.GCSUtils;
+import org.apache.asterix.runtime.writer.ExternalFileWriterConfiguration;
+import org.apache.asterix.runtime.writer.IExternalFileWriter;
+import org.apache.asterix.runtime.writer.IExternalFileWriterFactory;
+import org.apache.asterix.runtime.writer.IExternalFileWriterFactoryProvider;
+import org.apache.asterix.runtime.writer.IExternalPrinter;
+import org.apache.asterix.runtime.writer.IExternalPrinterFactory;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.IWarningCollector;
+import org.apache.hyracks.api.exceptions.SourceLocation;
+import org.apache.hyracks.api.util.ExceptionUtils;
+import org.apache.hyracks.data.std.primitive.LongPointable;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import com.google.cloud.storage.StorageException;
+
+import software.amazon.awssdk.core.exception.SdkException;
+
+public final class GCSExternalFileWriterFactory implements IExternalFileWriterFactory {
+ private static final long serialVersionUID = 4551318140901866805L;
+ private static final Logger LOGGER = LogManager.getLogger();
+ static final char SEPARATOR = '/';
+ public static final IExternalFileWriterFactoryProvider PROVIDER = new IExternalFileWriterFactoryProvider() {
+ @Override
+ public IExternalFileWriterFactory create(ExternalFileWriterConfiguration configuration) {
+ return new GCSExternalFileWriterFactory(configuration);
+ }
+
+ @Override
+ public char getSeparator() {
+ return SEPARATOR;
+ }
+ };
+ private final Map<String, String> configuration;
+ private final SourceLocation pathSourceLocation;
+ private final String staticPath;
+ private transient GCSCloudClient cloudClient;
+
+ private GCSExternalFileWriterFactory(ExternalFileWriterConfiguration externalConfig) {
+ configuration = externalConfig.getConfiguration();
+ pathSourceLocation = externalConfig.getPathSourceLocation();
+ staticPath = externalConfig.getStaticPath();
+ cloudClient = null;
+ }
+
+ @Override
+ public IExternalFileWriter createWriter(IHyracksTaskContext context, IExternalPrinterFactory printerFactory)
+ throws HyracksDataException {
+ buildClient();
+ String bucket = configuration.get(ExternalDataConstants.CONTAINER_NAME_FIELD_NAME);
+ IExternalPrinter printer = printerFactory.createPrinter();
+ IWarningCollector warningCollector = context.getWarningCollector();
+ return new GCSExternalFileWriter(printer, cloudClient, bucket, staticPath == null, warningCollector,
+ pathSourceLocation);
+ }
+
+ private void buildClient() throws HyracksDataException {
+ try {
+ synchronized (this) {
+ if (cloudClient == null) {
+ // only a single client should be built
+ GCSClientConfig config = GCSClientConfig.of(configuration);
+ cloudClient = new GCSCloudClient(config, GCSUtils.buildClient(configuration));
+ }
+ }
+ } catch (CompilationException e) {
+ throw HyracksDataException.create(e);
+ }
+ }
+
+ @Override
+ public char getSeparator() {
+ return SEPARATOR;
+ }
+
+ @Override
+ public void validate() throws AlgebricksException {
+ GCSClientConfig config = GCSClientConfig.of(configuration);
+ ICloudClient testClient = new GCSCloudClient(config, GCSUtils.buildClient(configuration));
+ String bucket = configuration.get(ExternalDataConstants.CONTAINER_NAME_FIELD_NAME);
+
+ if (bucket == null || bucket.isEmpty()) {
+ throw new CompilationException(ErrorCode.PARAMETERS_REQUIRED,
+ ExternalDataConstants.CONTAINER_NAME_FIELD_NAME);
+ }
+
+ try {
+ doValidate(testClient, bucket);
+ } catch (IOException e) {
+ if (e.getCause() instanceof StorageException) {
+ throw CompilationException.create(ErrorCode.EXTERNAL_SOURCE_CONTAINER_NOT_FOUND, bucket);
+ } else {
+ throw CompilationException.create(ErrorCode.EXTERNAL_SINK_ERROR,
+ ExceptionUtils.getMessageOrToString(e));
+ }
+ } catch (SdkException e) {
+ throw CompilationException.create(ErrorCode.EXTERNAL_SINK_ERROR, e, getMessageOrToString(e));
+ }
+ }
+
+ private void doValidate(ICloudClient testClient, String bucket) throws IOException, AlgebricksException {
+ if (staticPath != null) {
+ if (isExceedingMaxLength(staticPath, GCSExternalFileWriter.MAX_LENGTH_IN_BYTES)) {
+ throw new CompilationException(ErrorCode.WRITE_PATH_LENGTH_EXCEEDS_MAX_LENGTH, pathSourceLocation,
+ staticPath, GCSExternalFileWriter.MAX_LENGTH_IN_BYTES,
+ ExternalDataConstants.KEY_ADAPTER_NAME_GCS);
+ }
+
+ if (!testClient.isEmptyPrefix(bucket, staticPath)) {
+ // Ensure that the static path is empty
+ throw new CompilationException(ErrorCode.DIRECTORY_IS_NOT_EMPTY, pathSourceLocation, staticPath);
+ }
+ }
+
+ // do not validate write permissions if specified by the user not to do so
+ String validateWritePermissions = configuration
+ .getOrDefault(ExternalDataConstants.KEY_VALIDATE_WRITE_PERMISSION, Boolean.TRUE.toString());
+ if (!Boolean.parseBoolean(validateWritePermissions)) {
+ return;
+ }
+
+ Random random = new Random();
+ String pathPrefix = "testFile";
+ String path = pathPrefix + random.nextInt();
+ while (testClient.exists(bucket, path)) {
+ path = pathPrefix + random.nextInt();
+ }
+
+ long writeValue = random.nextLong();
+ byte[] data = new byte[Long.BYTES];
+ LongPointable.setLong(data, 0, writeValue);
+ ICloudBufferedWriter writer = testClient.createBufferedWriter(bucket, path);
+ CloudResettableInputStream stream = null;
+ boolean aborted = false;
+ try {
+ stream = new CloudResettableInputStream(writer, new WriterSingleBufferProvider());
+ stream.write(data, 0, data.length);
+ } catch (HyracksDataException e) {
+ stream.abort();
+ aborted = true;
+ } finally {
+ if (stream != null && !aborted) {
+ stream.finish();
+ stream.close();
+ }
+ }
+
+ try {
+ long readValue = LongPointable.getLong(testClient.readAllBytes(bucket, path), 0);
+ if (writeValue != readValue) {
+ // This should never happen unless GCS is messed up. But log for sanity check
+ LOGGER.warn(
+ "The writer can write but the written values wasn't successfully read back (wrote: {}, read:{})",
+ writeValue, readValue);
+ }
+ } finally {
+ // Delete the written file
+ testClient.deleteObjects(bucket, Collections.singleton(path));
+ }
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
index 98ccbce..5dfa803 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
@@ -320,7 +320,8 @@
static {
WRITER_SUPPORTED_FORMATS = Set.of(FORMAT_JSON_LOWER_CASE);
- WRITER_SUPPORTED_ADAPTERS = Set.of(ALIAS_LOCALFS_ADAPTER.toLowerCase(), KEY_ADAPTER_NAME_AWS_S3.toLowerCase());
+ WRITER_SUPPORTED_ADAPTERS = Set.of(ALIAS_LOCALFS_ADAPTER.toLowerCase(), KEY_ADAPTER_NAME_AWS_S3.toLowerCase(),
+ KEY_ADAPTER_NAME_GCS.toLowerCase());
WRITER_SUPPORTED_COMPRESSION = Set.of(KEY_COMPRESSION_GZIP);
}
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
index cf51200..8ca428b 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
@@ -22,6 +22,7 @@
import java.util.Map;
import java.util.zip.Deflater;
+import org.apache.asterix.cloud.writer.GCSExternalFileWriterFactory;
import org.apache.asterix.cloud.writer.S3ExternalFileWriterFactory;
import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.external.util.ExternalDataConstants;
@@ -51,6 +52,7 @@
CREATOR_MAP = new HashMap<>();
addCreator(ExternalDataConstants.KEY_ADAPTER_NAME_LOCALFS, LocalFSExternalFileWriterFactory.PROVIDER);
addCreator(ExternalDataConstants.KEY_ADAPTER_NAME_AWS_S3, S3ExternalFileWriterFactory.PROVIDER);
+ addCreator(ExternalDataConstants.KEY_ADAPTER_NAME_GCS, GCSExternalFileWriterFactory.PROVIDER);
}
public static IExternalFileWriterFactory createWriterFactory(ICcApplicationContext appCtx, IWriteDataSink sink,
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/18253
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I0b9cad99de2d32d1e672a3d396897807687685e8
Gerrit-Change-Number: 18253
Gerrit-PatchSet: 1
Gerrit-Owner: Savyasach Reddy <es...@gmail.com>
Gerrit-MessageType: newchange