github-actions[bot] commented on code in PR #66612:
URL: https://github.com/apache/doris/pull/66612#discussion_r3764930516
##########
be/src/exec/spill/spill_file_manager.cpp:
##########
@@ -158,21 +220,47 @@ std::vector<SpillDataDir*>
SpillFileManager::_get_stores_for_spill(
Status SpillFileManager::create_spill_file(const std::string& relative_path,
SpillFileSPtr& spill_file) {
- auto data_dirs = _get_stores_for_spill(TStorageMedium::type::SSD);
- if (data_dirs.empty()) {
- data_dirs = _get_stores_for_spill(TStorageMedium::type::HDD);
- }
- if (data_dirs.empty()) {
+ SpillDataDir* data_dir = _get_store_for_spill();
+ if (data_dir == nullptr) {
return Status::Error<ErrorCode::NO_AVAILABLE_ROOT_PATH>(
"no available disk can be used for spill.");
}
- // Select the first available data dir (sorted by usage ascending)
- SpillDataDir* data_dir = data_dirs.front();
spill_file = std::make_shared<SpillFile>(data_dir, relative_path);
return Status::OK();
}
+Status SpillFileManager::create_external_spill_directory(
+ const std::string& relative_path, QueryContext* query_context,
+ std::unique_ptr<ExternalSpillDirectory>* spill_directory) {
+ if (query_context == nullptr || spill_directory == nullptr) {
+ return Status::InvalidArgument(
+ "External spill directory requires QueryContext and output
session");
+ }
+
+ SpillDataDir* data_dir = _get_store_for_spill();
Review Comment:
[P1] Preserve capacity across eligible spill roots
Selecting one `SpillDataDir` here permanently pins the writer to that root:
every later reservation targets it, with no fallback when it reaches capacity.
Previously Paimon received all BE store paths and round-robined channel
creation, so a writer could use aggregate multi-disk capacity; concurrent opens
can now also choose the same initially least-used root before any bytes are
reserved. Please preserve managed sessions across eligible roots or add safe
re-selection/fallback, with a selected-root-full/peer-root-free test.
##########
be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp:
##########
@@ -269,20 +336,26 @@ Status JniPaimonWriteBackend::open(const
TPaimonTableSink& sink, RuntimeState* s
_jni_writer_obj = env->NewGlobalRef(local_obj);
env->DeleteLocalRef(local_obj);
- // Step 4: Build Java arguments and call PaimonJniWriter.open().
+ // Step 4: Allocate the query-scoped external spill session before
creating JNI arguments, so
+ // any storage error returns without leaving local references behind.
+ auto* spill_file_manager = state->exec_env()->spill_file_mgr();
+ if (spill_file_manager == nullptr) {
+ return Status::InternalError("Paimon JNI writer requires the Doris
spill file manager");
+ }
+ auto spill_relative_path =
+ fmt::format("{}/{}-{}", print_id(state->query_id()),
PAIMON_JNI_WRITER_IO_TMP_DIR,
+ spill_file_manager->next_id());
+ RETURN_IF_ERROR(spill_file_manager->create_external_spill_directory(
Review Comment:
[P1] Do not require spill storage for memory-only writers
This makes every writer depend on an available spill root before Java can
evaluate `write-buffer-spillable`. For the supported `false` setting,
`openMemoryResources()` returns without creating an IOManager, but this call
can fail first with `NO_AVAILABLE_ROOT_PATH` when spill storage is at its limit
or unavailable. That turns a memory-only write into an unnecessary
spill-storage failure. Please allocate the managed spill session lazily or
optionally after the table's spill setting is known, and cover the
disabled-spill/unavailable-root case.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/DorisIOManager.java:
##########
@@ -0,0 +1,289 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.paimon;
+
+import org.apache.paimon.disk.BufferFileReader;
+import org.apache.paimon.disk.BufferFileWriter;
+import org.apache.paimon.disk.FileIOChannel;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.memory.Buffer;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/** Paimon IOManager adapter which charges temporary channel I/O to Doris
spill management. */
+final class DorisIOManager implements IOManager {
+ interface SpillAccountant {
+ void reserve(long bytes) throws IOException;
+
+ void rollback(long bytes);
+
+ void commitWrite(long bytes);
+
+ void recordRead(long bytes);
+
+ void release(long bytes);
+ }
+
+ private final IOManager delegate;
+ private final SpillAccountant accountant;
+ private final Map<String, Long> channelBytes = new ConcurrentHashMap<>();
+
+ static DorisIOManager create(String[] tempDirs, long nativeSpillDirectory)
{
+ return new DorisIOManager(
+ IOManager.create(tempDirs), new
NativeSpillAccountant(nativeSpillDirectory));
+ }
+
+ DorisIOManager(IOManager delegate, SpillAccountant accountant) {
+ this.delegate = delegate;
+ this.accountant = accountant;
+ }
+
+ @Override
+ public FileIOChannel.ID createChannel() {
+ return delegate.createChannel();
+ }
+
+ @Override
+ public FileIOChannel.ID createChannel(String prefix) {
Review Comment:
[P1] Account for raw lookup and clustering SST files
This adapter exposes raw files without any accounting hook. In Paimon 1.4.2,
lookup compaction takes `createChannel(prefix).getPathFile()` and performs
lookup-SST I/O directly; primary-key clustering similarly builds
`SimpleLsmKvDb` under `pickTempDir()`. Their create/write/read/delete
operations never reach the wrapped buffer channels or native callbacks, so they
can exceed the spill limit while current usage and I/O metrics stay low. Please
provide accounting-aware raw-file integration (or reject these modes with the
managed adapter) and cover both lookup and clustering SST paths.
##########
regression-test/suites/paimon_write/test_paimon_write_thread_lifecycle.groovy:
##########
@@ -25,13 +25,6 @@ suite("test_paimon_write_thread_lifecycle",
"p0,external,paimon") {
return
}
- // Keep the reproducer opt-in until attached JNI writer threads are
released.
- String knownBugTestEnabled =
context.config.otherConfigs.get("enablePaimonKnownBugTest")
- if (knownBugTestEnabled == null ||
!knownBugTestEnabled.equalsIgnoreCase("true")) {
- logger.info("skip isolated Paimon known-bug thread regression")
- return
- }
-
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
Review Comment:
[P2] Make the regression fail on the old attachment leak
Now that this regression runs by default, its oracle needs to fail on the
old attach-without-detach code. That leak is bounded by the persistent
scheduler pool, so phase 1 can attach the remaining workers and phases 2-4 then
plateau; comparing only the last phase to phase 1 (+4) can pass unchanged old
code. `jvmBefore` and all process-thread samples are collected but never
asserted. Please compare post-write minima to the warmed baseline with a
justified allowance, and verify the test is red on the pre-fix implementation.
##########
be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp:
##########
@@ -150,6 +194,7 @@ Status JniPaimonWriteBackend::close() {
if (close_status.ok()) {
_memory_manager.reset();
+ _spill_directory.reset();
Review Comment:
[P1] Wait for compaction before freeing the spill callback
Paimon 1.4.2 does not make a successful `writer.close()` a quiescence proof.
`MergeTreeWriter.close()` cancels the compaction future, but
`CompactFutureManager` catches `CancellationException` and clears it without
joining the callable, and `AbstractFileStoreWrite.close()` uses `shutdownNow()`
without awaiting termination. A clustering task that has not stopped can
therefore call the Doris spill/IO callback after `_spill_directory` is freed
(and after the IOManager deletes its files). Please join/await every SDK task
before returning close success, or retain the callback owner and physical spill
root until those tasks actually exit.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/DorisIOManager.java:
##########
@@ -0,0 +1,289 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.paimon;
+
+import org.apache.paimon.disk.BufferFileReader;
+import org.apache.paimon.disk.BufferFileWriter;
+import org.apache.paimon.disk.FileIOChannel;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.memory.Buffer;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/** Paimon IOManager adapter which charges temporary channel I/O to Doris
spill management. */
+final class DorisIOManager implements IOManager {
+ interface SpillAccountant {
+ void reserve(long bytes) throws IOException;
+
+ void rollback(long bytes);
+
+ void commitWrite(long bytes);
+
+ void recordRead(long bytes);
+
+ void release(long bytes);
+ }
+
+ private final IOManager delegate;
+ private final SpillAccountant accountant;
+ private final Map<String, Long> channelBytes = new ConcurrentHashMap<>();
+
+ static DorisIOManager create(String[] tempDirs, long nativeSpillDirectory)
{
+ return new DorisIOManager(
+ IOManager.create(tempDirs), new
NativeSpillAccountant(nativeSpillDirectory));
+ }
+
+ DorisIOManager(IOManager delegate, SpillAccountant accountant) {
+ this.delegate = delegate;
+ this.accountant = accountant;
+ }
+
+ @Override
+ public FileIOChannel.ID createChannel() {
+ return delegate.createChannel();
+ }
+
+ @Override
+ public FileIOChannel.ID createChannel(String prefix) {
+ return delegate.createChannel(prefix);
+ }
+
+ @Override
+ public String[] tempDirs() {
+ return delegate.tempDirs();
+ }
+
+ @Override
+ public String pickTempDir() {
+ return delegate.pickTempDir();
+ }
+
+ @Override
+ public FileIOChannel.Enumerator createChannelEnumerator() {
+ return delegate.createChannelEnumerator();
+ }
+
+ @Override
+ public BufferFileWriter createBufferFileWriter(FileIOChannel.ID channelID)
throws IOException {
+ return new
AccountingBufferFileWriter(delegate.createBufferFileWriter(channelID), this);
+ }
+
+ @Override
+ public BufferFileReader createBufferFileReader(FileIOChannel.ID channelID)
throws IOException {
+ return new
AccountingBufferFileReader(delegate.createBufferFileReader(channelID), this);
+ }
+
+ @Override
+ public void close() throws Exception {
+ try {
+ delegate.close();
+ } finally {
+ releaseDeletedChannels();
+ }
+ }
+
+ private void releaseDeletedChannels() {
+ long released = 0;
+ for (Map.Entry<String, Long> entry : channelBytes.entrySet()) {
+ if (!new File(entry.getKey()).exists()
+ && channelBytes.remove(entry.getKey(), entry.getValue())) {
+ released += entry.getValue();
+ }
+ }
+ accountant.release(released);
+ }
+
+ private void reserveWrite(FileIOChannel.ID channelID, long bytes) throws
IOException {
Review Comment:
[P2] Release reservations after lookup-buffer reset
`channelBytes` is released only through the wrapper deletion methods or when
the whole manager closes, but Paimon 1.4.2 lookup merge bypasses them:
`LookupMergeFunction.reset()` resets its spillable `ExternalBuffer`, whose
`clearChannels()` directly `File.delete()`s the tracked channel paths. The live
writer then continues to later groups without reconciling those missing paths,
so deleted candidate buffers stay charged and can eventually cause a false
`DISK_REACH_CAPACITY_LIMIT`. Please integrate this SDK deletion with an exact
release or reconcile missing tracked paths before subsequent reservations, and
add a spill-reset-spill lookup test.
--
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]
