Copilot commented on code in PR #16865:
URL: https://github.com/apache/iceberg/pull/16865#discussion_r3472412352
##########
aliyun/src/main/java/org/apache/iceberg/aliyun/oss/OSSInputStream.java:
##########
@@ -82,89 +115,182 @@ public int read() throws IOException {
Preconditions.checkState(!closed, "Cannot read: already closed");
positionStream();
- pos += 1;
- next += 1;
- readBytes.increment();
- readOperations.increment();
+ try {
+ int bytesRead = Failsafe.with(retryPolicy).get(() -> stream.read());
+ if (bytesRead != -1) {
+ pos += 1;
+ next += 1;
+ readBytes.increment();
+ readOperations.increment();
+ }
+
+ return bytesRead;
+ } catch (FailsafeException ex) {
+ if (ex.getCause() instanceof IOException) {
+ throw (IOException) ex.getCause();
+ }
- return stream.read();
+ throw ex;
+ }
}
@Override
public int read(byte[] b, int off, int len) throws IOException {
Preconditions.checkState(!closed, "Cannot read: already closed");
positionStream();
- int bytesRead = stream.read(b, off, len);
- pos += bytesRead;
- next += bytesRead;
- readBytes.increment(bytesRead);
- readOperations.increment();
+ try {
+ int bytesRead = Failsafe.with(retryPolicy).get(() -> stream.read(b, off,
len));
+ if (bytesRead > 0) {
+ pos += bytesRead;
+ next += bytesRead;
+ readBytes.increment(bytesRead);
+ readOperations.increment();
+ }
+
+ return bytesRead;
+ } catch (FailsafeException ex) {
+ if (ex.getCause() instanceof IOException) {
+ throw (IOException) ex.getCause();
+ }
- return bytesRead;
+ throw ex;
+ }
}
@Override
- public void close() throws IOException {
- if (closed) {
- return;
+ public void readFully(long position, byte[] buffer, int offset, int length)
throws IOException {
+ Preconditions.checkPositionIndexes(offset, offset + length, buffer.length);
+
+ GetObjectRequest request =
+ new GetObjectRequest(uri.bucket(), uri.key()).withRange(position,
position + length - 1);
Review Comment:
`RangeReadable.readFully` can legally be called with `length == 0` (e.g.,
0-length ranges via `RangeReadable#readVectored`). The current implementation
will build an invalid range (`position` to `position - 1`) and still issue a
GET. Add an early return for `length == 0` to make this a no-op.
##########
aliyun/src/main/java/org/apache/iceberg/aliyun/oss/OSSInputStream.java:
##########
@@ -82,89 +115,182 @@ public int read() throws IOException {
Preconditions.checkState(!closed, "Cannot read: already closed");
positionStream();
- pos += 1;
- next += 1;
- readBytes.increment();
- readOperations.increment();
+ try {
+ int bytesRead = Failsafe.with(retryPolicy).get(() -> stream.read());
+ if (bytesRead != -1) {
+ pos += 1;
+ next += 1;
+ readBytes.increment();
+ readOperations.increment();
+ }
+
+ return bytesRead;
Review Comment:
`readOperations` is currently only incremented when `read()` returns a byte
(i.e., not at EOF). Other Iceberg `SeekableInputStream` implementations
increment the read-ops counter for each successful read call (e.g.,
`aws/.../S3InputStream`), and keep `readBytes` conditional. Consider
incrementing `readOperations` unconditionally once the read returns (even when
`bytesRead == -1`) to keep metrics consistent across FileIOs.
##########
aliyun/src/main/java/org/apache/iceberg/aliyun/oss/OSSInputStream.java:
##########
@@ -82,89 +115,182 @@ public int read() throws IOException {
Preconditions.checkState(!closed, "Cannot read: already closed");
positionStream();
- pos += 1;
- next += 1;
- readBytes.increment();
- readOperations.increment();
+ try {
+ int bytesRead = Failsafe.with(retryPolicy).get(() -> stream.read());
+ if (bytesRead != -1) {
+ pos += 1;
+ next += 1;
+ readBytes.increment();
+ readOperations.increment();
+ }
+
+ return bytesRead;
+ } catch (FailsafeException ex) {
+ if (ex.getCause() instanceof IOException) {
+ throw (IOException) ex.getCause();
+ }
- return stream.read();
+ throw ex;
+ }
}
@Override
public int read(byte[] b, int off, int len) throws IOException {
Preconditions.checkState(!closed, "Cannot read: already closed");
positionStream();
- int bytesRead = stream.read(b, off, len);
- pos += bytesRead;
- next += bytesRead;
- readBytes.increment(bytesRead);
- readOperations.increment();
+ try {
+ int bytesRead = Failsafe.with(retryPolicy).get(() -> stream.read(b, off,
len));
+ if (bytesRead > 0) {
+ pos += bytesRead;
+ next += bytesRead;
+ readBytes.increment(bytesRead);
+ readOperations.increment();
+ }
+
+ return bytesRead;
+ } catch (FailsafeException ex) {
+ if (ex.getCause() instanceof IOException) {
+ throw (IOException) ex.getCause();
+ }
- return bytesRead;
+ throw ex;
+ }
}
@Override
- public void close() throws IOException {
- if (closed) {
- return;
+ public void readFully(long position, byte[] buffer, int offset, int length)
throws IOException {
+ Preconditions.checkPositionIndexes(offset, offset + length, buffer.length);
+
+ GetObjectRequest request =
+ new GetObjectRequest(uri.bucket(), uri.key()).withRange(position,
position + length - 1);
+ OSSObject object = client.getObject(request);
+ try {
+ IOUtil.readFully(object.getObjectContent(), buffer, offset, length);
+ readBytes.increment(length);
+ readOperations.increment();
+ } finally {
+ // Use close() instead of forcedClose() here because the range response
data has been
+ // fully consumed. close() goes through InputStream ->
EofSensorInputStream path which
+ // detects EOF and returns the connection to Apache HttpClient's
connection pool.
+ // forcedClose() calls httpResponse.close() -> ConnectionHolder.close()
which hardcodes
+ // releaseConnection(false), always destroying the connection even if
data was consumed.
+ object.close();
}
+ }
+
+ @Override
+ public int readTail(byte[] buffer, int offset, int length) throws
IOException {
+ Preconditions.checkPositionIndexes(offset, offset + length, buffer.length);
+ GetObjectRequest request = new GetObjectRequest(uri.bucket(),
uri.key()).withRange(-1, length);
Review Comment:
`RangeReadable.readTail` should be able to handle `length == 0` without
issuing a request. Currently it always makes a range request with
`withRange(-1, 0)`, which may be rejected by OSS and is unnecessary. Return 0
early when `length == 0`.
##########
aliyun/src/test/java/org/apache/iceberg/aliyun/oss/mock/AliyunOSSMock.java:
##########
@@ -224,8 +224,8 @@ private void getObject(String bucketName, String
objectName, HttpExchange httpEx
long bytesToRead = Math.min(fileSize - 1, rangeEnd) - rangeStart + 1;
long skipSize = rangeStart;
if (rangeStart == -1) {
- bytesToRead = Math.min(fileSize - 1, rangeEnd);
- skipSize = fileSize - rangeEnd;
+ bytesToRead = Math.min(fileSize, rangeEnd);
+ skipSize = Math.max(0, fileSize - rangeEnd);
}
Review Comment:
For suffix range requests (`Range: bytes=-N`), `rangeStart` remains `-1`,
but it is later used in the 416 check and in the `Content-Range` response
header. This produces an invalid `Content-Range` like `bytes -1-.../size` and
can mask out-of-range handling. Compute an effective start/end for the response
(based on `skipSize`/`bytesToRead`) and use those for validation and
`Content-Range`.
##########
aliyun/src/main/java/org/apache/iceberg/aliyun/oss/OSSInputStream.java:
##########
@@ -82,89 +115,182 @@ public int read() throws IOException {
Preconditions.checkState(!closed, "Cannot read: already closed");
positionStream();
- pos += 1;
- next += 1;
- readBytes.increment();
- readOperations.increment();
+ try {
+ int bytesRead = Failsafe.with(retryPolicy).get(() -> stream.read());
+ if (bytesRead != -1) {
+ pos += 1;
+ next += 1;
+ readBytes.increment();
+ readOperations.increment();
+ }
+
+ return bytesRead;
+ } catch (FailsafeException ex) {
+ if (ex.getCause() instanceof IOException) {
+ throw (IOException) ex.getCause();
+ }
- return stream.read();
+ throw ex;
+ }
}
@Override
public int read(byte[] b, int off, int len) throws IOException {
Preconditions.checkState(!closed, "Cannot read: already closed");
positionStream();
- int bytesRead = stream.read(b, off, len);
- pos += bytesRead;
- next += bytesRead;
- readBytes.increment(bytesRead);
- readOperations.increment();
+ try {
+ int bytesRead = Failsafe.with(retryPolicy).get(() -> stream.read(b, off,
len));
+ if (bytesRead > 0) {
+ pos += bytesRead;
+ next += bytesRead;
+ readBytes.increment(bytesRead);
+ readOperations.increment();
+ }
+
+ return bytesRead;
Review Comment:
`readOperations` is only incremented when `read(byte[], ..)` returns `> 0`.
This means EOF reads (and 0-length reads) won't be counted as operations, which
differs from other Iceberg `SeekableInputStream` implementations. To keep
FileIO metrics comparable, increment `readOperations` unconditionally after a
successful read, while keeping `readBytes`/position updates conditional on
`bytesRead > 0`.
--
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]
