github-actions[bot] commented on code in PR #66612:
URL: https://github.com/apache/doris/pull/66612#discussion_r3861337756
##########
be/src/exec/spill/spill_file_writer.cpp:
##########
@@ -91,12 +91,27 @@ Status SpillFileWriter::_close_current_part(const
std::shared_ptr<SpillFile>& sp
_part_meta.append((const char*)&_part_max_sub_block_size,
sizeof(_part_max_sub_block_size));
_part_meta.append((const char*)&_part_written_blocks,
sizeof(_part_written_blocks));
+ int64_t meta_size = _part_meta.size();
+ if (!_data_dir->try_reserve_spill_data(meta_size)) {
Review Comment:
[P1] Propagate footer reservation failure before publishing the spill
This new capacity failure can occur after the data blocks consume the
remaining quota but before the footer, file close, part-count update, and
`finish_writing()`. The Iceberg `_do_spill()` and intermediate-merge paths
never explicitly close their local writers; the destructor only logs this
error, and intermediate merge deletes its input spill files before the output
writer is destroyed. The method can therefore return success with a replacement
whose part count is zero, which the reader treats as EOS, losing the spilled
rows. Please explicitly close and propagate status before publishing/deleting
inputs, make failed close state retry-safe, and cover footer-quota exhaustion.
##########
be/src/exec/spill/spill_file_manager.cpp:
##########
@@ -22,24 +22,238 @@
#include <algorithm>
#include <filesystem>
+#include <limits>
#include <memory>
#include <string>
+#include <unordered_set>
#include <utility>
#include "common/logging.h"
#include "common/metrics/doris_metrics.h"
#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 (_managed_paths.empty()) {
+ RETURN_IF_ERROR(_manager->_initialize_external_spill_session(this));
+ }
+ paths->clear();
+ for (const auto& managed_path : _managed_paths) {
+ paths->emplace_back(managed_path.path);
+ }
+ return Status::OK();
+}
+
+ExternalSpillSession::ManagedPath* ExternalSpillSession::_find_managed_path(
+ const std::string& path) {
+ for (auto& managed_path : _managed_paths) {
+ if (path == managed_path.path ||
+ (path.size() > managed_path.path.size() &&
path.starts_with(managed_path.path) &&
+ path[managed_path.path.size()] == '/')) {
+ return &managed_path;
+ }
+ }
+ return nullptr;
+}
+
+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);
+ auto* managed_path = _find_managed_path(path);
+ if (managed_path == nullptr) {
+ return Status::InvalidArgument("External spill path is not managed by
Doris: {}", path);
+ }
+ const int64_t accounted_bytes =
+ managed_path->buffer_accounted_bytes +
managed_path->direct_file_accounted_bytes;
+ if (bytes > std::numeric_limits<int64_t>::max() - accounted_bytes) {
+ return Status::InvalidArgument("External spill reservation overflows:
bytes={}", bytes);
+ }
+ if (!managed_path->data_dir->try_reserve_spill_data(bytes)) {
+ return Status::Error<ErrorCode::DISK_REACH_CAPACITY_LIMIT>(
+ "External spill write exceeds the Doris spill storage limit:
path={}, bytes={}",
+ path, bytes);
+ }
+ managed_path->buffer_accounted_bytes += bytes;
+ managed_path->buffer_accounted_bytes_by_path[path] += 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);
+ auto* managed_path = _find_managed_path(path);
+ if (managed_path == nullptr) {
+ LOG(WARNING) << "Ignoring accounting for unmanaged external spill
path: " << path;
+ return;
+ }
+ data_dir = managed_path->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;
+ auto path_it =
managed_path->buffer_accounted_bytes_by_path.find(path);
+ if (path_it == managed_path->buffer_accounted_bytes_by_path.end())
{
+ return;
+ }
+ released_bytes = std::min(requested_release, path_it->second);
+ path_it->second -= released_bytes;
+ if (path_it->second == 0) {
+ managed_path->buffer_accounted_bytes_by_path.erase(path_it);
+ }
+ managed_path->buffer_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);
+ }
+}
+
+Status ExternalSpillSession::reconcile_direct_file_usage(bool allow_release) {
+ struct ObservedPathUsage {
+ std::string path;
+ std::unordered_set<std::string> buffer_paths;
+ uintmax_t bytes = 0;
+ };
+ std::vector<ObservedPathUsage> observed_paths;
+ {
+ std::lock_guard lock(_mutex);
+ observed_paths.reserve(_managed_paths.size());
+ for (const auto& managed_path : _managed_paths) {
+ ObservedPathUsage observed_path {
+ .path = managed_path.path,
+ .buffer_paths = {},
+ .bytes = 0,
+ };
+ for (const auto& buffer_entry :
managed_path.buffer_accounted_bytes_by_path) {
+ observed_path.buffer_paths.emplace(buffer_entry.first);
+ }
+ observed_paths.emplace_back(std::move(observed_path));
+ }
+ }
+
+ // Do not hold the session mutex while walking the filesystem. Buffer
callbacks can continue to
+ // reserve and release capacity while a rate-limited raw-file observation
is in progress.
+ for (auto& observed_path : observed_paths) {
+ uintmax_t actual_bytes = 0;
+ std::error_code ec;
+ if (std::filesystem::exists(observed_path.path, ec)) {
+ std::filesystem::recursive_directory_iterator iterator(
+ observed_path.path,
std::filesystem::directory_options::skip_permission_denied,
+ ec);
+ const std::filesystem::recursive_directory_iterator end;
+ while (!ec && iterator != end) {
+ if (iterator->is_regular_file(ec)) {
+ const auto file_path = iterator->path().string();
+ if (!observed_path.buffer_paths.contains(file_path)) {
Review Comment:
[P1] Reclassify buffer paths after the unlocked scan
This exclusion set is only a snapshot. While the mutex is dropped for the
filesystem walk, Paimon's asynchronous lookup compaction can reserve and create
a new callback-managed channel; the scan then counts that file as direct data,
and the relocked code adds those bytes on top of the channel's existing
reservation without checking the current buffer map. Ordinary reconciliations
cannot release the overcount because `allow_release` is false, so a valid later
write can fail with `DISK_REACH_CAPACITY_LIMIT`. Please make the observation
generation-aware or reclassify observed paths after relocking, and add a
barrier-controlled concurrent scan/reserve regression.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java:
##########
@@ -576,13 +583,36 @@ private void closeWriter() throws Exception {
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) {
Review Comment:
[P1] Abort prepared files without globally fencing the BE
Native has already received the prepared payloads and passed its only abort
branch before this close runs. If the final filesystem/accounting
reconciliation returns a plain `IOException`, this branch clears Java's saved
messages and throws; native drops its local payloads without aborting them,
then sets `paimon_jni_close_failed` and rejects every later Paimon writer until
restart even though the SDK writer is closed and its executor terminated.
Please abort the prepared set on this close failure, transfer any residual
session/accounting to retryable cleanup without setting the SDK-liveness fence,
and test both file cleanup and subsequent writer admission.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java:
##########
@@ -547,6 +553,7 @@ private List<CommitMessage> prepareCommitMessages() throws
Exception {
List<CommitMessage> messages = commitIdentifier > 0
? writer.prepareCommit(true, commitIdentifier)
: writer.prepareCommit();
+ ioManager.reconcileNow(false);
Review Comment:
[P1] Never let reconciliation skip aborting drained files
`writer.prepareCommit()` drains Paimon's current increment before this new
fallible call. On the normal prepare path, a reconciliation error occurs before
the messages are saved, so the subsequent abort cannot recover them. A
write-time reconciliation error is worse: native enters `abortWriter()`
directly, its prepare reaches this same call, and another error skips
`InnerTableCommit.abort` even if the messages were published first. In both
cases the drained files are orphaned. Please publish the messages immediately
after SDK prepare and structure abort so it consumes that saved set despite
reconciliation failure; cover both normal prepare failure and
write-error/abort-time failure.
##########
be/src/exec/spill/spill_file_manager.cpp:
##########
@@ -183,32 +493,70 @@ void SpillFileManager::delete_spill_file(SpillFileSPtr
spill_file) {
void SpillFileManager::delete_query_spill_directory(const std::string&
query_id,
SpillDataDir* data_dir) {
- PendingQuerySpillDirectory pending_directory {
- .query_dir = data_dir->get_spill_data_path(query_id),
- };
+ auto query_dir = data_dir->get_spill_data_path(query_id);
+ {
+ std::lock_guard lock(_query_spill_directories_mutex);
+ auto it = _query_spill_directories
+ .try_emplace(query_dir,
+ QuerySpillDirectoryState {
+ .failed_count = 0,
+ .data_dir = data_dir,
+ .external_accounted_bytes = 0,
+ .external_leases = 0,
+ .delete_requested = false,
+ })
+ .first;
+ DCHECK(it->second.data_dir == data_dir);
+ it->second.delete_requested = true;
+ }
- auto status = _try_delete_query_spill_directory(pending_directory);
+ auto status = _try_delete_query_spill_directory(query_dir);
if (!status.ok()) {
- std::lock_guard lock(_pending_query_spill_directories_mutex);
- ++pending_directory.failed_count;
-
_pending_query_spill_directories.emplace_back(std::move(pending_directory));
+ std::lock_guard lock(_query_spill_directories_mutex);
+ auto it = _query_spill_directories.find(query_dir);
+ if (it != _query_spill_directories.end()) {
+ ++it->second.failed_count;
+ }
}
}
-Status SpillFileManager::_try_delete_query_spill_directory(
- const PendingQuerySpillDirectory& pending_directory) {
+Status SpillFileManager::_try_delete_query_spill_directory(const std::string&
query_dir) {
+ std::unique_lock lock(_query_spill_directories_mutex);
+ auto it = _query_spill_directories.find(query_dir);
+ if (it == _query_spill_directories.end() || !it->second.delete_requested) {
+ return Status::OK();
+ }
+ if (it->second.external_leases > 0) {
+ return Status::InternalError("external spill directory is still in
use: {}", 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");
});
const auto& fs = io::global_local_filesystem();
- return fs->delete_directory(pending_directory.query_dir);
+ auto status = fs->delete_directory(query_dir);
Review Comment:
[P2] Release the global registry lock during recursive deletion
`_query_spill_directories_mutex` remains held while `delete_directory()`
recursively removes every file under this query. The same global mutex is
required to initialize or release every external spill session and to tear down
every other query, so one large cleanup can stall unrelated Paimon writers and
query cleanup for the full `remove_all` duration. Please mark this directory as
deletion-in-progress, drop the global lock around filesystem I/O, and serialize
only competing operations for the same directory.
--
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]
