JingsongLi commented on code in PR #9301:
URL: https://github.com/apache/paimon/pull/9301#discussion_r3817978607
##########
paimon-format/src/main/java/org/apache/paimon/format/blob/AbstractBlobElementWriter.java:
##########
@@ -181,55 +188,202 @@ private SeekableInputStream openStream(Blob blob,
StreamOpener opener) throws IO
}
}
- protected final BlobDescriptor writeBlobData(BlobCopySource source) throws
IOException {
- long blobPosition = out.getPos();
- if (source.reused()) {
+ /**
+ * Fully stages {@code source} when fetch failures may be converted to
NULL. The returned source
+ * contains only bytes which were fetched successfully, so copying it can
never expose a partial
+ * source payload to the final BLOB output.
+ */
+ protected final @Nullable BlobCopySource
prepareBlobForWrite(BlobCopySource source)
+ throws IOException {
+ if (!writeNullOnFetchFailure) {
+ return source;
+ }
+
+ final BlobStaging staging;
+ try {
+ staging = stagingFactory.create();
Review Comment:
[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.
##########
paimon-format/src/main/java/org/apache/paimon/format/blob/AbstractBlobElementWriter.java:
##########
@@ -263,27 +417,53 @@ private void write(byte[] bytes, int length) throws
IOException {
}
private boolean shouldWriteNullOnFetchFailure(Throwable e) {
- return writeNullOnFetchFailure && !HttpClientUtils.isNotFoundError(e);
+ return writeNullOnFetchFailure
+ && !isTaskCancellation(e)
+ && !HttpClientUtils.isNotFoundError(e);
+ }
+
+ private static boolean isTaskCancellation(Throwable failure) {
+ if (Thread.currentThread().isInterrupted()) {
+ return true;
+ }
+
+ Throwable current = failure;
+ while (current != null) {
+ if (current instanceof InterruptedException
Review Comment:
[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.
--
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]
