This is an automated email from the ASF dual-hosted git repository.
SYaoJun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-graphar.git
The following commit(s) were added to refs/heads/main by this push:
new dd17c9b9 feat(java): add an optional S3 storage adapter (#966)
dd17c9b9 is described below
commit dd17c9b9f0fface85fedb02dcad9ba83cd6e9228
Author: alex <[email protected]>
AuthorDate: Tue Sep 8 02:36:01 2026 +0300
feat(java): add an optional S3 storage adapter (#966)
* feat(java): add an optional S3 storage adapter
S3 range reads and staged writes need to satisfy the storage API without
leaking AWS dependencies into metadata or core modules.
Relates to #965.
Rejected: a mandatory S3 or Hadoop dependency in the storage API.
Not-tested: a real MinIO endpoint; the follow-up opt-in transport test
remains separate.
* test(java): verify S3 storage against MinIO
Exercise staged conditional publication and range reads against an actual
S3-compatible endpoint, while keeping the test opt-in for normal SDK
verification.
Relates to #965.
* ci(java): build the SDK modules that the workflow already watches
The workflow triggers on storage-local, io-api and core but compiled only
storage-api, so every module merged after the storage API entered main
without being built or tested upstream. Add storage-s3 to the trigger paths
and list the shipped modules in SDK_MODULES.
* fix(java): reject an existing S3 object before staging bytes
create() deferred the existence check to the conditional publication, so a
caller could stream a whole object before learning about the conflict. Head
the object at open time to match the local adapter, and keep If-None-Match
as the atomic backstop when the head request cannot answer.
* style(java): use the canonical ASF header in the S3 input files
S3InputFile and S3SeekableInput carried a collapsed single-line header while
every other file in the SDK uses the multi-line form.
* ci(java): bind the javadoc step to a lifecycle phase
The step invoked javadoc:javadoc alone, so the reactor produced no jars and
a module resolved its siblings from the repository instead. With one module
in SDK_MODULES nothing was resolved across modules and the step passed; with
the shipped modules listed, storage-local could not find
graphar-storage-api.
Run package in the same invocation so the reactor supplies the dependency.
* fix(java): let S3 failures keep their identity and stop paying a round
trip per read
A head request that fails for any reason other than absence was read as
'object absent', so a permission or throttling error was deferred to close()
under a different message; only NoSuchKeyException and a 404 now mean
absent.
A conditional put rejected with 412 lost the race for an existing object but
surfaced as a plain IOException, so callers could not tell a conflict from a
transient failure; it now surfaces as FileAlreadyExistsException, matching
both the eager check and the OutputFile.create() contract.
Reads were served by one GetObject range request per call, so byte-level
readers of columnar headers and footers paid a round trip each. The input
now
keeps the last block in memory and serves subsequent reads from it, while a
read at least as large as the block bypasses the buffer.
---
.github/workflows/java-sdk.yml | 6 +-
maven-projects/pom.xml | 1 +
maven-projects/storage-s3/pom.xml | 75 ++++++++
.../org/apache/graphar/storage/s3/S3InputFile.java | 64 +++++++
.../apache/graphar/storage/s3/S3OutputFile.java | 179 +++++++++++++++++
.../apache/graphar/storage/s3/S3SeekableInput.java | 147 ++++++++++++++
.../org/apache/graphar/storage/s3/S3Storage.java | 110 +++++++++++
.../storage/s3/MinioS3StorageIntegrationIT.java | 135 +++++++++++++
.../apache/graphar/storage/s3/S3StorageTest.java | 212 +++++++++++++++++++++
9 files changed, 927 insertions(+), 2 deletions(-)
diff --git a/.github/workflows/java-sdk.yml b/.github/workflows/java-sdk.yml
index 755845da..a8779ee0 100644
--- a/.github/workflows/java-sdk.yml
+++ b/.github/workflows/java-sdk.yml
@@ -25,6 +25,7 @@ on:
- 'maven-projects/pom.xml'
- 'maven-projects/storage-api/**'
- 'maven-projects/storage-local/**'
+ - 'maven-projects/storage-s3/**'
- 'maven-projects/io-api/**'
- 'maven-projects/io-parquet/**'
- 'maven-projects/core/**'
@@ -38,6 +39,7 @@ on:
- 'maven-projects/pom.xml'
- 'maven-projects/storage-api/**'
- 'maven-projects/storage-local/**'
+ - 'maven-projects/storage-s3/**'
- 'maven-projects/io-api/**'
- 'maven-projects/io-parquet/**'
- 'maven-projects/core/**'
@@ -53,7 +55,7 @@ jobs:
test:
runs-on: ubuntu-latest
env:
- SDK_MODULES: storage-api
+ SDK_MODULES: storage-api,storage-local,io-api,storage-s3
steps:
- uses: actions/checkout@v7
with:
@@ -83,4 +85,4 @@ jobs:
working-directory: maven-projects
run: |
export JAVA_HOME=${JAVA_HOME_11_X64}
- mvn --no-transfer-progress -pl ${SDK_MODULES} -am javadoc:javadoc
+ mvn --no-transfer-progress -pl ${SDK_MODULES} -am package
javadoc:javadoc -DskipTests -Dspotless.check.skip=true
diff --git a/maven-projects/pom.xml b/maven-projects/pom.xml
index 56b5cd87..5ea0b89c 100644
--- a/maven-projects/pom.xml
+++ b/maven-projects/pom.xml
@@ -81,6 +81,7 @@
<module>storage-api</module>
<module>storage-local</module>
<module>io-api</module>
+ <module>storage-s3</module>
</modules>
<build>
diff --git a/maven-projects/storage-s3/pom.xml
b/maven-projects/storage-s3/pom.xml
new file mode 100644
index 00000000..98ac2b2d
--- /dev/null
+++ b/maven-projects/storage-s3/pom.xml
@@ -0,0 +1,75 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.graphar</groupId>
+ <artifactId>graphar-root</artifactId>
+ <version>${graphar.version}</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+ <artifactId>graphar-storage-s3</artifactId>
+ <packaging>jar</packaging>
+ <version>${graphar.version}</version>
+ <name>graphar-storage-s3</name>
+ <properties>
+ <maven.compiler.source>11</maven.compiler.source>
+ <maven.compiler.target>11</maven.compiler.target>
+ <aws.sdk.version>2.31.54</aws.sdk.version>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.graphar</groupId>
+ <artifactId>graphar-storage-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>s3</artifactId>
+ <version>${aws.sdk.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>4.13.2</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>com.diffplug.spotless</groupId>
+ <artifactId>spotless-maven-plugin</artifactId>
+ <version>${spotless-maven-plugin.version}</version>
+
<configuration><java><googleJavaFormat><version>1.7</version><style>AOSP</style></googleJavaFormat></java></configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-javadoc-plugin</artifactId>
+
<executions><execution><id>attach-javadocs</id><goals><goal>jar</goal></goals></execution></executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git
a/maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3InputFile.java
b/maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3InputFile.java
new file mode 100644
index 00000000..069fa032
--- /dev/null
+++
b/maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3InputFile.java
@@ -0,0 +1,64 @@
+/*
+ * 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.graphar.storage.s3;
+
+import java.io.IOException;
+import java.net.URI;
+import org.apache.graphar.storage.InputFile;
+import org.apache.graphar.storage.SeekableInput;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
+import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+
+final class S3InputFile implements InputFile {
+ private final S3Client client;
+ private final S3Storage.Location location;
+
+ S3InputFile(S3Client client, S3Storage.Location location) {
+ this.client = client;
+ this.location = location;
+ }
+
+ @Override
+ public URI uri() {
+ return location.uri;
+ }
+
+ @Override
+ public long size() throws IOException {
+ return head().contentLength();
+ }
+
+ @Override
+ public SeekableInput open() throws IOException {
+ HeadObjectResponse response = head();
+ return new S3SeekableInput(
+ client, location, response.contentLength(),
response.versionId(), response.eTag());
+ }
+
+ private HeadObjectResponse head() throws IOException {
+ try {
+ return client.headObject(
+
HeadObjectRequest.builder().bucket(location.bucket).key(location.key).build());
+ } catch (RuntimeException exception) {
+ throw new IOException("Cannot inspect S3 object " + location.uri,
exception);
+ }
+ }
+}
diff --git
a/maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3OutputFile.java
b/maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3OutputFile.java
new file mode 100644
index 00000000..8cf9da44
--- /dev/null
+++
b/maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3OutputFile.java
@@ -0,0 +1,179 @@
+/*
+ * 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.graphar.storage.s3;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import org.apache.graphar.storage.OutputFile;
+import org.apache.graphar.storage.PositionOutput;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+
+final class S3OutputFile implements OutputFile {
+ private final S3Client client;
+ private final S3Storage.Location location;
+ private final Path stagingDirectory;
+
+ S3OutputFile(S3Client client, S3Storage.Location location, Path
stagingDirectory) {
+ this.client = client;
+ this.location = location;
+ this.stagingDirectory = stagingDirectory;
+ }
+
+ @Override
+ public URI uri() {
+ return location.uri;
+ }
+
+ @Override
+ public PositionOutput create() throws IOException {
+ failIfPresent();
+ return open(true);
+ }
+
+ /**
+ * Rejects an existing object before any byte is staged, so that a caller
learns about the
+ * conflict at open time the way the local adapter does. Only an absent
object lets the caller
+ * proceed; a head request that fails for any other reason surfaces here
rather than after a
+ * full staging pass.
+ *
+ * @throws FileAlreadyExistsException when the object is already present
+ * @throws IOException when the object cannot be inspected
+ */
+ private void failIfPresent() throws IOException {
+ try {
+ client.headObject(
+
HeadObjectRequest.builder().bucket(location.bucket).key(location.key).build());
+ } catch (NoSuchKeyException absent) {
+ return;
+ } catch (S3Exception exception) {
+ if (exception.statusCode() == 404) {
+ return;
+ }
+ throw new IOException("Cannot inspect S3 object " + location.uri,
exception);
+ } catch (RuntimeException exception) {
+ throw new IOException("Cannot inspect S3 object " + location.uri,
exception);
+ }
+ throw new FileAlreadyExistsException(location.uri.toString());
+ }
+
+ @Override
+ public PositionOutput createOrOverwrite() throws IOException {
+ return open(false);
+ }
+
+ private PositionOutput open(boolean createOnly) throws IOException {
+ Path stage = Files.createTempFile(stagingDirectory, "graphar-s3-",
".stage");
+ try {
+ return new StagedOutput(client, location, stage, createOnly);
+ } catch (IOException | RuntimeException exception) {
+ Files.deleteIfExists(stage);
+ throw exception;
+ }
+ }
+
+ private static final class StagedOutput implements PositionOutput {
+ private final S3Client client;
+ private final S3Storage.Location location;
+ private final Path stage;
+ private final boolean createOnly;
+ private final OutputStream output;
+ private long position;
+ private boolean closed;
+
+ private StagedOutput(
+ S3Client client, S3Storage.Location location, Path stage,
boolean createOnly)
+ throws IOException {
+ this.client = client;
+ this.location = location;
+ this.stage = stage;
+ this.createOnly = createOnly;
+ this.output = Files.newOutputStream(stage);
+ }
+
+ @Override
+ public long position() throws IOException {
+ requireOpen();
+ return position;
+ }
+
+ @Override
+ public void write(ByteBuffer source) throws IOException {
+ requireOpen();
+ byte[] bytes = new byte[source.remaining()];
+ source.get(bytes);
+ write(bytes, 0, bytes.length);
+ }
+
+ @Override
+ public void write(byte[] source, int offset, int length) throws
IOException {
+ requireOpen();
+ output.write(source, offset, length);
+ position = Math.addExact(position, length);
+ }
+
+ @Override
+ public void flush() throws IOException {
+ requireOpen();
+ output.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ try {
+ output.close();
+ PutObjectRequest.Builder request =
+
PutObjectRequest.builder().bucket(location.bucket).key(location.key);
+ if (createOnly) {
+ request.ifNoneMatch("*");
+ }
+ client.putObject(request.build(), RequestBody.fromFile(stage));
+ } catch (S3Exception exception) {
+ if (createOnly && exception.statusCode() == 412) {
+ throw new
FileAlreadyExistsException(location.uri.toString());
+ }
+ throw new IOException("Cannot publish S3 object " +
location.uri, exception);
+ } catch (RuntimeException exception) {
+ throw new IOException("Cannot publish S3 object " +
location.uri, exception);
+ } finally {
+ Files.deleteIfExists(stage);
+ }
+ }
+
+ private void requireOpen() throws IOException {
+ if (closed) {
+ throw new IOException("S3 output is closed.");
+ }
+ }
+ }
+}
diff --git
a/maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3SeekableInput.java
b/maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3SeekableInput.java
new file mode 100644
index 00000000..4b9bef09
--- /dev/null
+++
b/maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3SeekableInput.java
@@ -0,0 +1,147 @@
+/*
+ * 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.graphar.storage.s3;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.apache.graphar.storage.SeekableInput;
+import software.amazon.awssdk.core.ResponseBytes;
+import software.amazon.awssdk.core.sync.ResponseTransformer;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+
+/**
+ * Reads an object through range requests, keeping the last fetched block in
memory so that the
+ * small sequential reads typical of columnar footers do not cost one round
trip each. A read at
+ * least as large as the block size bypasses the buffer and is served by a
single request.
+ */
+final class S3SeekableInput implements SeekableInput {
+ private static final int BLOCK_SIZE = 1 << 20;
+ private static final byte[] NO_BLOCK = new byte[0];
+
+ private final S3Client client;
+ private final S3Storage.Location location;
+ private final long size;
+ private final String versionId;
+ private final String eTag;
+ private long position;
+ private byte[] block = NO_BLOCK;
+ private long blockStart;
+ private int blockLength;
+ private boolean closed;
+
+ S3SeekableInput(
+ S3Client client,
+ S3Storage.Location location,
+ long size,
+ String versionId,
+ String eTag) {
+ this.client = client;
+ this.location = location;
+ this.size = size;
+ this.versionId = versionId;
+ this.eTag = eTag;
+ }
+
+ @Override
+ public long position() throws IOException {
+ requireOpen();
+ return position;
+ }
+
+ @Override
+ public void seek(long newPosition) throws IOException {
+ requireOpen();
+ if (newPosition < 0) {
+ throw new IllegalArgumentException("S3 seek position must be
non-negative.");
+ }
+ position = newPosition;
+ }
+
+ @Override
+ public int read(ByteBuffer destination) throws IOException {
+ requireOpen();
+ if (!destination.hasRemaining()) {
+ return 0;
+ }
+ if (position >= size) {
+ return -1;
+ }
+ int wanted = (int) Math.min(destination.remaining(), size - position);
+ if (wanted >= BLOCK_SIZE) {
+ byte[] bytes = fetch(position, wanted);
+ destination.put(bytes);
+ position += bytes.length;
+ return bytes.length;
+ }
+ if (position < blockStart || position >= blockStart + blockLength) {
+ int length = (int) Math.min(BLOCK_SIZE, size - position);
+ block = fetch(position, length);
+ blockStart = position;
+ blockLength = block.length;
+ }
+ int offset = (int) (position - blockStart);
+ int count = Math.min(wanted, blockLength - offset);
+ destination.put(block, offset, count);
+ position += count;
+ return count;
+ }
+
+ private byte[] fetch(long start, int length) throws IOException {
+ GetObjectRequest.Builder request =
+ GetObjectRequest.builder()
+ .bucket(location.bucket)
+ .key(location.key)
+ .range("bytes=" + start + "-" + (start + length - 1));
+ if (versionId != null) {
+ request.versionId(versionId);
+ } else if (eTag != null) {
+ request.ifMatch(eTag);
+ }
+ try {
+ ResponseBytes<GetObjectResponse> response =
+ client.getObject(request.build(),
ResponseTransformer.toBytes());
+ byte[] bytes = response.asByteArray();
+ if (bytes.length != length) {
+ throw new IOException(
+ "S3 returned " + bytes.length + " bytes for requested
range of " + length);
+ }
+ return bytes;
+ } catch (IOException exception) {
+ throw exception;
+ } catch (RuntimeException exception) {
+ throw new IOException("Cannot range-read S3 object " +
location.uri, exception);
+ }
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ block = NO_BLOCK;
+ blockLength = 0;
+ }
+
+ private void requireOpen() throws IOException {
+ if (closed) {
+ throw new IOException("S3 input is closed.");
+ }
+ }
+}
diff --git
a/maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3Storage.java
b/maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3Storage.java
new file mode 100644
index 00000000..8fdbbbe0
--- /dev/null
+++
b/maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3Storage.java
@@ -0,0 +1,110 @@
+/*
+ * 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.graphar.storage.s3;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Objects;
+import org.apache.graphar.storage.InputFile;
+import org.apache.graphar.storage.OutputFile;
+import org.apache.graphar.storage.Storage;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+
+/** S3-backed GraphAr storage using an injected AWS SDK client. */
+public final class S3Storage implements Storage {
+ private final S3Client client;
+ private final Path stagingDirectory;
+
+ /** Creates S3 storage. The caller owns the client and is responsible for
closing it. */
+ public S3Storage(S3Client client, Path stagingDirectory) throws
IOException {
+ this.client = Objects.requireNonNull(client, "S3 client cannot be
null.");
+ this.stagingDirectory =
+ Objects.requireNonNull(stagingDirectory, "Staging directory
cannot be null.");
+ Files.createDirectories(stagingDirectory);
+ if (!Files.isDirectory(stagingDirectory)) {
+ throw new IllegalArgumentException(
+ "S3 staging path is not a directory: " + stagingDirectory);
+ }
+ }
+
+ @Override
+ public InputFile inputFile(URI uri) {
+ Location location = Location.from(uri);
+ return new S3InputFile(client, location);
+ }
+
+ @Override
+ public OutputFile outputFile(URI uri) {
+ Location location = Location.from(uri);
+ return new S3OutputFile(client, location, stagingDirectory);
+ }
+
+ @Override
+ public boolean exists(URI uri) throws IOException {
+ Location location = Location.from(uri);
+ try {
+ client.headObject(
+
HeadObjectRequest.builder().bucket(location.bucket).key(location.key).build());
+ return true;
+ } catch (NoSuchKeyException exception) {
+ return false;
+ } catch (S3Exception exception) {
+ if (exception.statusCode() == 404) {
+ return false;
+ }
+ throw new IOException("Cannot inspect S3 object " + uri,
exception);
+ } catch (RuntimeException exception) {
+ throw new IOException("Cannot inspect S3 object " + uri,
exception);
+ }
+ }
+
+ static final class Location {
+ final URI uri;
+ final String bucket;
+ final String key;
+
+ private Location(URI uri, String bucket, String key) {
+ this.uri = uri;
+ this.bucket = bucket;
+ this.key = key;
+ }
+
+ private static Location from(URI uri) {
+ if (uri == null
+ || !"s3".equalsIgnoreCase(uri.getScheme())
+ || uri.getHost() == null
+ || uri.getHost().isEmpty()
+ || uri.getPath() == null
+ || uri.getPath().length() <= 1
+ || uri.getUserInfo() != null
+ || uri.getQuery() != null
+ || uri.getFragment() != null) {
+ throw new IllegalArgumentException(
+ "Expected an absolute s3://bucket/key URI: " + uri);
+ }
+ return new Location(uri, uri.getHost(),
uri.getPath().substring(1));
+ }
+ }
+}
diff --git
a/maven-projects/storage-s3/src/test/java/org/apache/graphar/storage/s3/MinioS3StorageIntegrationIT.java
b/maven-projects/storage-s3/src/test/java/org/apache/graphar/storage/s3/MinioS3StorageIntegrationIT.java
new file mode 100644
index 00000000..d4b16fb4
--- /dev/null
+++
b/maven-projects/storage-s3/src/test/java/org/apache/graphar/storage/s3/MinioS3StorageIntegrationIT.java
@@ -0,0 +1,135 @@
+/*
+ * 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.graphar.storage.s3;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.UUID;
+import org.apache.graphar.storage.PositionOutput;
+import org.apache.graphar.storage.SeekableInput;
+import org.junit.Assume;
+import org.junit.Test;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3Configuration;
+
+/**
+ * Opt-in MinIO transport test. Run with {@code
GRAPHAR_MINIO_ENDPOINT=http://localhost:19000}.
+ *
+ * <p>The default test lifecycle excludes {@code *IT}; this class
intentionally requires a
+ * separately started MinIO endpoint rather than adding Docker to normal SDK
verification.
+ */
+public class MinioS3StorageIntegrationIT {
+ private static final String ENDPOINT = "GRAPHAR_MINIO_ENDPOINT";
+ private static final String ACCESS_KEY = "GRAPHAR_MINIO_ACCESS_KEY";
+ private static final String SECRET_KEY = "GRAPHAR_MINIO_SECRET_KEY";
+
+ @Test
+ public void stagesConditionallyPublishesAndRangeReadsOverMinio() throws
Exception {
+ String endpoint = System.getenv(ENDPOINT);
+ Assume.assumeTrue(
+ "Set " + ENDPOINT + " to run the MinIO transport test.",
endpoint != null);
+
+ String bucket = "graphar-minio-" +
UUID.randomUUID().toString().replace("-", "");
+ Path staging = Files.createTempDirectory("graphar-minio-stage-");
+ try (S3Client client = client(endpoint)) {
+ client.createBucket(request -> request.bucket(bucket));
+ S3Storage storage = new S3Storage(client, staging);
+ URI object = URI.create("s3://" + bucket +
"/topology/chunk.parquet");
+
+ assertFalse(storage.exists(object));
+ write(storage.outputFile(object).create(), new byte[] {0, 1, 2, 3,
4, 5});
+ assertTrue(storage.exists(object));
+ assertEquals(6L, storage.inputFile(object).size());
+
+ ByteBuffer range = ByteBuffer.allocateDirect(3);
+ try (SeekableInput input = storage.inputFile(object).open()) {
+ input.seek(2);
+ assertEquals(3, input.read(range));
+ assertEquals(5L, input.position());
+ input.seek(6);
+ assertEquals(-1, input.read(ByteBuffer.allocate(1)));
+ }
+ range.flip();
+ byte[] actual = new byte[range.remaining()];
+ range.get(actual);
+ assertArrayEquals(new byte[] {2, 3, 4}, actual);
+
+ try {
+ write(storage.outputFile(object).create(), new byte[] {9});
+ fail("create() must not overwrite an existing S3 object.");
+ } catch (IOException expected) {
+ // MinIO returns the S3 conditional-publication failure
through the AWS SDK.
+ }
+ write(storage.outputFile(object).createOrOverwrite(), new byte[]
{9});
+ try (SeekableInput input = storage.inputFile(object).open()) {
+ ByteBuffer replacement = ByteBuffer.allocate(1);
+ assertEquals(1, input.read(replacement));
+ assertArrayEquals(new byte[] {9}, replacement.array());
+ }
+ } finally {
+ deleteTree(staging);
+ }
+ }
+
+ private static S3Client client(String endpoint) {
+ String accessKey = System.getenv().getOrDefault(ACCESS_KEY,
"minioadmin");
+ String secretKey = System.getenv().getOrDefault(SECRET_KEY,
"minioadmin");
+ return S3Client.builder()
+ .endpointOverride(URI.create(endpoint))
+ .credentialsProvider(
+ StaticCredentialsProvider.create(
+ AwsBasicCredentials.create(accessKey,
secretKey)))
+ .region(Region.US_EAST_1)
+ .serviceConfiguration(
+
S3Configuration.builder().pathStyleAccessEnabled(true).build())
+ .build();
+ }
+
+ private static void write(PositionOutput output, byte[] bytes) throws
IOException {
+ try (PositionOutput ignored = output) {
+ output.write(bytes);
+ }
+ }
+
+ private static void deleteTree(Path root) throws IOException {
+ if (!Files.exists(root)) {
+ return;
+ }
+ try (java.util.stream.Stream<Path> paths = Files.walk(root)) {
+ for (Path path :
+ paths.sorted(java.util.Comparator.reverseOrder())
+ .collect(java.util.stream.Collectors.toList())) {
+ Files.delete(path);
+ }
+ }
+ }
+}
diff --git
a/maven-projects/storage-s3/src/test/java/org/apache/graphar/storage/s3/S3StorageTest.java
b/maven-projects/storage-s3/src/test/java/org/apache/graphar/storage/s3/S3StorageTest.java
new file mode 100644
index 00000000..d60723d2
--- /dev/null
+++
b/maven-projects/storage-s3/src/test/java/org/apache/graphar/storage/s3/S3StorageTest.java
@@ -0,0 +1,212 @@
+/*
+ * 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.graphar.storage.s3;
+
+import java.io.IOException;
+import java.lang.reflect.Proxy;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.util.Arrays;
+import org.apache.graphar.storage.PositionOutput;
+import org.apache.graphar.storage.SeekableInput;
+import org.junit.Assert;
+import org.junit.Test;
+import software.amazon.awssdk.core.ResponseBytes;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
+import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.PutObjectResponse;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+
+/** Credential-free contract tests using an injected in-memory S3 SDK
transport. */
+public class S3StorageTest {
+ @Test
+ public void stagesOutputAndUsesVersionPinnedByteRanges() throws Exception {
+ byte[] object = "abcdef".getBytes();
+ RequestLog log = new RequestLog(object);
+ S3Storage storage =
+ new S3Storage(log.client(),
Files.createTempDirectory("graphar-s3-stage"));
+ URI uri = URI.create("s3://bucket/graph/a.parquet");
+
+ Assert.assertTrue(storage.exists(uri));
+ try (SeekableInput input = storage.inputFile(uri).open()) {
+ input.seek(2);
+ ByteBuffer bytes = ByteBuffer.allocate(3);
+ Assert.assertEquals(3, input.read(bytes));
+ Assert.assertArrayEquals("cde".getBytes(), bytes.array());
+ }
+ Assert.assertEquals("bytes=2-5", log.range);
+ Assert.assertEquals("version-1", log.versionId);
+
+ URI target = URI.create("s3://bucket/graph/new.bin");
+ try (PositionOutput output = storage.outputFile(target).create()) {
+ output.write(new byte[] {9, 8, 7});
+ }
+ Assert.assertEquals("*", log.ifNoneMatch);
+ Assert.assertArrayEquals(new byte[] {9, 8, 7}, log.putBytes);
+ }
+
+ @Test
+ public void rejectsAnExistingObjectBeforeStagingAnyByte() throws Exception
{
+ RequestLog log = new RequestLog("abcdef".getBytes());
+ S3Storage storage =
+ new S3Storage(log.client(),
Files.createTempDirectory("graphar-s3-stage"));
+
+ try {
+
storage.outputFile(URI.create("s3://bucket/graph/a.parquet")).create();
+ Assert.fail("create() must reject an object that already exists.");
+ } catch (FileAlreadyExistsException expected) {
+ Assert.assertNull("No object may be published on a rejected
create().", log.putBytes);
+ }
+ }
+
+ @Test
+ public void servesSequentialReadsFromOneRangeRequest() throws Exception {
+ RequestLog log = new RequestLog("abcdef".getBytes());
+ S3Storage storage =
+ new S3Storage(log.client(),
Files.createTempDirectory("graphar-s3-stage"));
+
+ try (SeekableInput input =
+
storage.inputFile(URI.create("s3://bucket/graph/a.parquet")).open()) {
+ for (int index = 0; index < 6; index++) {
+ ByteBuffer single = ByteBuffer.allocate(1);
+ Assert.assertEquals(1, input.read(single));
+ Assert.assertEquals("abcdef".charAt(index), (char)
single.array()[0]);
+ }
+ Assert.assertEquals(-1, input.read(ByteBuffer.allocate(1)));
+ }
+ Assert.assertEquals(
+ "Byte-level reads must share one range request.", 1,
log.getObjectCalls);
+ }
+
+ @Test
+ public void reportsALostPublicationRaceAsAnExistingObject() throws
Exception {
+ RequestLog log = new RequestLog("abcdef".getBytes());
+ log.putStatusCode = 412;
+ S3Storage storage =
+ new S3Storage(log.client(),
Files.createTempDirectory("graphar-s3-stage"));
+
+ PositionOutput output =
+
storage.outputFile(URI.create("s3://bucket/graph/new.bin")).create();
+ output.write(new byte[] {1, 2, 3});
+ try {
+ output.close();
+ Assert.fail("A conditional put rejected with 412 must surface as a
conflict.");
+ } catch (FileAlreadyExistsException expected) {
+ Assert.assertEquals("s3://bucket/graph/new.bin",
expected.getMessage());
+ }
+ }
+
+ @Test
+ public void surfacesAFailedExistenceCheckInsteadOfStaging() throws
Exception {
+ RequestLog log = new RequestLog("abcdef".getBytes());
+ log.headStatusCode = 403;
+ S3Storage storage =
+ new S3Storage(log.client(),
Files.createTempDirectory("graphar-s3-stage"));
+
+ try {
+
storage.outputFile(URI.create("s3://bucket/graph/new.bin")).create();
+ Assert.fail("A head request denied with 403 must not be read as an
absent object.");
+ } catch (FileAlreadyExistsException unexpected) {
+ throw new AssertionError("403 must not be reported as a
conflict.", unexpected);
+ } catch (IOException expected) {
+ Assert.assertNull("No object may be published on a failed check.",
log.putBytes);
+ }
+ }
+
+ private static final class RequestLog {
+ private final byte[] object;
+ private String range;
+ private String versionId;
+ private String ifNoneMatch;
+ private byte[] putBytes;
+ private int getObjectCalls;
+ private int headStatusCode;
+ private int putStatusCode;
+
+ private RequestLog(byte[] object) {
+ this.object = object;
+ }
+
+ private S3Client client() {
+ return (S3Client)
+ Proxy.newProxyInstance(
+ getClass().getClassLoader(),
+ new Class<?>[] {S3Client.class},
+ (proxy, method, arguments) -> {
+ if ("headObject".equals(method.getName())) {
+ HeadObjectRequest request =
(HeadObjectRequest) arguments[0];
+ Assert.assertEquals("bucket",
request.bucket());
+ if (headStatusCode != 0) {
+ throw S3Exception.builder()
+ .statusCode(headStatusCode)
+ .message("denied")
+ .build();
+ }
+ if
(!"graph/a.parquet".equals(request.key())) {
+ throw
NoSuchKeyException.builder().statusCode(404).build();
+ }
+ return HeadObjectResponse.builder()
+ .contentLength((long)
object.length)
+ .versionId("version-1")
+ .eTag("etag-1")
+ .build();
+ }
+ if ("getObject".equals(method.getName())) {
+ GetObjectRequest request =
(GetObjectRequest) arguments[0];
+ getObjectCalls++;
+ range = request.range();
+ versionId = request.versionId();
+ String[] bounds =
range.substring("bytes=".length()).split("-");
+ int start = Integer.parseInt(bounds[0]);
+ int end = Integer.parseInt(bounds[1]);
+ return ResponseBytes.fromByteArray(
+
GetObjectResponse.builder().build(),
+ Arrays.copyOfRange(object, start,
end + 1));
+ }
+ if ("putObject".equals(method.getName())) {
+ PutObjectRequest request =
(PutObjectRequest) arguments[0];
+ ifNoneMatch = request.ifNoneMatch();
+ if (putStatusCode != 0) {
+ throw S3Exception.builder()
+ .statusCode(putStatusCode)
+ .message("conflict")
+ .build();
+ }
+ RequestBody body = (RequestBody)
arguments[1];
+ putBytes =
+
body.contentStreamProvider().newStream().readAllBytes();
+ return PutObjectResponse.builder().build();
+ }
+ if ("serviceName".equals(method.getName())) {
+ return "s3";
+ }
+ throw new
UnsupportedOperationException(method.toString());
+ });
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
