Flink: rows larger than write.memory.segment.page.size are read back corrupted from the write buffer

Open

#19,797 opened on Aug 31, 2026 2 comments 0 reactions 1 assignee Claimed by @cshuo View on GitHub

type:bug

Dominant language: Java
Stars: 6.2k
Forks: 2.5k
Avg merge: 2d 8h
Merged PRs (30d): 111

Description

### 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 =
hoodie.datasource.write.partitionpath.field =
write.tasks = 32
hoodie.metadata.enable = false
write.memory.segment.page.size = 32768 (default)
write.task.max.size = 1024 (default)
```

### Logs and Stack Trace

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Flink: rows larger than write.memory.segment.page.size are read back corrupted from the write buffer

Description

Contributor guide

Assessment

Get new issues in your inbox