SemyonSinchenko commented on code in PR #966: URL: https://github.com/apache/incubator-graphar/pull/966#discussion_r3952289845
########## 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: When `createOnly` publication loses a concurrent race, S3 rejects the conditional put with 412, which the AWS SDK surfaces as an `S3Exception` — this generic catch wraps it into a plain `IOException`, so callers cannot distinguish "already exists" from a transient failure. This is inconsistent with the eager check in `failIfPresent()` (which throws `FileAlreadyExistsException`) and with the `OutputFile.create()` contract of failing on an existing file. The conditional put is the documented atomic guarantee, so its conflict outcome should surface as the same typed exception. Fully staged data is also dropped with a misleading message in this case. ########## 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: Every `read()` issues a full S3 `GetObject` range request with no read-ahead buffering, so byte-level readers (header/footer parsing and small-buffer scans typical of columnar formats, which this `SeekableInput` SPI is designed to serve) pay one network round-trip per read call. Consider buffering ahead (e.g., read a configurable block such as 64KB–1MB into memory and serve subsequent sequential reads from the buffer), so bulk and small reads both amortize to ~1 request per block. ########## 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: This catch treats every headObject failure (403 auth/permission errors, throttling, network faults) as "object absent", masking real environment errors and silently deferring the true failure to close() with a different message. Compare with S3Storage.exists() in the same module, which correctly distinguishes NoSuchKeyException/404 from other failures. Narrow the catch to the absence case so genuine errors surface at create() time instead of wasting staging work. -- 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]
