github-actions[bot] commented on code in PR #66810:
URL: https://github.com/apache/doris/pull/66810#discussion_r3795596210
##########
be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.cpp:
##########
@@ -218,10 +309,14 @@ jobject allocate_paimon_memory_page(JNIEnv* env, jclass,
jlong manager_handle, j
return nullptr;
}
try {
- return manager->allocate_page(env, bytes);
+ return manager->allocate_page(env, bytes, wait_for_memory == JNI_TRUE);
} catch (const std::exception& e) {
jclass exception_class = env->FindClass("java/lang/OutOfMemoryError");
- env->ThrowNew(exception_class, e.what());
+ // Avoid heap allocation while reporting an allocation failure.
+ char message[1024];
+ std::snprintf(message, sizeof(message), "Paimon JNI native page
allocation failed: %.900s",
+ e.what());
+ env->ThrowNew(exception_class, message);
Review Comment:
[P2] Preserve cancellation across the JNI allocation callback
The new wait loop returns `Status::Cancelled` for both query cancellation
and FragmentMgr shutdown, but this catch converts every native exception into
the same prefixed `OutOfMemoryError`. Java therefore records
`MEMORY_ERROR_PAIMON_PAGE`, and the C++ caller rewrites it to
`QUERY_MEMORY_EXCEEDED` while incrementing the page-memory counter. Because the
async writer keeps the first error, the memory error can win the race with
normal cancellation and report the wrong terminal reason. Please carry a
distinct cancellation/shutdown category across JNI and translate it back to
`Status::Cancelled`; reserve this OOM prefix and counter for actual page
allocation/limit failures.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/DorisMemorySegmentPool.java:
##########
@@ -17,41 +17,160 @@
package org.apache.doris.paimon;
-import org.apache.paimon.memory.AbstractMemorySegmentPool;
import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.memory.MemorySegmentPool;
import java.nio.ByteBuffer;
+import java.util.ArrayDeque;
+import java.util.List;
+import java.util.Objects;
/**
* Paimon write-buffer pool backed by memory allocated and tracked by Doris BE.
*
* <p>This class only adapts the Paimon page interface to the BE allocator.
The native memory
* manager owns every returned page and releases them after the Java writer
has closed.
*/
-final class DorisMemorySegmentPool extends AbstractMemorySegmentPool {
+final class DorisMemorySegmentPool implements MemorySegmentPool {
+ @FunctionalInterface
+ interface PageAllocator {
+ ByteBuffer allocate(long nativeMemoryManager, int pageSize, boolean
waitForMemory);
+ }
+
private final long nativeMemoryManager;
+ private final int pageSize;
+ private final int maxPages;
+ private final PageAllocator pageAllocator;
+ private final ArrayDeque<MemorySegment> availableSegments = new
ArrayDeque<>();
+ private int allocatedPages;
+ private boolean nativeMemoryPressure;
DorisMemorySegmentPool(long maxMemory, int pageSize, long
nativeMemoryManager) {
- super(maxMemory, pageSize);
+ this(maxMemory, pageSize, nativeMemoryManager,
+ PaimonJniWriter::allocatePaimonMemoryPage);
+ }
+
+ DorisMemorySegmentPool(long maxMemory, int pageSize, long
nativeMemoryManager,
+ PageAllocator pageAllocator) {
if (nativeMemoryManager == 0) {
throw new IllegalArgumentException("Doris native memory manager
must not be null");
}
+ if (pageSize <= 0) {
+ throw new IllegalArgumentException(
+ "Doris-managed Paimon memory page size must be positive: "
+ pageSize);
+ }
if (maxMemory < pageSize) {
throw new IllegalArgumentException(
"Doris-managed Paimon memory pool must contain at least
one page: maxMemory="
+ maxMemory + ", pageSize=" + pageSize);
}
+ long pages = maxMemory / pageSize;
+ if (pages > Integer.MAX_VALUE) {
+ throw new IllegalArgumentException(
+ "Doris-managed Paimon memory pool has too many pages: " +
pages);
+ }
this.nativeMemoryManager = nativeMemoryManager;
+ this.pageSize = pageSize;
+ this.maxPages = (int) pages;
+ this.pageAllocator = Objects.requireNonNull(pageAllocator,
"pageAllocator");
}
@Override
- protected MemorySegment allocateMemory() {
- ByteBuffer buffer =
- PaimonJniWriter.allocatePaimonMemoryPage(nativeMemoryManager,
pageSize);
+ public MemorySegment nextSegment() {
+ synchronized (this) {
+ MemorySegment available = availableSegments.pollFirst();
+ if (available != null) {
+ nativeMemoryPressure = false;
+ return available;
+ }
+ if (allocatedPages >= maxPages) {
+ return null;
+ }
+ ++allocatedPages;
+ }
+
+ final ByteBuffer buffer;
+ try {
+ buffer = pageAllocator.allocate(nativeMemoryManager, pageSize,
false);
+ } catch (Throwable t) {
+ rollbackAllocation();
+ throw t;
+ }
+
if (buffer == null) {
- throw new OutOfMemoryError(
- "Doris failed to allocate a native Paimon memory page of "
+ pageSize + " bytes");
+ recordNativeMemoryPressure();
+ return null;
Review Comment:
[P1] Retry lazy-owner construction before this null escapes
On the first row (or another lazy buffered-owner creation), Paimon's
`OwnerMemoryPool.nextSegment()` calls this pool twice with owner preemption
between the calls. Under transient Doris reservation pressure both calls reach
this branch; with no established owner to preempt, the second null is passed
into `SimpleCollectingOutputView`, whose constructor immediately throws
`NullPointerException("Initial Segment may not be null")`. Control never
reaches the next `writeRow()` boundary where `waitForMemoryIfNeeded()` is
called, so the intended park-on-pressure behavior becomes a generic first-write
failure. The new test only works because it manually inserts that
otherwise-unreachable boundary. Please provision/retry specifically around lazy
owner construction before its required initial pages remain null; blocking
every established owner's post-preemption retry would suppress Paimon's own
spill/flush path.
##########
be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp:
##########
@@ -421,64 +549,122 @@ Status
JniPaimonWriter::_write_projected_block(RuntimeState* state, Block& block
block.get_by_position(i).name = _sink.column_names[i];
}
- // Pipeline: Doris Block → Arrow Schema → Arrow RecordBatch → IPC Stream →
JNI direct buffer
- //
- // Step 1: Build Arrow schema from the projected Block.
+ // Build the schema once, then convert row ranges independently. Each
range owns its
+ // RecordBatch and IPC buffer, so those transient allocations are released
before the next
+ // range is converted.
// Paimon write timestamps are transported as civil-time fields. The Java
writer uses the
// pinned Paimon target type to preserve NTZ values or convert LTZ values
with the session zone.
// Variant V2 is transported losslessly as its value/metadata pair,
including nested Variant.
std::shared_ptr<arrow::Schema> arrow_schema;
RETURN_IF_ERROR(get_paimon_arrow_schema_from_block(block, &arrow_schema));
- // Step 2: Convert Doris Block columns to an Arrow RecordBatch.
+ // This is a best-effort batch-size target, not a hard native-memory
limit. The finite Java
+ // allocator remains the hard decode boundary. Half of its limit leaves
room for Arrow offsets,
+ // validity buffers, allocator rounding and representation expansion
during decode.
+ const size_t arrow_batch_memory_budget =
+ std::max<size_t>(1, static_cast<size_t>(_arrow_memory_limit_bytes)
/ 2);
+ const size_t target_batch_bytes =
+ std::min(state->preferred_block_size_bytes(),
arrow_batch_memory_budget);
+ const size_t block_bytes = block.bytes();
+ const size_t average_row_bytes =
+ std::max<size_t>(1, block_bytes / block_rows + (block_bytes %
block_rows != 0));
+ const BlockBudget batch_budget(static_cast<size_t>(state->batch_size()),
target_batch_bytes);
+ const size_t rows_per_batch =
batch_budget.effective_max_rows(average_row_bytes);
+
+ for (size_t start_row = 0; start_row < block_rows;) {
+ const size_t range_rows = std::min(rows_per_batch, block_rows -
start_row);
+ const size_t end_row = start_row + range_rows;
+ const size_t estimated_ipc_bytes = average_row_bytes >
target_batch_bytes / range_rows
+ ? target_batch_bytes
+ : average_row_bytes *
range_rows;
+ RETURN_IF_ERROR(_write_row_range(state, block, arrow_schema,
start_row, end_row,
+ estimated_ipc_bytes));
+ start_row = end_row;
+ }
+ return Status::OK();
+}
+
+Status JniPaimonWriter::_write_row_range(RuntimeState* state, const Block&
block,
+ const std::shared_ptr<arrow::Schema>&
arrow_schema,
+ size_t start_row, size_t end_row,
+ size_t estimated_ipc_bytes) {
std::shared_ptr<arrow::RecordBatch> record_batch;
RETURN_IF_ERROR(convert_to_arrow_batch(block, arrow_schema,
_arrow_pool.get(), &record_batch,
- state->timezone_obj()));
-
- // Step 3: Serialize the RecordBatch to Arrow IPC Stream format in memory.
- auto out_stream_res = arrow::io::BufferOutputStream::Create(4096,
_arrow_pool.get());
+ state->timezone_obj(), start_row,
end_row));
Review Comment:
[P2] Cover the whole Arrow range with the OOM translator
This conversion is the first allocation phase of the new range pipeline, but
Arrow OOM returned by `MakeBuilder`/serde append/`Finish` is already collapsed
to generic internal or fatal status before `_write_row_range()` sees it. There
is a second gap in the same boundary: `ArrowMemoryPool::Allocate/Reallocate`
lets Doris allocator exceptions escape, so actual allocation failures can
bypass both this call and all of the later IPC calls before
`convert_cpp_arrow_status()` can inspect a status. In either case the query
does not get the intended `QUERY_MEMORY_EXCEEDED` classification and
`CppArrowMemoryErrorCount` remains unchanged. Please preserve Arrow OOM through
conversion and catch Doris allocator failures around the complete per-range
conversion/IPC pipeline.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java:
##########
@@ -602,7 +674,128 @@ private void abortWriter() throws Exception {
// Utilities
// ────────────────────────────────────────────────────────────
- static native ByteBuffer allocatePaimonMemoryPage(long
nativeMemoryManager, int bytes);
+ static WriterMemoryBudget splitWriterMemoryBudget(
+ long writerBudgetBytes, long configuredWriteBufferBytes, int
pageSize) {
+ if (writerBudgetBytes <= 0) {
+ throw new IllegalArgumentException(
+ "Paimon JNI writer memory budget must be positive: " +
writerBudgetBytes);
+ }
+ if (configuredWriteBufferBytes <= 0) {
+ throw new IllegalArgumentException(
+ "Paimon write buffer size must be positive: " +
configuredWriteBufferBytes);
+ }
+ if (pageSize <= 0) {
+ throw new IllegalArgumentException("Paimon page size must be
positive: " + pageSize);
+ }
+
+ long arrowHeadroomBytes = Math.max(
+ writerBudgetBytes / ARROW_HEADROOM_DIVISOR,
ARROW_MIN_HEADROOM_BYTES);
+ if (arrowHeadroomBytes > writerBudgetBytes - pageSize) {
Review Comment:
[P1] Preserve Paimon's three-page minimum after the split
This check only reserves one Paimon page, but Paimon 1.4.2's merge-tree
`SortBufferWriteBuffer` rejects any pool whose `freePages()` is below three.
Because `page-size` is table-configurable, this regresses valid configurations:
with a 32 MiB per-writer budget, 8 MiB pages, and a configured write buffer of
at least 32 MiB, the old pool exposed four pages; the new 16 MiB Arrow minimum
leaves two, passes `open()` here, and then fails deterministically when the
first merge-tree writer for an ordinary primary-key table is created. Please
make the split preserve the merge-tree three-page minimum (or validate a
writer-type-aware minimum) and reject insufficient budgets during open.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonCommitCodec.java:
##########
@@ -94,10 +103,19 @@ byte[][] encode(List<CommitMessage> messages) throws
Exception {
chunkSize = Math.max(1, chunkSize / 2);
continue;
}
- throw new IOException("A single Paimon commit message exceeds
the "
- + maxPayloadBytes + " byte framed payload limit", e);
+ throw new CommitPayloadMemoryException(
+ "A single Paimon commit message exceeds the "
+ + maxPayloadBytes + " byte framed payload
limit",
+ e);
+ }
+ if (payload.length > maxTotalPayloadBytes - totalPayloadBytes) {
Review Comment:
[P1] Enforce the payload cap before allocating the next copy
This check runs only after `encodeChunk()` has grown its
`ByteArrayOutputStream` and `toByteArray()` has allocated a second array, while
all earlier chunks remain retained. With the minimum 16 MiB cap and two
approximately 8 MiB chunks, producing the accepted second chunk can already
hold roughly 8 MiB previous payload + 8 MiB backing buffer + 8 MiB result = 24
MiB; an attempted third can reach roughly 32 MiB before being rejected here. A
JVM OOM can therefore happen before `CommitPayloadMemoryException`, defeating
the hard cap this path is meant to provide. Please enforce the remaining budget
during serialization and account for/eliminate the backing-buffer copy before
starting the chunk.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
