You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2026/03/04 06:07:13 UTC
(doris) branch spill_repartition updated (05670c557c4 -> a3fd36e7ed4)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a change to branch spill_repartition
in repository https://gitbox.apache.org/repos/asf/doris.git
from 05670c557c4 fix probe revokeable memory size bug
new c0380ad4594 change spill file to shared ptr
new a3fd36e7ed4 fix agg profile bug
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/pipeline/exec/multi_cast_data_streamer.cpp | 2 +-
be/src/pipeline/exec/multi_cast_data_streamer.h | 2 +-
.../exec/partitioned_aggregation_sink_operator.cpp | 11 ++++++-
.../exec/partitioned_aggregation_sink_operator.h | 2 +-
.../partitioned_aggregation_source_operator.cpp | 5 ++++
.../exec/partitioned_aggregation_source_operator.h | 2 +-
.../exec/partitioned_hash_join_probe_operator.cpp | 20 +++++++++++--
.../exec/partitioned_hash_join_probe_operator.h | 6 ++--
.../exec/partitioned_hash_join_sink_operator.cpp | 9 ++++++
.../exec/partitioned_hash_join_sink_operator.h | 2 +-
be/src/pipeline/exec/spill_sort_sink_operator.cpp | 5 ++++
be/src/pipeline/exec/spill_sort_sink_operator.h | 2 +-
.../pipeline/exec/spill_sort_source_operator.cpp | 13 +++++++-
be/src/pipeline/exec/spill_sort_source_operator.h | 2 +-
be/src/vec/spill/spill_file.cpp | 9 +++---
be/src/vec/spill/spill_file.h | 12 ++++----
be/src/vec/spill/spill_file_reader.h | 2 +-
be/src/vec/spill/spill_file_writer.cpp | 35 +++++++++++++++-------
be/src/vec/spill/spill_file_writer.h | 10 +++----
be/src/vec/spill/spill_repartitioner.cpp | 2 +-
be/src/vec/spill/spill_repartitioner.h | 6 ++--
21 files changed, 114 insertions(+), 45 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 02/02: fix agg profile bug
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch spill_repartition
in repository https://gitbox.apache.org/repos/asf/doris.git
commit a3fd36e7ed454c713e9b4e48065c39c7f063462a
Author: yiguolei <gu...@selectdb.com>
AuthorDate: Wed Mar 4 14:06:55 2026 +0800
fix agg profile bug
---
be/src/pipeline/exec/partitioned_aggregation_sink_operator.cpp | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/be/src/pipeline/exec/partitioned_aggregation_sink_operator.cpp b/be/src/pipeline/exec/partitioned_aggregation_sink_operator.cpp
index ac729bf911c..fd785f78335 100644
--- a/be/src/pipeline/exec/partitioned_aggregation_sink_operator.cpp
+++ b/be/src/pipeline/exec/partitioned_aggregation_sink_operator.cpp
@@ -427,7 +427,7 @@ Status PartitionedAggSinkLocalState::revoke_memory(RuntimeState* state) {
custom_profile()->add_info_string("Spilled", "true");
update_profile<false>(sink_local_state->custom_profile());
} else {
- update_profile<true>(sink_local_state->custom_profile());
+ update_profile<false>(sink_local_state->custom_profile());
}
DBUG_EXECUTE_IF("fault_inject::partitioned_agg_sink::revoke_memory_submit_func", {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 01/02: change spill file to shared ptr
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch spill_repartition
in repository https://gitbox.apache.org/repos/asf/doris.git
commit c0380ad4594c74e444d707e43537a15b98c61bf1
Author: yiguolei <gu...@selectdb.com>
AuthorDate: Wed Mar 4 13:11:14 2026 +0800
change spill file to shared ptr
---
be/src/pipeline/exec/multi_cast_data_streamer.cpp | 2 +-
be/src/pipeline/exec/multi_cast_data_streamer.h | 2 +-
.../exec/partitioned_aggregation_sink_operator.cpp | 9 ++++++
.../exec/partitioned_aggregation_sink_operator.h | 2 +-
.../partitioned_aggregation_source_operator.cpp | 5 ++++
.../exec/partitioned_aggregation_source_operator.h | 2 +-
.../exec/partitioned_hash_join_probe_operator.cpp | 20 +++++++++++--
.../exec/partitioned_hash_join_probe_operator.h | 6 ++--
.../exec/partitioned_hash_join_sink_operator.cpp | 9 ++++++
.../exec/partitioned_hash_join_sink_operator.h | 2 +-
be/src/pipeline/exec/spill_sort_sink_operator.cpp | 5 ++++
be/src/pipeline/exec/spill_sort_sink_operator.h | 2 +-
.../pipeline/exec/spill_sort_source_operator.cpp | 13 +++++++-
be/src/pipeline/exec/spill_sort_source_operator.h | 2 +-
be/src/vec/spill/spill_file.cpp | 9 +++---
be/src/vec/spill/spill_file.h | 12 ++++----
be/src/vec/spill/spill_file_reader.h | 2 +-
be/src/vec/spill/spill_file_writer.cpp | 35 +++++++++++++++-------
be/src/vec/spill/spill_file_writer.h | 10 +++----
be/src/vec/spill/spill_repartitioner.cpp | 2 +-
be/src/vec/spill/spill_repartitioner.h | 6 ++--
21 files changed, 113 insertions(+), 44 deletions(-)
diff --git a/be/src/pipeline/exec/multi_cast_data_streamer.cpp b/be/src/pipeline/exec/multi_cast_data_streamer.cpp
index b98763746fe..9006a236d28 100644
--- a/be/src/pipeline/exec/multi_cast_data_streamer.cpp
+++ b/be/src/pipeline/exec/multi_cast_data_streamer.cpp
@@ -252,7 +252,7 @@ Status MultiCastDataStreamer::_start_spill_task(RuntimeState* state,
auto spill_func = [state, blocks = std::move(blocks), spill_file = std::move(spill_file),
sink_profile]() mutable {
const auto blocks_count = blocks.size();
- vectorized::SpillFileWriterUPtr writer;
+ vectorized::SpillFileWriterSPtr writer;
RETURN_IF_ERROR(spill_file->create_writer(state, sink_profile, writer));
for (auto& block : blocks) {
if (state->is_cancelled()) break;
diff --git a/be/src/pipeline/exec/multi_cast_data_streamer.h b/be/src/pipeline/exec/multi_cast_data_streamer.h
index 5e8526c8dda..6c123148319 100644
--- a/be/src/pipeline/exec/multi_cast_data_streamer.h
+++ b/be/src/pipeline/exec/multi_cast_data_streamer.h
@@ -45,7 +45,7 @@ struct MultiCastBlock {
};
struct SpillingReader {
- vectorized::SpillFileReaderUPtr reader;
+ vectorized::SpillFileReaderSPtr reader;
vectorized::SpillFileSPtr spill_file;
int64_t block_offset {0};
bool all_data_read {false};
diff --git a/be/src/pipeline/exec/partitioned_aggregation_sink_operator.cpp b/be/src/pipeline/exec/partitioned_aggregation_sink_operator.cpp
index 25d49935fde..ac729bf911c 100644
--- a/be/src/pipeline/exec/partitioned_aggregation_sink_operator.cpp
+++ b/be/src/pipeline/exec/partitioned_aggregation_sink_operator.cpp
@@ -77,6 +77,15 @@ Status PartitionedAggSinkLocalState::close(RuntimeState* state, Status exec_stat
if (Base::_closed) {
return Status::OK();
}
+
+ for (auto& writer : _spill_writers) {
+ if (writer) {
+ RETURN_IF_ERROR(writer->close());
+ writer.reset();
+ }
+ }
+ _spill_writers.clear();
+
return Base::close(state, exec_status);
}
diff --git a/be/src/pipeline/exec/partitioned_aggregation_sink_operator.h b/be/src/pipeline/exec/partitioned_aggregation_sink_operator.h
index 6ac9759ebfe..5538c296562 100644
--- a/be/src/pipeline/exec/partitioned_aggregation_sink_operator.h
+++ b/be/src/pipeline/exec/partitioned_aggregation_sink_operator.h
@@ -97,7 +97,7 @@ public:
RuntimeProfile::Counter* _spill_serialize_hash_table_timer = nullptr;
- std::vector<vectorized::SpillFileWriterUPtr> _spill_writers;
+ std::vector<vectorized::SpillFileWriterSPtr> _spill_writers;
std::atomic<bool> _eos = false;
};
diff --git a/be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp b/be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp
index fd3f6c284be..fe2ebb2df78 100644
--- a/be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp
+++ b/be/src/pipeline/exec/partitioned_aggregation_source_operator.cpp
@@ -101,6 +101,11 @@ Status PartitionedAggLocalState::close(RuntimeState* state) {
return Status::OK();
}
+ if (_current_reader) {
+ RETURN_IF_ERROR(_current_reader->close());
+ _current_reader.reset();
+ }
+
// Clean up partition queue resources.
for (auto& partition : _partition_queue) {
if (partition.spill_file) {
diff --git a/be/src/pipeline/exec/partitioned_aggregation_source_operator.h b/be/src/pipeline/exec/partitioned_aggregation_source_operator.h
index ff40cd411d4..518e59b5ac5 100644
--- a/be/src/pipeline/exec/partitioned_aggregation_source_operator.h
+++ b/be/src/pipeline/exec/partitioned_aggregation_source_operator.h
@@ -116,7 +116,7 @@ private:
SpillRepartitioner _repartitioner;
// Persistent reader for _recover_blocks_from_partition (survives across yield calls)
- vectorized::SpillFileReaderUPtr _current_reader;
+ vectorized::SpillFileReaderSPtr _current_reader;
};
class AggSourceOperatorX;
diff --git a/be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp b/be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp
index 49156b83c9e..b24f86846d5 100644
--- a/be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp
+++ b/be/src/pipeline/exec/partitioned_hash_join_probe_operator.cpp
@@ -186,6 +186,24 @@ Status PartitionedHashJoinProbeLocalState::close(RuntimeState* state) {
if (_closed) {
return Status::OK();
}
+
+ for (auto& writer : _probe_writers) {
+ if (writer) {
+ RETURN_IF_ERROR(writer->close());
+ writer.reset();
+ }
+ }
+ _probe_writers.clear();
+
+ if (_current_build_reader) {
+ RETURN_IF_ERROR(_current_build_reader->close());
+ _current_build_reader.reset();
+ }
+ if (_current_probe_reader) {
+ RETURN_IF_ERROR(_current_probe_reader->close());
+ _current_probe_reader.reset();
+ }
+
// Clean up any remaining spill partition queue entries
for (auto& entry : _spill_partition_queue) {
if (entry.build_file) {
@@ -203,8 +221,6 @@ Status PartitionedHashJoinProbeLocalState::close(RuntimeState* state) {
ExecEnv::GetInstance()->spill_file_mgr()->delete_spill_file(_current_partition.probe_file);
}
_current_partition = JoinSpillPartitionInfo {};
- _current_build_reader.reset();
- _current_probe_reader.reset();
_queue_probe_blocks.clear();
RETURN_IF_ERROR(PipelineXSpillLocalState::close(state));
diff --git a/be/src/pipeline/exec/partitioned_hash_join_probe_operator.h b/be/src/pipeline/exec/partitioned_hash_join_probe_operator.h
index a375fe1d1f7..bbaccf5906b 100644
--- a/be/src/pipeline/exec/partitioned_hash_join_probe_operator.h
+++ b/be/src/pipeline/exec/partitioned_hash_join_probe_operator.h
@@ -158,14 +158,14 @@ private:
std::map<uint32_t, std::vector<vectorized::Block>> _probe_blocks;
std::vector<vectorized::SpillFileSPtr> _probe_spilling_groups;
- std::vector<vectorized::SpillFileWriterUPtr> _probe_writers;
+ std::vector<vectorized::SpillFileWriterSPtr> _probe_writers;
std::unique_ptr<vectorized::PartitionerBase> _partitioner;
std::unique_ptr<RuntimeProfile> _internal_runtime_profile;
// Persistent readers for recovery across scheduling slices
- vectorized::SpillFileReaderUPtr _current_build_reader;
- vectorized::SpillFileReaderUPtr _current_probe_reader;
+ vectorized::SpillFileReaderSPtr _current_build_reader;
+ vectorized::SpillFileReaderSPtr _current_probe_reader;
// ---- Spill partition queue state ----
// Whether _spill_partition_queue has been initialized from spilled build groups +
diff --git a/be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp b/be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp
index 5f8526b4317..cac453c8b69 100644
--- a/be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp
+++ b/be/src/pipeline/exec/partitioned_hash_join_sink_operator.cpp
@@ -84,6 +84,15 @@ Status PartitionedHashJoinSinkLocalState::close(RuntimeState* state, Status exec
RETURN_IF_ERROR(p._inner_sink_operator->close(_shared_state->_inner_runtime_state.get(),
exec_status));
}
+
+ for (auto& writer : _build_writers) {
+ if (writer) {
+ RETURN_IF_ERROR(writer->close());
+ writer.reset();
+ }
+ }
+ _build_writers.clear();
+
return PipelineXSpillSinkLocalState::close(state, exec_status);
}
diff --git a/be/src/pipeline/exec/partitioned_hash_join_sink_operator.h b/be/src/pipeline/exec/partitioned_hash_join_sink_operator.h
index 61f275c7156..12d4810bbdd 100644
--- a/be/src/pipeline/exec/partitioned_hash_join_sink_operator.h
+++ b/be/src/pipeline/exec/partitioned_hash_join_sink_operator.h
@@ -94,7 +94,7 @@ protected:
RuntimeProfile::Counter* _in_mem_rows_counter = nullptr;
RuntimeProfile::Counter* _memory_usage_reserved = nullptr;
- std::vector<vectorized::SpillFileWriterUPtr> _build_writers;
+ std::vector<vectorized::SpillFileWriterSPtr> _build_writers;
};
class PartitionedHashJoinSinkOperatorX
diff --git a/be/src/pipeline/exec/spill_sort_sink_operator.cpp b/be/src/pipeline/exec/spill_sort_sink_operator.cpp
index 87aa89021b4..5ce2130eff4 100644
--- a/be/src/pipeline/exec/spill_sort_sink_operator.cpp
+++ b/be/src/pipeline/exec/spill_sort_sink_operator.cpp
@@ -66,6 +66,11 @@ void SpillSortSinkLocalState::update_profile(RuntimeProfile* child_profile) {
#undef UPDATE_PROFILE
Status SpillSortSinkLocalState::close(RuntimeState* state, Status execsink_status) {
+ if (_spilling_writer) {
+ RETURN_IF_ERROR(_spilling_writer->close());
+ _spilling_writer.reset();
+ }
+ _spilling_file.reset();
return Base::close(state, execsink_status);
}
diff --git a/be/src/pipeline/exec/spill_sort_sink_operator.h b/be/src/pipeline/exec/spill_sort_sink_operator.h
index 19f9a0c85c9..36710e14582 100644
--- a/be/src/pipeline/exec/spill_sort_sink_operator.h
+++ b/be/src/pipeline/exec/spill_sort_sink_operator.h
@@ -62,7 +62,7 @@ private:
RuntimeProfile::Counter* _spill_merge_sort_timer = nullptr;
vectorized::SpillFileSPtr _spilling_file;
- vectorized::SpillFileWriterUPtr _spilling_writer;
+ vectorized::SpillFileWriterSPtr _spilling_writer;
std::atomic<bool> _eos = false;
};
diff --git a/be/src/pipeline/exec/spill_sort_source_operator.cpp b/be/src/pipeline/exec/spill_sort_source_operator.cpp
index 06150cce856..9b4eb4b0d0a 100644
--- a/be/src/pipeline/exec/spill_sort_source_operator.cpp
+++ b/be/src/pipeline/exec/spill_sort_source_operator.cpp
@@ -64,6 +64,17 @@ Status SpillSortLocalState::close(RuntimeState* state) {
if (_closed) {
return Status::OK();
}
+
+ for (auto& reader : _current_merging_readers) {
+ if (reader) {
+ RETURN_IF_ERROR(reader->close());
+ reader.reset();
+ }
+ }
+ _current_merging_readers.clear();
+ _current_merging_files.clear();
+ _merger.reset();
+
return Base::close(state);
}
@@ -100,7 +111,7 @@ Status SpillSortLocalState::execute_merge_sort_spill_files(RuntimeState* state)
ExecEnv::GetInstance()->spill_file_mgr()->next_id());
RETURN_IF_ERROR(ExecEnv::GetInstance()->spill_file_mgr()->create_spill_file(relative_path,
tmp_file));
- vectorized::SpillFileWriterUPtr tmp_writer;
+ vectorized::SpillFileWriterSPtr tmp_writer;
RETURN_IF_ERROR(tmp_file->create_writer(state, operator_profile(), tmp_writer));
_shared_state->sorted_spill_groups.emplace_back(tmp_file);
diff --git a/be/src/pipeline/exec/spill_sort_source_operator.h b/be/src/pipeline/exec/spill_sort_source_operator.h
index 3b2943033a2..fe13c2aad20 100644
--- a/be/src/pipeline/exec/spill_sort_source_operator.h
+++ b/be/src/pipeline/exec/spill_sort_source_operator.h
@@ -63,7 +63,7 @@ protected:
std::vector<vectorized::SpillFileSPtr> _current_merging_files;
/// Readers held alive during merge; one per SpillFile, reads parts sequentially.
- std::vector<vectorized::SpillFileReaderUPtr> _current_merging_readers;
+ std::vector<vectorized::SpillFileReaderSPtr> _current_merging_readers;
std::unique_ptr<vectorized::VSortedRunMerger> _merger;
std::unique_ptr<RuntimeProfile> _internal_runtime_profile;
diff --git a/be/src/vec/spill/spill_file.cpp b/be/src/vec/spill/spill_file.cpp
index 7be97d738c3..81d8d322478 100644
--- a/be/src/vec/spill/spill_file.cpp
+++ b/be/src/vec/spill/spill_file.cpp
@@ -65,17 +65,18 @@ void SpillFile::gc() {
}
Status SpillFile::create_writer(RuntimeState* state, RuntimeProfile* profile,
- SpillFileWriterUPtr& writer) {
- writer = std::make_unique<SpillFileWriter>(this, state, profile, _data_dir, _spill_dir);
+ SpillFileWriterSPtr& writer) {
+ writer = std::make_shared<SpillFileWriter>(shared_from_this(), state, profile, _data_dir,
+ _spill_dir);
// record active writer
_active_writer = writer.get();
return Status::OK();
}
-SpillFileReaderUPtr SpillFile::create_reader(RuntimeState* state, RuntimeProfile* profile) const {
+SpillFileReaderSPtr SpillFile::create_reader(RuntimeState* state, RuntimeProfile* profile) const {
// It's a programming error to create a reader while a writer is still active.
DCHECK(_active_writer == nullptr) << "create_reader() called while writer still active";
- return std::make_unique<SpillFileReader>(state, profile, _spill_dir, _part_count);
+ return std::make_shared<SpillFileReader>(state, profile, _spill_dir, _part_count);
}
void SpillFile::finish_writing(int64_t total_written_bytes, size_t part_count) {
diff --git a/be/src/vec/spill/spill_file.h b/be/src/vec/spill/spill_file.h
index 2d938d98fdb..362b522f10c 100644
--- a/be/src/vec/spill/spill_file.h
+++ b/be/src/vec/spill/spill_file.h
@@ -32,8 +32,8 @@ class Block;
class SpillDataDir;
class SpillFileWriter;
class SpillFileReader;
-using SpillFileWriterUPtr = std::unique_ptr<SpillFileWriter>;
-using SpillFileReaderUPtr = std::unique_ptr<SpillFileReader>;
+using SpillFileWriterSPtr = std::shared_ptr<SpillFileWriter>;
+using SpillFileReaderSPtr = std::shared_ptr<SpillFileReader>;
/// SpillFile represents a logical spill file that may consist of multiple
/// physical "part" files on disk. Parts are managed automatically by
@@ -46,7 +46,7 @@ using SpillFileReaderUPtr = std::unique_ptr<SpillFileReader>;
/// +-- ...
///
/// Writing workflow:
-/// SpillFileWriterUPtr writer;
+/// SpillFileWriterSPtr writer;
/// RETURN_IF_ERROR(spill_file->create_writer(state, profile, writer));
/// RETURN_IF_ERROR(writer->write_block(state, block)); // auto-rotates parts
/// RETURN_IF_ERROR(writer->close()); // finalizes all parts
@@ -55,7 +55,7 @@ using SpillFileReaderUPtr = std::unique_ptr<SpillFileReader>;
/// auto reader = spill_file->create_reader(state, profile);
/// RETURN_IF_ERROR(reader->open());
/// while (!eos) { RETURN_IF_ERROR(reader->read(&block, &eos)); }
-class SpillFile {
+class SpillFile : public std::enable_shared_from_this<SpillFile> {
public:
// to avoid too many small file writes
static constexpr size_t MIN_SPILL_WRITE_BATCH_MEM = 512 * 1024;
@@ -80,11 +80,11 @@ public:
/// Create a SpillFileWriter that automatically manages multi-part rotation.
/// Only one writer should exist per SpillFile at a time.
/// Part size threshold is read from config::spill_file_part_size_bytes.
- Status create_writer(RuntimeState* state, RuntimeProfile* profile, SpillFileWriterUPtr& writer);
+ Status create_writer(RuntimeState* state, RuntimeProfile* profile, SpillFileWriterSPtr& writer);
/// Create a SpillFileReader that reads sequentially across all parts.
/// The caller should call reader->open() before reading.
- SpillFileReaderUPtr create_reader(RuntimeState* state, RuntimeProfile* profile) const;
+ SpillFileReaderSPtr create_reader(RuntimeState* state, RuntimeProfile* profile) const;
private:
friend class SpillFileWriter;
diff --git a/be/src/vec/spill/spill_file_reader.h b/be/src/vec/spill/spill_file_reader.h
index e76bf8355db..c77b58a3290 100644
--- a/be/src/vec/spill/spill_file_reader.h
+++ b/be/src/vec/spill/spill_file_reader.h
@@ -107,7 +107,7 @@ private:
std::shared_ptr<ResourceContext> _resource_ctx = nullptr;
};
-using SpillFileReaderUPtr = std::unique_ptr<SpillFileReader>;
+using SpillFileReaderSPtr = std::shared_ptr<SpillFileReader>;
} // namespace doris::vectorized
#include "common/compile_check_end.h"
diff --git a/be/src/vec/spill/spill_file_writer.cpp b/be/src/vec/spill/spill_file_writer.cpp
index 3413142f9c8..d0ef50dbe60 100644
--- a/be/src/vec/spill/spill_file_writer.cpp
+++ b/be/src/vec/spill/spill_file_writer.cpp
@@ -32,10 +32,10 @@
namespace doris::vectorized {
#include "common/compile_check_begin.h"
-SpillFileWriter::SpillFileWriter(SpillFile* spill_file, RuntimeState* state,
+SpillFileWriter::SpillFileWriter(const std::shared_ptr<SpillFile>& spill_file, RuntimeState* state,
RuntimeProfile* profile, SpillDataDir* data_dir,
const std::string& spill_dir)
- : _spill_file(spill_file),
+ : _spill_file_wptr(spill_file),
_data_dir(data_dir),
_spill_dir(spill_dir),
_max_part_size(config::spill_file_part_size_bytes),
@@ -46,8 +46,9 @@ SpillFileWriter::SpillFileWriter(SpillFile* spill_file, RuntimeState* state,
_memory_used_counter = common_profile->get_counter("MemoryUsage");
// Register this writer as the active writer for the SpillFile.
- if (_spill_file) {
- _spill_file->_active_writer = this;
+ auto spill_file_locked = _spill_file_wptr.lock();
+ if (spill_file_locked) {
+ spill_file_locked->_active_writer = this;
}
// Custom (spill-specific) counters
@@ -63,7 +64,14 @@ SpillFileWriter::SpillFileWriter(SpillFile* spill_file, RuntimeState* state,
}
SpillFileWriter::~SpillFileWriter() {
- DCHECK(_closed) << "SpillFileWriter destroyed without close(), possible memory leak";
+ if (_closed) {
+ return;
+ }
+ Status st = close();
+ if (!st.ok()) {
+ LOG(WARNING) << "SpillFileWriter::~SpillFileWriter() failed: " << st.to_string()
+ << ", spill_dir=" << _spill_dir;
+ }
}
Status SpillFileWriter::_open_next_part() {
@@ -158,13 +166,18 @@ Status SpillFileWriter::close() {
RETURN_IF_ERROR(_close_current_part());
- if (!_spill_file || _spill_file->_active_writer != this) {
- return Status::Error<INTERNAL_ERROR>(
- "SpillFileWriter close() called but not registered as active writer, possible "
- "double close or logic error");
+ // Use weak_ptr lock to safely access SpillFile during close.
+ // If the SpillFile has already been destroyed, we skip the callback.
+ auto spill_file = _spill_file_wptr.lock();
+ if (spill_file) {
+ if (spill_file->_active_writer != this) {
+ return Status::Error<INTERNAL_ERROR>(
+ "SpillFileWriter close() called but not registered as active writer, possible "
+ "double close or logic error");
+ }
+ spill_file->finish_writing(_total_written_bytes, _total_parts);
+ spill_file->_active_writer = nullptr;
}
- _spill_file->finish_writing(_total_written_bytes, _total_parts);
- _spill_file->_active_writer = nullptr;
return Status::OK();
}
diff --git a/be/src/vec/spill/spill_file_writer.h b/be/src/vec/spill/spill_file_writer.h
index 2e45c8ae32c..8492d11ba13 100644
--- a/be/src/vec/spill/spill_file_writer.h
+++ b/be/src/vec/spill/spill_file_writer.h
@@ -38,7 +38,7 @@ class SpillFile;
/// (config::spill_file_part_size_bytes).
///
/// Usage:
-/// SpillFileWriterUPtr writer;
+/// SpillFileWriterSPtr writer;
/// RETURN_IF_ERROR(spill_file->create_writer(state, profile, writer));
/// RETURN_IF_ERROR(writer->write_block(state, block));
/// RETURN_IF_ERROR(writer->close());
@@ -48,8 +48,8 @@ class SpillFile;
/// directory.
class SpillFileWriter {
public:
- SpillFileWriter(SpillFile* spill_file, RuntimeState* state, RuntimeProfile* profile,
- SpillDataDir* data_dir, const std::string& spill_dir);
+ SpillFileWriter(const std::shared_ptr<SpillFile>& spill_file, RuntimeState* state,
+ RuntimeProfile* profile, SpillDataDir* data_dir, const std::string& spill_dir);
~SpillFileWriter();
@@ -75,7 +75,7 @@ private:
Status _write_internal(const Block& block);
// ── Back-reference ──
- SpillFile* _spill_file; // non-owning; for close() callback
+ std::weak_ptr<SpillFile> _spill_file_wptr; // weak ref; use lock() in close()
// ── Configuration ──
SpillDataDir* _data_dir = nullptr;
@@ -109,7 +109,7 @@ private:
std::shared_ptr<ResourceContext> _resource_ctx = nullptr;
};
-using SpillFileWriterUPtr = std::unique_ptr<SpillFileWriter>;
+using SpillFileWriterSPtr = std::shared_ptr<SpillFileWriter>;
} // namespace vectorized
} // namespace doris
diff --git a/be/src/vec/spill/spill_repartitioner.cpp b/be/src/vec/spill/spill_repartitioner.cpp
index 452ce9a8c71..780927e117c 100644
--- a/be/src/vec/spill/spill_repartitioner.cpp
+++ b/be/src/vec/spill/spill_repartitioner.cpp
@@ -134,7 +134,7 @@ Status SpillRepartitioner::repartition(RuntimeState* state,
return Status::OK();
}
-Status SpillRepartitioner::repartition(RuntimeState* state, vectorized::SpillFileReaderUPtr& reader,
+Status SpillRepartitioner::repartition(RuntimeState* state, vectorized::SpillFileReaderSPtr& reader,
bool* done) {
DCHECK(_output_spill_files != nullptr) << "setup_output() must be called first";
DCHECK(reader != nullptr) << "reader must not be null";
diff --git a/be/src/vec/spill/spill_repartitioner.h b/be/src/vec/spill/spill_repartitioner.h
index 0effe325f54..f6894dc3637 100644
--- a/be/src/vec/spill/spill_repartitioner.h
+++ b/be/src/vec/spill/spill_repartitioner.h
@@ -107,7 +107,7 @@ public:
/// and wants to repartition only the remaining data without re-reading
/// from the beginning. Ownership of the reader is transferred on completion.
/// Call repeatedly until done == true.
- Status repartition(RuntimeState* state, vectorized::SpillFileReaderUPtr& reader, bool* done);
+ Status repartition(RuntimeState* state, vectorized::SpillFileReaderSPtr& reader, bool* done);
/// Route a single in-memory block into output files via persistent writers.
Status route_block(RuntimeState* state, vectorized::Block& block);
@@ -158,11 +158,11 @@ private:
// ── Persistent state across repartition/route_block calls ──────
// Output writers (one per partition), created by setup_output()
- std::vector<vectorized::SpillFileWriterUPtr> _output_writers;
+ std::vector<vectorized::SpillFileWriterSPtr> _output_writers;
// Pointer to caller's output SpillFiles vector (for finalize)
std::vector<vectorized::SpillFileSPtr>* _output_spill_files = nullptr;
// Input reader for repartition(), persists across yield calls
- vectorized::SpillFileReaderUPtr _input_reader;
+ vectorized::SpillFileReaderSPtr _input_reader;
vectorized::SpillFileSPtr _current_input_file;
};
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org