[format] Safely write NULL for BLOB body fetch failures

[wwj6591812]

Purpose

blob-write-null-on-fetch-failure=true can safely turn a BLOB fetch/open failure into NULL only while no bytes for that BLOB have reached the managed BLOB output. An HTTP-backed BLOB write can also fail later, while the response body is being consumed. One observed failure was:

org.apache.paimon.shade.hc.core5.http.ConnectionClosedException:
Premature end of Content-Length delimited message body
(expected: 990434; received: 89075)

At that point a raw BLOB, ARRAY element, or MAP value may already have appended a record header and a partial payload to the shared BLOB file. Catching the exception and returning NULL directly would leave unindexed bytes behind and could corrupt subsequent offsets, CRCs, and records. PositionOutputStream also has no portable record-level rollback across local, distributed, and object-store implementations.

This PR adds a per-BLOB-element staging boundary so the existing opt-in option can safely cover terminal response-body fetch failures.

Changes

When blob-write-null-on-fetch-failure=true:

• Bypass staging for exact, already-materialized BlobData values, including inline payloads larger than the spill threshold. Subclasses remain staged because they may override stream behavior.
• Read each fetch-capable BLOB payload into a private staging buffer before appending any of its payload bytes to the final managed BLOB output.
• Keep up to 1 MiB in memory and spill larger payloads to the writer task's IOManager temp directory when available; standalone callers without an engine-provided directory fall back to the process temp directory.
• Append the staged payload to the final output only after the source body completes and its declared length is satisfied.
• If the source body ultimately fails while being read, close the source, discard the stage, and encode that scalar/ARRAY/MAP element as NULL.
• Keep staging creation/write/read/close/delete failures, source cleanup failures, final output failures, footer/CRC failures, and consumer failures fatal. These are not remote fetch failures and must not be hidden as NULL.
• Preserve task cancellation, including a plain InterruptedIOException whose underlying wait cleared the thread flag: restore the interrupt as soon as the source read fails, before cleanup can replace the original failure. Timeout subclasses such as SocketTimeoutException remain eligible for the configured fetch fallback.
• Discard a failed reusable source so the next descriptor reopens it instead of continuing from an unknown offset.

When the option is false, the existing direct-streaming fast path and failure behavior are unchanged.

Relationship to #9271

This PR and the now-merged #9271 address different layers and are intentionally independent:

Case	#9271: HTTP body recovery	This PR: atomic NULL fallback
A truncated body is recoverable	Resume/replay and return the complete body	Commit the complete staged body, not NULL
Recovery is exhausted	Propagate the terminal read failure	If the existing option is enabled, discard the stage and write NULL
The option is disabled	Recovery is still attempted	Terminal failure still fails the write
This PR is used without #9271	No transport-level recovery	The first terminal body-read failure can safely become NULL when opted in

#9271 recovers transport interruptions: it uses validated Range + If-Range with a strong ETag, or a complete HTTP 200 replay with SHA-256 prefix verification when no strong ETag is available. It deliberately does not change NULL semantics.

This PR supplies the output atomicity needed for the separate terminal policy. With both changes, Paimon first tries bounded body recovery; only after recovery is exhausted does the existing opt-in setting write NULL. #9271 is now merged into master; this PR remains independently reviewable at the format/output layer.

Scope and performance

• The new staging path is enabled only by blob-write-null-on-fetch-failure=true; other writers keep the existing direct path. Exact BlobData inline values still bypass staging when the option is enabled.
• The change applies to the append-only managed-BLOB writer path. Primary-key BLOB externalization currently keeps both NULL-on-fetch options disabled and is not expanded by this PR.
• Each active fetch-capable BLOB writer stages one element at a time. It uses at most approximately 1 MiB of heap before spilling, then performs one local write/read before the final output write.
• Normal success, handled fetch failure, abort, and close paths delete spill files. A process hard kill can leave an unreferenced temporary file for host-level cleanup.
• There is no existing public API signature or managed BLOB on-disk format change. Existing public constructors remain available; magic bytes, length/index encoding, CRC handling, descriptors, and NULL encoding remain unchanged.

Tests

Format-layer unit tests cover scalar, ARRAY, and MAP layouts; good/fail/good sequencing; known-length EOF and unknown-length read errors; option-disabled behavior; reusable-source reopening; exact metrics; cleared-flag cancellation (including a later cleanup failure) versus socket timeout; exact BlobData inline bypass above 1 MiB; in-memory and spilled staging; cleanup on success/failure/abort/close; and fatal staging/final-output/consumer failures.

mvn -pl paimon-format -am \
  -DskipITs -DwildcardSuites=none \
  -Dtest=BlobFormatWriterTest \
  -Dsurefire.failIfNoSpecifiedTests=false test

Tests run: 59, Failures: 0, Errors: 0, Skipped: 0
BUILD SUCCESS

Additional focused tests cover BlobFileFormatTest (44/44), DedicatedFormatRollingFileWriterTest (15/15), the full AppendOnlyWriter -> Dedicated -> Multiple -> BlobFileFormat IOManager spill path (1/1, including cleanup), and generated configuration documentation completeness (1/1).

Flink integration tests use deterministic truncated HTTP responses and cover scalar, ARRAY, and MAP NULL fallback, a following BLOB in the same writer, disabled fallback with no committed row/snapshot, and separation from the 404 option.

Tests run: 5, Failures: 0, Errors: 0, Skipped: 0
BUILD SUCCESS

Latest master already contains merged #9271. The combined code also passed:

• HttpClientUtilsTest: 32 tests
• HTTP recovery plus terminal-NULL Flink integration tests: 7 tests (2 recovery + 5 fallback)
• All failures/errors/skips: 0

The combined integration coverage verifies both successful recovery staying non-NULL and recovery exhaustion falling back to NULL only when explicitly enabled.

API and format

No existing public API signature or storage-format change.

[JingsongLi]

I found two blocking issues in the new staging path. The focused BlobFormatWriterTest suite passes 57/57, but a local counterexample confirms that an InterruptedIOException with a cleared interrupt flag is committed as NULL. The two failing CI jobs are unrelated: Maven Central returned 429, and PrimaryKeyFileStoreTableITCase had an unrelated changelog assertion.

[JingsongLi]

[P1] Preserve cancellation when InterruptedIOException clears the flag

A plain InterruptedIOException is the standard signal for interrupted I/O, and implementations such as PipedInputStream throw it after the underlying wait has consumed the thread interrupt flag. In that case copyToStaging returns the exception, this method sees neither a set flag nor either listed exception type, and handleSourceReadFailure converts it to NULL when blob-write-null-on-fetch-failure is enabled. The cancelled task can therefore continue and commit a substituted NULL. The existing interruption test misses this because its test stream explicitly re-sets the flag before throwing. I reproduced the cleared-flag case locally: the writer completed and the row read back as NULL. Please preserve cancellation provenance, restore the interrupt, and propagate it. A blanket base-class check needs to exempt timeout subclasses such as SocketTimeoutException, which are intentionally eligible for fallback.

[wwj6591812]

Thanks, fixed. A plain InterruptedIOException is now recognized at the source-read catch point, so Paimon restores the thread interrupt before source/staging cleanup can replace the original failure. The check matches the exact base class, so timeout subclasses such as SocketTimeoutException remain eligible for the configured fetch fallback. I added regressions for both a cleared interrupt flag and a subsequent source-close failure, and kept the socket-timeout-to-NULL case covered.

[JingsongLi]

[P2] Do not spill already-materialized inline BLOBs

This creates staging for every payload whenever the option is enabled, including exact BlobData values produced by ordinary inline BYTES writes. BlobData already owns a byte array and reads through ByteArraySeekableStream, so there is no remote fetch to make atomic. With the default 1 MiB threshold, every larger inline value is nevertheless copied to java.io.tmpdir, read back, and then written to the final output. Mixed descriptor/inline workloads therefore gain a full extra disk round trip and can fail valid inline rows with local ENOSPC or heavy cross-subtask temp-directory contention solely because a descriptor failure policy is enabled. Please bypass staging for exact BlobData and other provably in-memory sources. For sources that must spill, prefer an engine/task-configured local directory over the process-global default, and cover an inline payload above the default threshold.

[wwj6591812]

Thanks, fixed. Exact BlobData now bypasses staging, including payloads above the 1 MiB threshold; subclasses remain staged because they may override stream behavior. Descriptor/stream staging now uses the writer task's IOManager temp directory when available, falling back to the process temp directory only when no engine directory exists. I added >1 MiB inline-bypass coverage plus a core-level test that observes a real spill through AppendOnlyWriter -> Dedicated -> Multiple -> BlobFileFormat and verifies cleanup.

[wwj6591812]

@JingsongLi hi, please cc, thx