keksmd commented on code in PR #966: URL: https://github.com/apache/incubator-graphar/pull/966#discussion_r3952410136
########## maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3OutputFile.java: ########## @@ -0,0 +1,164 @@ +/* + * 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.PutObjectRequest; + +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. A head request that cannot answer is + * not treated as a conflict: the conditional publication in {@link StagedOutput#close()} + * remains the atomic guarantee. + * + * @throws FileAlreadyExistsException when the object is already present + */ + private void failIfPresent() throws IOException { + try { + client.headObject( + HeadObjectRequest.builder().bucket(location.bucket).key(location.key).build()); + } catch (RuntimeException absentOrUnknown) { Review Comment: Narrowed to the absence case as you suggested: only `NoSuchKeyException` and a 404 `S3Exception` mean absent, everything else is now raised as an `IOException` from `create()` rather than deferred to `close()` under a different message. Same shape as `S3Storage.exists()` in this module. New test `surfacesAFailedExistenceCheckInsteadOfStaging` drives a 403 head and asserts nothing was published. ########## maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3OutputFile.java: ########## @@ -0,0 +1,164 @@ +/* + * 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.PutObjectRequest; + +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. A head request that cannot answer is + * not treated as a conflict: the conditional publication in {@link StagedOutput#close()} + * remains the atomic guarantee. + * + * @throws FileAlreadyExistsException when the object is already present + */ + private void failIfPresent() throws IOException { + try { + client.headObject( + HeadObjectRequest.builder().bucket(location.bucket).key(location.key).build()); + } catch (RuntimeException absentOrUnknown) { + return; + } + 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 (RuntimeException exception) { Review Comment: Agreed, the conditional put is the atomic guarantee, so its conflict should be typed. A 412 on a `createOnly` publication now surfaces as `FileAlreadyExistsException` with the object URI, matching `failIfPresent()` and the `OutputFile.create()` contract; any other `S3Exception` keeps the `IOException` wrapping. New test `reportsALostPublicationRaceAsAnExistingObject` covers the raced path. ########## maven-projects/storage-s3/src/main/java/org/apache/graphar/storage/s3/S3SeekableInput.java: ########## @@ -0,0 +1,116 @@ +/* + * 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; + +final class S3SeekableInput implements SeekableInput { + 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 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 count = (int) Math.min(destination.remaining(), size - position); + GetObjectRequest.Builder request = + GetObjectRequest.builder() + .bucket(location.bucket) + .key(location.key) + .range("bytes=" + position + "-" + (position + count - 1)); Review Comment: Added read-ahead. The input now fetches a 1 MiB block and serves subsequent reads from memory; a read at least as large as the block bypasses the buffer so bulk reads stay at one request and do not pay a copy. New test `servesSequentialReadsFromOneRangeRequest` reads six single bytes and asserts exactly one `GetObject` call, where the old code issued six. The existing range assertion moved from `bytes=2-4` to `bytes=2-5` because the block is now clipped to the object end rather than to the caller buffer. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
