This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 53764232f00 branch-4.1: [fix](paimon) manage JNI writer lifecycle and
spill (#66612)
53764232f00 is described below
commit 53764232f004d090d887f02b3033774a13898f08
Author: Socrates <[email protected]>
AuthorDate: Fri Aug 28 01:03:52 2026 +0800
branch-4.1: [fix](paimon) manage JNI writer lifecycle and spill (#66612)
### What problem does this PR solve?
Paimon JNI writers can create write-buffer, lookup, and clustering
temporary files while background compaction tasks are still running.
Previously, those files were not owned by Doris query spill lifecycle,
and native callbacks or memory could be released before all Java tasks
had stopped.
This PR fixes the lifecycle and spill-management issues:
- Places each Paimon writer's temporary directory lazily under one
Doris-managed, query-scoped spill root, following the same root
selection, capacity check, usage accounting, and query cleanup model as
internal spill.
- Accounts Paimon buffer-channel writes through Doris spill callbacks
and protects the query spill directory with a lease while asynchronous
SDK work may still access it.
- Stops and joins Paimon compaction work before releasing JNI resources,
native memory, or spill callbacks; if task termination cannot be
confirmed, dependent resources remain retained.
- Uses scoped JNI references on writer open and error paths and avoids
leaking native thread attachments.
- Adds deterministic coverage for managed spill paths, cleanup retry,
disabled-spill behavior, and writer thread lifecycle.
---
.../writer/paimon/jni_paimon_write_backend.cpp | 276 ++++++++++-----
.../sink/writer/paimon/jni_paimon_write_backend.h | 4 +-
be/src/exec/spill/spill_file_manager.cpp | 169 +++++++++
be/src/exec/spill/spill_file_manager.h | 49 +++
be/src/exec/spill/spill_file_writer.cpp | 2 +-
be/src/util/jni-util.h | 3 +
.../writer/paimon/paimon_write_backend_test.cpp | 2 +-
be/test/vec/spill/spill_file_test.cpp | 205 ++++++++++-
.../org/apache/doris/paimon/DorisIOManager.java | 392 +++++++++++++++++++++
.../org/apache/doris/paimon/PaimonJniWriter.java | 126 +++++--
.../apache/doris/paimon/DorisIOManagerTest.java | 223 ++++++++++++
.../apache/doris/paimon/PaimonJniWriterTest.java | 109 +++++-
.../write/test_paimon_write_external_paths.groovy | 45 ++-
.../test_paimon_write_thread_lifecycle.groovy | 41 ++-
14 files changed, 1479 insertions(+), 167 deletions(-)
diff --git a/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp
b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp
index 8a255d9ceba..d46fb1d99f6 100644
--- a/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp
+++ b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp
@@ -22,6 +22,7 @@
#include <arrow/io/memory.h>
#include <arrow/ipc/reader.h>
#include <arrow/record_batch.h>
+#include <fmt/format.h>
#include <algorithm>
#include <atomic>
@@ -33,44 +34,156 @@
#include "common/check.h"
#include "common/logging.h"
#include "exec/sink/writer/paimon/paimon_jni_memory_manager.h"
+#include "exec/spill/spill_file_manager.h"
#include "format/arrow/arrow_block_convertor.h"
#include "runtime/exec_env.h"
+#include "runtime/query_context.h"
#include "runtime/runtime_state.h"
#include "util/defer_op.h"
#include "util/jni-util.h"
#include "util/pretty_printer.h"
-#include "util/string_util.h"
namespace doris {
namespace {
constexpr std::string_view PAIMON_JNI_WRITER_IO_TMP_DIR =
"paimon_jni_writer_io_tmp";
+void throw_java_io_exception(JNIEnv* env, const std::string& message) {
+ jclass exception_class = env->FindClass("java/io/IOException");
+ env->ThrowNew(exception_class, message.c_str());
+ env->DeleteLocalRef(exception_class);
+}
+
+jobjectArray get_paimon_spill_directories(JNIEnv* env, jclass, jlong
spill_session_handle) {
+ auto* spill_session =
reinterpret_cast<ExternalSpillSession*>(spill_session_handle);
+ if (spill_session == nullptr) {
+ throw_java_io_exception(env, "Paimon external spill session is null");
+ return nullptr;
+ }
+
+ std::vector<std::string> paths;
+ Status st = spill_session->get_paths(&paths);
+ if (!st.ok()) {
+ throw_java_io_exception(env, st.to_string());
+ return nullptr;
+ }
+ jclass string_class = env->FindClass("java/lang/String");
+ if (string_class == nullptr) {
+ return nullptr;
+ }
+ jobjectArray result =
+ env->NewObjectArray(static_cast<jsize>(paths.size()),
string_class, nullptr);
+ env->DeleteLocalRef(string_class);
+ if (result == nullptr) {
+ return nullptr;
+ }
+ for (jsize i = 0; i < static_cast<jsize>(paths.size()); ++i) {
+ jstring path = env->NewStringUTF(paths[i].c_str());
+ if (path == nullptr) {
+ return nullptr;
+ }
+ env->SetObjectArrayElement(result, i, path);
+ env->DeleteLocalRef(path);
+ if (env->ExceptionCheck()) {
+ return nullptr;
+ }
+ }
+ return result;
+}
+
+void reserve_paimon_spill(JNIEnv* env, jclass, jlong spill_session_handle,
jstring path,
+ jlong bytes) {
+ auto* spill_session =
reinterpret_cast<ExternalSpillSession*>(spill_session_handle);
+ if (spill_session == nullptr || path == nullptr) {
+ throw_java_io_exception(env, "Paimon external spill session or path is
null");
+ return;
+ }
+ const char* path_chars = env->GetStringUTFChars(path, nullptr);
+ if (path_chars == nullptr) {
+ return;
+ }
+ std::string native_path(path_chars);
+ env->ReleaseStringUTFChars(path, path_chars);
+ Status st = spill_session->reserve(native_path, bytes);
+ if (!st.ok()) {
+ throw_java_io_exception(env, st.to_string());
+ }
+}
+
+void update_paimon_spill_accounting(JNIEnv* env, jclass, jlong
spill_session_handle, jstring path,
+ jlong current_bytes_delta, jlong
write_bytes,
+ jlong read_bytes) {
+ auto* spill_session =
reinterpret_cast<ExternalSpillSession*>(spill_session_handle);
+ if (spill_session == nullptr || path == nullptr) {
+ return;
+ }
+ const char* path_chars = env->GetStringUTFChars(path, nullptr);
+ if (path_chars == nullptr) {
+ return;
+ }
+ std::string native_path(path_chars);
+ env->ReleaseStringUTFChars(path, path_chars);
+ spill_session->update_accounting(native_path, current_bytes_delta,
write_bytes, read_bytes);
+}
+
+Status register_paimon_spill_natives(JNIEnv* env, jclass writer_class) {
+ static char get_spill_directories_name[] = "getPaimonSpillDirectories";
+ static char get_spill_directories_signature[] = "(J)[Ljava/lang/String;";
+ static char reserve_spill_name[] = "reservePaimonSpill";
+ static char reserve_spill_signature[] = "(JLjava/lang/String;J)V";
+ static char update_spill_name[] = "updatePaimonSpillAccounting";
+ static char update_spill_signature[] = "(JLjava/lang/String;JJJ)V";
+ static ::JNINativeMethod methods[] = {
+ {get_spill_directories_name, get_spill_directories_signature,
+ reinterpret_cast<void*>(&get_paimon_spill_directories)},
+ {reserve_spill_name, reserve_spill_signature,
+ reinterpret_cast<void*>(&reserve_paimon_spill)},
+ {update_spill_name, update_spill_signature,
+ reinterpret_cast<void*>(&update_paimon_spill_accounting)},
+ };
+ if (env->RegisterNatives(writer_class, methods,
+ static_cast<jint>(sizeof(methods) /
sizeof(methods[0]))) != JNI_OK) {
+ RETURN_IF_ERROR(Jni::Env::GetJniExceptionMsg(
+ env, true, "JNI exception registering Paimon spill native
methods: "));
+ return Status::JniError("Failed to register Paimon spill native
methods");
+ }
+ return Status::OK();
+}
+
std::atomic<bool>& paimon_jni_close_failed() {
static std::atomic<bool> failed {false};
return failed;
}
-std::mutex& retained_memory_managers_mutex() {
+struct RetainedPaimonResources {
+ std::unique_ptr<PaimonJniMemoryManager> memory_manager;
+ std::unique_ptr<ExternalSpillSession> spill_session;
+};
+
+std::mutex& retained_resources_mutex() {
static auto* mutex = new std::mutex();
return *mutex;
}
-std::vector<std::unique_ptr<PaimonJniMemoryManager>>&
retained_memory_managers() {
- static auto* managers = new
std::vector<std::unique_ptr<PaimonJniMemoryManager>>();
- return *managers;
+std::vector<RetainedPaimonResources>& retained_resources() {
+ static auto* resources = new std::vector<RetainedPaimonResources>();
+ return *resources;
}
-void retain_memory_after_failed_close(std::unique_ptr<PaimonJniMemoryManager>
manager) {
+void
retain_resources_after_failed_close(std::unique_ptr<PaimonJniMemoryManager>
memory_manager,
+ std::unique_ptr<ExternalSpillSession>
spill_session) {
// An unconfirmed Java close means a background Paimon task may still
reference this manager's
- // native pages. Quarantine the manager and stop admitting new writers so
repeated failures
- // cannot accumulate process-lifetime native memory without a bound.
+ // native pages or spill callbacks. Quarantine both resources and stop
admitting new writers so
+ // repeated failures cannot accumulate process-lifetime resources without
a bound.
paimon_jni_close_failed().store(true, std::memory_order_release);
- if (manager == nullptr) {
+ if (memory_manager == nullptr && spill_session == nullptr) {
return;
}
- std::lock_guard<std::mutex> lock(retained_memory_managers_mutex());
- retained_memory_managers().emplace_back(std::move(manager));
+ std::lock_guard<std::mutex> lock(retained_resources_mutex());
+ retained_resources().emplace_back(RetainedPaimonResources {
+ .memory_manager = std::move(memory_manager),
+ .spill_session = std::move(spill_session),
+ });
}
} // namespace
@@ -80,12 +193,9 @@ void
retain_memory_after_failed_close(std::unique_ptr<PaimonJniMemoryManager> ma
// ────────────────────────────────────────────────────────────
static constexpr const char* PAIMON_JNI_WRITER_CLASS =
"org/apache/doris/paimon/PaimonJniWriter";
-static constexpr const char* SCANNER_LOADER_CLASS =
- "org/apache/doris/common/classloader/ScannerLoader";
-
const char* const PAIMON_JNI_WRITER_OPEN_SIGNATURE =
"(Ljava/lang/String;Ljava/util/Map;[Ljava/lang/String;JLjava/lang/String;ZZLjava/lang/"
- "String;Ljava/lang/String;JJ)V";
+ "String;JJJ)V";
PaimonJniWriterOpenMode PaimonJniWriterOpenMode::from_write_mode(
TPaimonWriteMode::type write_mode) {
@@ -93,6 +203,8 @@ PaimonJniWriterOpenMode
PaimonJniWriterOpenMode::from_write_mode(
static_cast<jboolean>(write_mode == TPaimonWriteMode::CHANGELOG)};
}
+JniPaimonWriteBackend::JniPaimonWriteBackend() = default;
+
JniPaimonWriteBackend::~JniPaimonWriteBackend() {
Status st = close();
if (!st.ok()) {
@@ -104,6 +216,7 @@ Status JniPaimonWriteBackend::close() {
if (_jni_writer_obj == nullptr && _jni_writer_cls == nullptr) {
_memory_manager.reset();
_arrow_schema.reset();
+ _spill_session.reset();
_opened = false;
return Status::OK();
}
@@ -117,9 +230,11 @@ Status JniPaimonWriteBackend::close() {
_jni_writer_obj = nullptr;
_jni_writer_cls = nullptr;
if (java_users_may_exist) {
- retain_memory_after_failed_close(std::move(_memory_manager));
+ retain_resources_after_failed_close(std::move(_memory_manager),
+ std::move(_spill_session));
} else {
_memory_manager.reset();
+ _spill_session.reset();
}
_arrow_schema.reset();
_opened = false;
@@ -145,6 +260,7 @@ Status JniPaimonWriteBackend::close() {
if (close_status.ok()) {
_memory_manager.reset();
+ _spill_session.reset();
} else {
if (_memory_manager != nullptr) {
LOG(WARNING)
@@ -152,10 +268,9 @@ Status JniPaimonWriteBackend::close() {
<<
PrettyPrinter::print_bytes(_memory_manager->memory_limit()) << ", peak="
<<
PrettyPrinter::print_bytes(_memory_manager->native_peak_allocated_bytes());
}
- // Paimon may still have asynchronous flush or compaction tasks using
MemorySegments backed
- // by these pages. Retain this failed writer's ownership until process
exit to prevent UAF;
- // retain_memory_after_failed_close also fences subsequent Paimon JNI
writer admission.
- retain_memory_after_failed_close(std::move(_memory_manager));
+ // Paimon may still have asynchronous tasks using Doris-backed pages
or spill callbacks.
+ // Retain ownership until process exit and fence subsequent writer
admission.
+ retain_resources_after_failed_close(std::move(_memory_manager),
std::move(_spill_session));
}
_arrow_schema.reset();
_opened = false;
@@ -172,46 +287,6 @@ Status JniPaimonWriteBackend::_check_jni_exception(JNIEnv*
env, const std::strin
return Status::OK();
}
-Status JniPaimonWriteBackend::_load_writer_class(JNIEnv* env, jclass*
writer_class) {
- jclass loader_class = env->FindClass(SCANNER_LOADER_CLASS);
- RETURN_IF_ERROR(_check_jni_exception(env, "find ScannerLoader"));
-
- jmethodID loader_constructor = env->GetMethodID(loader_class, "<init>",
"()V");
- jmethodID get_loaded_class = env->GetMethodID(loader_class,
"getLoadedClass",
-
"(Ljava/lang/String;)Ljava/lang/Class;");
- RETURN_IF_ERROR(_check_jni_exception(env, "resolve ScannerLoader
methods"));
-
- jobject loader = env->NewObject(loader_class, loader_constructor);
- jstring class_name = env->NewStringUTF(PAIMON_JNI_WRITER_CLASS);
- auto* loaded_class =
- static_cast<jclass>(env->CallObjectMethod(loader,
get_loaded_class, class_name));
- RETURN_IF_ERROR(_check_jni_exception(env, "load PaimonJniWriter"));
-
- *writer_class = loaded_class;
- env->DeleteLocalRef(class_name);
- env->DeleteLocalRef(loader);
- env->DeleteLocalRef(loader_class);
- return Status::OK();
-}
-
-static jobject _to_java_options(JNIEnv* env, const std::map<std::string,
std::string>& options) {
- jclass map_cls = env->FindClass("java/util/HashMap");
- jmethodID map_ctor = env->GetMethodID(map_cls, "<init>", "()V");
- jmethodID put_method = env->GetMethodID(
- map_cls, "put",
"(Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;");
-
- jobject map_obj = env->NewObject(map_cls, map_ctor);
- for (const auto& kv : options) {
- jstring key = env->NewStringUTF(kv.first.c_str());
- jstring val = env->NewStringUTF(kv.second.c_str());
- env->CallObjectMethod(map_obj, put_method, key, val);
- env->DeleteLocalRef(key);
- env->DeleteLocalRef(val);
- }
- env->DeleteLocalRef(map_cls);
- return map_obj;
-}
-
static Status _get_paimon_arrow_schema(JNIEnv* env, jobject writer, jmethodID
get_schema_id,
std::shared_ptr<arrow::Schema>* schema)
{
auto schema_bytes = static_cast<jbyteArray>(env->CallObjectMethod(writer,
get_schema_id));
@@ -243,7 +318,6 @@ static Status _get_paimon_arrow_schema(JNIEnv* env, jobject
writer, jmethodID ge
*schema = reader_result.ValueOrDie()->schema();
return Status::OK();
}
-
Status JniPaimonWriteBackend::open(const TPaimonTableSink& sink, RuntimeState*
state,
RuntimeProfile* profile) {
if (paimon_jni_close_failed().load(std::memory_order_acquire)) {
@@ -269,14 +343,25 @@ Status JniPaimonWriteBackend::open(const
TPaimonTableSink& sink, RuntimeState* s
JNIEnv* env = nullptr;
RETURN_IF_ERROR(Jni::Env::Get(&env));
+ if (env->PushLocalFrame(32) != JNI_OK) {
+ Status st = _check_jni_exception(env, "create PaimonJniWriter open
local reference frame");
+ return st.ok() ? Status::InternalError("Failed to create JNI local
reference frame") : st;
+ }
+ Defer pop_local_frame([&]() { env->PopLocalFrame(nullptr); });
// Step 1: Load PaimonJniWriter class through ScannerLoader (Paimon jars
are
// not on the default application classpath, so FindClass won't work).
- jclass local_cls = nullptr;
- RETURN_IF_ERROR(_load_writer_class(env, &local_cls));
- _jni_writer_cls = static_cast<jclass>(env->NewGlobalRef(local_cls));
- env->DeleteLocalRef(local_cls);
+ Jni::LocalObject local_writer_class;
+ RETURN_IF_ERROR(
+ Jni::Util::get_jni_scanner_class(env, PAIMON_JNI_WRITER_CLASS,
&local_writer_class));
+ auto writer_class = static_cast<jclass>(local_writer_class.get());
+ _jni_writer_cls = static_cast<jclass>(env->NewGlobalRef(writer_class));
+ RETURN_IF_ERROR(_check_jni_exception(env, "create global PaimonJniWriter
class reference"));
+ if (_jni_writer_cls == nullptr) {
+ return Status::JniError("Failed to create global PaimonJniWriter class
reference");
+ }
RETURN_IF_ERROR(PaimonJniMemoryManager::register_natives(env,
_jni_writer_cls));
+ RETURN_IF_ERROR(register_paimon_spill_natives(env, _jni_writer_cls));
// Step 2: Cache JNI method IDs for write, prepareCommit, abort, close.
jmethodID open_id = env->GetMethodID(_jni_writer_cls, "open",
PAIMON_JNI_WRITER_OPEN_SIGNATURE);
@@ -285,54 +370,55 @@ Status JniPaimonWriteBackend::open(const
TPaimonTableSink& sink, RuntimeState* s
_prepare_commit_id = env->GetMethodID(_jni_writer_cls, "prepareCommit",
"()[[B");
_abort_id = env->GetMethodID(_jni_writer_cls, "abort", "()V");
_close_id = env->GetMethodID(_jni_writer_cls, "close", "()V");
- RETURN_IF_ERROR(_check_jni_exception(env, "GetMethodID"));
+ RETURN_IF_ERROR(_check_jni_exception(env, "resolve PaimonJniWriter
methods"));
// Step 3: Create the Java PaimonJniWriter instance.
jmethodID ctor_id = env->GetMethodID(_jni_writer_cls, "<init>", "()V");
jobject local_obj = env->NewObject(_jni_writer_cls, ctor_id);
- RETURN_IF_ERROR(_check_jni_exception(env, "NewObject"));
+ RETURN_IF_ERROR(_check_jni_exception(env, "create PaimonJniWriter"));
_jni_writer_obj = env->NewGlobalRef(local_obj);
- env->DeleteLocalRef(local_obj);
+ RETURN_IF_ERROR(_check_jni_exception(env, "create global PaimonJniWriter
object reference"));
+ if (_jni_writer_obj == nullptr) {
+ return Status::JniError("Failed to create global PaimonJniWriter
object reference");
+ }
- // Step 4: Build Java arguments and call PaimonJniWriter.open().
+ // Step 4: Create a lazy query-scoped spill session. Java requests its
path only when Paimon
+ // first uses the IOManager, so a memory-only writer does not depend on
spill storage.
+ auto* spill_file_manager = state->exec_env()->spill_file_mgr();
+ if (spill_file_manager != nullptr) {
+ auto spill_relative_path =
+ fmt::format("{}-{}", PAIMON_JNI_WRITER_IO_TMP_DIR,
spill_file_manager->next_id());
+ RETURN_IF_ERROR(spill_file_manager->create_external_spill_session(
+ spill_relative_path, state->get_query_ctx(), &_spill_session));
+ }
+
+ // Step 5: Build Java arguments and call PaimonJniWriter.open().
const std::map<std::string, std::string> empty_config;
jstring j_serialized_table =
env->NewStringUTF(sink.serialized_table.c_str());
- jobject j_hadoop_config =
- _to_java_options(env, sink.__isset.hadoop_config ?
sink.hadoop_config : empty_config);
+ Jni::LocalObject j_hadoop_config;
+ RETURN_IF_ERROR(Jni::Util::convert_to_java_map(
+ env, sink.__isset.hadoop_config ? sink.hadoop_config :
empty_config, &j_hadoop_config));
jstring j_commit_user = env->NewStringUTF(sink.commit_user.c_str());
jstring j_time_zone = env->NewStringUTF(state->timezone().c_str());
- std::vector<std::string> spill_directories;
- for (const auto& store_path : state->exec_env()->store_paths()) {
- spill_directories.push_back(store_path.path + "/" +
- std::string(PAIMON_JNI_WRITER_IO_TMP_DIR));
- }
- DORIS_CHECK(!spill_directories.empty());
- jstring j_spill_directories = env->NewStringUTF(join(spill_directories,
":").c_str());
jclass string_cls = env->FindClass("java/lang/String");
jobjectArray j_cols =
env->NewObjectArray(static_cast<jsize>(sink.column_names.size()),
string_cls, nullptr);
for (size_t i = 0; i < sink.column_names.size(); ++i) {
- jstring str = env->NewStringUTF(sink.column_names[i].c_str());
- env->SetObjectArrayElement(j_cols, static_cast<jsize>(i), str);
- env->DeleteLocalRef(str);
+ jstring column_name = env->NewStringUTF(sink.column_names[i].c_str());
+ env->SetObjectArrayElement(j_cols, static_cast<jsize>(i), column_name);
+ env->DeleteLocalRef(column_name);
}
+ RETURN_IF_ERROR(_check_jni_exception(env, "build PaimonJniWriter open
arguments"));
PaimonJniWriterOpenMode open_mode =
PaimonJniWriterOpenMode::from_write_mode(sink.write_mode);
- env->CallVoidMethod(_jni_writer_obj, open_id, j_serialized_table,
j_hadoop_config, j_cols,
- static_cast<jlong>(sink.transaction_id),
j_commit_user, open_mode.overwrite,
- open_mode.changelog, j_time_zone, j_spill_directories,
- static_cast<jlong>(_memory_manager->memory_limit()),
- reinterpret_cast<jlong>(_memory_manager.get()));
- Status st = _check_jni_exception(env, "open");
-
- env->DeleteLocalRef(j_serialized_table);
- env->DeleteLocalRef(j_hadoop_config);
- env->DeleteLocalRef(j_commit_user);
- env->DeleteLocalRef(j_time_zone);
- env->DeleteLocalRef(j_spill_directories);
- env->DeleteLocalRef(j_cols);
- env->DeleteLocalRef(string_cls);
+ env->CallVoidMethod(
+ _jni_writer_obj, open_id, j_serialized_table,
j_hadoop_config.get(), j_cols,
+ static_cast<jlong>(sink.transaction_id), j_commit_user,
open_mode.overwrite,
+ open_mode.changelog, j_time_zone,
static_cast<jlong>(_memory_manager->memory_limit()),
+ reinterpret_cast<jlong>(_memory_manager.get()),
+ _spill_session == nullptr ? 0 :
reinterpret_cast<jlong>(_spill_session.get()));
+ Status st = _check_jni_exception(env, "open PaimonJniWriter");
if (st.ok()) {
st = _get_paimon_arrow_schema(env, _jni_writer_obj,
get_arrow_schema_id, &_arrow_schema);
diff --git a/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h
b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h
index 918a7afe232..d5a5de3b9ea 100644
--- a/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h
+++ b/be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h
@@ -35,6 +35,7 @@ class Schema;
namespace doris {
+class ExternalSpillSession;
class RuntimeState;
extern const char* const PAIMON_JNI_WRITER_OPEN_SIGNATURE;
@@ -57,6 +58,7 @@ struct PaimonJniWriterOpenMode {
/// common backend contract.
class JniPaimonWriteBackend final : public IPaimonWriteBackend {
public:
+ JniPaimonWriteBackend();
~JniPaimonWriteBackend() override;
Status open(const TPaimonTableSink& sink, RuntimeState* state,
@@ -67,7 +69,6 @@ public:
private:
Status _check_jni_exception(JNIEnv* env, const std::string& method_name);
- Status _load_writer_class(JNIEnv* env, jclass* writer_class);
void _refresh_memory_profile();
// JNI global references — live for the duration of this backend.
@@ -82,6 +83,7 @@ private:
std::unique_ptr<PaimonJniMemoryManager> _memory_manager;
std::shared_ptr<arrow::Schema> _arrow_schema;
+ std::unique_ptr<ExternalSpillSession> _spill_session;
RuntimeProfile::Counter* _native_page_memory_limit = nullptr;
RuntimeProfile::Counter* _native_page_memory_peak = nullptr;
bool _opened = false;
diff --git a/be/src/exec/spill/spill_file_manager.cpp
b/be/src/exec/spill/spill_file_manager.cpp
index 048e1d2e551..a5243bbc28d 100644
--- a/be/src/exec/spill/spill_file_manager.cpp
+++ b/be/src/exec/spill/spill_file_manager.cpp
@@ -22,6 +22,7 @@
#include <algorithm>
#include <filesystem>
+#include <limits>
#include <memory>
#include <string>
#include <utility>
@@ -31,15 +32,106 @@
#include "exec/spill/spill_file.h"
#include "io/fs/file_system.h"
#include "io/fs/local_file_system.h"
+#include "runtime/query_context.h"
#include "storage/olap_define.h"
#include "util/debug_points.h"
#include "util/parse_util.h"
#include "util/pretty_printer.h"
#include "util/time.h"
+#include "util/uid_util.h"
namespace doris {
#include "common/compile_check_begin.h"
+ExternalSpillSession::ExternalSpillSession(SpillFileManager* manager,
QueryContext* query_context,
+ std::string relative_path)
+ : _manager(manager),
+ _query_context(query_context->weak_from_this()),
+ _resource_context(query_context->resource_ctx()),
+ _query_id(print_id(query_context->query_id())),
+ _relative_path(std::move(relative_path)) {
+ DCHECK(_manager != nullptr);
+ DCHECK(!_query_context.expired());
+ DCHECK(_resource_context != nullptr);
+}
+
+ExternalSpillSession::~ExternalSpillSession() {
+ _manager->_release_external_spill_session(this);
+}
+
+Status ExternalSpillSession::get_paths(std::vector<std::string>* paths) {
+ if (paths == nullptr) {
+ return Status::InvalidArgument("External spill paths output must not
be null");
+ }
+ std::lock_guard lock(_mutex);
+ if (_data_dir == nullptr) {
+ RETURN_IF_ERROR(_manager->_initialize_external_spill_session(this));
+ }
+ *paths = {_path};
+ return Status::OK();
+}
+
+bool ExternalSpillSession::_contains(const std::string& path) const {
+ return path == _path ||
+ (path.size() > _path.size() && path.starts_with(_path) &&
path[_path.size()] == '/');
+}
+
+Status ExternalSpillSession::reserve(const std::string& path, int64_t bytes) {
+ if (bytes <= 0) {
+ return Status::InvalidArgument("External spill reservation must be
positive: {}", bytes);
+ }
+
+ std::lock_guard lock(_mutex);
+ if (_data_dir == nullptr || !_contains(path)) {
+ return Status::InvalidArgument("External spill path is not managed by
Doris: {}", path);
+ }
+ if (bytes > std::numeric_limits<int64_t>::max() - _accounted_bytes) {
+ return Status::InvalidArgument("External spill reservation overflows:
bytes={}", bytes);
+ }
+ if (_data_dir->reach_capacity_limit(bytes)) {
+ return Status::Error<ErrorCode::DISK_REACH_CAPACITY_LIMIT>(
+ "External spill write exceeds the Doris spill storage limit:
path={}, bytes={}",
+ path, bytes);
+ }
+ // Match SpillFileWriter: check capacity before the write, then account
the accepted bytes.
+ _data_dir->update_spill_data_usage(bytes);
+ _accounted_bytes += bytes;
+ return Status::OK();
+}
+
+void ExternalSpillSession::update_accounting(const std::string& path, int64_t
current_bytes_delta,
+ int64_t write_bytes, int64_t
read_bytes) {
+ int64_t released_bytes = 0;
+ SpillDataDir* data_dir = nullptr;
+ {
+ std::lock_guard lock(_mutex);
+ if (_data_dir == nullptr || !_contains(path)) {
+ LOG(WARNING) << "Ignoring accounting for unmanaged external spill
path: " << path;
+ return;
+ }
+ data_dir = _data_dir;
+ if (current_bytes_delta < 0) {
+ const int64_t requested_release =
+ current_bytes_delta == std::numeric_limits<int64_t>::min()
+ ? std::numeric_limits<int64_t>::max()
+ : -current_bytes_delta;
+ released_bytes = std::min(requested_release, _accounted_bytes);
+ _accounted_bytes -= released_bytes;
+ }
+ }
+ if (released_bytes > 0) {
+ data_dir->update_spill_data_usage(-released_bytes);
+ }
+ if (write_bytes > 0) {
+
_resource_context->io_context()->update_spill_write_bytes_to_local_storage(write_bytes);
+ _manager->update_spill_write_bytes(write_bytes);
+ }
+ if (read_bytes > 0) {
+
_resource_context->io_context()->update_spill_read_bytes_from_local_storage(read_bytes);
+ _manager->update_spill_read_bytes(read_bytes);
+ }
+}
+
SpillFileManager::~SpillFileManager() {
// QueryContext destruction can still queue failed deletions after stop(),
for example while
// VDataStreamMgr is being destroyed. Retry them once more before dropping
the in-memory state.
@@ -173,6 +265,76 @@ Status SpillFileManager::create_spill_file(const
std::string& relative_path,
return Status::OK();
}
+Status SpillFileManager::create_external_spill_session(
+ const std::string& relative_path, QueryContext* query_context,
+ std::unique_ptr<ExternalSpillSession>* spill_session) {
+ if (query_context == nullptr || spill_session == nullptr) {
+ return Status::InvalidArgument(
+ "External spill session requires QueryContext and output
session");
+ }
+
+ spill_session->reset(new ExternalSpillSession(this, query_context,
relative_path));
+ return Status::OK();
+}
+
+Status
SpillFileManager::_initialize_external_spill_session(ExternalSpillSession*
spill_session) {
+ auto query_context = spill_session->_query_context.lock();
+ if (query_context == nullptr) {
+ return Status::Cancelled("Query ended before the external spill
session was initialized");
+ }
+ auto* 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.");
+ }
+
+ const auto query_dir =
data_dir->get_spill_data_path(spill_session->_query_id);
+ {
+ // QueryContext teardown uses the regular pending-deletion path while
this lease is live.
+ std::lock_guard lock(_pending_query_spill_directories_mutex);
+ ++_external_spill_directory_leases[query_dir];
+ }
+ query_context->record_spill_data_dir(data_dir);
+ spill_session->_data_dir = data_dir;
+ spill_session->_path = query_dir + "/" + spill_session->_relative_path;
+ return Status::OK();
+}
+
+void SpillFileManager::_release_external_spill_session(ExternalSpillSession*
spill_session) {
+ std::lock_guard session_lock(spill_session->_mutex);
+ if (spill_session->_data_dir == nullptr) {
+ return;
+ }
+
+ if (spill_session->_accounted_bytes > 0) {
+ // Match SpillFile::gc(): QueryContext owns physical cleanup and its
retry path, while the
+ // writer releases logical usage when its lifetime ends.
+
spill_session->_data_dir->update_spill_data_usage(-spill_session->_accounted_bytes);
+ spill_session->_accounted_bytes = 0;
+ }
+
+ const auto query_dir =
spill_session->_data_dir->get_spill_data_path(spill_session->_query_id);
+ std::lock_guard directory_lock(_pending_query_spill_directories_mutex);
+ auto it = _external_spill_directory_leases.find(query_dir);
+ DCHECK(it != _external_spill_directory_leases.end());
+ if (it == _external_spill_directory_leases.end()) {
+ return;
+ }
+ DCHECK_GT(it->second, 0);
+ if (--it->second == 0) {
+ _external_spill_directory_leases.erase(it);
+ }
+}
+
+SpillDataDir* SpillFileManager::_get_store_for_spill() {
+ auto data_dirs = _get_stores_for_spill(TStorageMedium::type::SSD);
+ if (data_dirs.empty()) {
+ data_dirs = _get_stores_for_spill(TStorageMedium::type::HDD);
+ }
+ // Select the first available data dir (sorted by usage ascending).
+ return data_dirs.empty() ? nullptr : data_dirs.front();
+}
+
void SpillFileManager::delete_spill_file(SpillFileSPtr spill_file) {
if (!spill_file) {
LOG(WARNING) << "[spill][delete] null spill_file";
@@ -197,6 +359,13 @@ void SpillFileManager::delete_query_spill_directory(const
std::string& query_id,
Status SpillFileManager::_try_delete_query_spill_directory(
const PendingQuerySpillDirectory& pending_directory) {
+ {
+ std::lock_guard lock(_pending_query_spill_directories_mutex);
+ if
(_external_spill_directory_leases.contains(pending_directory.query_dir)) {
+ return Status::InternalError("external spill directory is still in
use: {}",
+ pending_directory.query_dir);
+ }
+ }
DBUG_EXECUTE_IF("fault_inject::spill_file_manager::delete_query_spill_directory",
{
return Status::Error<INTERNAL_ERROR>("injected query spill directory
deletion failure");
});
diff --git a/be/src/exec/spill/spill_file_manager.h
b/be/src/exec/spill/spill_file_manager.h
index 7455789c216..6d3c34a9059 100644
--- a/be/src/exec/spill/spill_file_manager.h
+++ b/be/src/exec/spill/spill_file_manager.h
@@ -41,6 +41,8 @@ class AtomicGauge;
using UIntGauge = AtomicGauge<uint64_t>;
class MetricEntity;
struct MetricPrototype;
+class QueryContext;
+class ResourceContext;
class SpillFileManager;
class SpillDataDir {
@@ -113,6 +115,38 @@ private:
IntGauge* spill_disk_has_spill_data = nullptr;
IntGauge* spill_disk_has_spill_gc_data = nullptr;
};
+
+// Adapts one external writer to the same root selection, capacity accounting
and query cleanup
+// used by Doris spill files.
+class ExternalSpillSession {
+public:
+ ~ExternalSpillSession();
+
+ Status get_paths(std::vector<std::string>* paths);
+
+ Status reserve(const std::string& path, int64_t bytes);
+
+ void update_accounting(const std::string& path, int64_t
current_bytes_delta,
+ int64_t write_bytes, int64_t read_bytes);
+
+private:
+ friend class SpillFileManager;
+
+ ExternalSpillSession(SpillFileManager* manager, QueryContext*
query_context,
+ std::string relative_path);
+ bool _contains(const std::string& path) const;
+
+ SpillFileManager* _manager;
+ std::weak_ptr<QueryContext> _query_context;
+ std::shared_ptr<ResourceContext> _resource_context;
+ std::string _query_id;
+ std::string _relative_path;
+ SpillDataDir* _data_dir = nullptr;
+ std::string _path;
+ int64_t _accounted_bytes = 0;
+ std::mutex _mutex;
+};
+
class SpillFileManager {
public:
~SpillFileManager();
@@ -128,6 +162,12 @@ public:
// e.g. "query_id/sort-node_id-task_id-unique_id"
Status create_spill_file(const std::string& relative_path, SpillFileSPtr&
spill_file);
+ // Create a lazy managed session for an external spill implementation. A
spill root is selected
+ // and registered only when the external implementation first requests its
path.
+ Status create_external_spill_session(const std::string& relative_path,
+ QueryContext* query_context,
+
std::unique_ptr<ExternalSpillSession>* spill_session);
+
/// Get a unique ID for constructing spill file paths.
uint64_t next_id() { return id_++; }
@@ -145,6 +185,8 @@ public:
void update_spill_read_bytes(int64_t bytes) {
_spill_read_bytes_counter->increment(bytes); }
private:
+ friend class ExternalSpillSession;
+
struct PendingQuerySpillDirectory {
int failed_count {0};
std::string query_dir;
@@ -155,15 +197,22 @@ private:
void _spill_gc_thread_callback();
Status _try_delete_query_spill_directory(const PendingQuerySpillDirectory&
pending_directory);
void _retry_pending_query_spill_directories();
+ Status _initialize_external_spill_session(ExternalSpillSession*
spill_session);
+ void _release_external_spill_session(ExternalSpillSession* spill_session);
std::vector<SpillDataDir*> _get_stores_for_spill(TStorageMedium::type
storage_medium);
+ SpillDataDir* _get_store_for_spill();
std::unordered_map<std::string, std::unique_ptr<SpillDataDir>>
_spill_store_map;
CountDownLatch _stop_background_threads_latch;
std::shared_ptr<Thread> _spill_gc_thread;
+ // Query cleanup uses the regular pending-deletion path. External leases
only defer deletion
+ // while an SDK task can still access the same query directory; filesystem
I/O never holds this
+ // mutex.
std::mutex _pending_query_spill_directories_mutex;
std::vector<PendingQuerySpillDirectory> _pending_query_spill_directories;
+ std::unordered_map<std::string, size_t> _external_spill_directory_leases;
std::atomic_uint64_t id_ = 0;
diff --git a/be/src/exec/spill/spill_file_writer.cpp
b/be/src/exec/spill/spill_file_writer.cpp
index 62320ef9c05..441fe4aaac5 100644
--- a/be/src/exec/spill/spill_file_writer.cpp
+++ b/be/src/exec/spill/spill_file_writer.cpp
@@ -273,4 +273,4 @@ Status SpillFileWriter::_write_internal(const Block& block,
return status;
}
-} // namespace doris
\ No newline at end of file
+} // namespace doris
diff --git a/be/src/util/jni-util.h b/be/src/util/jni-util.h
index 16fbe1b587f..956e4c4df7d 100644
--- a/be/src/util/jni-util.h
+++ b/be/src/util/jni-util.h
@@ -606,6 +606,9 @@ public:
bool uninitialized() const { return _obj == nullptr; }
+ // Access the JNI handle without changing ownership.
+ jobject get() const { return _obj; }
+
void reset(JNIEnv* env) {
if (_obj == nullptr) {
return;
diff --git a/be/test/exec/sink/writer/paimon/paimon_write_backend_test.cpp
b/be/test/exec/sink/writer/paimon/paimon_write_backend_test.cpp
index e19104f16da..387e8932793 100644
--- a/be/test/exec/sink/writer/paimon/paimon_write_backend_test.cpp
+++ b/be/test/exec/sink/writer/paimon/paimon_write_backend_test.cpp
@@ -34,7 +34,7 @@ TEST(PaimonWriteBackendFactoryTest, SelectBackendType) {
TEST(JniPaimonWriteBackendTest, OpenAbiAndWriteModes) {
EXPECT_STREQ(
"(Ljava/lang/String;Ljava/util/Map;[Ljava/lang/String;JLjava/lang/String;ZZLjava/lang/"
- "String;Ljava/lang/String;JJ)V",
+ "String;JJJ)V",
PAIMON_JNI_WRITER_OPEN_SIGNATURE);
auto append =
PaimonJniWriterOpenMode::from_write_mode(TPaimonWriteMode::APPEND);
diff --git a/be/test/vec/spill/spill_file_test.cpp
b/be/test/vec/spill/spill_file_test.cpp
index 286a056c3cf..dff4a427d69 100644
--- a/be/test/vec/spill/spill_file_test.cpp
+++ b/be/test/vec/spill/spill_file_test.cpp
@@ -93,7 +93,7 @@ protected:
auto st =
io::global_local_filesystem()->create_directory(spill_data_dir->path(), false);
ASSERT_TRUE(st.ok()) << "create directory failed: " << st.to_string();
auto second_spill_data_dir = std::make_unique<SpillDataDir>(
- _second_spill_dir, 1024L * 1024 * 128, TStorageMedium::HDD);
+ _second_spill_dir, 1024L * 1024 * 128, TStorageMedium::SSD);
st =
io::global_local_filesystem()->create_directory(second_spill_data_dir->path(),
false);
ASSERT_TRUE(st.ok()) << "create directory failed: " << st.to_string();
@@ -417,10 +417,15 @@ TEST_F(SpillFileTest, OpenCanRetryAfterFailure) {
ASSERT_TRUE(st.ok());
}
- const auto part_path =
+ const auto first_part_path =
std::filesystem::path(_spill_dir) / "spill" / "test_query" /
"open_retry" / "0";
- const auto backup_path =
- std::filesystem::path(_spill_dir) / "spill" / "test_query" /
"open_retry" / "0.bak";
+ const auto second_part_path =
+ std::filesystem::path(_second_spill_dir) / "spill" / "test_query"
/ "open_retry" / "0";
+ const auto part_path =
+ std::filesystem::exists(first_part_path) ? first_part_path :
second_part_path;
+ ASSERT_TRUE(std::filesystem::exists(part_path));
+ auto backup_path = part_path;
+ backup_path += ".bak";
std::filesystem::rename(part_path, backup_path);
@@ -935,13 +940,17 @@ TEST_F(SpillFileTest, GCCleansUpFiles) {
st = writer->close();
ASSERT_TRUE(st.ok());
- // Remember the spill directory path
- spill_file_dir = _data_dir_ptr->get_spill_data_path() +
"/test_query/gc_test";
-
- // Verify directory exists
+ // Remember the selected spill directory path.
bool exists = false;
- st = io::global_local_filesystem()->exists(spill_file_dir, &exists);
- ASSERT_TRUE(st.ok());
+ for (auto* data_dir : {_data_dir_ptr, _second_data_dir_ptr}) {
+ auto candidate = data_dir->get_spill_data_path() +
"/test_query/gc_test";
+ st = io::global_local_filesystem()->exists(candidate, &exists);
+ ASSERT_TRUE(st.ok());
+ if (exists) {
+ spill_file_dir = std::move(candidate);
+ break;
+ }
+ }
ASSERT_TRUE(exists);
// spill_file goes out of scope here, destructor calls gc()
@@ -1295,10 +1304,17 @@ TEST_F(SpillFileTest,
DeleteSpillFileThroughManagerSynchronously) {
st = writer->close();
ASSERT_TRUE(st.ok());
- auto spill_file_dir =
_data_dir_ptr->get_spill_data_path("test_query/mgr_delete");
+ std::string spill_file_dir;
bool exists = false;
- st = io::global_local_filesystem()->exists(spill_file_dir, &exists);
- ASSERT_TRUE(st.ok());
+ for (auto* data_dir : {_data_dir_ptr, _second_data_dir_ptr}) {
+ auto candidate =
data_dir->get_spill_data_path("test_query/mgr_delete");
+ st = io::global_local_filesystem()->exists(candidate, &exists);
+ ASSERT_TRUE(st.ok());
+ if (exists) {
+ spill_file_dir = std::move(candidate);
+ break;
+ }
+ }
ASSERT_TRUE(exists);
ExecEnv::GetInstance()->spill_file_mgr()->delete_spill_file(spill_file);
@@ -1321,6 +1337,163 @@ TEST_F(SpillFileTest, ManagerNextId) {
ASSERT_EQ(id3, id2 + 1);
}
+TEST_F(SpillFileTest, ManagerAllocatesExternalSpillSessionOnManagedRoot) {
+ TUniqueId query_id;
+ query_id.hi = 21;
+ query_id.lo = 22;
+ auto query_id_str = print_id(query_id);
+ auto query_ctx = MockQueryContext::create(query_id);
+
+ std::unique_ptr<ExternalSpillSession> spill_session;
+ auto st =
ExecEnv::GetInstance()->spill_file_mgr()->create_external_spill_session(
+ "paimon", query_ctx.get(), &spill_session);
+
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ std::vector<std::string> paths;
+ st = spill_session->get_paths(&paths);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ ASSERT_EQ(paths.size(), 1);
+ const std::string first_path =
_data_dir_ptr->get_spill_data_path(query_id_str) + "/paimon";
+ const std::string second_path =
+ _second_data_dir_ptr->get_spill_data_path(query_id_str) +
"/paimon";
+ ASSERT_TRUE(paths.front() == first_path || paths.front() == second_path);
+ bool exists = false;
+ for (const auto& path : paths) {
+ st = io::global_local_filesystem()->exists(path, &exists);
+ ASSERT_TRUE(st.ok());
+ ASSERT_FALSE(exists);
+ }
+
+ const std::string& selected_path = paths.front();
+ const std::string channel = selected_path + "/paimon-io-test/channel";
+ ASSERT_TRUE(spill_session->reserve(channel, 1024).ok());
+ auto* selected_data_dir = selected_path == first_path ? _data_dir_ptr :
_second_data_dir_ptr;
+ auto* unselected_data_dir =
+ selected_data_dir == _data_dir_ptr ? _second_data_dir_ptr :
_data_dir_ptr;
+ ASSERT_EQ(selected_data_dir->get_spill_data_bytes(), 1024);
+ ASSERT_EQ(unselected_data_dir->get_spill_data_bytes(), 0);
+ spill_session->update_accounting(channel, -256, 0, 0);
+ ASSERT_EQ(selected_data_dir->get_spill_data_bytes(), 768);
+ _create_residual_file(channel);
+
+ // Query teardown must not remove a directory while an asynchronous
external writer can still
+ // use its native callback. The regular spill GC handles deferred cleanup
after lease release.
+ query_ctx.reset();
+ auto query_dir = selected_data_dir->get_spill_data_path(query_id_str);
+ st = io::global_local_filesystem()->exists(query_dir, &exists);
+ ASSERT_TRUE(st.ok());
+ ASSERT_TRUE(exists);
+
+ spill_session.reset();
+ // Match SpillFile::gc(): logical usage is released with the writer, while
QueryContext owns
+ // physical deletion and retries.
+ ASSERT_EQ(selected_data_dir->get_spill_data_bytes(), 0);
+ ASSERT_EQ(unselected_data_dir->get_spill_data_bytes(), 0);
+
+ st = io::global_local_filesystem()->exists(query_dir, &exists);
+ ASSERT_TRUE(st.ok());
+ ASSERT_TRUE(exists);
+ ExecEnv::GetInstance()->spill_file_mgr()->gc(10000);
+ st = io::global_local_filesystem()->exists(query_dir, &exists);
+ ASSERT_TRUE(st.ok());
+ ASSERT_FALSE(exists);
+ ASSERT_EQ(selected_data_dir->get_spill_data_bytes(), 0);
+}
+
+TEST_F(SpillFileTest, ExternalSpillSessionSkipsFullManagedRoot) {
+ TUniqueId query_id;
+ query_id.hi = 23;
+ query_id.lo = 24;
+ auto query_id_str = print_id(query_id);
+ auto query_ctx = MockQueryContext::create(query_id);
+
+ const int64_t unavailable_bytes = _data_dir_ptr->get_spill_data_limit() +
1;
+ _data_dir_ptr->update_spill_data_usage(unavailable_bytes);
+ Defer release_full_root([&]() {
_data_dir_ptr->update_spill_data_usage(-unavailable_bytes); });
+
+ std::unique_ptr<ExternalSpillSession> spill_session;
+ auto st =
ExecEnv::GetInstance()->spill_file_mgr()->create_external_spill_session(
+ "paimon", query_ctx.get(), &spill_session);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+
+ std::vector<std::string> paths;
+ st = spill_session->get_paths(&paths);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ ASSERT_EQ(paths.size(), 1);
+ ASSERT_EQ(paths.front(),
_second_data_dir_ptr->get_spill_data_path(query_id_str) + "/paimon");
+}
+
+TEST_F(SpillFileTest, ExternalSpillDirectoryCleanupRetriesAfterLeaseRelease) {
+ ExecEnv::GetInstance()->spill_file_mgr()->stop();
+ TUniqueId query_id;
+ query_id.hi = 33;
+ query_id.lo = 34;
+ auto query_ctx = MockQueryContext::create(query_id);
+
+ std::unique_ptr<ExternalSpillSession> spill_session;
+ ASSERT_TRUE(ExecEnv::GetInstance()
+ ->spill_file_mgr()
+ ->create_external_spill_session("paimon",
query_ctx.get(), &spill_session)
+ .ok());
+ std::vector<std::string> paths;
+ ASSERT_TRUE(spill_session->get_paths(&paths).ok());
+ auto* selected_data_dir =
+
paths.front().starts_with(_data_dir_ptr->get_spill_data_path(print_id(query_id)))
+ ? _data_dir_ptr
+ : _second_data_dir_ptr;
+ ASSERT_TRUE(spill_session->reserve(paths.front() + "/paimon-io/channel",
1024).ok());
+ _create_residual_file(paths.front() + "/paimon-io/channel");
+
+ const bool previous_enable_debug_points = config::enable_debug_points;
+ constexpr auto debug_point_name =
+ "fault_inject::spill_file_manager::delete_query_spill_directory";
+ Defer restore_debug_point([&] {
+ DebugPoints::instance()->remove(debug_point_name);
+ config::enable_debug_points = previous_enable_debug_points;
+ });
+ auto debug_point = std::make_shared<DebugPoint>();
+ debug_point->execute_limit = 1;
+ config::enable_debug_points = true;
+ DebugPoints::instance()->add(debug_point_name, debug_point);
+
+ query_ctx.reset();
+ spill_session.reset();
+ ASSERT_EQ(selected_data_dir->get_spill_data_bytes(), 0);
+
+ const auto query_dir =
selected_data_dir->get_spill_data_path(print_id(query_id));
+ bool exists = false;
+ ASSERT_TRUE(io::global_local_filesystem()->exists(query_dir,
&exists).ok());
+ ASSERT_TRUE(exists);
+
+ ExecEnv::GetInstance()->spill_file_mgr()->gc(10000);
+ ASSERT_TRUE(io::global_local_filesystem()->exists(query_dir,
&exists).ok());
+ ASSERT_TRUE(exists);
+ ExecEnv::GetInstance()->spill_file_mgr()->gc(10000);
+ ASSERT_TRUE(io::global_local_filesystem()->exists(query_dir,
&exists).ok());
+ ASSERT_FALSE(exists);
+}
+
+TEST_F(SpillFileTest, ExternalSpillSessionIsLazyWhenNoRootAvailable) {
+ TUniqueId query_id;
+ query_id.hi = 25;
+ query_id.lo = 26;
+ auto query_ctx = MockQueryContext::create(query_id);
+
+
_data_dir_ptr->update_spill_data_usage(_data_dir_ptr->get_spill_data_limit());
+
_second_data_dir_ptr->update_spill_data_usage(_second_data_dir_ptr->get_spill_data_limit());
+ Defer release_full_roots([&]() {
+
_data_dir_ptr->update_spill_data_usage(-_data_dir_ptr->get_spill_data_limit());
+ _second_data_dir_ptr->update_spill_data_usage(
+ -_second_data_dir_ptr->get_spill_data_limit());
+ });
+
+ std::unique_ptr<ExternalSpillSession> spill_session;
+ auto st =
ExecEnv::GetInstance()->spill_file_mgr()->create_external_spill_session(
+ "paimon", query_ctx.get(), &spill_session);
+ ASSERT_TRUE(st.ok()) << st.to_string();
+ ASSERT_NE(spill_session, nullptr);
+}
+
TEST_F(SpillFileTest, ManagerCreateMultipleFiles) {
const int num_files = 5;
std::vector<SpillFileSPtr> files;
@@ -1511,7 +1684,8 @@ TEST_F(SpillFileTest, DataDirCapacityTracking) {
spill_file);
ASSERT_TRUE(st.ok());
- auto initial_bytes = _data_dir_ptr->get_spill_data_bytes();
+ auto initial_bytes =
+ _data_dir_ptr->get_spill_data_bytes() +
_second_data_dir_ptr->get_spill_data_bytes();
SpillFileWriterSPtr writer;
st = spill_file->create_writer(_runtime_state.get(), _profile.get(),
writer);
@@ -1527,7 +1701,8 @@ TEST_F(SpillFileTest, DataDirCapacityTracking) {
st = writer->close();
ASSERT_TRUE(st.ok());
- auto after_write_bytes = _data_dir_ptr->get_spill_data_bytes();
+ auto after_write_bytes =
+ _data_dir_ptr->get_spill_data_bytes() +
_second_data_dir_ptr->get_spill_data_bytes();
ASSERT_GT(after_write_bytes, initial_bytes);
}
diff --git
a/fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/DorisIOManager.java
b/fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/DorisIOManager.java
new file mode 100644
index 00000000000..7f9cba65517
--- /dev/null
+++
b/fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/DorisIOManager.java
@@ -0,0 +1,392 @@
+// 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.io.UncheckedIOException;
+import java.nio.channels.FileChannel;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/** Paimon IOManager adapter which charges temporary I/O to Doris spill
management. */
+final class DorisIOManager implements IOManager {
+ interface SpillAccountant {
+ String[] getSpillDirectories() throws IOException;
+
+ void reserve(String path, long bytes) throws IOException;
+
+ void rollback(String path, long bytes);
+
+ void commitWrite(String path, long bytes);
+
+ void recordRead(String path, long bytes);
+
+ void release(String path, long bytes);
+ }
+
+ static final class SpillDirectoryCleanupException extends IOException {
+ private SpillDirectoryCleanupException(Exception cause) {
+ super("Failed to eagerly clean a Paimon spill directory", cause);
+ }
+ }
+
+ private final SpillAccountant accountant;
+ private final Map<String, Long> channelBytes = new ConcurrentHashMap<>();
+ private final Map<String, Integer> activeChannelWriters = new
ConcurrentHashMap<>();
+ private volatile IOManager delegate;
+
+ static DorisIOManager create(long nativeSpillSession) {
+ return new DorisIOManager(new
NativeSpillAccountant(nativeSpillSession));
+ }
+
+ DorisIOManager(SpillAccountant accountant) {
+ this(null, accountant);
+ }
+
+ DorisIOManager(IOManager delegate, SpillAccountant accountant) {
+ this.accountant = accountant;
+ this.delegate = delegate;
+ }
+
+ private IOManager delegate() throws IOException {
+ if (delegate == null) {
+ synchronized (this) {
+ if (delegate == null) {
+ String[] spillDirectories =
accountant.getSpillDirectories();
+ if (spillDirectories == null || spillDirectories.length ==
0) {
+ throw new IOException("Doris spill manager returned no
available directories");
+ }
+ delegate = IOManager.create(spillDirectories);
+ }
+ }
+ }
+ return delegate;
+ }
+
+ private IOManager uncheckedDelegate() {
+ try {
+ return delegate();
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to initialize the Doris
spill directory", e);
+ }
+ }
+
+ @Override
+ public FileIOChannel.ID createChannel() {
+ return uncheckedDelegate().createChannel();
+ }
+
+ @Override
+ public FileIOChannel.ID createChannel(String prefix) {
+ return uncheckedDelegate().createChannel(prefix);
+ }
+
+ @Override
+ public String[] tempDirs() {
+ return uncheckedDelegate().tempDirs();
+ }
+
+ @Override
+ public String pickTempDir() {
+ return uncheckedDelegate().pickTempDir();
+ }
+
+ @Override
+ public FileIOChannel.Enumerator createChannelEnumerator() {
+ return uncheckedDelegate().createChannelEnumerator();
+ }
+
+ @Override
+ public BufferFileWriter createBufferFileWriter(FileIOChannel.ID channelID)
throws IOException {
+ // ExternalBuffer clears old buffer channels with File.delete(),
bypassing IOManager
+ // deletion. Release those known channels before reserving more space.
+ releaseDeletedChannels();
+ 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 {
+ IOManager initializedDelegate = delegate;
+ if (initializedDelegate == null) {
+ return;
+ }
+
+ try {
+ initializedDelegate.close();
+ } catch (Exception e) {
+ releaseDeletedChannels();
+ throw new SpillDirectoryCleanupException(e);
+ }
+ releaseDeletedChannels();
+ }
+
+ private boolean releaseDeletedChannels() {
+ boolean releasedAny = false;
+ for (Map.Entry<String, Long> entry : channelBytes.entrySet()) {
+ if (!activeChannelWriters.containsKey(entry.getKey())
+ && !new File(entry.getKey()).exists()
+ && channelBytes.remove(entry.getKey(), entry.getValue())) {
+ accountant.release(entry.getKey(), entry.getValue());
+ releasedAny = true;
+ }
+ }
+ return releasedAny;
+ }
+
+ private void reserveWrite(FileIOChannel.ID channelID, long bytes) throws
IOException {
+ String path = channelID.getPath();
+ activeChannelWriters.merge(path, 1, Integer::sum);
+ boolean accounted = false;
+ boolean tracked = false;
+ try {
+ try {
+ accountant.reserve(path, bytes);
+ } catch (IOException reserveFailure) {
+ if (!releaseDeletedChannels()) {
+ throw reserveFailure;
+ }
+ accountant.reserve(path, bytes);
+ }
+ accounted = true;
+ channelBytes.merge(path, bytes, Long::sum);
+ tracked = true;
+ } finally {
+ if (!tracked) {
+ if (accounted) {
+ accountant.rollback(path, bytes);
+ }
+ finishWrite(channelID);
+ }
+ }
+ }
+
+ private void finishWrite(FileIOChannel.ID channelID) {
+ activeChannelWriters.computeIfPresent(channelID.getPath(),
+ (ignored, writers) -> writers == 1 ? null : writers - 1);
+ }
+
+ private void releaseChannel(FileIOChannel.ID channelID) {
+ Long released = channelBytes.remove(channelID.getPath());
+ if (released != null) {
+ accountant.release(channelID.getPath(), released);
+ }
+ }
+
+ private static final class NativeSpillAccountant implements
SpillAccountant {
+ private final long nativeSpillSession;
+
+ private NativeSpillAccountant(long nativeSpillSession) {
+ this.nativeSpillSession = nativeSpillSession;
+ }
+
+ @Override
+ public String[] getSpillDirectories() throws IOException {
+ return
PaimonJniWriter.getPaimonSpillDirectories(nativeSpillSession);
+ }
+
+ @Override
+ public void reserve(String path, long bytes) throws IOException {
+ PaimonJniWriter.reservePaimonSpill(nativeSpillSession, path,
bytes);
+ }
+
+ @Override
+ public void rollback(String path, long bytes) {
+ PaimonJniWriter.updatePaimonSpillAccounting(
+ nativeSpillSession, path, -bytes, 0, 0);
+ }
+
+ @Override
+ public void commitWrite(String path, long bytes) {
+ PaimonJniWriter.updatePaimonSpillAccounting(
+ nativeSpillSession, path, 0, bytes, 0);
+ }
+
+ @Override
+ public void recordRead(String path, long bytes) {
+ PaimonJniWriter.updatePaimonSpillAccounting(
+ nativeSpillSession, path, 0, 0, bytes);
+ }
+
+ @Override
+ public void release(String path, long bytes) {
+ PaimonJniWriter.updatePaimonSpillAccounting(
+ nativeSpillSession, path, -bytes, 0, 0);
+ }
+ }
+
+ private static final class AccountingBufferFileWriter implements
BufferFileWriter {
+ private final BufferFileWriter delegate;
+ private final DorisIOManager manager;
+
+ private AccountingBufferFileWriter(BufferFileWriter delegate,
DorisIOManager manager) {
+ this.delegate = delegate;
+ this.manager = manager;
+ }
+
+ @Override
+ public void writeBlock(Buffer buffer) throws IOException {
+ long bytes = Integer.BYTES + buffer.getSize();
+ manager.reserveWrite(getChannelID(), bytes);
+ try {
+ delegate.writeBlock(buffer);
+ manager.accountant.commitWrite(getChannelID().getPath(),
bytes);
+ } catch (IOException | RuntimeException writeFailure) {
+ try {
+ delegate.closeAndDelete();
+ } catch (IOException | RuntimeException cleanupFailure) {
+ writeFailure.addSuppressed(cleanupFailure);
+ }
+ if (!getChannelID().getPathFile().exists()) {
+ manager.releaseChannel(getChannelID());
+ }
+ throw writeFailure;
+ } finally {
+ manager.finishWrite(getChannelID());
+ }
+ }
+
+ @Override
+ public FileIOChannel.ID getChannelID() {
+ return delegate.getChannelID();
+ }
+
+ @Override
+ public long getSize() throws IOException {
+ return delegate.getSize();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return delegate.isClosed();
+ }
+
+ @Override
+ public void close() throws IOException {
+ delegate.close();
+ }
+
+ @Override
+ public void deleteChannel() {
+ try {
+ delegate.deleteChannel();
+ } finally {
+ if (!getChannelID().getPathFile().exists()) {
+ manager.releaseChannel(getChannelID());
+ }
+ }
+ }
+
+ @Override
+ public FileChannel getNioFileChannel() {
+ return delegate.getNioFileChannel();
+ }
+
+ @Override
+ public void closeAndDelete() throws IOException {
+ try {
+ delegate.closeAndDelete();
+ } finally {
+ if (!getChannelID().getPathFile().exists()) {
+ manager.releaseChannel(getChannelID());
+ }
+ }
+ }
+ }
+
+ private static final class AccountingBufferFileReader implements
BufferFileReader {
+ private final BufferFileReader delegate;
+ private final DorisIOManager manager;
+
+ private AccountingBufferFileReader(BufferFileReader delegate,
DorisIOManager manager) {
+ this.delegate = delegate;
+ this.manager = manager;
+ }
+
+ @Override
+ public void readInto(Buffer buffer) throws IOException {
+ long position = delegate.getNioFileChannel().position();
+ delegate.readInto(buffer);
+ manager.accountant.recordRead(
+ getChannelID().getPath(),
delegate.getNioFileChannel().position() - position);
+ }
+
+ @Override
+ public boolean hasReachedEndOfFile() {
+ return delegate.hasReachedEndOfFile();
+ }
+
+ @Override
+ public FileIOChannel.ID getChannelID() {
+ return delegate.getChannelID();
+ }
+
+ @Override
+ public long getSize() throws IOException {
+ return delegate.getSize();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return delegate.isClosed();
+ }
+
+ @Override
+ public void close() throws IOException {
+ delegate.close();
+ }
+
+ @Override
+ public void deleteChannel() {
+ try {
+ delegate.deleteChannel();
+ } finally {
+ if (!getChannelID().getPathFile().exists()) {
+ manager.releaseChannel(getChannelID());
+ }
+ }
+ }
+
+ @Override
+ public FileChannel getNioFileChannel() {
+ return delegate.getNioFileChannel();
+ }
+
+ @Override
+ public void closeAndDelete() throws IOException {
+ try {
+ delegate.closeAndDelete();
+ } finally {
+ if (!getChannelID().getPathFile().exists()) {
+ manager.releaseChannel(getChannelID());
+ }
+ }
+ }
+ }
+}
diff --git
a/fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java
b/fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java
index 0d2511660d7..cf2536b7a5f 100644
---
a/fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java
+++
b/fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java
@@ -32,8 +32,6 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.crosspartition.IndexBootstrap;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.disk.IOManager;
-import org.apache.paimon.disk.IOManagerImpl;
import org.apache.paimon.index.BucketAssigner;
import org.apache.paimon.index.HashBucketAssigner;
import org.apache.paimon.index.SimpleHashBucketAssigner;
@@ -46,12 +44,12 @@ import org.apache.paimon.table.sink.PartitionKeyExtractor;
import org.apache.paimon.table.sink.RowPartitionKeyExtractor;
import org.apache.paimon.table.sink.SinkRecord;
import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.utils.ExecutorThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
import java.nio.ByteBuffer;
-import java.nio.file.Files;
-import java.nio.file.Paths;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Collections;
@@ -61,6 +59,9 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
/**
* JNI entry point for Paimon write operations.
@@ -90,6 +91,7 @@ public class PaimonJniWriter {
private static final Logger LOG =
LoggerFactory.getLogger(PaimonJniWriter.class);
private static final int APPEND_ONLY_WRITER_MIN_PAGES = 1;
private static final int MERGE_TREE_WRITER_MIN_PAGES = 3;
+ private static final long COMPACTION_CLOSE_TIMEOUT_SECONDS = 60;
private final ClassLoader classLoader;
private final PaimonCommitCodec commitCodec = new PaimonCommitCodec();
@@ -101,7 +103,8 @@ public class PaimonJniWriter {
private PaimonWriteSchema writeSchema;
private FileStoreTable table;
private TableWriteImpl<?> writer;
- private IOManager ioManager;
+ private DorisIOManager ioManager;
+ private ExecutorService compactionExecutor;
private long commitIdentifier;
private String commitUser;
private BucketMode bucketMode;
@@ -144,14 +147,15 @@ public class PaimonJniWriter {
* @param overwrite whether this is an overwrite write
* @param changelogWrite whether the first input column contains a row
change operation
* @param timeZone normalized Doris session timezone used for Paimon
LTZ values
- * @param spillDirectories Doris storage-root scoped directories for
Paimon write-buffer spill
* @param nativePageMemoryLimitBytes maximum Doris-managed Paimon page
memory
* @param nativeMemoryManager opaque BE manager used to allocate tracked
native pages
+ * @param nativeSpillSession opaque managed spill session used for
capacity and I/O accounting
*/
public void open(String serializedTable, Map<String, String> hadoopConfig,
String[] columnNames, long transactionId, String
commitUser,
- boolean overwrite, boolean changelogWrite, String
timeZone, String spillDirectories,
- long nativePageMemoryLimitBytes, long
nativeMemoryManager) throws Exception {
+ boolean overwrite, boolean changelogWrite, String
timeZone,
+ long nativePageMemoryLimitBytes, long nativeMemoryManager,
+ long nativeSpillSession) throws Exception {
try (ThreadClassLoaderContext ignored = new
ThreadClassLoaderContext(classLoader)) {
if (nativePageMemoryLimitBytes <= 0) {
throw new IllegalArgumentException(
@@ -187,10 +191,10 @@ public class PaimonJniWriter {
table,
commitUser,
overwrite,
- spillDirectories,
coreOptions,
nativePageMemoryLimitBytes,
- nativeMemoryManager);
+ nativeMemoryManager,
+ nativeSpillSession);
return null;
} catch (Throwable t) {
try {
@@ -312,14 +316,17 @@ public class PaimonJniWriter {
// ────────────────────────────────────────────────────────────
private void openFileStoreWriter(FileStoreTable table, String commitUser,
boolean overwrite,
- String spillDirectories, CoreOptions coreOptions, long
nativePageMemoryLimitBytes,
- long nativeMemoryManager) throws Exception {
+ CoreOptions coreOptions, long nativePageMemoryLimitBytes,
+ long nativeMemoryManager, long nativeSpillSession) throws
Exception {
writer = table.newWrite(commitUser);
+ compactionExecutor = Executors.newSingleThreadExecutor(
+ new ExecutorThreadFactory("doris-paimon-compaction"));
+ writer.withCompactExecutor(compactionExecutor);
if (overwrite) {
writer.withIgnorePreviousFiles(true);
}
- openMemoryResources(table, coreOptions, spillDirectories,
nativePageMemoryLimitBytes,
- nativeMemoryManager);
+ openMemoryResources(table, coreOptions, nativePageMemoryLimitBytes,
+ nativeMemoryManager, nativeSpillSession);
openDynamicBucketAssigner(table, commitUser, overwrite, coreOptions);
}
@@ -340,9 +347,9 @@ public class PaimonJniWriter {
private void openMemoryResources(
FileStoreTable table,
CoreOptions coreOptions,
- String spillDirectories,
long nativePageMemoryLimitBytes,
- long nativeMemoryManager) throws Exception {
+ long nativeMemoryManager,
+ long nativeSpillSession) throws Exception {
int pageSize = coreOptions.pageSize();
long writeBufferSize = coreOptions.writeBufferSize();
// Paimon creates merge-tree bucket writers lazily on the first write.
Their
@@ -359,17 +366,12 @@ public class PaimonJniWriter {
LOG.info("Paimon writer uses Doris-managed memory pool: limit={}
bytes, pageSize={}",
memoryPoolFactory.totalBufferSize(), pageSize);
- if (!coreOptions.writeBufferSpillable()) {
- return;
- }
-
- String[] splitDirectories = IOManagerImpl.splitPaths(spillDirectories);
- for (String directory : splitDirectories) {
- Files.createDirectories(Paths.get(directory));
- }
- ioManager = IOManager.create(splitDirectories);
+ // All Paimon temporary files, including lookup and clustering files
written directly by
+ // Paimon, use the same Doris-managed directory. DorisIOManager
requests that directory only
+ // on its first actual use, so a memory-only writer does not depend on
spill storage.
+ ioManager = DorisIOManager.create(nativeSpillSession);
writer.withIOManager(ioManager);
- LOG.info("Paimon writer spill enabled: dirs={}", spillDirectories);
+ LOG.info("Paimon writer uses a lazy Doris-managed spill session");
}
static long validateAndGetMemoryPoolLimit(long writeBufferSize,
@@ -576,13 +578,36 @@ public class PaimonJniWriter {
throw new IllegalStateException(
"A previous Paimon SDK close failed; native memory cannot
be released safely");
}
- Exception failure = closeResource(writer, null);
- failure = closeResource(globalIndexAssigner, failure);
- failure = closeResource(ioManager, failure);
+ Exception lifecycleFailure = closeResource(writer, null);
+ Exception compactionFailure = closeCompactionExecutor();
+ if (compactionFailure != null) {
+ lifecycleFailure = appendFailure(lifecycleFailure,
compactionFailure);
+ // The task may still reference Doris-backed memory and spill
files. Leave all dependent
+ // Java resources reachable and open; the native backend will
retain their handles.
+ sdkCloseFailed = true;
+ throw lifecycleFailure;
+ }
+ lifecycleFailure = closeResource(globalIndexAssigner,
lifecycleFailure);
+ Exception cleanupFailure = closeResource(ioManager, null);
+ boolean physicalCleanupFailure =
+ cleanupFailure instanceof
DorisIOManager.SpillDirectoryCleanupException;
+ if (cleanupFailure != null && !physicalCleanupFailure) {
+ lifecycleFailure = appendFailure(lifecycleFailure, cleanupFailure);
+ }
clearWriterState();
- if (failure != null) {
+ if (lifecycleFailure != null) {
+ if (physicalCleanupFailure) {
+ lifecycleFailure.addSuppressed(cleanupFailure);
+ }
sdkCloseFailed = true;
- throw failure;
+ throw lifecycleFailure;
+ }
+ if (physicalCleanupFailure) {
+ // The QueryContext owns the parent spill directory and its GC
retry path. Failure to
+ // eagerly remove Paimon's nested directory is not evidence that
Java tasks still hold
+ // native memory, so it must not fence every later Paimon writer
on this BE.
+ LOG.warn("Failed to eagerly clean a Paimon spill directory; Doris
spill GC will retry",
+ cleanupFailure);
}
}
@@ -601,6 +626,35 @@ public class PaimonJniWriter {
return previousFailure;
}
+ private Exception closeCompactionExecutor() {
+ if (compactionExecutor == null) {
+ return null;
+ }
+ ExecutorService executor = compactionExecutor;
+ compactionExecutor = null;
+ executor.shutdownNow();
+ try {
+ if (!executor.awaitTermination(COMPACTION_CLOSE_TIMEOUT_SECONDS,
TimeUnit.SECONDS)) {
+ IllegalStateException failure = new IllegalStateException(
+ "Paimon compaction did not stop within "
+ + COMPACTION_CLOSE_TIMEOUT_SECONDS + "
seconds");
+ return failure;
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return e;
+ }
+ return null;
+ }
+
+ private static Exception appendFailure(Exception previousFailure,
Exception failure) {
+ if (previousFailure == null) {
+ return failure;
+ }
+ previousFailure.addSuppressed(failure);
+ return previousFailure;
+ }
+
private void clearWriterState() {
writer = null;
table = null;
@@ -641,6 +695,16 @@ public class PaimonJniWriter {
static native ByteBuffer allocatePaimonMemoryPage(long
nativeMemoryManager, int bytes);
+ static native String[] getPaimonSpillDirectories(long nativeSpillSession)
+ throws IOException;
+
+ static native void reservePaimonSpill(long nativeSpillSession, String
path, long bytes)
+ throws IOException;
+
+ static native void updatePaimonSpillAccounting(
+ long nativeSpillSession, String path,
+ long currentBytesDelta, long writeBytes, long readBytes);
+
private static class PartitionBucket {
private final BinaryRow partition;
private final int bucket;
diff --git
a/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/DorisIOManagerTest.java
b/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/DorisIOManagerTest.java
new file mode 100644
index 00000000000..fd684288d80
--- /dev/null
+++
b/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/DorisIOManagerTest.java
@@ -0,0 +1,223 @@
+// 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.disk.IOManagerImpl;
+import org.apache.paimon.memory.Buffer;
+import org.apache.paimon.memory.MemorySegment;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+public class DorisIOManagerTest {
+ @TempDir
+ private Path tempDir;
+
+ @Test
+ public void testSpillDirectoryIsLazyAndVisibleIoIsAccounted() throws
Exception {
+ Path managedSpillPath = tempDir.resolve("query/paimon");
+ RecordingSpillAccountant accountant = new
RecordingSpillAccountant(managedSpillPath);
+ Assertions.assertFalse(Files.exists(managedSpillPath));
+
+ try (DorisIOManager manager = new DorisIOManager(accountant)) {
+ Assertions.assertEquals(0, accountant.directoryRequests);
+ Assertions.assertFalse(Files.exists(managedSpillPath));
+
+ FileIOChannel.ID channel = manager.createChannel();
+ Assertions.assertEquals(1, accountant.directoryRequests);
+ Assertions.assertTrue(Files.isDirectory(managedSpillPath));
+ Buffer buffer = Buffer.create(MemorySegment.wrap(new byte[16]),
16);
+
+ BufferFileWriter writer = manager.createBufferFileWriter(channel);
+ writer.writeBlock(buffer);
+ writer.close();
+
+ Assertions.assertEquals(20, accountant.reservedBytes);
+ Assertions.assertEquals(20, accountant.currentBytes);
+ Assertions.assertEquals(20, accountant.writtenBytes);
+
+ BufferFileReader reader = manager.createBufferFileReader(channel);
+ reader.readInto(Buffer.create(MemorySegment.wrap(new byte[16]),
0));
+ Assertions.assertEquals(20, accountant.readBytes);
+ reader.closeAndDelete();
+
+ Assertions.assertEquals(0, accountant.currentBytes);
+ Assertions.assertEquals(20, accountant.releasedBytes);
+ }
+ }
+
+ @Test
+ public void testCloseBeforeUseDoesNotRequestSpillDirectory() throws
Exception {
+ Path managedSpillPath = tempDir.resolve("query/paimon");
+ RecordingSpillAccountant accountant = new
RecordingSpillAccountant(managedSpillPath);
+
+ new DorisIOManager(accountant).close();
+
+ Assertions.assertEquals(0, accountant.directoryRequests);
+ Assertions.assertFalse(Files.exists(managedSpillPath));
+ }
+
+ @Test
+ public void testDirectDeletionIsReleasedBeforeNextWriter() throws
Exception {
+ RecordingSpillAccountant accountant = new
RecordingSpillAccountant(tempDir);
+ try (DorisIOManager manager = new DorisIOManager(accountant)) {
+ FileIOChannel.ID deletedChannel = manager.createChannel();
+ BufferFileWriter firstWriter =
manager.createBufferFileWriter(deletedChannel);
+ firstWriter.writeBlock(Buffer.create(MemorySegment.wrap(new
byte[8]), 8));
+ firstWriter.close();
+ Assertions.assertTrue(deletedChannel.getPathFile().delete());
+
+ FileIOChannel.ID nextChannel = manager.createChannel();
+ BufferFileWriter nextWriter =
manager.createBufferFileWriter(nextChannel);
+ nextWriter.writeBlock(Buffer.create(MemorySegment.wrap(new
byte[4]), 4));
+ nextWriter.close();
+
+ Assertions.assertEquals(20, accountant.reservedBytes);
+ Assertions.assertEquals(8, accountant.currentBytes);
+ Assertions.assertEquals(12, accountant.releasedBytes);
+ }
+ }
+
+ @Test
+ public void testAllManagedSpillDirectoriesArePassedToPaimon() throws
Exception {
+ Path first = tempDir.resolve("spill-a");
+ Path second = tempDir.resolve("spill-b");
+ RecordingSpillAccountant accountant = new
RecordingSpillAccountant(first, second);
+
+ try (DorisIOManager manager = new DorisIOManager(accountant)) {
+ boolean usedFirst = false;
+ boolean usedSecond = false;
+ for (int i = 0; i < 4; i++) {
+ Path channel = manager.createChannel().getPathFile().toPath();
+ usedFirst |= channel.startsWith(first);
+ usedSecond |= channel.startsWith(second);
+ }
+ Assertions.assertTrue(usedFirst);
+ Assertions.assertTrue(usedSecond);
+ Assertions.assertEquals(1, accountant.directoryRequests);
+ }
+ }
+
+ @Test
+ public void testWriteFailureRollsBackWhenChannelIsDeleted() throws
Exception {
+ RecordingSpillAccountant accountant = new
RecordingSpillAccountant(tempDir);
+ IOManager delegate = IOManager.create(tempDir.toString());
+ FileIOChannel.ID channel = delegate.createChannel();
+ BufferFileWriter closedWriter =
delegate.createBufferFileWriter(channel);
+ closedWriter.close();
+ IOManager failingWriteManager = new IOManagerImpl(tempDir.toString()) {
+ @Override
+ public BufferFileWriter createBufferFileWriter(FileIOChannel.ID
ignored) {
+ return closedWriter;
+ }
+ };
+
+ try (DorisIOManager manager = new DorisIOManager(failingWriteManager,
accountant)) {
+ BufferFileWriter writer = manager.createBufferFileWriter(channel);
+ Assertions.assertThrows(IOException.class,
+ () ->
writer.writeBlock(Buffer.create(MemorySegment.wrap(new byte[8]), 8)));
+ Assertions.assertFalse(channel.getPathFile().exists());
+ Assertions.assertEquals(12, accountant.reservedBytes);
+ Assertions.assertEquals(0, accountant.currentBytes);
+ } finally {
+ delegate.close();
+ }
+ }
+
+ @Test
+ public void testManagerCloseFailureKeepsExistingChannelAccounted() throws
Exception {
+ RecordingSpillAccountant accountant = new
RecordingSpillAccountant(tempDir);
+ IOManager failingCloseManager = new IOManagerImpl(tempDir.toString()) {
+ @Override
+ public void close() throws Exception {
+ throw new IOException("injected close failure");
+ }
+ };
+ DorisIOManager manager = new DorisIOManager(failingCloseManager,
accountant);
+ FileIOChannel.ID channel = manager.createChannel();
+ BufferFileWriter writer = manager.createBufferFileWriter(channel);
+ writer.writeBlock(Buffer.create(MemorySegment.wrap(new byte[8]), 8));
+ writer.close();
+
+ Assertions.assertThrows(
+ DorisIOManager.SpillDirectoryCleanupException.class,
manager::close);
+ Assertions.assertTrue(channel.getPathFile().exists());
+ Assertions.assertEquals(12, accountant.currentBytes);
+
+ writer.deleteChannel();
+ Assertions.assertEquals(0, accountant.currentBytes);
+ Assertions.assertEquals(12, accountant.releasedBytes);
+ }
+
+ private static final class RecordingSpillAccountant implements
DorisIOManager.SpillAccountant {
+ private final Path[] spillDirectories;
+ private long directoryRequests;
+ private long reservedBytes;
+ private long currentBytes;
+ private long writtenBytes;
+ private long readBytes;
+ private long releasedBytes;
+
+ private RecordingSpillAccountant(Path... spillDirectories) {
+ this.spillDirectories = spillDirectories;
+ }
+
+ @Override
+ public String[] getSpillDirectories() {
+ directoryRequests++;
+ return java.util.Arrays.stream(spillDirectories)
+ .map(Path::toString)
+ .toArray(String[]::new);
+ }
+
+ @Override
+ public void reserve(String path, long bytes) {
+ reservedBytes += bytes;
+ currentBytes += bytes;
+ }
+
+ @Override
+ public void rollback(String path, long bytes) {
+ currentBytes -= bytes;
+ }
+
+ @Override
+ public void commitWrite(String path, long bytes) {
+ writtenBytes += bytes;
+ }
+
+ @Override
+ public void recordRead(String path, long bytes) {
+ readBytes += bytes;
+ }
+
+ @Override
+ public void release(String path, long bytes) {
+ releasedBytes += bytes;
+ currentBytes -= bytes;
+ }
+ }
+}
diff --git
a/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/PaimonJniWriterTest.java
b/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/PaimonJniWriterTest.java
index c552d61c885..5b91e363f7b 100644
---
a/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/PaimonJniWriterTest.java
+++
b/fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/PaimonJniWriterTest.java
@@ -17,14 +17,31 @@
package org.apache.doris.paimon;
+import org.apache.paimon.disk.BufferFileWriter;
+import org.apache.paimon.disk.FileIOChannel;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.disk.IOManagerImpl;
+import org.apache.paimon.memory.Buffer;
+import org.apache.paimon.memory.MemorySegment;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import java.io.IOException;
+import java.lang.reflect.Field;
import java.net.URL;
import java.net.URLClassLoader;
+import java.nio.file.Path;
import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
public class PaimonJniWriterTest {
+ @TempDir
+ private Path tempDir;
@Test
public void testManagedMemoryPoolRequiresAtLeastOnePage() {
@@ -65,8 +82,7 @@ public class PaimonJniWriterTest {
try {
Assertions.assertThrows(Exception.class, () -> writer.open(
"not-a-serialized-table", Collections.emptyMap(), new
String[0],
- 1L, "test-user", false, false, "UTC",
System.getProperty("java.io.tmpdir"),
- 64L * 1024 * 1024, 1L));
+ 1L, "test-user", false, false, "UTC", 64L * 1024 * 1024,
1L, 1L));
Assertions.assertSame(testClassLoader,
thread.getContextClassLoader());
} finally {
try {
@@ -118,4 +134,93 @@ public class PaimonJniWriterTest {
testClassLoader.close();
}
}
+
+ @Test
+ public void testCloseWaitsForCompactionExecutor() throws Exception {
+ PaimonJniWriter writer = new PaimonJniWriter();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch stopped = new CountDownLatch(1);
+ try {
+ executor.execute(() -> {
+ started.countDown();
+ try {
+ new CountDownLatch(1).await();
+ } catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
+ } finally {
+ stopped.countDown();
+ }
+ });
+ Assertions.assertTrue(started.await(10, TimeUnit.SECONDS));
+
+ Field executorField =
PaimonJniWriter.class.getDeclaredField("compactionExecutor");
+ executorField.setAccessible(true);
+ executorField.set(writer, executor);
+
+ writer.close();
+ Assertions.assertTrue(stopped.await(10, TimeUnit.SECONDS));
+ Assertions.assertTrue(executor.isTerminated());
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testSpillDirectoryCleanupFailureDoesNotFenceWriter() throws
Exception {
+ IOManager failingCloseManager = new IOManagerImpl(tempDir.toString()) {
+ @Override
+ public void close() throws Exception {
+ throw new IOException("injected cleanup failure");
+ }
+ };
+ AtomicLong currentBytes = new AtomicLong();
+ DorisIOManager.SpillAccountant accountant = new
DorisIOManager.SpillAccountant() {
+ @Override
+ public String[] getSpillDirectories() {
+ return new String[] {tempDir.toString()};
+ }
+
+ @Override
+ public void reserve(String path, long bytes) {
+ currentBytes.addAndGet(bytes);
+ }
+
+ @Override
+ public void rollback(String path, long bytes) {
+ currentBytes.addAndGet(-bytes);
+ }
+
+ @Override
+ public void commitWrite(String path, long bytes) {
+ }
+
+ @Override
+ public void recordRead(String path, long bytes) {
+ }
+
+ @Override
+ public void release(String path, long bytes) {
+ currentBytes.addAndGet(-bytes);
+ }
+ };
+ DorisIOManager ioManager = new DorisIOManager(failingCloseManager,
accountant);
+ FileIOChannel.ID channel = ioManager.createChannel();
+ BufferFileWriter spillWriter =
ioManager.createBufferFileWriter(channel);
+ spillWriter.writeBlock(Buffer.create(MemorySegment.wrap(new byte[8]),
8));
+ spillWriter.close();
+ Assertions.assertEquals(12, currentBytes.get());
+
+ PaimonJniWriter writer = new PaimonJniWriter();
+ Field ioManagerField =
PaimonJniWriter.class.getDeclaredField("ioManager");
+ ioManagerField.setAccessible(true);
+ ioManagerField.set(writer, ioManager);
+
+ Assertions.assertDoesNotThrow(writer::close);
+ Assertions.assertTrue(channel.getPathFile().exists());
+ Assertions.assertEquals(12, currentBytes.get());
+ Assertions.assertDoesNotThrow(writer::close);
+ spillWriter.deleteChannel();
+ Assertions.assertEquals(0, currentBytes.get());
+ }
}
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_external_paths.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_external_paths.groovy
index a8e057c33d3..d12b3e3a751 100644
---
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_external_paths.groovy
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_external_paths.groovy
@@ -46,6 +46,19 @@ suite("test_paimon_write_external_paths",
"p0,external,paimon") {
'data-file.external-paths.strategy' = 'round-robin'
);
+ DROP TABLE IF EXISTS paimon.${dbName}.t_round_robin_roll;
+ CREATE TABLE paimon.${dbName}.t_round_robin_roll (
+ id INT, payload STRING
+ ) USING paimon
+ TBLPROPERTIES (
+ 'primary-key' = 'id',
+ 'bucket' = '1',
+ 'write-only' = 'true',
+ 'target-file-size' = '1 kb',
+ 'data-file.external-paths' =
'${pathRoot}/round-roll-a,${pathRoot}/round-roll-b',
+ 'data-file.external-paths.strategy' = 'round-robin'
+ );
+
DROP TABLE IF EXISTS paimon.${dbName}.t_weight_robin;
CREATE TABLE paimon.${dbName}.t_weight_robin (
id INT, payload STRING
@@ -127,11 +140,6 @@ suite("test_paimon_write_external_paths",
"p0,external,paimon") {
sql """INSERT INTO t_round_robin VALUES ('p1', 2, 'two')"""
sql """INSERT INTO t_round_robin VALUES ('p2', 3, 'three')"""
sql """INSERT INTO t_round_robin VALUES ('p2', 4, 'four')"""
- sql """
- INSERT INTO t_round_robin
- SELECT 'p-bulk', CAST(number + 100 AS INT), repeat('x', 2048)
- FROM numbers("number" = "16")
- """
def oldRoundFiles = dataFiles("t_round_robin")
assertFalse(oldRoundFiles.isEmpty())
// Each lifecycle randomly initializes its round-robin position, so
independent
@@ -143,6 +151,33 @@ suite("test_paimon_write_external_paths",
"p0,external,paimon") {
assertDorisSparkRows("external_round_robin_initial", "t_round_robin",
"pt, id, length(payload)", "ORDER BY pt, id")
+ // Isolate the round-robin oracle from the independent writers above.
One fixed bucket and
+ // one pipeline task keep all rows in a single Paimon writer. Paimon
checks file rolling
+ // every 1000 rows; 4000 deterministic high-entropy rows leave enough
margin to roll
+ // repeatedly and therefore visit both roots regardless of its random
start.
+ sql """SET parallel_pipeline_task_num = 1"""
+ try {
+ sql """
+ INSERT INTO t_round_robin_roll
+ SELECT CAST(number AS INT),
+ concat(md5(CAST(number AS STRING)),
+ md5(CAST(number + 100000 AS STRING)))
+ FROM numbers("number" = "4000")
+ """
+ } finally {
+ sql """SET parallel_pipeline_task_num = 0"""
+ }
+ def rolledFiles = dataFiles("t_round_robin_roll")
+ assertTrue(rolledFiles.size() >= 2)
+ assertTrue(rolledFiles.every {
+ it.startsWith("${pathRoot}/round-roll-a/") ||
+ it.startsWith("${pathRoot}/round-roll-b/")
+ })
+ assertTrue(rolledFiles.any {
it.startsWith("${pathRoot}/round-roll-a/") })
+ assertTrue(rolledFiles.any {
it.startsWith("${pathRoot}/round-roll-b/") })
+ assertDorisSparkRows("external_round_robin_roll", "t_round_robin_roll",
+ "count(*), sum(id), min(length(payload)),
max(length(payload))", "")
+
spark_paimon """
ALTER TABLE paimon.${dbName}.t_round_robin SET TBLPROPERTIES (
'data-file.external-paths' =
'${pathRoot}/round-c,${pathRoot}/round-d'
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_thread_lifecycle.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_thread_lifecycle.groovy
index b13fdd317e6..e5f34149a76 100644
---
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_thread_lifecycle.groovy
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_thread_lifecycle.groovy
@@ -18,20 +18,13 @@
// Http is a framework utility class, not an injected Suite DSL property.
import org.apache.doris.regression.util.Http
-suite("test_paimon_write_thread_lifecycle", "p0,external,paimon") {
+suite("test_paimon_write_thread_lifecycle",
"p0,external,paimon,nonConcurrent") {
String enabled = context.config.otherConfigs.get("enablePaimonTest")
if (enabled == null || !enabled.equalsIgnoreCase("true")) {
logger.info("disable paimon test.")
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")
String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
String catalogName = "test_pw_thread_lifecycle_catalog"
@@ -104,7 +97,7 @@ suite("test_paimon_write_thread_lifecycle",
"p0,external,paimon") {
try {
// Warm all writer and metrics paths before taking the baseline. This
keeps
// one-time JVM attachment and SDK class initialization out of the
leak oracle.
- for (int round = 0; round < 2; round++) {
+ for (int round = 0; round < 12; round++) {
sql """
INSERT INTO t_thread_lifecycle
SELECT number + ${round * 1000}, repeat('w', 32)
@@ -130,7 +123,7 @@ suite("test_paimon_write_thread_lifecycle",
"p0,external,paimon") {
def jvmPhases = []
def processPhases = []
for (int phase = 0; phase < 4; phase++) {
- writePhase(2 + phase * 12)
+ writePhase(12 + phase * 12)
sleep(5000)
jvmPhases.add(minimumThreadCounts(jvmThreadCounts))
processPhases.add(minimumThreadCounts(processThreadCounts))
@@ -138,15 +131,31 @@ suite("test_paimon_write_thread_lifecycle",
"p0,external,paimon") {
+ "process=${processPhases[-1]}")
}
- assertEquals(50000L,
+ assertEquals(60000L,
(sql """SELECT COUNT(*) FROM t_thread_lifecycle""")[0][0] as
long)
backendEndpoints.keySet().each { backendId ->
- // Equal steady-state phases must reuse or detach JNI writer
threads.
- // Comparing later phases excludes the cold shared writer-pool
expansion.
- assertTrue(jvmPhases[-1][backendId] <= jvmPhases[0][backendId] + 4,
- "JVM threads kept growing on backend ${backendId}: phases="
- + jvmPhases.collect { counts -> counts[backendId]
})
+ // Warm-up performs the same workload as every measured phase.
Judge persistent growth
+ // from the actual pre-phase baseline and phase low-water marks
instead of failing on
+ // an isolated background-thread spike: a leaked thread cannot
disappear in a later
+ // phase, while an unrelated transient thread can.
+ def jvmCounts = jvmPhases.collect { sample -> sample[backendId] as
long }
+ def processCounts = processPhases.collect { sample ->
sample[backendId] as long }
+ def earlyJvmFloor = jvmCounts.take(2).min()
+ def lateJvmFloor = jvmCounts.drop(2).min()
+ def earlyProcessFloor = processCounts.take(2).min()
+ def lateProcessFloor = processCounts.drop(2).min()
+
+ assertTrue(jvmCounts.min() <= jvmBefore[backendId] + 2,
+ "JVM threads never returned to the warm-up baseline on
backend ${backendId}: "
+ + "baseline=${jvmBefore[backendId]},
phases=${jvmCounts}")
+ assertTrue(lateJvmFloor <= earlyJvmFloor + 2,
+ "JVM threads kept growing on backend ${backendId}:
phases=${jvmCounts}")
+ assertTrue(processCounts.min() <= processBefore[backendId] + 4,
+ "Process threads never returned to the warm-up baseline on
backend ${backendId}: "
+ + "baseline=${processBefore[backendId]},
phases=${processCounts}")
+ assertTrue(lateProcessFloor <= earlyProcessFloor + 4,
+ "Process threads kept growing on backend ${backendId}:
phases=${processCounts}")
}
} finally {
sql """drop catalog if exists ${catalogName}"""
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
