DanielLeens commented on PR #10931:
URL: https://github.com/apache/seatunnel/pull/10931#issuecomment-4519962536
Thanks for working on this! I went through the full diff, checked the latest
head `f2b4dac187987e29c0c309e0e60b98742991220b` locally against `upstream/dev`,
and traced the real call paths in both StarRocks and Doris because this common
Arrow reader is shared by both connectors. I did not run Maven locally in this
pass; this is a source-level review only.
# What problem this PR solves
- User pain point
When StarRocks source reads Arrow batches, the old implementation keeps
Arrow reader / allocator state on the `ArrowToSeatunnelRowReader` instance, so
delayed or missing cleanup can eventually surface as Arrow-memory-related
exceptions.
- Fix approach
This PR moves Arrow parsing into try-with-resources inside `readArrow()`
so `RootAllocator`, `ByteArrayInputStream`, and `ArrowStreamReader` are closed
immediately after the batch is converted to `SeaTunnelRow`.
- One sentence
The goal is good: shrink the Arrow off-heap lifetime to a single
deserialize pass instead of the whole row-batch object lifetime.
# Runtime chain I checked
```text
StarRocks source
-> StarRocksBeReadClient.hasNext()
[connector-starrocks/.../StarRocksBeReadClient.java:145-170]
-> new ArrowToSeatunnelRowReader(result.getRows(),
seaTunnelRowType).readArrow()
-> ArrowToSeatunnelRowReader.readArrow()
[connector-common/.../ArrowToSeatunnelRowReader.java:94-116]
-> ArrowStreamReader.loadNextBatch()
-> convertSeatunnelRow()
-> keep seatunnelRowBatch for later row-by-row consumption
-> old batch is closed at the next-batch boundary [145-147]
Doris source async path
-> DorisValueReader async thread
[connector-doris/.../DorisValueReader.java:154-170]
-> new ArrowToSeatunnelRowReader(nextResult.getRows(),
seaTunnelRowType).readArrow()
-> rowBatch.close()
-> rowBatchBlockingQueue.put(rowBatch)
-> consumer thread later takes the row batch and iterates rows [199-218]
```
# Core logic review
Before this PR, the reader kept Arrow objects on the instance:
```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);
}
...
finally {
close();
}
```
After this PR, Arrow resources are local to `readArrow()`, but the raw
payload is now stored on the object and `close()` is a no-op:
```java
private byte[] byteArray;
private void initArrowReader(byte[] byteArray) {
this.byteArray = byteArray;
}
public ArrowToSeatunnelRowReader readArrow() {
try (RootAllocator rootAllocator = new RootAllocator(Integer.MAX_VALUE);
ByteArrayInputStream byteArrayInputStream = new
ByteArrayInputStream(byteArray);
ArrowStreamReader arrowStreamReader =
new ArrowStreamReader(byteArrayInputStream,
rootAllocator)) {
...
}
}
@Override
public void close() {}
```
Key findings:
1. The normal runtime path definitely hits this change in both StarRocks and
Doris.
2. The PR does improve off-heap Arrow cleanup timing.
3. The current implementation still keeps the raw Arrow payload attached to
the row-batch object, so it does not fully solve the memory-pressure story.
4. Doris async mode is the most sensitive path because queued row batches
can accumulate both the fully materialized rows and the original Arrow bytes at
the same time.
# Findings
Issue 1: `close()` no longer releases the batch-side retained payload, so
the main-path release hook is weakened
- Location:
`seatunnel-connectors-v2/connector-common/src/main/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/reader/ArrowToSeatunnelRowReader.java:64,90-116,317-318`
- Problem description:
This class sits on the common Arrow deserialize path for StarRocks and
Doris. The PR stores the raw `byte[]` in the instance field `byteArray`, but
`close()` is now empty. At the same time, callers still rely on the old
batch-boundary cleanup pattern:
- StarRocks closes the previous batch before fetching the next one at
`connector-starrocks/.../StarRocksBeReadClient.java:145-147`
- Doris async mode closes the batch before queueing it at
`connector-doris/.../DorisValueReader.java:161-169`
- Doris sync mode closes the previous batch before advancing at
`connector-doris/.../DorisValueReader.java:227-230`
So the off-heap reader is closed earlier, but the row-batch object still
keeps the original Arrow payload alive for the rest of its lifetime.
- Potential risk:
This can still create a high heap peak on the real source path, especially
in Doris async mode where multiple row batches may sit in the queue at once. In
practice that means one extra full copy of each batch can remain reachable even
after conversion, which directly weakens the memory-fix goal of this PR.
- Best improvement suggestion:
Option A: keep the try-with-resources change, but explicitly clear
`byteArray`, `root`, and `fieldVectors` after conversion, and make `close()` at
least release those helper references.
Option B: do not store the raw `byte[]` on the instance at all; keep Arrow
parsing fully method-local and only retain the final `seatunnelRowBatch`.
- Severity: High
- Already raised by others: No
Issue 2: there is no regression test for the new lifecycle contract on this
shared reader
- Location:
`seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/common/source/arrow/ArrowToSeatunnelRowReaderTest.java:310-326,447-458`
- Problem description:
The existing tests cover the happy path of reading rows out of Arrow, but
this PR changes the lifecycle semantics of a shared connector-common reader.
There is still no regression test that locks down the new “read first, then
close, then continue consuming rows” contract, and no test that protects
against retaining the original Arrow payload after conversion.
- Potential risk:
Since both StarRocks and Doris reuse this class, a lifecycle regression
here can come back later without CI catching it.
- Best improvement suggestion:
Please add at least one regression test that reflects the real caller
contract. For example:
1. call `readArrow()`, then `close()`, and verify rows are still readable,
and/or
2. verify the reader no longer retains Arrow-side helper state after
conversion.
- Severity: Medium
- Already raised by others: No
# Compatibility
- API: compatible
- Config/defaults: no change
- Protocol/serialization: no change to Arrow payload format
- Historical behavior: the implementation-level meaning of `close()`
changed, which is exactly why issue 1 matters on the existing call path
# Tests / coverage
- No new UT/E2E test code was added in this PR.
- The main gap is coverage, not flaky-test structure.
# CI status
- Current GitHub state on this head:
- `mergeable=MERGEABLE`
- `mergeStateStatus=BLOCKED`
- `Build` is still `QUEUED`
- So from the GitHub side, CI is not green yet, but the source-level blocker
above is independent of the queued Build.
### Conclusion: can merge after fixes
1. Blocking items
- Issue 1: the shared reader still retains the raw Arrow payload on the
normal StarRocks / Doris path, so the memory-fix story is not complete yet.
2. Suggested but non-blocking follow-up
- Issue 2: add a regression test that locks down the new lifecycle contract
of this common reader.
Overall, the direction makes sense and the earlier Arrow-resource cleanup is
a real improvement. But for the current head, I do not think we should merge
before the retained-payload problem is closed as well, because that is still on
the exact memory-sensitive 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]
