You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Samrat002 (via GitHub)" <gi...@apache.org> on 2026/05/04 19:01:03 UTC
[PR] [FLINK-39603] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Samrat002 opened a new pull request, #28112:
URL: https://github.com/apache/flink/pull/28112
## What is the purpose of the change
`NativeS3InputStream.seek()` unconditionally aborts the HTTP connection and opens a new Range GET for every seek, even for small forward seeks that land inside already-buffered data. During state restore, this creates `O(N)` HTTP round-trips for N state entries. each call costing TCP handshake + TLS + S3 response latency.
This PR aim to optimise an skip all HTTP work is deferred to the next `read()` call. Forward seeks within `max(readBufferSize, bufferedStream.available())` skip in-buffer instead of reopening the connection.
## Brief change log
- `seek()` and `skip()` now only update `nextReadPos` without any I/O
- Added `lazySeek()` to reconcile `nextReadPos` vs `streamPos` on the next `read()` call
- Forward seeks within the dynamic threshold `max(readBufferSize, bufferedStream.available())` skip in-buffer; backward or large forward seeks reopen via Range GET
- Multiple seeks between reads coalesce — only the final position matters
## Verifying this change
This change added tests and can be verified wit Unit Test and E2E working for high state flink application
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: yes
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
##### Was generative AI tooling used to co-author this PR?
- [ ] Yes (please specify the tool below)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "gaborgsomogyi (via GitHub)" <gi...@apache.org>.
gaborgsomogyi commented on PR #28112:
URL: https://github.com/apache/flink/pull/28112#issuecomment-4388127239
Hi @Samrat002, great work reducing IOPS via lazy seek and range requests! While reviewing the implementation I found one remaining efficiency issue worth addressing.
What's working well
A seek that lands within the read-ahead buffer is served entirely from the buffer - no extra `GetObject` call, no bytes pulled from S3. The attached patch includes `seekWithinBuffer_afterSmallRead_doesNotTouchUnderlyingStream` which demonstrates this and passes.
Remaining issue: bulk reads bypass the `BufferedInputStream` buffer
`BufferedInputStream.read(byte[], off, len)` has an internal fast-path: when `len >= bufferSize` it reads directly from the underlying stream, skipping the local buffer entirely. After such a read the buffer is empty.
This means a forward seek that follows a bulk read cannot be satisfied from the buffer and falls through to `bufferedStream.skip()`, which consumes bytes from the live HTTP connection. No new `GetObject` is issued (IOPS are fine), but unnecessary bytes are downloaded and discarded.
The attached patch includes `seekWithinBuffer_afterLargeRead_touchesUnderlyingStream` which demonstrates the bug - it currently fails because `bytesSkippedFromUnderlying` is `10` instead of `0`.
Suggested fix
Replace `BufferedInputStream` with a plain `byte[]` buffer managed directly, tracking `bufferStart` (file offset of the first byte in the buffer), `bufferOffset` (current read position within the array), and `bufferLength` (valid bytes). Then `lazySeek()` can check `[bufferStart, bufferStart + bufferLength)` to decide whether the seek is satisfiable in-buffer, independently of whether the preceding read was a byte-at-a-time or a bulk read.
The patch file adds the `AtomicLong` counters to `TrackingInputStream` and both test methods.
[buffer-efficiency-tests.patch](https://github.com/user-attachments/files/27439474/buffer-efficiency-tests.patch)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603][s3] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "gaborgsomogyi (via GitHub)" <gi...@apache.org>.
gaborgsomogyi commented on code in PR #28112:
URL: https://github.com/apache/flink/pull/28112#discussion_r3217216017
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3InputStream.java:
##########
@@ -86,8 +95,9 @@ public NativeS3InputStream(
this.bucketName = bucketName;
this.key = key;
this.contentLength = contentLength;
- this.readBufferSize = Math.min(readBufferSize, MAX_READ_BUFFER_SIZE);
- this.position = 0;
+ this.readBufferSize = readBufferSize;
Review Comment:
The incoming `readBufferSize` is already capped but we've agreed not to cap, right?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "gaborgsomogyi (via GitHub)" <gi...@apache.org>.
gaborgsomogyi commented on code in PR #28112:
URL: https://github.com/apache/flink/pull/28112#discussion_r3201573763
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3InputStreamTest.java:
##########
@@ -333,15 +468,116 @@ void skipToEofWithOpenStreamReleasesStream() throws Exception {
assertThat(in.skip(DATA.length)).isEqualTo(DATA.length - 1);
assertThat(in.getPos()).isEqualTo(DATA.length);
+ assertThat(first.wasAborted()).isFalse();
- assertThat(first.wasAborted()).isTrue();
- assertThat(first.wasClosed()).isTrue();
- assertThat(first.wasAbortedBeforeClose()).isTrue();
+ assertThat(in.read()).isEqualTo(-1);
+ }
+ }
+
+ @Test
+ void skipZeroAndNegativeAreNoOps() throws Exception {
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in = new NativeS3InputStream(client, BUCKET, KEY, DATA.length)) {
+ in.read();
+ assertThat(in.skip(0)).isZero();
+ assertThat(in.skip(-5)).isZero();
assertThat(client.getObjectCalls()).isEqualTo(1);
+ }
+ }
+
+ // --- read + seek integration ---
+
+ @Test
+ void readAndSeekReturnCorrectData() throws Exception {
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in = new NativeS3InputStream(client, BUCKET, KEY, DATA.length)) {
+ assertThat(in.read()).isEqualTo(0);
+ assertThat(in.getPos()).isEqualTo(1);
+ byte[] buf = new byte[10];
+ assertThat(in.read(buf, 0, 10)).isEqualTo(10);
+ assertThat(in.getPos()).isEqualTo(11);
+ for (int i = 0; i < 10; i++) {
+ assertThat(buf[i]).isEqualTo(DATA[i + 1]);
+ }
+ assertThat(in.available()).isEqualTo(DATA.length - 11);
+ in.seek(200);
+ assertThat(in.read()).isEqualTo(200);
+ assertThat(in.getPos()).isEqualTo(201);
+ in.seek(250);
+ byte[] tail = new byte[20];
+ assertThat(in.read(tail, 0, 20)).isEqualTo(6);
+ assertThat(in.getPos()).isEqualTo(256);
assertThat(in.read()).isEqualTo(-1);
+ assertThat(in.read(new byte[1], 0, 1)).isEqualTo(-1);
+ }
+ }
+
+ // --- buffer-efficiency: skip stays in local buffer vs. underlying stream ---
+
+ @Test
+ void seekWithinBuffer_afterSmallRead_doesNotTouchUnderlyingStream() throws Exception {
+ int smallBuffer = 32;
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in =
+ new NativeS3InputStream(client, BUCKET, KEY, DATA.length, smallBuffer)) {
+ // single-byte read fills the local buffer (up to smallBuffer bytes buffered)
+ in.read();
+ TrackingInputStream underlying = client.lastStream();
+
+ // forward seek of 9 bytes — fits in the local buffer, no underlying access needed
+ in.seek(10);
+ in.read();
+
+ assertThat(client.getObjectCalls()).isEqualTo(1);
+ assertThat(underlying.bytesSkippedFromUnderlying()).isEqualTo(0);
Review Comment:
I mean `bytesReadFromUnderlying` is the key part here. Cool that we checked skip but another aspect that we've read exactly the expected amount of data from wrapper stream
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "gaborgsomogyi (via GitHub)" <gi...@apache.org>.
gaborgsomogyi commented on code in PR #28112:
URL: https://github.com/apache/flink/pull/28112#discussion_r3199962035
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3InputStreamTest.java:
##########
@@ -57,9 +58,16 @@ private static class TrackingInputStream extends InputStream implements Abortabl
private final AtomicBoolean aborted = new AtomicBoolean();
private final AtomicBoolean closed = new AtomicBoolean();
private volatile boolean abortedBeforeClose;
+ private final int maxAvailable;
+ private final AtomicLong bytesSkipped = new AtomicLong();
TrackingInputStream(byte[] data, int offset) {
+ this(data, offset, Integer.MAX_VALUE);
+ }
+
+ TrackingInputStream(byte[] data, int offset, int maxAvailable) {
this.delegate = new ByteArrayInputStream(data, offset, data.length - offset);
+ this.maxAvailable = maxAvailable;
}
TrackingInputStream(byte[] data) {
Review Comment:
This is unused
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3InputStreamTest.java:
##########
@@ -333,15 +468,116 @@ void skipToEofWithOpenStreamReleasesStream() throws Exception {
assertThat(in.skip(DATA.length)).isEqualTo(DATA.length - 1);
assertThat(in.getPos()).isEqualTo(DATA.length);
+ assertThat(first.wasAborted()).isFalse();
- assertThat(first.wasAborted()).isTrue();
- assertThat(first.wasClosed()).isTrue();
- assertThat(first.wasAbortedBeforeClose()).isTrue();
+ assertThat(in.read()).isEqualTo(-1);
+ }
+ }
+
+ @Test
+ void skipZeroAndNegativeAreNoOps() throws Exception {
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in = new NativeS3InputStream(client, BUCKET, KEY, DATA.length)) {
+ in.read();
+ assertThat(in.skip(0)).isZero();
+ assertThat(in.skip(-5)).isZero();
assertThat(client.getObjectCalls()).isEqualTo(1);
+ }
+ }
+
+ // --- read + seek integration ---
+
+ @Test
+ void readAndSeekReturnCorrectData() throws Exception {
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in = new NativeS3InputStream(client, BUCKET, KEY, DATA.length)) {
+ assertThat(in.read()).isEqualTo(0);
+ assertThat(in.getPos()).isEqualTo(1);
+ byte[] buf = new byte[10];
+ assertThat(in.read(buf, 0, 10)).isEqualTo(10);
+ assertThat(in.getPos()).isEqualTo(11);
+ for (int i = 0; i < 10; i++) {
+ assertThat(buf[i]).isEqualTo(DATA[i + 1]);
+ }
+ assertThat(in.available()).isEqualTo(DATA.length - 11);
+ in.seek(200);
+ assertThat(in.read()).isEqualTo(200);
+ assertThat(in.getPos()).isEqualTo(201);
+ in.seek(250);
+ byte[] tail = new byte[20];
+ assertThat(in.read(tail, 0, 20)).isEqualTo(6);
+ assertThat(in.getPos()).isEqualTo(256);
assertThat(in.read()).isEqualTo(-1);
+ assertThat(in.read(new byte[1], 0, 1)).isEqualTo(-1);
+ }
+ }
+
+ // --- buffer-efficiency: skip stays in local buffer vs. underlying stream ---
+
+ @Test
+ void seekWithinBuffer_afterSmallRead_doesNotTouchUnderlyingStream() throws Exception {
+ int smallBuffer = 32;
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in =
+ new NativeS3InputStream(client, BUCKET, KEY, DATA.length, smallBuffer)) {
+ // single-byte read fills the local buffer (up to smallBuffer bytes buffered)
+ in.read();
+ TrackingInputStream underlying = client.lastStream();
+
+ // forward seek of 9 bytes — fits in the local buffer, no underlying access needed
+ in.seek(10);
+ in.read();
+
+ assertThat(client.getObjectCalls()).isEqualTo(1);
+ assertThat(underlying.bytesSkippedFromUnderlying()).isEqualTo(0);
+ }
+ }
+
+ @Test
+ void seekWithinBuffer_afterLargeRead_touchesUnderlyingStream() throws Exception {
+ int smallBuffer = 16;
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in =
+ new NativeS3InputStream(client, BUCKET, KEY, DATA.length, smallBuffer)) {
+ // bulk read >= bufferSize bypasses BufferedInputStream's local array entirely,
+ // leaving the local buffer empty afterward
+ byte[] buf = new byte[smallBuffer];
+ in.read(buf, 0, smallBuffer);
+ TrackingInputStream underlying = client.lastStream();
+
+ // forward seek of 10 bytes — within readBufferSize but buffer is empty, so
+ // BufferedInputStream.skip() delegates directly to the underlying stream
+ in.seek((long) smallBuffer + 10);
+ in.read();
+
+ assertThat(client.getObjectCalls()).isEqualTo(1);
+ assertThat(underlying.bytesSkippedFromUnderlying()).isEqualTo(10);
}
}
+ @Test
+ void seekBeyondReadBufferSize_inflatedByUnderlyingAvailable_reopensWithRangeRequest()
+ throws Exception {
+ int smallBuffer = 16;
+ // default maxAvailable = Integer.MAX_VALUE: underlying stream reports a large available(),
+ // which used to inflate forwardSeekLimit and incorrectly suppress range-request reopens
Review Comment:
This will tell almost nothing to devs who is not read this discussion so I think we can remove it.
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3InputStream.java:
##########
@@ -98,39 +110,60 @@ public NativeS3InputStream(
this.readBufferSize / 1024);
}
+ /** Reconciles {@link #nextReadPos} and {@link #streamPos} before reading bytes. */
@GuardedBy("lock")
- private void lazyInitialize() throws IOException {
- assert lock.isHeldByCurrentThread() : "lazyInitialize() requires lock to be held";
- if (currentStream == null && !closed) {
- openStreamAtCurrentPosition();
- }
- }
+ private void lazySeek() throws IOException {
+ assert lock.isHeldByCurrentThread() : "lazySeek() requires lock to be held";
+ long targetPos = nextReadPos;
- /** At EOF, release instead of reopening: {@code bytes=contentLength-} returns S3 416. */
- @GuardedBy("lock")
- private void repositionOpenStream() throws IOException {
- assert lock.isHeldByCurrentThread() : "repositionOpenStream() requires lock to be held";
if (currentStream == null) {
+ streamPos = targetPos;
return;
}
- if (position >= contentLength) {
+
+ if (targetPos == streamPos) {
+ return;
+ }
+
+ long diff = targetPos - streamPos;
+ streamPos = targetPos;
+
+ if (targetPos >= contentLength) {
releaseStreams();
- } else {
+ return;
+ }
+
+ if (diff > 0 && diff <= (long) readBufferSize) {
+ skipBytesInBuffer(diff);
+ return;
+ }
+
+ openStreamAtCurrentPosition();
+ }
+
+ @GuardedBy("lock")
+ private void ensureStreamOpen() throws IOException {
+ assert lock.isHeldByCurrentThread() : "ensureStreamOpen() requires lock to be held";
+ if (currentStream == null && !closed) {
openStreamAtCurrentPosition();
}
}
- /**
- * Opens (or reopens) the S3 stream at the current position.
- *
- * <p>This method:
- *
- * <ul>
- * <li>Closes any existing stream
- * <li>Opens a new stream starting at {@link #position}
- * <li>Uses HTTP range requests for non-zero positions
- * </ul>
- */
+ @GuardedBy("lock")
+ private void skipBytesInBuffer(long n) throws IOException {
+ assert lock.isHeldByCurrentThread() : "skipBytesInBuffer() requires lock to be held";
+ long remaining = n;
+ while (remaining > 0) {
+ long skipped = bufferedStream.skip(remaining);
+ if (skipped <= 0) {
+ openStreamAtCurrentPosition();
Review Comment:
`skipBytesInBuffer` has an untested recovery path: if `bufferedStream.skip()` returns <= 0 mid-skip (underlying HTTP connection stalled or closed), it falls back to `openStreamAtCurrentPosition()`. This is correct - streamPos is already set to targetPos before skipBytesInBuffer is called, so the fresh range request lands at the right position - but it is never exercised by the current test suite.
A targeted test would extend `TrackingInputStream` to `return 0` from `skip()` on the first call, then delegate normally. The setup: do a small read (fills buffer), seek forward within `readBufferSize`, trigger the skip failure, then assert the correct byte is returned and `getObjectCalls() == 2` (new range request was issued at the seek target).
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3InputStream.java:
##########
@@ -98,39 +110,60 @@ public NativeS3InputStream(
this.readBufferSize / 1024);
}
+ /** Reconciles {@link #nextReadPos} and {@link #streamPos} before reading bytes. */
@GuardedBy("lock")
- private void lazyInitialize() throws IOException {
- assert lock.isHeldByCurrentThread() : "lazyInitialize() requires lock to be held";
- if (currentStream == null && !closed) {
- openStreamAtCurrentPosition();
- }
- }
+ private void lazySeek() throws IOException {
+ assert lock.isHeldByCurrentThread() : "lazySeek() requires lock to be held";
+ long targetPos = nextReadPos;
- /** At EOF, release instead of reopening: {@code bytes=contentLength-} returns S3 416. */
- @GuardedBy("lock")
- private void repositionOpenStream() throws IOException {
- assert lock.isHeldByCurrentThread() : "repositionOpenStream() requires lock to be held";
if (currentStream == null) {
+ streamPos = targetPos;
return;
}
- if (position >= contentLength) {
+
+ if (targetPos == streamPos) {
+ return;
+ }
+
+ long diff = targetPos - streamPos;
+ streamPos = targetPos;
+
+ if (targetPos >= contentLength) {
releaseStreams();
- } else {
+ return;
+ }
+
+ if (diff > 0 && diff <= (long) readBufferSize) {
Review Comment:
I would put comment here explaining that we can't really find out the data size sitting inside the buffer without heavy or shaky development effort(extending JDK class) so the tradeoff is to accept max buffer size amount of skips during seeks.
As a side note which we don't need to mention as comment. If that's not performing well we can add a new config which controls the seek behavior, like `s3.max-forward-seek-bytes` or something similar.
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3InputStreamTest.java:
##########
@@ -333,15 +468,116 @@ void skipToEofWithOpenStreamReleasesStream() throws Exception {
assertThat(in.skip(DATA.length)).isEqualTo(DATA.length - 1);
assertThat(in.getPos()).isEqualTo(DATA.length);
+ assertThat(first.wasAborted()).isFalse();
- assertThat(first.wasAborted()).isTrue();
- assertThat(first.wasClosed()).isTrue();
- assertThat(first.wasAbortedBeforeClose()).isTrue();
+ assertThat(in.read()).isEqualTo(-1);
+ }
+ }
+
+ @Test
+ void skipZeroAndNegativeAreNoOps() throws Exception {
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in = new NativeS3InputStream(client, BUCKET, KEY, DATA.length)) {
+ in.read();
+ assertThat(in.skip(0)).isZero();
+ assertThat(in.skip(-5)).isZero();
assertThat(client.getObjectCalls()).isEqualTo(1);
+ }
+ }
+
+ // --- read + seek integration ---
+
+ @Test
+ void readAndSeekReturnCorrectData() throws Exception {
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in = new NativeS3InputStream(client, BUCKET, KEY, DATA.length)) {
+ assertThat(in.read()).isEqualTo(0);
+ assertThat(in.getPos()).isEqualTo(1);
+ byte[] buf = new byte[10];
+ assertThat(in.read(buf, 0, 10)).isEqualTo(10);
+ assertThat(in.getPos()).isEqualTo(11);
+ for (int i = 0; i < 10; i++) {
+ assertThat(buf[i]).isEqualTo(DATA[i + 1]);
+ }
+ assertThat(in.available()).isEqualTo(DATA.length - 11);
+ in.seek(200);
+ assertThat(in.read()).isEqualTo(200);
+ assertThat(in.getPos()).isEqualTo(201);
+ in.seek(250);
+ byte[] tail = new byte[20];
+ assertThat(in.read(tail, 0, 20)).isEqualTo(6);
+ assertThat(in.getPos()).isEqualTo(256);
assertThat(in.read()).isEqualTo(-1);
+ assertThat(in.read(new byte[1], 0, 1)).isEqualTo(-1);
+ }
+ }
+
+ // --- buffer-efficiency: skip stays in local buffer vs. underlying stream ---
+
+ @Test
+ void seekWithinBuffer_afterSmallRead_doesNotTouchUnderlyingStream() throws Exception {
+ int smallBuffer = 32;
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in =
+ new NativeS3InputStream(client, BUCKET, KEY, DATA.length, smallBuffer)) {
+ // single-byte read fills the local buffer (up to smallBuffer bytes buffered)
+ in.read();
Review Comment:
I think for correctness we should assert on read values in all tests (we read from proper place and proper values). Additionally this comment is promising that the buffer is filled up which is not asserted.
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3InputStreamTest.java:
##########
@@ -333,15 +468,116 @@ void skipToEofWithOpenStreamReleasesStream() throws Exception {
assertThat(in.skip(DATA.length)).isEqualTo(DATA.length - 1);
assertThat(in.getPos()).isEqualTo(DATA.length);
+ assertThat(first.wasAborted()).isFalse();
- assertThat(first.wasAborted()).isTrue();
- assertThat(first.wasClosed()).isTrue();
- assertThat(first.wasAbortedBeforeClose()).isTrue();
+ assertThat(in.read()).isEqualTo(-1);
+ }
+ }
+
+ @Test
+ void skipZeroAndNegativeAreNoOps() throws Exception {
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in = new NativeS3InputStream(client, BUCKET, KEY, DATA.length)) {
+ in.read();
+ assertThat(in.skip(0)).isZero();
+ assertThat(in.skip(-5)).isZero();
assertThat(client.getObjectCalls()).isEqualTo(1);
+ }
+ }
+
+ // --- read + seek integration ---
+
+ @Test
+ void readAndSeekReturnCorrectData() throws Exception {
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in = new NativeS3InputStream(client, BUCKET, KEY, DATA.length)) {
+ assertThat(in.read()).isEqualTo(0);
+ assertThat(in.getPos()).isEqualTo(1);
+ byte[] buf = new byte[10];
+ assertThat(in.read(buf, 0, 10)).isEqualTo(10);
+ assertThat(in.getPos()).isEqualTo(11);
+ for (int i = 0; i < 10; i++) {
+ assertThat(buf[i]).isEqualTo(DATA[i + 1]);
+ }
+ assertThat(in.available()).isEqualTo(DATA.length - 11);
+ in.seek(200);
+ assertThat(in.read()).isEqualTo(200);
+ assertThat(in.getPos()).isEqualTo(201);
+ in.seek(250);
+ byte[] tail = new byte[20];
+ assertThat(in.read(tail, 0, 20)).isEqualTo(6);
+ assertThat(in.getPos()).isEqualTo(256);
assertThat(in.read()).isEqualTo(-1);
+ assertThat(in.read(new byte[1], 0, 1)).isEqualTo(-1);
+ }
+ }
+
+ // --- buffer-efficiency: skip stays in local buffer vs. underlying stream ---
+
+ @Test
+ void seekWithinBuffer_afterSmallRead_doesNotTouchUnderlyingStream() throws Exception {
+ int smallBuffer = 32;
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in =
+ new NativeS3InputStream(client, BUCKET, KEY, DATA.length, smallBuffer)) {
+ // single-byte read fills the local buffer (up to smallBuffer bytes buffered)
+ in.read();
+ TrackingInputStream underlying = client.lastStream();
+
+ // forward seek of 9 bytes — fits in the local buffer, no underlying access needed
+ in.seek(10);
+ in.read();
+
+ assertThat(client.getObjectCalls()).isEqualTo(1);
+ assertThat(underlying.bytesSkippedFromUnderlying()).isEqualTo(0);
Review Comment:
Here it's not checked that more data is not read.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603][s3] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "gaborgsomogyi (via GitHub)" <gi...@apache.org>.
gaborgsomogyi merged PR #28112:
URL: https://github.com/apache/flink/pull/28112
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on code in PR #28112:
URL: https://github.com/apache/flink/pull/28112#discussion_r3201406435
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3InputStreamTest.java:
##########
@@ -333,15 +468,116 @@ void skipToEofWithOpenStreamReleasesStream() throws Exception {
assertThat(in.skip(DATA.length)).isEqualTo(DATA.length - 1);
assertThat(in.getPos()).isEqualTo(DATA.length);
+ assertThat(first.wasAborted()).isFalse();
- assertThat(first.wasAborted()).isTrue();
- assertThat(first.wasClosed()).isTrue();
- assertThat(first.wasAbortedBeforeClose()).isTrue();
+ assertThat(in.read()).isEqualTo(-1);
+ }
+ }
+
+ @Test
+ void skipZeroAndNegativeAreNoOps() throws Exception {
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in = new NativeS3InputStream(client, BUCKET, KEY, DATA.length)) {
+ in.read();
+ assertThat(in.skip(0)).isZero();
+ assertThat(in.skip(-5)).isZero();
assertThat(client.getObjectCalls()).isEqualTo(1);
+ }
+ }
+
+ // --- read + seek integration ---
+
+ @Test
+ void readAndSeekReturnCorrectData() throws Exception {
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in = new NativeS3InputStream(client, BUCKET, KEY, DATA.length)) {
+ assertThat(in.read()).isEqualTo(0);
+ assertThat(in.getPos()).isEqualTo(1);
+ byte[] buf = new byte[10];
+ assertThat(in.read(buf, 0, 10)).isEqualTo(10);
+ assertThat(in.getPos()).isEqualTo(11);
+ for (int i = 0; i < 10; i++) {
+ assertThat(buf[i]).isEqualTo(DATA[i + 1]);
+ }
+ assertThat(in.available()).isEqualTo(DATA.length - 11);
+ in.seek(200);
+ assertThat(in.read()).isEqualTo(200);
+ assertThat(in.getPos()).isEqualTo(201);
+ in.seek(250);
+ byte[] tail = new byte[20];
+ assertThat(in.read(tail, 0, 20)).isEqualTo(6);
+ assertThat(in.getPos()).isEqualTo(256);
assertThat(in.read()).isEqualTo(-1);
+ assertThat(in.read(new byte[1], 0, 1)).isEqualTo(-1);
+ }
+ }
+
+ // --- buffer-efficiency: skip stays in local buffer vs. underlying stream ---
+
+ @Test
+ void seekWithinBuffer_afterSmallRead_doesNotTouchUnderlyingStream() throws Exception {
+ int smallBuffer = 32;
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in =
+ new NativeS3InputStream(client, BUCKET, KEY, DATA.length, smallBuffer)) {
+ // single-byte read fills the local buffer (up to smallBuffer bytes buffered)
+ in.read();
+ TrackingInputStream underlying = client.lastStream();
+
+ // forward seek of 9 bytes — fits in the local buffer, no underlying access needed
+ in.seek(10);
+ in.read();
+
+ assertThat(client.getObjectCalls()).isEqualTo(1);
+ assertThat(underlying.bytesSkippedFromUnderlying()).isEqualTo(0);
Review Comment:
Added
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "gaborgsomogyi (via GitHub)" <gi...@apache.org>.
gaborgsomogyi commented on PR #28112:
URL: https://github.com/apache/flink/pull/28112#issuecomment-4390447597
Hi @Samrat002, great work on this PR - the lazy seek and range request approach is a solid foundation. While reviewing I found one correctness issue in `lazySeek()` worth fixing.
What works well
A forward seek that lands within the read-ahead buffer is served entirely from the local buffer with zero underlying stream access. The attached patch includes `seekWithinBuffer_afterSmallRead_doesNotTouchUnderlyingStream` which demonstrates and passes this case.
Known trade-off: bulk reads bypass the BufferedInputStream buffer
`BufferedInputStream.read(byte[], off, len)` has an internal fast-path: when `len >= bufferSize` it reads directly from the underlying stream, bypassing the local array entirely. After such a read the buffer is empty, so a small forward seek costs exactly the seek distance in underlying stream bytes. `seekWithinBuffer_afterLargeRead_touchesUnderlyingStream` in the patch documents and passes this, asserting `bytesSkippedFromUnderlying == 10`. This is an acceptable trade-off - bulk reads perform well for large sequential access.
Actual bug: `forwardSeekLimit` is unbounded
`long forwardSeekLimit = Math.max((long) readBufferSize, (long) bufferedStream.available());`
`bufferedStream.available()` returns `(localBufferBytes) + in.available()`, where `in` is the live `ResponseInputStream`. If the underlying HTTP client reports a large `available()` - in the worst case the full remaining content length - `forwardSeekLimit` grows proportionally to the file size. A seek that exceeds readBufferSize but falls under this inflated limit triggers `skipBytesInBuffer()` instead of reopening with a range request, downloading and discarding an unbounded number of bytes from the live HTTP connection.
`seekBeyondReadBufferSize_inflatedByUnderlyingAvailable_reopensWithRangeRequest` in the patch demonstrates this - it currently fails because `getObjectCalls()` is 1 instead of 2 and `bytesSkippedFromUnderlying` is non-zero.
[buffer-efficiency-tests.patch](https://github.com/user-attachments/files/27450058/buffer-efficiency-tests.patch)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on code in PR #28112:
URL: https://github.com/apache/flink/pull/28112#discussion_r3200878247
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3InputStreamTest.java:
##########
@@ -57,9 +58,16 @@ private static class TrackingInputStream extends InputStream implements Abortabl
private final AtomicBoolean aborted = new AtomicBoolean();
private final AtomicBoolean closed = new AtomicBoolean();
private volatile boolean abortedBeforeClose;
+ private final int maxAvailable;
+ private final AtomicLong bytesSkipped = new AtomicLong();
TrackingInputStream(byte[] data, int offset) {
+ this(data, offset, Integer.MAX_VALUE);
+ }
+
+ TrackingInputStream(byte[] data, int offset, int maxAvailable) {
this.delegate = new ByteArrayInputStream(data, offset, data.length - offset);
+ this.maxAvailable = maxAvailable;
}
TrackingInputStream(byte[] data) {
Review Comment:
removed
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on PR #28112:
URL: https://github.com/apache/flink/pull/28112#issuecomment-4373699935
@gaborgsomogyi PTAL
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603][s3] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on code in PR #28112:
URL: https://github.com/apache/flink/pull/28112#discussion_r3217523748
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3InputStream.java:
##########
@@ -86,8 +95,9 @@ public NativeS3InputStream(
this.bucketName = bucketName;
this.key = key;
this.contentLength = contentLength;
- this.readBufferSize = Math.min(readBufferSize, MAX_READ_BUFFER_SIZE);
- this.position = 0;
+ this.readBufferSize = readBufferSize;
Review Comment:
My bad. I missed it completely. What i changed was half of the code changes.
i have updated with right changes now. PTAL
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #28112:
URL: https://github.com/apache/flink/pull/28112#issuecomment-4373771794
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c3d49db458ad51a557590d204fe5e7a6df91c7ff",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c3d49db458ad51a557590d204fe5e7a6df91c7ff",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c3d49db458ad51a557590d204fe5e7a6df91c7ff UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on code in PR #28112:
URL: https://github.com/apache/flink/pull/28112#discussion_r3200879849
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3InputStreamTest.java:
##########
@@ -333,15 +468,116 @@ void skipToEofWithOpenStreamReleasesStream() throws Exception {
assertThat(in.skip(DATA.length)).isEqualTo(DATA.length - 1);
assertThat(in.getPos()).isEqualTo(DATA.length);
+ assertThat(first.wasAborted()).isFalse();
- assertThat(first.wasAborted()).isTrue();
- assertThat(first.wasClosed()).isTrue();
- assertThat(first.wasAbortedBeforeClose()).isTrue();
+ assertThat(in.read()).isEqualTo(-1);
+ }
+ }
+
+ @Test
+ void skipZeroAndNegativeAreNoOps() throws Exception {
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in = new NativeS3InputStream(client, BUCKET, KEY, DATA.length)) {
+ in.read();
+ assertThat(in.skip(0)).isZero();
+ assertThat(in.skip(-5)).isZero();
assertThat(client.getObjectCalls()).isEqualTo(1);
+ }
+ }
+
+ // --- read + seek integration ---
+
+ @Test
+ void readAndSeekReturnCorrectData() throws Exception {
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in = new NativeS3InputStream(client, BUCKET, KEY, DATA.length)) {
+ assertThat(in.read()).isEqualTo(0);
+ assertThat(in.getPos()).isEqualTo(1);
+ byte[] buf = new byte[10];
+ assertThat(in.read(buf, 0, 10)).isEqualTo(10);
+ assertThat(in.getPos()).isEqualTo(11);
+ for (int i = 0; i < 10; i++) {
+ assertThat(buf[i]).isEqualTo(DATA[i + 1]);
+ }
+ assertThat(in.available()).isEqualTo(DATA.length - 11);
+ in.seek(200);
+ assertThat(in.read()).isEqualTo(200);
+ assertThat(in.getPos()).isEqualTo(201);
+ in.seek(250);
+ byte[] tail = new byte[20];
+ assertThat(in.read(tail, 0, 20)).isEqualTo(6);
+ assertThat(in.getPos()).isEqualTo(256);
assertThat(in.read()).isEqualTo(-1);
+ assertThat(in.read(new byte[1], 0, 1)).isEqualTo(-1);
+ }
+ }
+
+ // --- buffer-efficiency: skip stays in local buffer vs. underlying stream ---
+
+ @Test
+ void seekWithinBuffer_afterSmallRead_doesNotTouchUnderlyingStream() throws Exception {
+ int smallBuffer = 32;
+ StubS3Client client = new StubS3Client(DATA);
+ try (NativeS3InputStream in =
+ new NativeS3InputStream(client, BUCKET, KEY, DATA.length, smallBuffer)) {
+ // single-byte read fills the local buffer (up to smallBuffer bytes buffered)
+ in.read();
Review Comment:
made changes PTAL
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3InputStream.java:
##########
@@ -98,39 +110,60 @@ public NativeS3InputStream(
this.readBufferSize / 1024);
}
+ /** Reconciles {@link #nextReadPos} and {@link #streamPos} before reading bytes. */
@GuardedBy("lock")
- private void lazyInitialize() throws IOException {
- assert lock.isHeldByCurrentThread() : "lazyInitialize() requires lock to be held";
- if (currentStream == null && !closed) {
- openStreamAtCurrentPosition();
- }
- }
+ private void lazySeek() throws IOException {
+ assert lock.isHeldByCurrentThread() : "lazySeek() requires lock to be held";
+ long targetPos = nextReadPos;
- /** At EOF, release instead of reopening: {@code bytes=contentLength-} returns S3 416. */
- @GuardedBy("lock")
- private void repositionOpenStream() throws IOException {
- assert lock.isHeldByCurrentThread() : "repositionOpenStream() requires lock to be held";
if (currentStream == null) {
+ streamPos = targetPos;
return;
}
- if (position >= contentLength) {
+
+ if (targetPos == streamPos) {
+ return;
+ }
+
+ long diff = targetPos - streamPos;
+ streamPos = targetPos;
+
+ if (targetPos >= contentLength) {
releaseStreams();
- } else {
+ return;
+ }
+
+ if (diff > 0 && diff <= (long) readBufferSize) {
+ skipBytesInBuffer(diff);
+ return;
+ }
+
+ openStreamAtCurrentPosition();
+ }
+
+ @GuardedBy("lock")
+ private void ensureStreamOpen() throws IOException {
+ assert lock.isHeldByCurrentThread() : "ensureStreamOpen() requires lock to be held";
+ if (currentStream == null && !closed) {
openStreamAtCurrentPosition();
}
}
- /**
- * Opens (or reopens) the S3 stream at the current position.
- *
- * <p>This method:
- *
- * <ul>
- * <li>Closes any existing stream
- * <li>Opens a new stream starting at {@link #position}
- * <li>Uses HTTP range requests for non-zero positions
- * </ul>
- */
+ @GuardedBy("lock")
+ private void skipBytesInBuffer(long n) throws IOException {
+ assert lock.isHeldByCurrentThread() : "skipBytesInBuffer() requires lock to be held";
+ long remaining = n;
+ while (remaining > 0) {
+ long skipped = bufferedStream.skip(remaining);
+ if (skipped <= 0) {
+ openStreamAtCurrentPosition();
Review Comment:
added one more extra test to validate it. PTAL
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on PR #28112:
URL: https://github.com/apache/flink/pull/28112#issuecomment-4396484742
@gaborgsomogyi . I have addressed to review comments. PTAL
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on code in PR #28112:
URL: https://github.com/apache/flink/pull/28112#discussion_r3201408323
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3InputStream.java:
##########
@@ -98,39 +110,60 @@ public NativeS3InputStream(
this.readBufferSize / 1024);
}
+ /** Reconciles {@link #nextReadPos} and {@link #streamPos} before reading bytes. */
@GuardedBy("lock")
- private void lazyInitialize() throws IOException {
- assert lock.isHeldByCurrentThread() : "lazyInitialize() requires lock to be held";
- if (currentStream == null && !closed) {
- openStreamAtCurrentPosition();
- }
- }
+ private void lazySeek() throws IOException {
+ assert lock.isHeldByCurrentThread() : "lazySeek() requires lock to be held";
+ long targetPos = nextReadPos;
- /** At EOF, release instead of reopening: {@code bytes=contentLength-} returns S3 416. */
- @GuardedBy("lock")
- private void repositionOpenStream() throws IOException {
- assert lock.isHeldByCurrentThread() : "repositionOpenStream() requires lock to be held";
if (currentStream == null) {
+ streamPos = targetPos;
return;
}
- if (position >= contentLength) {
+
+ if (targetPos == streamPos) {
+ return;
+ }
+
+ long diff = targetPos - streamPos;
+ streamPos = targetPos;
+
+ if (targetPos >= contentLength) {
releaseStreams();
- } else {
+ return;
+ }
+
+ if (diff > 0 && diff <= (long) readBufferSize) {
Review Comment:
added now . sorry for the miss 😄
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39603] Optimize NativeS3InputStream via seeks and reduce IOPS [flink]
Posted by "gaborgsomogyi (via GitHub)" <gi...@apache.org>.
gaborgsomogyi commented on PR #28112:
URL: https://github.com/apache/flink/pull/28112#issuecomment-4386229075
I think we need an explicit config for buffer size. There are use-cases where we need to set 16MB buffer for proper IOPS usage.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org