DanielLeens commented on PR #10931:
URL: https://github.com/apache/seatunnel/pull/10931#issuecomment-4525096374
Thanks for the update. I re-reviewed the latest head
`c3d585c95e04631fd17c5981751dd186e30e9e7d` locally against `upstream/dev`, and
I traced the real StarRocks / Doris call paths again. I did not run Maven
locally in this pass; this is a source-level review only.
# What problem this PR solves
- User pain point
The shared Arrow row-batch reader sits on the StarRocks and Doris source
path. If Arrow-side resources stay alive longer than necessary, long-running
reads can accumulate memory pressure and eventually surface as Arrow-related
memory exceptions.
- Fix approach
This revision keeps `ByteArrayInputStream` as an explicit field and closes
`ArrowStreamReader`, `ByteArrayInputStream`, and `RootAllocator` in `close()`,
so the Arrow reader side is released immediately after `readArrow()`.
- One sentence
The direction is good, but the latest head still keeps the raw Arrow
payload reachable through the row-batch object lifetime, so the memory-fix
story is not complete yet.
# Runtime chain I checked
```text
StarRocks source
-> StarRocksBeReadClient.hasNext()
[connector-starrocks/.../StarRocksBeReadClient.java:146-172]
-> previous rowBatch.close() [147-149]
-> new ArrowToSeatunnelRowReader(result.getRows(),
seaTunnelRowType).readArrow() [170-172]
-> caller keeps consuming the same rowBatch object
Doris async path
-> DorisValueReader background fetch loop
[connector-doris/.../DorisValueReader.java:156-170]
-> new ArrowToSeatunnelRowReader(nextResult.getRows(),
seaTunnelRowType).readArrow() [161-165]
-> rowBatch.close() [166-167]
-> rowBatchBlockingQueue.put(rowBatch) [168-169]
-> consumer thread later drains the queued rowBatch
Doris sync path
-> DorisValueReader.hasNext() [226-241]
-> previous rowBatch.close() [228-230]
-> read next Arrow batch into ArrowToSeatunnelRowReader [238-241]
```
# Core logic review
Before this PR, the input stream was created inline:
```java
private ArrowStreamReader arrowStreamReader;
private RootAllocator rootAllocator;
private void initArrowReader(byte[] byteArray) {
this.rootAllocator = new RootAllocator(Integer.MAX_VALUE);
this.arrowStreamReader =
new ArrowStreamReader(new ByteArrayInputStream(byteArray),
rootAllocator);
}
```
Now the latest head stores the input stream on the object and closes it in
`close()`:
```java
private ArrowStreamReader arrowStreamReader;
private ByteArrayInputStream byteArrayInputStream;
private RootAllocator rootAllocator;
private void initArrowReader(byte[] byteArray) {
this.rootAllocator = new RootAllocator(Integer.MAX_VALUE);
this.byteArrayInputStream = new ByteArrayInputStream(byteArray);
this.arrowStreamReader = new
ArrowStreamReader(this.byteArrayInputStream, rootAllocator);
}
@Override
public void close() {
if (arrowStreamReader != null) {
arrowStreamReader.close();
root = null;
}
if (byteArrayInputStream != null) {
byteArrayInputStream.close();
}
if (rootAllocator != null) {
rootAllocator.close();
}
}
```
Key findings:
1. The normal runtime path definitely hits this change in both StarRocks and
Doris.
2. The PR does improve the Arrow reader / allocator cleanup timing.
3. The latest head still keeps the raw Arrow payload reachable through the
row-batch object.
4. Doris async mode is still the most memory-sensitive path because the
closed rowBatch object is queued and consumed later.
# Findings
Issue 1: the row batch still retains the raw Arrow payload, so the main-path
memory fix is not complete
- Location:
`seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/reader/ArrowToSeatunnelRowReader.java:64`
- Problem description:
This class is on the shared Arrow deserialize path for both StarRocks and
Doris. The latest head stores `ByteArrayInputStream` as an instance field, and
`close()` never clears that field. Looking at the real callers, they continue
to hold and consume the `ArrowToSeatunnelRowReader` instance after
`readArrow()` returns. So even though `ArrowStreamReader` and `RootAllocator`
are closed earlier, the raw Arrow `byte[]` is still reachable through the batch
object lifetime.
- Potential risk:
On the exact memory-sensitive path this PR is trying to fix, the process
can still keep two copies of the same batch alive at once: the converted
`SeaTunnelRow` batch and the original Arrow payload. Doris async mode makes
this worse because closed row-batch objects are still queued cross-thread.
- Best improvement suggestion:
Option A: keep `ByteArrayInputStream` and `ArrowStreamReader`
method-local, and retain only the final `seatunnelRowBatch`.
Option B: if you want to keep the current structure, then `close()` needs
to clear all Arrow-side helper references explicitly, including the input
stream field, so the raw payload becomes unreachable after conversion.
- Severity: High
- Already raised by others: No
Issue 2: there is still no regression test for the new lifecycle contract of
this shared reader
- Location:
`seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/ArrowToSeatunnelRowReaderTest.java:319`
- Problem description:
This PR changes lifecycle behavior in a connector-common reader shared by
multiple sources, but there is still no regression test that locks down the new
contract after `readArrow()` triggers `close()`.
- Potential risk:
A later cleanup change can easily re-break this path without CI catching
it.
- Best improvement suggestion:
Please add at least one regression test that reflects the real caller
contract: read the batch, confirm rows are still consumable, and verify
Arrow-side helper state is no longer retained after conversion.
- Severity: Medium
- Already raised by others: No
# Compatibility
- API: compatible
- Config/defaults: no change
- Protocol/serialization: no change
- Historical behavior: still compatible from the caller side, but the
intended memory-fix behavior is incomplete
# Tests / coverage
- No new UT/E2E test code was added in this PR.
- So the main test gap here is coverage, not flaky-test structure.
# CI status
- GitHub currently reports:
- `mergeable=MERGEABLE`
- `mergeStateStatus=BLOCKED`
- `Build=FAILURE`
- I have not yet tied that Build failure back to a specific source-level
error in this rereview, and issue 1 is already a blocker independently of CI.
### Conclusion: can merge after fixes
1. Blocking items
- Issue 1: the shared row-batch object still retains the raw Arrow payload
on the real StarRocks / Doris path, so the memory leak / pressure fix is not
complete yet.
2. Suggested but non-blocking follow-up
- Issue 2: add a regression test for the updated lifecycle contract of this
common Arrow reader.
Overall, the direction makes sense and the earlier reader cleanup is a real
improvement. But for the current head, I do not think we should merge yet,
because the batch object still keeps the original Arrow payload alive on the
exact path this PR is trying to fix.
--
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]
