Flink: rows larger than write.memory.segment.page.size are read back corrupted from the write buffer #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)