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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
