This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 9177f0cbd6d [refactor](be) Separate S3 spill file I/O from common
readers and writers (#69781)
9177f0cbd6d is described below
commit 9177f0cbd6d0faa0ea809cc1c0afe375a47965f0
Author: Jerry Hu <[email protected]>
AuthorDate: Mon Nov 2 09:54:30 2026 +0800
[refactor](be) Separate S3 spill file I/O from common readers and writers
(#69781)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: `SpillFileWriter` and `SpillFileReader` mixed their
common spill-part lifecycle and serialization/deserialization with
object-storage-specific policies. The writer included S3 upload queue
admission, cancellation checks, timing, and remote accounting; the
reader included S3 footer prefetch, adjacent-block coalescing, and
remote byte/request accounting.
This change introduces `SpillS3FileWriter` and `SpillS3FileReader`,
selected by the spill file factory for remote files. The common classes
retain part lifecycle, footer format and validation, block
serialization/deserialization, part rotation/traversal, and seek
handling. Narrow virtual hooks preserve the S3 policies, failed-part
reconciliation, query-buffer cap, and local/remote metric separation.
The derived writer closes before base destruction so implicit close
still dispatches its S3 hooks; reader cleanup remains entirely in the
base class and does not require derived hooks. Public writer/reader APIs
and the spill file format remain unchanged.
### Release note
None
### Check List (For Author)
- Test: Unit Test / Build and static checks
- `BUILD_TYPE=ASAN ./build.sh --be -j16` passed on the updated
reader/writer head.
- `./run-be-ut.sh -j16 --run
--filter='SpillFileS3Test.*:SpillFileTest.*:SpillRepartitionerTest.*:SpillSortSourceOperatorTest.*'`
passed (109 tests from 4 suites).
- New coverage checks S3 reader factory selection and empty-file
handling, the query-buffer cap on coalesced GET lengths, and local
reader type/accounting isolation. Existing suites exercise multi-part
roundtrips, seeking, footer overflow, oversized blocks, and disabled
coalescing.
- BE formatting/check-format, build hygiene, and `git diff --check`
passed.
- Changed-line clang-tidy passed for all 7 files in the reader follow-up
using the UT compilation database and the Clang resource directory.
- S3 unit tests use an in-memory object-storage mock. SQL regression and
real-object-store E2E were not rerun for the reader follow-up; these
results are not remote CI results.
- Behavior changed: No; this is an internal responsibility split
preserving the existing local and S3 spill behavior.
- Does this need documentation: No
---
be/src/exec/spill/spill_file.cpp | 15 +++-
be/src/exec/spill/spill_file.h | 2 +
be/src/exec/spill/spill_file_reader.cpp | 61 ++++----------
be/src/exec/spill/spill_file_reader.h | 31 ++++---
be/src/exec/spill/spill_file_writer.cpp | 106 ++++--------------------
be/src/exec/spill/spill_file_writer.h | 57 ++++++-------
be/src/exec/spill/spill_s3_file_reader.cpp | 82 +++++++++++++++++++
be/src/exec/spill/spill_s3_file_reader.h | 46 +++++++++++
be/src/exec/spill/spill_s3_file_writer.cpp | 127 +++++++++++++++++++++++++++++
be/src/exec/spill/spill_s3_file_writer.h | 47 +++++++++++
be/test/vec/spill/spill_file_s3_test.cpp | 95 +++++++++++++++++++++
be/test/vec/spill/spill_file_test.cpp | 11 +++
12 files changed, 496 insertions(+), 184 deletions(-)
diff --git a/be/src/exec/spill/spill_file.cpp b/be/src/exec/spill/spill_file.cpp
index 9cda254ec98..9849c8561fe 100644
--- a/be/src/exec/spill/spill_file.cpp
+++ b/be/src/exec/spill/spill_file.cpp
@@ -27,6 +27,8 @@
#include "exec/spill/spill_file_manager.h"
#include "exec/spill/spill_file_reader.h"
#include "exec/spill/spill_file_writer.h"
+#include "exec/spill/spill_s3_file_reader.h"
+#include "exec/spill/spill_s3_file_writer.h"
#include "io/fs/local_file_system.h"
#include "io/fs/path.h"
#include "io/fs/s3_file_system.h"
@@ -137,8 +139,13 @@ std::string SpillFile::_log_context() const {
Status SpillFile::create_writer(RuntimeState* state, RuntimeProfile* profile,
SpillFileWriterSPtr& writer) {
- writer = std::make_shared<SpillFileWriter>(shared_from_this(), state,
profile, _data_dir, _fs,
- _spill_dir);
+ if (_data_dir->is_remote()) {
+ writer = std::make_shared<SpillS3FileWriter>(shared_from_this(),
state, profile, _data_dir,
+ _fs, _spill_dir);
+ } else {
+ writer = std::make_shared<SpillFileWriter>(shared_from_this(), state,
profile, _data_dir,
+ _fs, _spill_dir);
+ }
// _active_writer is set in SpillFileWriter constructor via the shared_ptr
return Status::OK();
}
@@ -146,6 +153,10 @@ Status SpillFile::create_writer(RuntimeState* state,
RuntimeProfile* profile,
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";
+ if (_data_dir->is_remote()) {
+ return std::make_shared<SpillS3FileReader>(state, profile, _data_dir,
_fs, _spill_dir,
+ _part_sizes);
+ }
return std::make_shared<SpillFileReader>(state, profile, _data_dir, _fs,
_spill_dir,
_part_sizes);
}
diff --git a/be/src/exec/spill/spill_file.h b/be/src/exec/spill/spill_file.h
index 6665e1ae585..bc3d77714a8 100644
--- a/be/src/exec/spill/spill_file.h
+++ b/be/src/exec/spill/spill_file.h
@@ -30,6 +30,7 @@ class RuntimeState;
class Block;
class SpillDataDir;
class SpillFileWriter;
+class SpillS3FileWriter;
class SpillFileReader;
using SpillFileWriterSPtr = std::shared_ptr<SpillFileWriter>;
using SpillFileReaderSPtr = std::shared_ptr<SpillFileReader>;
@@ -91,6 +92,7 @@ public:
private:
friend class SpillFileWriter;
+ friend class SpillS3FileWriter;
friend class SpillFileManager;
/// Called by SpillFileWriter::close() to mark writing as complete.
diff --git a/be/src/exec/spill/spill_file_reader.cpp
b/be/src/exec/spill/spill_file_reader.cpp
index 7e8082b41e1..287f8924c9a 100644
--- a/be/src/exec/spill/spill_file_reader.cpp
+++ b/be/src/exec/spill/spill_file_reader.cpp
@@ -19,11 +19,10 @@
#include <glog/logging.h>
-#include <algorithm>
#include <cstring>
+#include <utility>
#include "common/cast_set.h"
-#include "common/config.h"
#include "common/exception.h"
#include "core/block/block.h"
#include "exec/spill/spill_file_manager.h"
@@ -34,6 +33,7 @@
#include "runtime/query_context.h"
#include "runtime/runtime_profile_counter_names.h"
#include "runtime/runtime_state.h"
+#include "runtime/workload_management/io_context.h"
#include "util/debug_points.h"
#include "util/slice.h"
namespace doris {
@@ -49,7 +49,6 @@ SpillFileReader::SpillFileReader(RuntimeState* state,
RuntimeProfile* profile,
_spill_dir(std::move(spill_dir)),
_part_sizes(std::move(part_sizes)),
_part_count(_part_sizes.size()),
- _is_remote(data_dir != nullptr && data_dir->is_remote()),
_resource_ctx(state->get_query_ctx()->resource_ctx()) {
// Internalize counter setup. The counters themselves are registered by
the owning
// operator (SpillReadCounters::init), so look them up by the shared name
constants:
@@ -69,34 +68,28 @@ SpillFileReader::SpillFileReader(RuntimeState* state,
RuntimeProfile* profile,
_read_file_size = get_counter(profile::SPILL_READ_FILE_BYTES);
_read_rows_count = get_counter(profile::SPILL_READ_ROWS);
_read_file_count = get_counter(profile::SPILL_READ_FILE_COUNT);
- // Optional: older profiles may not register it.
- _remote_read_requests =
custom_profile->get_counter(profile::SPILL_REMOTE_READ_REQUESTS);
- if (_is_remote) {
- _coalesce_bytes = static_cast<size_t>(std::max<int64_t>(
- 0,
- std::min(config::spill_s3_read_coalesce_bytes,
state->spill_buffer_size_bytes())));
- }
}
void SpillFileReader::_record_read(size_t bytes_read) {
COUNTER_UPDATE(_read_file_size, bytes_read);
ExecEnv::GetInstance()->spill_file_mgr()->update_spill_read_bytes(bytes_read);
- if (_is_remote) {
- // One successful read_at() is one logical GET. Retries inside the
object storage client
- // and reads that finally failed are not counted.
- if (_remote_read_requests != nullptr) {
- COUNTER_UPDATE(_remote_read_requests, 1);
- }
- if (_resource_ctx) {
-
_resource_ctx->io_context()->update_spill_read_bytes_from_remote_storage(bytes_read);
- _resource_ctx->io_context()->update_spill_remote_read_requests(1);
- }
-
ExecEnv::GetInstance()->spill_file_mgr()->update_spill_remote_read(bytes_read,
1);
- } else if (_resource_ctx) {
+ _record_spill_read(bytes_read);
+}
+
+void SpillFileReader::_record_spill_read(size_t bytes_read) {
+ if (_resource_ctx) {
_resource_ctx->io_context()->update_spill_read_bytes_from_local_storage(bytes_read);
}
}
+size_t SpillFileReader::_footer_probe_size(size_t, bool) const {
+ return 2 * sizeof(size_t);
+}
+
+size_t SpillFileReader::_read_window_end(size_t index) const {
+ return _block_start_offsets[index + 1];
+}
+
Status SpillFileReader::open() {
if (_is_open || _part_count == 0) {
return Status::OK();
@@ -120,7 +113,7 @@ Status SpillFileReader::_open_part(size_t part_index, bool
fetch_small_part) {
}
io::FileReaderOptions opts;
opts.cache_type = io::FileCachePolicy::NO_CACHE;
- // The writer recorded the part size; on object storage this saves a HEAD
request.
+ // The writer recorded the part size, so opening needs no size lookup.
opts.file_size = _part_sizes[part_index];
{
SCOPED_TIMER(_read_file_timer);
@@ -138,23 +131,12 @@ Status SpillFileReader::_read_footer(size_t file_size,
bool fetch_small_part) {
// [block start offsets array (_part_block_count * size_t)]
// [serialized blocks]
constexpr size_t kFooterTailBytes = 2 * sizeof(size_t);
- // Enough for the offsets of a few thousand blocks, so the footer usually
takes one GET.
- constexpr size_t kRemoteFooterProbeBytes = 64 * 1024;
if (file_size < kFooterTailBytes) {
return Status::InternalError("spill part {} is too small: {} bytes",
_file_reader->path().native(), file_size);
}
- // Without coalescing (local disk) the footer is read exactly. Otherwise a
larger tail, bounded
- // by the coalesced read size like every other read, is read in one
request, or the whole
- // part when it fits in one coalesced read.
- size_t probe_size = kFooterTailBytes;
- if (_coalesce_bytes > 0) {
- probe_size = fetch_small_part && file_size <= _coalesce_bytes
- ? file_size
- : std::min({file_size, kRemoteFooterProbeBytes,
- std::max(_coalesce_bytes,
kFooterTailBytes)});
- }
+ const size_t probe_size = _footer_probe_size(file_size, fetch_small_part);
_ensure_read_buff(probe_size);
RETURN_IF_ERROR(_read_exact(file_size - probe_size, _read_buff.data(),
probe_size));
@@ -217,14 +199,7 @@ Status SpillFileReader::_block_slice(size_t index, Slice*
out) {
const size_t begin = _block_start_offsets[index];
const size_t end = _block_start_offsets[index + 1];
if (begin < _window_begin || end > _window_end) {
- // Take this block and the following ones as long as the read stays
within
- // _coalesce_bytes. A block larger than that is still read whole.
- size_t last = index + 1;
- while (last < _part_block_count &&
- _block_start_offsets[last + 1] - begin <= _coalesce_bytes) {
- ++last;
- }
- const size_t window_end = _block_start_offsets[last];
+ const size_t window_end = _read_window_end(index);
// A failed read may leave the buffer partly overwritten.
_window_begin = _window_end = 0;
_ensure_read_buff(window_end - begin);
diff --git a/be/src/exec/spill/spill_file_reader.h
b/be/src/exec/spill/spill_file_reader.h
index bbc5f4e3dfb..2c8468dd56c 100644
--- a/be/src/exec/spill/spill_file_reader.h
+++ b/be/src/exec/spill/spill_file_reader.h
@@ -50,18 +50,12 @@ class SpillDataDir;
///
/// Parts are opened on the file system pinned in the SpillFile (local disk or
object storage).
/// Part sizes are known from the writer, so no size lookup is needed on open.
-///
-/// On object storage every read is a GET, so the reader keeps the request
count low:
-/// the part footer is fetched with one tail read of at most `_coalesce_bytes`
(plus one more
-/// read for the rest of a block offset array that does not fit), adjacent
blocks are coalesced
-/// into one read of at most `_coalesce_bytes`, and a part no larger than that
is fetched whole.
-/// A single block, or the rest of the offset array, larger than
`_coalesce_bytes` is read whole.
class SpillFileReader {
public:
SpillFileReader(RuntimeState* state, RuntimeProfile* profile,
SpillDataDir* data_dir,
io::FileSystemSPtr fs, std::string spill_dir,
std::vector<int64_t> part_sizes);
- ~SpillFileReader() { (void)close(); }
+ virtual ~SpillFileReader() { (void)close(); }
/// Open the first part and read its footer metadata.
Status open();
@@ -77,16 +71,24 @@ public:
Status close();
+protected:
+ /// Storage-specific read sizing. Local files read the footer tail and one
block at a time.
+ virtual size_t _footer_probe_size(size_t file_size, bool fetch_small_part)
const;
+ virtual size_t _read_window_end(size_t index) const;
+
+ /// Account a successful store read in the storage-specific counters.
+ virtual void _record_spill_read(size_t bytes_read);
+
private:
- /// Open a specific part file and read its footer. With
`fetch_small_part`, a part that
- /// fits in one coalesced read is fetched whole, so its blocks need no
further reads.
+ /// Open a specific part file and read its footer. `fetch_small_part`
allows the storage
+ /// policy to fetch the whole part when it fits in one read.
Status _open_part(size_t part_index, bool fetch_small_part);
/// Read the footer (block offsets, max sub block size, block count) of
the current part.
Status _read_footer(size_t file_size, bool fetch_small_part);
/// Point `out` at the serialized bytes of block `index` of the current
part, reading
- /// them (and the following blocks that fit in the coalesce window) if not
buffered.
+ /// them (and any following blocks selected by the storage policy) if not
buffered.
Status _block_slice(size_t index, Slice* out);
/// Read exactly `len` bytes at `offset` of the current part into `data`.
@@ -110,9 +112,6 @@ private:
std::string _spill_dir;
std::vector<int64_t> _part_sizes;
size_t _part_count;
- bool _is_remote = false;
- // Upper bound of one coalesced read; 0 reads block by block (local disk).
- size_t _coalesce_bytes = 0;
// ── Current part state ──
size_t _current_part_index = 0;
@@ -126,7 +125,6 @@ private:
PaddedPODArray<char> _read_buff;
size_t _window_begin = 0;
size_t _window_end = 0;
- std::vector<size_t> _block_start_offsets;
PBlock _pb_block;
@@ -138,9 +136,10 @@ private:
RuntimeProfile::Counter* _read_file_size = nullptr;
RuntimeProfile::Counter* _read_rows_count = nullptr;
RuntimeProfile::Counter* _read_file_count = nullptr;
- // Remote only, may be null when the profile does not register it.
- RuntimeProfile::Counter* _remote_read_requests = nullptr;
+protected:
+ // Includes the footer start as the final offset, so every block has an
end offset.
+ std::vector<size_t> _block_start_offsets;
std::shared_ptr<ResourceContext> _resource_ctx = nullptr;
};
diff --git a/be/src/exec/spill/spill_file_writer.cpp
b/be/src/exec/spill/spill_file_writer.cpp
index d780123eb1e..abb9b61a1e4 100644
--- a/be/src/exec/spill/spill_file_writer.cpp
+++ b/be/src/exec/spill/spill_file_writer.cpp
@@ -18,8 +18,6 @@
#include "exec/spill/spill_file_writer.h"
#include <algorithm>
-#include <chrono>
-#include <thread>
#include "agent/be_exec_version_manager.h"
#include "common/config.h"
@@ -31,12 +29,10 @@
#include "io/fs/local_file_writer.h"
#include "runtime/exec_env.h"
#include "runtime/query_context.h"
-#include "runtime/runtime_profile_counter_names.h"
#include "runtime/runtime_state.h"
#include "runtime/thread_context.h"
#include "util/defer_op.h"
#include "util/stopwatch.hpp"
-#include "util/threadpool.h"
namespace doris {
@@ -67,9 +63,6 @@ SpillFileWriter::SpillFileWriter(const
std::shared_ptr<SpillFile>& spill_file, R
_write_file_current_size =
custom_profile->get_counter("SpillWriteFileCurrentBytes");
_write_rows_counter = custom_profile->get_counter("SpillWriteRows");
_total_file_count =
custom_profile->get_counter("SpillWriteFileTotalCount");
- _remote_upload_bytes =
custom_profile->get_counter(profile::SPILL_REMOTE_UPLOAD_BYTES);
- _remote_upload_timer =
custom_profile->get_counter(profile::SPILL_REMOTE_UPLOAD_TIME);
- _remote_upload_wait_timer =
custom_profile->get_counter(profile::SPILL_REMOTE_UPLOAD_WAIT_TIME);
}
SpillFileWriter::~SpillFileWriter() {
@@ -100,10 +93,8 @@ void SpillFileWriter::_discard(SpillFile* spill_file) {
// without publishing the object, before the SpillFile deletes its prefix.
Incomplete
// multipart uploads are left to the bucket lifecycle rule.
std::unique_ptr<io::FileWriter> writer = std::move(_file_writer);
- if (_data_dir->is_remote()) {
- _account_remote_part(/*closed=*/false);
- }
writer.reset();
+ _on_part_closed(nullptr, /*closed=*/false, /*close_ns=*/0);
_advance_part();
}
@@ -157,13 +148,10 @@ Status SpillFileWriter::_close_current_part(const
std::shared_ptr<SpillFile>& sp
if (status.ok()) {
_part_written_bytes += meta_size;
COUNTER_UPDATE(_write_file_total_size, meta_size);
- if (_resource_ctx) {
- if (_data_dir->is_remote()) {
-
_resource_ctx->io_context()->update_spill_write_bytes_to_remote_storage(meta_size);
- } else {
-
_resource_ctx->io_context()->update_spill_write_bytes_to_local_storage(meta_size);
- }
- }
+ // The S3 destructor closes before base destruction; a standalone base
writer
+ // dispatches to its local implementation during its own destructor.
+ // NOLINTNEXTLINE(clang-analyzer-optin.cplusplus.VirtualCall)
+ _record_spill_write_bytes(meta_size);
if (_write_file_current_size) {
COUNTER_UPDATE(_write_file_current_size, meta_size);
}
@@ -181,16 +169,12 @@ Status SpillFileWriter::_close_current_part(const
std::shared_ptr<SpillFile>& sp
// tail of the uploads plus one CompleteMultipartUpload) is negligible
against the part
// itself, so overlapping it with the next part is not worth an async
close pipeline.
std::unique_ptr<io::FileWriter> writer = std::move(_file_writer);
+ int64_t close_ns = 0;
if (status.ok()) {
MonotonicStopWatch close_watch;
close_watch.start();
status = writer->close();
- if (_data_dir->is_remote()) {
- _part_upload_ns += close_watch.elapsed_time();
- }
- }
- if (_data_dir->is_remote()) {
- _account_remote_part(status.ok());
+ close_ns = close_watch.elapsed_time();
}
if (!status.ok()) {
// A part that failed before close() (an append failed or the query
was cancelled) must
@@ -203,13 +187,11 @@ Status SpillFileWriter::_close_current_part(const
std::shared_ptr<SpillFile>& sp
if (!status.ok()) {
LOG(WARNING) << "failed to close spill part " << _current_part_path <<
": " << status;
- // A failed append or close can still leave an object in S3. The
writer has drained
- // in-flight uploads above, so list the prefix after accounting a
conservative upper
- // bound rather than relying on the close status to prove whether PUT
committed.
- if (spill_file && _data_dir->is_remote()) {
- spill_file->account_potential_part(_part_written_bytes);
- }
- } else if (spill_file) {
+ }
+ // S3 destruction closes while derived hooks still exist; base-only
writers use this no-op.
+ // NOLINTNEXTLINE(clang-analyzer-optin.cplusplus.VirtualCall)
+ _on_part_closed(spill_file, status.ok(), close_ns);
+ if (status.ok() && spill_file) {
spill_file->add_part(_part_written_bytes);
}
@@ -217,62 +199,17 @@ Status SpillFileWriter::_close_current_part(const
std::shared_ptr<SpillFile>& sp
return status;
}
-void SpillFileWriter::_account_remote_part(bool closed) {
- // A successful close() publishes the whole part, including its footer.
Failed uploads
- // can still be in flight, so their bytes are not counted as successfully
uploaded.
- const int64_t uploaded_bytes = closed ? _part_written_bytes : 0;
- if (_remote_upload_bytes != nullptr) {
- COUNTER_UPDATE(_remote_upload_bytes, uploaded_bytes);
- COUNTER_UPDATE(_remote_upload_timer, _part_upload_ns);
+void SpillFileWriter::_record_spill_write_bytes(int64_t bytes) {
+ if (_resource_ctx) {
+
_resource_ctx->io_context()->update_spill_write_bytes_to_local_storage(bytes);
}
-
ExecEnv::GetInstance()->spill_file_mgr()->update_spill_remote_write(uploaded_bytes);
}
-Status SpillFileWriter::_wait_for_upload_queue() {
- auto is_cancelled = [this]() {
- return _resource_ctx != nullptr &&
_resource_ctx->task_controller()->is_cancelled();
- };
- // A cancelled query must not start another (billed) upload.
- if (is_cancelled()) {
- return Status::Cancelled("query cancelled before uploading spill
data");
- }
- auto* upload_pool = ExecEnv::GetInstance()->s3_file_upload_thread_pool();
- auto queue_full = [upload_pool]() {
- return static_cast<int64_t>(upload_pool->get_queue_size()) *
config::s3_write_buffer_size >=
- config::spill_s3_max_inflight_upload_bytes;
- };
- if (!queue_full()) {
- return Status::OK();
- }
- MonotonicStopWatch wait_watch;
- wait_watch.start();
- Defer record_wait {
- [&]() { COUNTER_UPDATE(_remote_upload_wait_timer,
wait_watch.elapsed_time()); }};
- // Queued uploads always finish, so the wait ends without help from other
writers.
- while (queue_full()) {
- if (is_cancelled()) {
- return Status::Cancelled("query cancelled while waiting for the
spill upload queue");
- }
- std::this_thread::sleep_for(std::chrono::milliseconds(5));
- }
- return Status::OK();
-}
+void SpillFileWriter::_on_part_closed(const std::shared_ptr<SpillFile>&, bool,
int64_t) {}
Status SpillFileWriter::_append_to_part(const Slice& data) {
RETURN_IF_ERROR(_part_append_status);
- if (!_data_dir->is_remote()) {
- _part_append_status = _file_writer->append(data);
- return _part_append_status;
- }
- // S3FileWriter splits `data` into upload buffers and queues every full
one without waiting,
- // so the queue is checked once per append: `data` is already in memory,
and the wait keeps
- // the next append from producing faster than object storage takes the
data.
- _part_append_status = _wait_for_upload_queue();
- RETURN_IF_ERROR(_part_append_status);
- MonotonicStopWatch append_watch;
- append_watch.start();
_part_append_status = _file_writer->append(data);
- _part_upload_ns += append_watch.elapsed_time();
return _part_append_status;
}
@@ -282,7 +219,6 @@ void SpillFileWriter::_advance_part() {
_part_written_bytes = 0;
_part_max_sub_block_size = 0;
_part_meta.clear();
- _part_upload_ns = 0;
_part_append_status = Status::OK();
}
@@ -427,15 +363,7 @@ Status SpillFileWriter::_write_internal(const Block& block,
_part_meta.append((const char*)&_part_written_bytes,
sizeof(size_t));
COUNTER_UPDATE(_write_file_total_size, buff_size);
- if (_resource_ctx) {
- if (_data_dir->is_remote()) {
-
_resource_ctx->io_context()->update_spill_write_bytes_to_remote_storage(
- buff_size);
- } else {
-
_resource_ctx->io_context()->update_spill_write_bytes_to_local_storage(
- buff_size);
- }
- }
+ _record_spill_write_bytes(buff_size);
if (_write_file_current_size) {
COUNTER_UPDATE(_write_file_current_size, buff_size);
}
diff --git a/be/src/exec/spill/spill_file_writer.h
b/be/src/exec/spill/spill_file_writer.h
index ac60b5f67c0..dd17e5480c8 100644
--- a/be/src/exec/spill/spill_file_writer.h
+++ b/be/src/exec/spill/spill_file_writer.h
@@ -47,19 +47,15 @@ class SpillFile;
/// directory. A part is closed synchronously when it is rotated out or when
close() is called.
/// If the SpillFile is destroyed before the writer is closed, the unfinished
part is discarded.
///
-/// Files are created on the file system pinned in the SpillFile, which is
either a local disk or
-/// the object storage of a cloud storage vault. For object storage the writer
additionally
-/// pauses before every append while the S3 upload queue holds at least
-/// spill_s3_max_inflight_upload_bytes of buffers, which slows down a spill
that
-/// produces faster than object storage takes it, and reports request
statistics derived from
-/// the closed part.
+/// Files are created on the file system pinned in the SpillFile. This base
class handles
+/// common part lifecycle and local writes; SpillS3FileWriter provides
object-storage writes.
class SpillFileWriter {
public:
SpillFileWriter(const std::shared_ptr<SpillFile>& spill_file,
RuntimeState* state,
RuntimeProfile* profile, SpillDataDir* data_dir,
io::FileSystemSPtr fs,
std::string spill_dir);
- ~SpillFileWriter();
+ virtual ~SpillFileWriter();
/// Write a block. Automatically opens the first part, splits large blocks,
/// and rotates to a new part when the current one exceeds max_part_size.
@@ -76,6 +72,19 @@ public:
size_t max_block_bytes() const { return _max_block_bytes; }
size_t max_footer_bytes() const { return _max_footer_bytes; }
+protected:
+ /// Preserve the first append failure for the whole part, including its
footer.
+ virtual Status _append_to_part(const Slice& data);
+
+ /// Record bytes in the storage-specific query I/O context.
+ virtual void _record_spill_write_bytes(int64_t bytes);
+
+ /// Account a closed or discarded part after its writer has drained on
failure.
+ virtual void _on_part_closed(const std::shared_ptr<SpillFile>& spill_file,
bool closed,
+ int64_t close_ns);
+
+ const std::string& _spill_directory() const { return _spill_dir; }
+
private:
Status _close_impl(const std::shared_ptr<SpillFile>& spill_file);
@@ -87,26 +96,11 @@ private:
Status _open_next_part(const std::shared_ptr<SpillFile>& spill_file);
/// Close the current part: write the footer, close the file writer
synchronously
- /// (draining in-flight uploads on failure), account its remote
statistics, register
+ /// (draining in-flight uploads on failure), account its statistics,
register
/// the successful part with the SpillFile, and advance to the next part
index.
/// Statistics are accounted whether it succeeds or not.
Status _close_current_part(const std::shared_ptr<SpillFile>& spill_file);
- /// Append to the current part. For object storage `data` is appended after
- /// _wait_for_upload_queue(). After a failed append the part is
- /// broken: every later append, the footer included, returns that failure
without reaching
- /// the file writer (S3FileWriter must not be appended to after a failed
append).
- Status _append_to_part(const Slice& data);
-
- /// Object storage only: refuse a cancelled query, and wait while the S3
upload queue holds
- /// at least spill_s3_max_inflight_upload_bytes of buffers. Buffers that
are still being
- /// filled are not queued, so idle writers cannot block each other.
- Status _wait_for_upload_queue();
-
- /// Object storage only: account the upload time and, after a successful
close(), the bytes
- /// of the whole part including its footer. Does not inspect S3 upload
state.
- void _account_remote_part(bool closed);
-
/// Reset the per-part state and move on to the next part index.
void _advance_part();
@@ -134,16 +128,17 @@ private:
// ── Current part state (reset on rotation) ──
size_t _current_part_index = 0;
std::string _current_part_path;
- std::unique_ptr<doris::io::FileWriter> _file_writer;
size_t _part_written_blocks = 0;
- int64_t _part_written_bytes = 0;
size_t _part_max_sub_block_size = 0;
std::string _part_meta;
- // Object storage only: time spent in append() and close() of the current
part.
- int64_t _part_upload_ns = 0;
- // First failed append of the current part.
+
+protected:
+ std::unique_ptr<doris::io::FileWriter> _file_writer;
+ int64_t _part_written_bytes = 0;
Status _part_append_status;
+ std::shared_ptr<ResourceContext> _resource_ctx = nullptr;
+private:
bool _closed = false;
// First failed write_block(); every later write and close() returns it.
Status _write_status;
@@ -161,12 +156,6 @@ private:
RuntimeProfile::Counter* _write_rows_counter = nullptr;
RuntimeProfile::Counter* _memory_used_counter = nullptr;
RuntimeProfile::Counter* _total_file_count = nullptr;
- // Remote only, may be null when the profile does not register them.
- RuntimeProfile::Counter* _remote_upload_bytes = nullptr;
- RuntimeProfile::Counter* _remote_upload_timer = nullptr;
- RuntimeProfile::Counter* _remote_upload_wait_timer = nullptr;
-
- std::shared_ptr<ResourceContext> _resource_ctx = nullptr;
};
using SpillFileWriterSPtr = std::shared_ptr<SpillFileWriter>;
} // namespace doris
diff --git a/be/src/exec/spill/spill_s3_file_reader.cpp
b/be/src/exec/spill/spill_s3_file_reader.cpp
new file mode 100644
index 00000000000..bdb943529c2
--- /dev/null
+++ b/be/src/exec/spill/spill_s3_file_reader.cpp
@@ -0,0 +1,82 @@
+// 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.
+
+#include "exec/spill/spill_s3_file_reader.h"
+
+#include <algorithm>
+#include <utility>
+
+#include "common/config.h"
+#include "exec/spill/spill_file_manager.h"
+#include "runtime/exec_env.h"
+#include "runtime/runtime_profile_counter_names.h"
+#include "runtime/runtime_state.h"
+#include "runtime/workload_management/io_context.h"
+
+namespace doris {
+
+SpillS3FileReader::SpillS3FileReader(RuntimeState* state, RuntimeProfile*
profile,
+ SpillDataDir* data_dir,
io::FileSystemSPtr fs,
+ std::string spill_dir,
std::vector<int64_t> part_sizes)
+ : SpillFileReader(state, profile, data_dir, std::move(fs),
std::move(spill_dir),
+ std::move(part_sizes)),
+ _coalesce_bytes(static_cast<size_t>(
+ std::max<int64_t>(0,
std::min(config::spill_s3_read_coalesce_bytes,
+
state->spill_buffer_size_bytes())))) {
+ RuntimeProfile* custom_profile =
profile->get_child(profile::CUSTOM_COUNTERS);
+ _remote_read_requests =
custom_profile->get_counter(profile::SPILL_REMOTE_READ_REQUESTS);
+}
+
+size_t SpillS3FileReader::_footer_probe_size(size_t file_size, bool
fetch_small_part) const {
+ constexpr size_t kFooterTailBytes = 2 * sizeof(size_t);
+ // Enough for the offsets of a few thousand blocks, so the footer usually
takes one GET.
+ constexpr size_t kRemoteFooterProbeBytes = 64 * 1024;
+ if (_coalesce_bytes == 0) {
+ return kFooterTailBytes;
+ }
+ return fetch_small_part && file_size <= _coalesce_bytes
+ ? file_size
+ : std::min({file_size, kRemoteFooterProbeBytes,
+ std::max(_coalesce_bytes, kFooterTailBytes)});
+}
+
+size_t SpillS3FileReader::_read_window_end(size_t index) const {
+ // Take this block and the following ones as long as the read stays within
+ // _coalesce_bytes. A block larger than that is still read whole.
+ const size_t begin = _block_start_offsets[index];
+ size_t last = index + 1;
+ while (last + 1 < _block_start_offsets.size() &&
+ _block_start_offsets[last + 1] - begin <= _coalesce_bytes) {
+ ++last;
+ }
+ return _block_start_offsets[last];
+}
+
+void SpillS3FileReader::_record_spill_read(size_t bytes_read) {
+ // One successful read_at() is one logical GET. Retries inside the object
storage client
+ // and reads that finally failed are not counted.
+ if (_remote_read_requests != nullptr) {
+ COUNTER_UPDATE(_remote_read_requests, 1);
+ }
+ if (_resource_ctx) {
+
_resource_ctx->io_context()->update_spill_read_bytes_from_remote_storage(bytes_read);
+ _resource_ctx->io_context()->update_spill_remote_read_requests(1);
+ }
+
ExecEnv::GetInstance()->spill_file_mgr()->update_spill_remote_read(bytes_read,
1);
+}
+
+} // namespace doris
diff --git a/be/src/exec/spill/spill_s3_file_reader.h
b/be/src/exec/spill/spill_s3_file_reader.h
new file mode 100644
index 00000000000..46e1c7d7e80
--- /dev/null
+++ b/be/src/exec/spill/spill_s3_file_reader.h
@@ -0,0 +1,46 @@
+// 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.
+
+#pragma once
+
+#include "exec/spill/spill_file_reader.h"
+
+namespace doris {
+
+/// S3-specific coalesced reads and request accounting for spill parts.
+/// The footer is fetched with one tail read (plus one more for offsets that
do not fit),
+/// adjacent blocks are coalesced, and a part that fits in the read window is
fetched whole.
+/// A single block or the rest of the offset array larger than the window is
read whole.
+class SpillS3FileReader final : public SpillFileReader {
+public:
+ SpillS3FileReader(RuntimeState* state, RuntimeProfile* profile,
SpillDataDir* data_dir,
+ io::FileSystemSPtr fs, std::string spill_dir,
+ std::vector<int64_t> part_sizes);
+
+protected:
+ size_t _footer_probe_size(size_t file_size, bool fetch_small_part) const
override;
+ size_t _read_window_end(size_t index) const override;
+ void _record_spill_read(size_t bytes_read) override;
+
+private:
+ // Upper bound of one coalesced read; 0 reads block by block.
+ size_t _coalesce_bytes = 0;
+ // Optional: older profiles may not register it.
+ RuntimeProfile::Counter* _remote_read_requests = nullptr;
+};
+
+} // namespace doris
diff --git a/be/src/exec/spill/spill_s3_file_writer.cpp
b/be/src/exec/spill/spill_s3_file_writer.cpp
new file mode 100644
index 00000000000..053761cbc52
--- /dev/null
+++ b/be/src/exec/spill/spill_s3_file_writer.cpp
@@ -0,0 +1,127 @@
+// 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.
+
+#include "exec/spill/spill_s3_file_writer.h"
+
+#include <chrono>
+#include <thread>
+#include <utility>
+
+#include "common/config.h"
+#include "exec/spill/spill_file.h"
+#include "exec/spill/spill_file_manager.h"
+#include "runtime/exec_env.h"
+#include "runtime/runtime_profile_counter_names.h"
+#include "runtime/workload_management/io_context.h"
+#include "util/defer_op.h"
+#include "util/stopwatch.hpp"
+#include "util/threadpool.h"
+
+namespace doris {
+
+SpillS3FileWriter::SpillS3FileWriter(const std::shared_ptr<SpillFile>&
spill_file,
+ RuntimeState* state, RuntimeProfile*
profile,
+ SpillDataDir* data_dir,
io::FileSystemSPtr fs,
+ std::string spill_dir)
+ : SpillFileWriter(spill_file, state, profile, data_dir, std::move(fs),
+ std::move(spill_dir)) {
+ RuntimeProfile* custom_profile = profile->get_child("CustomCounters");
+ _remote_upload_bytes =
custom_profile->get_counter(profile::SPILL_REMOTE_UPLOAD_BYTES);
+ _remote_upload_timer =
custom_profile->get_counter(profile::SPILL_REMOTE_UPLOAD_TIME);
+ _remote_upload_wait_timer =
custom_profile->get_counter(profile::SPILL_REMOTE_UPLOAD_WAIT_TIME);
+}
+
+SpillS3FileWriter::~SpillS3FileWriter() {
+ // Base destruction cannot dispatch virtual hooks to this class. Close
while the S3
+ // counters and queue admission hook are still alive, then let the base
clear registration.
+ Status st = close();
+ if (!st.ok()) {
+ LOG(WARNING) << "SpillS3FileWriter::~SpillS3FileWriter() failed: " <<
st.to_string()
+ << ", spill_dir=" << _spill_directory();
+ }
+}
+
+Status SpillS3FileWriter::_wait_for_upload_queue() {
+ auto is_cancelled = [this]() {
+ return _resource_ctx != nullptr &&
_resource_ctx->task_controller()->is_cancelled();
+ };
+ // A cancelled query must not start another (billed) upload.
+ if (is_cancelled()) {
+ return Status::Cancelled("query cancelled before uploading spill
data");
+ }
+ auto* upload_pool = ExecEnv::GetInstance()->s3_file_upload_thread_pool();
+ auto queue_full = [upload_pool]() {
+ return static_cast<int64_t>(upload_pool->get_queue_size()) *
config::s3_write_buffer_size >=
+ config::spill_s3_max_inflight_upload_bytes;
+ };
+ if (!queue_full()) {
+ return Status::OK();
+ }
+ MonotonicStopWatch wait_watch;
+ wait_watch.start();
+ Defer record_wait {
+ [&]() { COUNTER_UPDATE(_remote_upload_wait_timer,
wait_watch.elapsed_time()); }};
+ // Queued uploads always finish, so the wait ends without help from other
writers.
+ while (queue_full()) {
+ if (is_cancelled()) {
+ return Status::Cancelled("query cancelled while waiting for the
spill upload queue");
+ }
+ std::this_thread::sleep_for(std::chrono::milliseconds(5));
+ }
+ return Status::OK();
+}
+
+Status SpillS3FileWriter::_append_to_part(const Slice& data) {
+ RETURN_IF_ERROR(_part_append_status);
+ // S3FileWriter splits `data` into upload buffers and queues every full
one without waiting,
+ // so the queue is checked once per append: `data` is already in memory,
and the wait keeps
+ // the next append from producing faster than object storage takes the
data.
+ _part_append_status = _wait_for_upload_queue();
+ RETURN_IF_ERROR(_part_append_status);
+ MonotonicStopWatch append_watch;
+ append_watch.start();
+ _part_append_status = _file_writer->append(data);
+ _part_upload_ns += append_watch.elapsed_time();
+ return _part_append_status;
+}
+
+void SpillS3FileWriter::_record_spill_write_bytes(int64_t bytes) {
+ if (_resource_ctx) {
+
_resource_ctx->io_context()->update_spill_write_bytes_to_remote_storage(bytes);
+ }
+}
+
+void SpillS3FileWriter::_on_part_closed(const std::shared_ptr<SpillFile>&
spill_file, bool closed,
+ int64_t close_ns) {
+ _part_upload_ns += close_ns;
+ // A successful close publishes the whole part, including its footer.
Failed uploads
+ // can still be in flight, so their bytes are not counted as successfully
uploaded.
+ const int64_t uploaded_bytes = closed ? _part_written_bytes : 0;
+ if (_remote_upload_bytes != nullptr) {
+ COUNTER_UPDATE(_remote_upload_bytes, uploaded_bytes);
+ COUNTER_UPDATE(_remote_upload_timer, _part_upload_ns);
+ }
+
ExecEnv::GetInstance()->spill_file_mgr()->update_spill_remote_write(uploaded_bytes);
+ // A failed close may have published an object despite returning an error.
The caller
+ // has already drained its writer, so reconcile the prefix before charging
the file.
+ if (!closed && spill_file) {
+ spill_file->account_potential_part(_part_written_bytes);
+ }
+ _part_upload_ns = 0;
+}
+
+} // namespace doris
diff --git a/be/src/exec/spill/spill_s3_file_writer.h
b/be/src/exec/spill/spill_s3_file_writer.h
new file mode 100644
index 00000000000..7bea2e771d4
--- /dev/null
+++ b/be/src/exec/spill/spill_s3_file_writer.h
@@ -0,0 +1,47 @@
+// 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.
+
+#pragma once
+
+#include "exec/spill/spill_file_writer.h"
+
+namespace doris {
+
+/// S3-specific upload admission and accounting for a spill part.
+class SpillS3FileWriter final : public SpillFileWriter {
+public:
+ SpillS3FileWriter(const std::shared_ptr<SpillFile>& spill_file,
RuntimeState* state,
+ RuntimeProfile* profile, SpillDataDir* data_dir,
io::FileSystemSPtr fs,
+ std::string spill_dir);
+ ~SpillS3FileWriter() override;
+
+protected:
+ Status _append_to_part(const Slice& data) override;
+ void _record_spill_write_bytes(int64_t bytes) override;
+ void _on_part_closed(const std::shared_ptr<SpillFile>& spill_file, bool
closed,
+ int64_t close_ns) override;
+
+private:
+ Status _wait_for_upload_queue();
+
+ int64_t _part_upload_ns = 0;
+ RuntimeProfile::Counter* _remote_upload_bytes = nullptr;
+ RuntimeProfile::Counter* _remote_upload_timer = nullptr;
+ RuntimeProfile::Counter* _remote_upload_wait_timer = nullptr;
+};
+
+} // namespace doris
diff --git a/be/test/vec/spill/spill_file_s3_test.cpp
b/be/test/vec/spill/spill_file_s3_test.cpp
index fef54874ae4..c4496a5cc4e 100644
--- a/be/test/vec/spill/spill_file_s3_test.cpp
+++ b/be/test/vec/spill/spill_file_s3_test.cpp
@@ -51,6 +51,8 @@
#include "exec/spill/spill_file_manager.h"
#include "exec/spill/spill_file_reader.h"
#include "exec/spill/spill_file_writer.h"
+#include "exec/spill/spill_s3_file_reader.h"
+#include "exec/spill/spill_s3_file_writer.h"
#include "io/fs/file_system.h"
#include "io/fs/file_writer.h"
#include "io/fs/local_file_system.h"
@@ -464,6 +466,7 @@ protected:
_saved_inflight = config::spill_s3_max_inflight_upload_bytes;
_saved_check_after_upload =
config::enable_s3_object_check_after_upload;
_saved_read_coalesce = config::spill_s3_read_coalesce_bytes;
+ _saved_enable_file_cache = config::enable_file_cache;
// The manager talks to meta-service through the cloud storage engine
only in cloud
// mode; there is no engine in this test, so pin the mode regardless
of earlier tests.
_saved_deploy_mode = config::deploy_mode;
@@ -477,6 +480,8 @@ protected:
config::s3_write_buffer_size = 8 * 1024;
config::spill_file_part_size_bytes = 32 * 1024;
config::spill_storage_limit_bytes = 0;
+ // The mock S3 fixture has no FileCacheFactory; S3 spill explicitly
bypasses file cache.
+ config::enable_file_cache = false;
_runtime_state = std::make_unique<MockRuntimeState>();
_profile = std::make_unique<RuntimeProfile>("test");
@@ -513,6 +518,7 @@ protected:
config::spill_s3_max_inflight_upload_bytes = _saved_inflight;
config::enable_s3_object_check_after_upload =
_saved_check_after_upload;
config::spill_s3_read_coalesce_bytes = _saved_read_coalesce;
+ config::enable_file_cache = _saved_enable_file_cache;
config::deploy_mode = _saved_deploy_mode;
config::cloud_unique_id = _saved_cloud_unique_id;
BackendOptions::set_localhost(_saved_localhost);
@@ -793,11 +799,67 @@ protected:
int64_t _saved_inflight = 0;
bool _saved_check_after_upload = true;
int64_t _saved_read_coalesce = 0;
+ bool _saved_enable_file_cache = true;
std::string _saved_deploy_mode;
std::string _saved_cloud_unique_id;
std::string _saved_localhost;
};
+TEST_F(SpillFileS3Test, WriterFactoryUsesS3Subclass) {
+ _create_manager();
+ SpillFileSPtr spill_file;
+ ASSERT_TRUE(_manager->create_spill_file("query_factory/sort-1",
spill_file).ok());
+ SpillFileWriterSPtr writer;
+ ASSERT_TRUE(spill_file->create_writer(_runtime_state.get(),
_profile.get(), writer).ok());
+ ASSERT_NE(dynamic_cast<SpillS3FileWriter*>(writer.get()), nullptr);
+ ASSERT_TRUE(writer->close().ok());
+ writer.reset();
+ spill_file.reset();
+}
+
+TEST_F(SpillFileS3Test, ReaderFactoryUsesS3Subclass) {
+ _create_manager();
+ Status st;
+ auto spill_file = _write_blocks("query_reader_factory/sort-1", {}, &st);
+ ASSERT_TRUE(st.ok()) << st;
+ auto reader = spill_file->create_reader(_runtime_state.get(),
_profile.get());
+ ASSERT_NE(dynamic_cast<SpillS3FileReader*>(reader.get()), nullptr);
+ ASSERT_TRUE(reader->open().ok());
+ Block block;
+ bool eos = false;
+ ASSERT_TRUE(reader->read(&block, &eos).ok());
+ ASSERT_TRUE(eos);
+ ASSERT_TRUE(reader->close().ok());
+ ASSERT_EQ(mock_store().get_requests, 0);
+ ASSERT_EQ(_counter(profile::SPILL_REMOTE_READ_REQUESTS), 0);
+}
+
+TEST_F(SpillFileS3Test, ImplicitClosePreservesRemoteAccounting) {
+ _create_manager();
+ SpillFileSPtr spill_file;
+ ASSERT_TRUE(_manager->create_spill_file("query_implicit/sort-1",
spill_file).ok());
+ SpillFileWriterSPtr writer;
+ ASSERT_TRUE(spill_file->create_writer(_runtime_state.get(),
_profile.get(), writer).ok());
+ std::mt19937 rng(42);
+ auto block = _random_string_block(rng, 8, 100);
+ ASSERT_TRUE(writer->write_block(_runtime_state.get(), block).ok());
+
+ // Destruction must dispatch the S3 footer, upload and accounting hooks
before the
+ // base destructor runs; no explicit close is issued here.
+ writer.reset();
+ ASSERT_TRUE(spill_file->ready_for_reading());
+ const auto object_bytes =
+ static_cast<int64_t>(_object(spill_root() +
"/query_implicit/sort-1/0").size());
+ ASSERT_GT(object_bytes, 0);
+ ASSERT_EQ(_counter(profile::SPILL_REMOTE_UPLOAD_BYTES), object_bytes);
+ ASSERT_EQ(_manager->remote_spill_data_bytes(), object_bytes);
+ ASSERT_EQ(_data_dir->get_spill_data_bytes(), object_bytes);
+ ASSERT_EQ(_read_all(spill_file), _column_values(block));
+ spill_file.reset();
+ ASSERT_EQ(_manager->remote_spill_data_bytes(), 0);
+ ASSERT_EQ(_data_dir->get_spill_data_bytes(), 0);
+}
+
TEST_F(SpillFileS3Test, RemoteStoreLayout) {
_create_manager();
ASSERT_TRUE(_data_dir->is_remote());
@@ -1423,6 +1485,39 @@ TEST_F(SpillFileS3Test, ReadCoalescesAdjacentBlocks) {
}
}
+TEST_F(SpillFileS3Test, ReadCoalesceClampedBySpillBufferSize) {
+ constexpr int64_t kWindow = 1024 * 1024;
+ config::spill_file_part_size_bytes = 8 * kWindow;
+ config::spill_s3_read_coalesce_bytes = 8 * kWindow;
+ auto query_options = _runtime_state->query_options();
+ query_options.__set_spill_buffer_size_bytes(kWindow);
+ _runtime_state->set_query_options(query_options);
+ _create_manager();
+ std::mt19937 rng(61);
+ std::vector<Block> blocks;
+ for (int i = 0; i < 8; ++i) {
+ blocks.push_back(_random_string_block(rng, 256, 1024));
+ }
+ Status st;
+ auto spill_file = _write_blocks("query_read_buffer_limit/sort-1", blocks,
&st);
+ ASSERT_TRUE(st.ok()) << st;
+ auto keys = mock_store().keys_with_prefix(kBucket, spill_root() +
"/query_read_buffer_limit/");
+ ASSERT_EQ(keys.size(), 1);
+ const std::string object = _object(keys[0]);
+ ASSERT_GT(object.size(), kWindow);
+ const auto offsets = _block_offsets(object);
+ std::vector<size_t> expected_gets {64 * 1024};
+ auto reads = _coalesced_reads(offsets, kWindow);
+ expected_gets.insert(expected_gets.end(), reads.begin(), reads.end());
+
+ _reset_get_stats();
+ ASSERT_EQ(_read_all(spill_file), _values_of(blocks));
+ ASSERT_EQ(_get_sizes(), expected_gets);
+ ASSERT_EQ(_counter(profile::SPILL_REMOTE_READ_REQUESTS),
expected_gets.size());
+ auto* io_ctx =
_runtime_state->get_query_ctx()->resource_ctx()->io_context();
+ ASSERT_EQ(io_ctx->spill_read_bytes_from_local_storage(), 0);
+}
+
// A read size smaller than the block offset array: the probe reads only its
tail and one
// more GET fetches the rest of the footer, then every block is read alone.
TEST_F(SpillFileS3Test, FooterLargerThanProbe) {
diff --git a/be/test/vec/spill/spill_file_test.cpp
b/be/test/vec/spill/spill_file_test.cpp
index be0c5375648..003aba1baf4 100644
--- a/be/test/vec/spill/spill_file_test.cpp
+++ b/be/test/vec/spill/spill_file_test.cpp
@@ -27,6 +27,7 @@
#include <numeric>
#include <set>
#include <string>
+#include <typeinfo>
#include <vector>
#include "common/config.h"
@@ -45,6 +46,7 @@
#include "runtime/exec_env.h"
#include "runtime/fragment_mgr.h"
#include "runtime/runtime_profile.h"
+#include "runtime/workload_management/io_context.h"
#include "storage/olap_define.h"
#include "testutil/column_helper.h"
#include "testutil/mock/mock_query_context.h"
@@ -213,6 +215,8 @@ TEST_F(SpillFileTest, CreateWriterAndReader) {
st = spill_file->create_writer(_runtime_state.get(), _profile.get(),
writer);
ASSERT_TRUE(st.ok()) << st.to_string();
ASSERT_TRUE(writer != nullptr);
+ SpillFileWriter* raw_writer = writer.get();
+ ASSERT_EQ(typeid(*raw_writer), typeid(SpillFileWriter));
// Close writer with no data written
st = writer->close();
@@ -222,6 +226,8 @@ TEST_F(SpillFileTest, CreateWriterAndReader) {
// Create reader on empty file (0 parts)
auto reader = spill_file->create_reader(_runtime_state.get(),
_profile.get());
ASSERT_TRUE(reader != nullptr);
+ SpillFileReader* raw_reader = reader.get();
+ ASSERT_EQ(typeid(*raw_reader), typeid(SpillFileReader));
st = reader->open();
ASSERT_TRUE(st.ok()) << st.to_string();
@@ -1894,6 +1900,11 @@ TEST_F(SpillFileTest, ReadCounters) {
auto* read_file_size = _custom_profile->get_counter("SpillReadFileBytes");
ASSERT_TRUE(read_file_size != nullptr);
ASSERT_GT(read_file_size->value(), 0);
+ auto* io_ctx =
_runtime_state->get_query_ctx()->resource_ctx()->io_context();
+ ASSERT_EQ(io_ctx->spill_read_bytes_from_local_storage(),
read_file_size->value());
+ ASSERT_EQ(io_ctx->spill_read_bytes_from_remote_storage(), 0);
+ ASSERT_EQ(io_ctx->spill_remote_read_requests(), 0);
+
ASSERT_EQ(_custom_profile->get_counter(profile::SPILL_REMOTE_READ_REQUESTS)->value(),
0);
}
// Regression test: SpillFileReader used to look up the deserialize timer
under a
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
