You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "ericyuan915 (via GitHub)" <gi...@apache.org> on 2026/08/31 00:21:07 UTC
[I] Flink: rows larger than write.memory.segment.page.size are read back corrupted from the write buffer [hudi]
ericyuan915 opened a new issue, #19797:
URL: https://github.com/apache/hudi/issues/19797
### Bug Description
## Bug Description
**What happened:**
On a COPY_ON_WRITE table written by the Flink sink with `write.operation = insert_overwrite`, rows whose serialized size substantially exceeds `write.memory.segment.page.size` (32 KB default) are read back from the write buffer corrupted. The failure surfaces during Avro conversion as an `ArrayIndexOutOfBoundsException` with a nonsensical segment index:
```
java.lang.ArrayIndexOutOfBoundsException: Index 51817 out of bounds for length 8
at org.apache.flink.table.data.binary.BinarySegmentUtils.getByteMultiSegments(BinarySegmentUtils.java:601)
at org.apache.flink.table.data.binary.NestedRowData.isNullAt(NestedRowData.java:199)
at org.apache.hudi.util.RowDataToAvroConverters$13.convert(RowDataToAvroConverters.java:316)
```
The rows in question are ~230 KB, so at the default page size each spans 8 memory segments — which matches the `length 8` in the message. The reported index (51817) implies a nested-row offset of roughly 1.7 GB, i.e. the packed `offsetAndSize` for the nested `ROW` field was read from the wrong place.
**What you expected:**
Rows larger than a single memory-segment page should round-trip through the write buffer intact, or the write should fail with a clear error naming the limit.
**Steps to reproduce:**
1. COW table, Flink sink via `HoodiePipeline`, `write.operation = insert_overwrite`.
2. Schema containing a nested `ROW` field, at least one member of which is a large `STRING` (a JSON blob of ~200 KB per row in our case), so that the serialized `BinaryRowData` is comfortably larger than one page.
3. Leave `write.memory.segment.page.size` at its 32 KB default.
4. Write a few million such rows.
I do **not** have a self-contained reproducer yet — see the caveat at the end.
## Analysis
`StreamWriteFunction` buffers rows per bucket in a `RowDataBucket`, which wraps a Flink `BinaryInMemorySortBuffer` built by `BufferUtils.createBuffer` over a `MemorySegmentPool` whose page size comes from `write.memory.segment.page.size`:
- `StreamWriteFunction.doBufferRecord` → `RowDataBucket.writeRow` → `BinaryInMemorySortBuffer.write`, which calls `RowDataSerializer.serializeToPages`.
- On flush, `RowDataBucket.getDataIterator` → `BinaryInMemorySortBuffer.getIterator` → `getRecordFromBuffer` → `BinaryRowDataSerializer.mapFromPages` → `pointTo`.
`pointTo` takes a single-segment fast path when the row fits in the remainder of the current page, and otherwise goes through `pointToMultiSegments`, which assembles a `MemorySegment[]` for the row. With 32 KB pages and ~230 KB rows, **every** row takes the multi-segment path.
Reading the nested `ROW` field then goes `BinaryRowData.getRow` → `BinarySegmentUtils.readRowData`, which unpacks a `long` from the fixed-length part into `(offset, size)` and calls `NestedRowData.pointTo(segments, offset + baseOffset, size)`. The observed ~1.7 GB offset means that `long` is not the value that was written, so the parent row's fixed-length part is being addressed incorrectly once the row is spread across segments.
I read `BinaryRowDataSerializer.pointToMultiSegments` looking for an off-by-one in the array sizing and did not find one — it allocates `varSegSize + 1` segments computed from `remainLength`, which does account for the row starting mid-segment. So I have localized the failure but not root-caused it.
**Scope:** this only affects operations that route through `StreamWriteFunction` — `upsert`, `insert_overwrite`, `insert_overwrite_table`. Plain `insert` is unaffected, because `OptionsResolver.isAppendMode` sends it to `Pipelines.append`, whose `AppendWriteFunction` hands each row straight to `BulkInsertWriterHelper` and never touches this buffer. We confirmed that empirically: the identical data and schema write fine under `insert`.
**Partial mitigations that do not hold:** raising `write.memory.segment.page.size` reduces how often rows straddle a page but does not eliminate it, because `checkSkipWriteForFixLengthPart` only guarantees the row's *fixed* part fits the current page — the variable-length part can still cross the boundary. Our actual workaround was to project the large column out of the write schema so rows fit in a single page.
## Caveat
This is a report from a production-shaped workload, not from a minimal reproducer, and I have not yet confirmed it against a released artifact — only against a build from `release-1.2` in which the classes above are unmodified. Happy to put together a standalone test (COW + `insert_overwrite` + a nested `ROW` with a multi-hundred-KB `STRING` member at the default page size) if that would help triage.
### Environment
- **Hudi version:** 1.2.x, built from `release-1.2`. The classes in the stack are unmodified from upstream.
- **Flink version:** 1.18 (`hudi-flink1.18-bundle`), BATCH execution mode, bounded source
- **Table type:** COPY_ON_WRITE
- **Storage:** object store (GCS-backed Hadoop FS)
- **Running on Docker?** no (YARN)
Relevant options — everything not listed is at its default:
```
write.operation = insert_overwrite
table.type = COPY_ON_WRITE
hoodie.datasource.write.recordkey.field = <a top-level STRING column>
hoodie.datasource.write.partitionpath.field = <a top-level DOUBLE column>
write.tasks = 32
hoodie.metadata.enable = false
write.memory.segment.page.size = 32768 (default)
write.task.max.size = 1024 (default)
```
### Logs and Stack Trace
```
org.apache.hudi.exception.HoodieUpsertException: Failed to upsert for commit time <instant>
at org.apache.hudi.table.action.commit.FlinkWriteHelper.write(FlinkWriteHelper.java:81)
at org.apache.hudi.table.action.commit.FlinkInsertOverwriteCommitActionExecutor.execute(FlinkInsertOverwriteCommitActionExecutor.java:71)
at org.apache.hudi.table.HoodieFlinkCopyOnWriteTable.insertOverwrite(HoodieFlinkCopyOnWriteTable.java:241)
at org.apache.hudi.client.HoodieFlinkWriteClient.insertOverwrite(HoodieFlinkWriteClient.java:298)
at org.apache.hudi.sink.StreamWriteFunction$WriteFunction.write(StreamWriteFunction.java:516)
at org.apache.hudi.sink.StreamWriteFunction.writeRecords(StreamWriteFunction.java:445)
at org.apache.hudi.sink.StreamWriteFunction.flushBucket(StreamWriteFunction.java:381)
at org.apache.hudi.sink.StreamWriteFunction.bufferRecord(StreamWriteFunction.java:323)
at org.apache.hudi.sink.StreamWriteFunction.processElement(StreamWriteFunction.java:184)
...
Caused by: org.apache.hudi.exception.HoodieException: operation has failed
at org.apache.hudi.common.util.queue.BoundedInMemoryQueue.throwExceptionIfFailed(BoundedInMemoryQueue.java:274)
at org.apache.hudi.common.util.queue.BoundedInMemoryQueue.insertRecord(BoundedInMemoryQueue.java:202)
at org.apache.hudi.common.util.queue.IteratorBasedQueueProducer.produce(IteratorBasedQueueProducer.java:45)
...
Caused by: org.apache.hudi.exception.HoodieException: Index 51817 out of bounds for length 8
at org.apache.hudi.io.BaseCreateHandle.doWrite(BaseCreateHandle.java:123)
at org.apache.hudi.io.HoodieWriteHandle.write(HoodieWriteHandle.java:241)
at org.apache.hudi.execution.ExplicitWriteHandler.consume(ExplicitWriteHandler.java:48)
...
Caused by: java.lang.ArrayIndexOutOfBoundsException: Index 51817 out of bounds for length 8
at org.apache.flink.table.data.binary.BinarySegmentUtils.getByteMultiSegments(BinarySegmentUtils.java:601)
at org.apache.flink.table.data.binary.BinarySegmentUtils.getByte(BinarySegmentUtils.java:593)
at org.apache.flink.table.data.binary.BinarySegmentUtils.bitGet(BinarySegmentUtils.java:536)
at org.apache.flink.table.data.binary.NestedRowData.isNullAt(NestedRowData.java:199)
at org.apache.flink.table.data.RowData.lambda$createFieldGetter$1(RowData.java:293)
at org.apache.hudi.util.RowDataToAvroConverters$13.convert(RowDataToAvroConverters.java:316)
at org.apache.hudi.util.RowDataToAvroConverters$12.convert(RowDataToAvroConverters.java:284)
at org.apache.hudi.util.RowDataToAvroConverters$13.convert(RowDataToAvroConverters.java:315)
at org.apache.hudi.util.RowDataToAvroConverters$12.convert(RowDataToAvroConverters.java:284)
at org.apache.hudi.client.model.HoodieFlinkRecord.toIndexedRecord(HoodieFlinkRecord.java:286)
at org.apache.hudi.io.storage.HoodieAvroFileWriter.writeWithMetadata(HoodieAvroFileWriter.java:44)
at org.apache.hudi.io.BaseCreateHandle.writeRecordToFile(BaseCreateHandle.java:165)
at org.apache.hudi.io.BaseCreateHandle.doWrite(BaseCreateHandle.java:103)
```
--
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: commits-unsubscribe@hudi.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Flink: rows larger than write.memory.segment.page.size are read back corrupted from the write buffer [hudi]
Posted by "cshuo (via GitHub)" <gi...@apache.org>.
cshuo commented on issue #19797:
URL: https://github.com/apache/hudi/issues/19797#issuecomment-5473133096
@ericyuan915 I see two issues here:
1. Is the AVRO writer path expected? On release-1.2.0, FlinkCreateHandle should normally use HoodieRowDataParquetWriter. If this is not intentional, please check the effective record merger, merge strategy, payload class, and related settings in hoodie.properties, as these may cause fallback to HoodieAvroFileWriter.
2. The corruption closely matches #19664, fixed by #19692 and #19728. The actual trigger is memory exhaustion partially mutating a BinaryInMemorySortBuffer, followed by reuse of that diverged buffer—not merely a row spanning multiple pages. Could you test with the fix PRs?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Flink: rows larger than write.memory.segment.page.size are read back corrupted from the write buffer [hudi]
Posted by "ericyuan915 (via GitHub)" <gi...@apache.org>.
ericyuan915 commented on issue #19797:
URL: https://github.com/apache/hudi/issues/19797#issuecomment-5689127539
@cshuo @vinothchandar Really appreciate the info, I haven't got a chance to test it out yet, but once I get the chance, I will and let you know the result!
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Flink: rows larger than write.memory.segment.page.size are read back corrupted from the write buffer [hudi]
Posted by "vinothchandar (via GitHub)" <gi...@apache.org>.
vinothchandar commented on issue #19797:
URL: https://github.com/apache/hudi/issues/19797#issuecomment-5592254579
@ericyuan915 any updates for us
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org