github-actions[bot] commented on code in PR #67292:
URL: https://github.com/apache/doris/pull/67292#discussion_r4278847462
##########
be/src/storage/segment/column_reader.cpp:
##########
@@ -1518,6 +1590,15 @@ Status MapFileColumnIterator::next_batch(size_t* n,
MutableColumnPtr& dst, bool*
column_offsets.get_data().back() - column_offsets.get_data()[start
- 1]; // -1 is valid
if (num_items > 0) {
+ if (!read_offset_only()) {
Review Comment:
[P1] Avoid comparing ordinals of pruned map children. For a supported
map_keys/map_values scan, set_access_paths marks the other child SKIP, and
FileColumnIterator::seek_to_ordinal/next_batch leave that child's ordinal
unchanged. After one nonempty batch (or a seek to a nonzero item offset), the
active child and skipped child therefore differ and this DORIS_CHECK aborts the
BE. This runs even with enable_query_read_ahead=false. Derive the item start
from offsets/an active child and only compare participating children; keep the
read-ahead preparation behind its context gate.
##########
be/src/io/cache/cached_remote_file_reader.cpp:
##########
@@ -132,6 +135,32 @@
CachedRemoteFileReader::CachedRemoteFileReader(FileReaderSPtr remote_file_reader
}
}
+std::shared_ptr<RangeCacheWriteback>
CachedRemoteFileReader::make_range_cache_writeback(
+ const IOContext& io_context, PartialBlockWritebackManager*
partial_block_manager) const {
+ DORIS_CHECK(_cache != nullptr);
+ DORIS_CHECK(partial_block_manager != nullptr);
+ if (_resolve_cache_write_mode(&io_context) == CacheWriteMode::NO_WRITE ||
+ use_remote_only_on_cache_miss(&io_context)) {
+ return nullptr;
+ }
+ auto* write_manager = _cache->async_write_manager();
+ DORIS_CHECK(write_manager != nullptr);
+ CacheContext cache_context(&io_context);
+ return std::make_shared<RangeCacheWriteback>(RangeCacheWritebackOptions {
+ .write_manager = write_manager,
+ .partial_block_manager = partial_block_manager,
+ .inflight_index =
config::enable_async_file_cache_write_inflight_write_buffer_index
+ ? _cache->inflight_write_buffer_index()
+ : nullptr,
+ .source_reader = _remote_file_reader,
Review Comment:
[P1] Keep the cached reader owner alive through delayed hole fills.
Retaining _remote_file_reader alone does not keep it open:
CachedRemoteFileReader's destructor unconditionally calls its close(). With
disable_segment_cache=true, a short query can release its segment/cached
wrapper before the default 2-second merge delay expires, leaving
Task::source_reader closed when the hole worker reads it. This aborts at
S3FileReader's !closed DCHECK; PackedFileReader instead rejects the reads even
in release builds, so writeback is lost. Retain the cached wrapper separately
for the lifetime of every queued/active task while using the raw reader for
hole IO.
##########
be/src/io/cache/range_cache_writeback.cpp:
##########
@@ -0,0 +1,176 @@
+// 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 "io/cache/range_cache_writeback.h"
+
+#include <utility>
+
+#include "common/logging.h"
+#include "io/cache/partial_block_writeback_manager.h"
+#include "io/fs/read_ahead_metrics.h"
+#include "io/fs/read_io_trace.h"
+#include "util/time.h"
+
+namespace doris::io {
+
+RangeCacheWriteback::RangeCacheWriteback(RangeCacheWritebackOptions options)
+ : _options(std::move(options)) {
+ DORIS_CHECK(_options.write_manager != nullptr);
+ DORIS_CHECK(_options.partial_block_manager != nullptr);
+ DORIS_CHECK(_options.source_reader != nullptr);
+ DORIS_CHECK(_options.file_size > 0);
+ DORIS_CHECK(_options.source_reader->size() == _options.file_size);
+ DORIS_CHECK(_options.block_size > 0);
+}
+
+std::optional<AsyncCacheWriteEpoch> RangeCacheWriteback::capture_write_epoch()
const {
+ if (!_options.write_manager->accepting()) {
Review Comment:
[P2] Start a cache writer before replacing the cache-populating read path.
With the default enable_async_file_cache_write=false, BlockFileCache leaves its
writer unstarted; only enabling that separate flag starts it. Turning on
enable_query_read_ahead alone still forces range reads to NO_WRITE, but this
guard then returns no epoch and the consumed-range callback skips every
writeback. Successfully prefetched cold data pages consequently never populate
File Cache (and repeat remote GETs whenever Page Cache does not retain them),
whereas the original synchronous path did populate it. Arrange a functioning
writer for read-ahead or preserve the normal cache-population path when the
writer is unavailable; cover this default flag combination.
##########
be/src/storage/segment/segment_iterator.cpp:
##########
@@ -384,6 +385,33 @@ Status SegmentIterator::init(const StorageReadOptions&
opts) {
return status;
}
+Status SegmentIterator::_init_query_read_ahead() {
+ if (!SegmentReadAhead::enabled() || _opts.io_ctx.reader_type !=
ReaderType::READER_QUERY) {
+ return Status::OK();
+ }
+
+ DORIS_CHECK(_opts.runtime_state != nullptr);
+ auto* exec_env = _opts.runtime_state->exec_env();
+ DORIS_CHECK(exec_env != nullptr);
+ auto* scheduler = exec_env->file_range_read_scheduler();
+ DORIS_CHECK(scheduler != nullptr);
+ auto* query_context = _opts.runtime_state->get_query_ctx();
+ DORIS_CHECK(query_context != nullptr);
+ auto read_context =
query_context->get_or_create_file_range_read_context(scheduler);
+ const auto status = SegmentReadAhead::create_for_query(
+ _file_reader, exec_env, std::move(read_context), _opts,
&_segment_read_ahead);
+ if (!status.ok()) {
+ LOG_EVERY_N(WARNING, 100)
+ << "failed to initialize query read-ahead; use the original
read path: " << status;
+ return Status::OK();
+ }
+ DORIS_CHECK(_segment_read_ahead != nullptr);
Review Comment:
[P1] Handle read-ahead being disabled during initialization.
enable_query_read_ahead is mutable: this method can pass enabled(), then
create_for_query can observe a concurrent disable and return OK with its output
reset to nullptr. The new DORIS_CHECK then aborts the BE when an operator turns
the feature off during a scan. The row-store point-query helper has the same
assumption after creating its request context. Treat the factory's documented
OK/null result as synchronous fallback, or keep a single initialization
decision for the request.
##########
be/src/io/cache/partial_block_writeback_manager.cpp:
##########
@@ -0,0 +1,1035 @@
+// 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 "io/cache/partial_block_writeback_manager.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cstring>
+#include <iterator>
+#include <optional>
+#include <utility>
+#include <vector>
+
+#include "common/logging.h"
+#include "cpp/sync_point.h"
+#include "io/cache/hole_fill_planner.h"
+#include "io/cache/inflight_write_buffer_index.h"
+#include "io/fs/read_ahead_metrics.h"
+#include "io/fs/read_io_trace.h"
+#include "util/countdown_latch.h"
+#include "util/defer_op.h"
+#include "util/thread.h"
+#include "util/threadpool.h"
+#include "util/time.h"
+
+namespace doris::io {
+namespace {
+
+using namespace std::chrono_literals;
+
+constexpr auto kCapacityRetryInterval = 10ms;
+constexpr size_t kMaxHoleFillWorkerCount = 128;
+
+} // namespace
+
+size_t PartialBlockWritebackManager::BlockKeyHash::operator()(const BlockKey&
key) const {
+ const size_t manager_hash = std::hash<AsyncCacheWriteManager*>
{}(key.write_manager);
+ const size_t file_hash = KeyHash {}(key.cache_hash);
+ const size_t offset_hash = std::hash<size_t> {}(key.block_offset);
+ return manager_hash ^ (file_hash << 1) ^ (offset_hash << 2);
+}
+
+// Per-call diagnostics only. Keep the request identity alive across moves
into a Task and emit
+// once, after every manager/fragment lock has left scope. The disabled path
reads no clocks.
+struct PartialBlockWritebackManager::SubmitTrace : HoleSubmitTiming {
+ explicit SubmitTrace(const PartialBlockWritebackRequest& request) {
+ if (ReadIOTrace::enabled()) {
+ start_ns = MonotonicNanos();
+ context = request.io_context;
+ reader = request.source_reader;
+ offset = request.block_offset + request.fragment_offset;
+ size = request.data.size;
+ }
+ }
+
+ ~SubmitTrace() {
+ if (start_ns != 0) {
+ ReadIOTrace::record({.event = "hole_submit",
+ .context = &context.io_context,
+ .file = reader->path().native(),
+ .id = task_id,
+ .parent_id = context.io_context.read_trace_id,
+ .offset = offset,
+ .size = size,
+ .time_ns = MonotonicNanos(),
+ .start_ns = start_ns,
+ .outcome = outcome,
+ .hole_submit_timing = this});
+ }
+ }
+
+ int64_t now() const { return start_ns != 0 ? MonotonicNanos() : 0; }
+
+ PartialBlockSubmitResult finish(PartialBlockSubmitResult result) {
+ switch (result) {
+ case PartialBlockSubmitResult::QUEUED:
+ outcome = "queued";
+ break;
+ case PartialBlockSubmitResult::MERGED:
+ outcome = "merged";
+ break;
+ case PartialBlockSubmitResult::ACTIVE_DEDUPLICATED:
+ outcome = "active_deduplicated";
+ break;
+ case PartialBlockSubmitResult::CACHE_BLOCK_PRESENT:
+ outcome = "cache_block_present";
+ break;
+ case PartialBlockSubmitResult::STALE_EPOCH:
+ outcome = "stale_epoch";
+ break;
+ case PartialBlockSubmitResult::BUFFER_ALLOCATION_FAILED:
+ outcome = "allocation_failed";
+ break;
+ case PartialBlockSubmitResult::REJECTED:
+ outcome = "rejected";
+ break;
+ }
+ return result;
+ }
+
+ int64_t start_ns {0};
+ FileRangeReadIOContext context;
+ FileReaderSPtr reader;
+ size_t offset {0};
+ size_t size {0};
+ uint64_t task_id {0};
+ std::string_view outcome;
+};
+
+// One record per locked scan, not per entry. No JSON formatting or
trace-writer locking is done
+// under the queue mutex. CV sleep and its mutex reacquisition are excluded
from scan time.
+struct PartialBlockWritebackManager::QueueScanTrace : HoleQueueScanStats {
+ int64_t start_ns {0};
+ int64_t end_ns {0};
+
+ void record(bool activated) const {
+ if (start_ns != 0) {
+ ReadIOTrace::record({.event = "hole_queue_scan",
+ .time_ns = end_ns,
+ .start_ns = start_ns,
+ .outcome = activated ? "activated" :
"no_runnable_task",
+ .queue_scan = this});
+ }
+ }
+};
+
+struct PartialBlockWritebackManager::Task {
+ // Allocate one tracked block buffer and copy the first fragment into it.
+ static TaskPtr create(PartialBlockWritebackRequest request, const
BlockKey& task_key,
+ size_t block_size, SubmitTrace& trace);
+
+ // A queued task may merge while the manager lock is free. Once activate()
wins, later
+ // fragments are deduplicated and the worker obtains their bytes from the
source read instead.
+ std::optional<PartialBlockSubmitResult> try_merge(size_t fragment_offset,
Slice data,
+ const IOContext&
fragment_context,
+ SubmitTrace& trace) {
+ const auto lock_start = trace.now();
+ std::lock_guard lock(fragment_mutex);
+ const auto locked_at = trace.now();
+ trace.fragment_lock_wait_ns += locked_at - lock_start;
+ Defer hold_time {[&] { trace.fragment_lock_hold_ns += trace.now() -
locked_at; }};
+ trace.task_id = io_context.io_context.read_trace_id;
+ if (is_active()) {
+ const auto trace_start = trace.now();
+ trace_fragment("fragment_ignored_active", fragment_offset,
data.size, fragment_context);
+ trace.lifecycle_trace_ns += trace.now() - trace_start;
+ return PartialBlockSubmitResult::ACTIVE_DEDUPLICATED;
+ }
+ if (!key.write_manager->is_current_write_epoch(write_epoch)) {
+ return std::nullopt;
+ }
+
+
TEST_SYNC_POINT("PartialBlockWritebackManager::try_submit:before_merge_copy");
+ const auto copy_start = trace.now();
+ std::memcpy(buffer->data() + fragment_offset, data.data, data.size);
+ trace.copy_ns += trace.now() - copy_start;
+ trace.copied_bytes += data.size;
+ covered_intervals.emplace_back(
Review Comment:
[P2] Coalesce coverage while merging queued fragments. Each merge appends
another interval even when the same bytes are already covered. A hot partial
block can stay queued indefinitely while other writes exhaust writer capacity;
repeated point/scanner reads with Page Cache disabled or evicted keep growing
this vector although pending_bytes still accounts for one block. Activation
then copies and sorts the entire submission history in HoleFillPlanner, adding
another allocation/CPU spike. Maintain bounded disjoint coverage (or charge/cap
fragment metadata) so repeated identical fragments cannot bypass the queue's
memory bound.
##########
be/src/io/fs/file_range_read_scheduler.cpp:
##########
@@ -0,0 +1,657 @@
+// 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 "io/fs/file_range_read_scheduler.h"
+
+#include <exception>
+#include <limits>
+#include <utility>
+
+#include "common/exception.h"
+#include "common/logging.h"
+#include "core/allocator.h"
+#include "cpp/sync_point.h"
+#include "io/fs/read_ahead_metrics.h"
+#include "io/fs/read_io_trace.h"
+#include "runtime/memory/mem_tracker_limiter.h"
+#include "runtime/thread_context.h"
+#include "util/defer_op.h"
+#include "util/threadpool.h"
+#include "util/time.h"
+
+namespace doris::io {
+
+using FileRangeReadAllocator = Allocator<false, false, false,
DefaultMemoryAllocator, true>;
+
+namespace {
+
+bool add_overflows(size_t left, size_t right) {
+ return right > std::numeric_limits<size_t>::max() - left;
+}
+
+} // namespace
+
+FileRangeReadSubmitResult
FileRangeReadSubmitResult::rejected(FileRangeReadRejectReason reason,
+ Status status) {
+ return FileRangeReadSubmitResult {
+ .reads = {},
+ .reject_reason = reason,
+ .status = std::move(status),
+ };
+}
+
+FileRangeReadBudget::FileRangeReadBudget(size_t max_bytes) :
_max_bytes(max_bytes) {
+ DORIS_CHECK(_max_bytes > 0);
+}
+
+size_t FileRangeReadBudget::resident_bytes() const {
+ std::lock_guard lock(_mutex);
+ return _resident_bytes;
+}
+
+bool FileRangeReadBudget::_can_reserve_unlocked(size_t bytes) const {
+ return bytes <= _max_bytes - _resident_bytes;
+}
+
+void FileRangeReadBudget::_reserve_unlocked(size_t bytes) {
+ DORIS_CHECK(_can_reserve_unlocked(bytes));
+ _resident_bytes += bytes;
+}
+
+void FileRangeReadBudget::_release(size_t bytes) {
+ std::lock_guard lock(_mutex);
+ DORIS_CHECK(bytes <= _resident_bytes);
+ _resident_bytes -= bytes;
+}
+
+FileRangeReadContext::FileRangeReadContext(size_t max_bytes) :
_budget(max_bytes) {}
+
+void FileRangeReadContext::cancel() {
+ if (_cancelled.exchange(true, std::memory_order_acq_rel)) {
+ return;
+ }
+
+ std::lock_guard lock(_reads_mutex);
+ for (const auto& weak_read : _reads) {
+ if (auto read = weak_read.lock(); read != nullptr) {
+ read->request_cancel();
+ }
+ }
+ _reads.clear();
+}
+
+void FileRangeReadContext::_register_reads(
+ const std::vector<std::shared_ptr<FileRangeRead>>& reads) {
+ bool cancel_reads = false;
+ {
+ std::lock_guard lock(_reads_mutex);
+ cancel_reads = _cancelled.load(std::memory_order_acquire);
+ if (!cancel_reads) {
+ std::erase_if(_reads, [](const auto& read) { return
read.expired(); });
+ _reads.reserve(_reads.size() + reads.size());
+ for (const auto& read : reads) {
+ _reads.emplace_back(read);
+ }
+ }
+ }
+ if (cancel_reads) {
+ for (const auto& read : reads) {
+ read->request_cancel();
+ }
+ }
+}
+
+FileRangeReadReservation::FileRangeReadReservation(
+ std::shared_ptr<FileRangeReadContext> context,
+ std::shared_ptr<FileRangeReadBudget> global_budget, size_t bytes)
noexcept
+ : _context(std::move(context)),
_global_budget(std::move(global_budget)), _bytes(bytes) {
+ DORIS_CHECK(_context != nullptr);
+ DORIS_CHECK(_global_budget != nullptr);
+ DORIS_CHECK(_bytes > 0);
+}
+
+FileRangeReadReservation::FileRangeReadReservation(FileRangeReadReservation&&
other) noexcept
+ : _context(std::move(other._context)),
+ _global_budget(std::move(other._global_budget)),
+ _bytes(std::exchange(other._bytes, 0)) {}
+
+FileRangeReadReservation& FileRangeReadReservation::operator=(
+ FileRangeReadReservation&& other) noexcept {
+ if (this != &other) {
+ _reset();
+ _context = std::move(other._context);
+ _global_budget = std::move(other._global_budget);
+ _bytes = std::exchange(other._bytes, 0);
+ }
+ return *this;
+}
+
+FileRangeReadReservation::~FileRangeReadReservation() {
+ _reset();
+}
+
+void FileRangeReadReservation::_reset() {
+ if (_context == nullptr) {
+ return;
+ }
+ _context->_budget._release(_bytes);
+ _global_budget->_release(_bytes);
+ read_ahead_bvars().resident_bytes << -static_cast<int64_t>(_bytes);
+ _context.reset();
+ _global_budget.reset();
+ _bytes = 0;
+}
+
+FileRangeRead::FileRangeRead(FileRange range,
std::shared_ptr<MemTrackerLimiter> tracker,
+ FileRangeReadReservation reservation,
+ std::shared_ptr<ReadAheadStatistics> statistics)
+ : _range(range),
+ _trace_id(ReadIOTrace::next_id()),
+ _tracker(std::move(tracker)),
+ _reservation(std::move(reservation)),
+ _statistics(std::move(statistics)) {
+ FileRangeReadAllocator allocator;
+ _data = reinterpret_cast<char*>(allocator.alloc(_range.size));
+}
+
+FileRangeRead::~FileRangeRead() {
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_tracker);
+ FileRangeReadAllocator allocator;
+ allocator.free(_data, _range.size);
+}
+
+Status FileRangeRead::create(FileRange range,
std::shared_ptr<MemTrackerLimiter> tracker,
+ FileRangeReadReservation reservation,
+ std::shared_ptr<FileRangeRead>* output,
+ std::shared_ptr<ReadAheadStatistics> statistics) {
+ DORIS_CHECK(range.size > 0);
+ DORIS_CHECK(tracker != nullptr);
+ DORIS_CHECK(output != nullptr);
+
+ Status injected_status;
+ TEST_SYNC_POINT_CALLBACK("FileRangeRead::create:inject_failure",
&injected_status);
+ if (!injected_status.ok()) {
+ return injected_status;
+ }
+
+ SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(tracker);
+ try {
+ output->reset(new FileRangeRead(range, std::move(tracker),
std::move(reservation),
+ std::move(statistics)));
+ } catch (const Exception& exception) {
+ return exception.to_status();
+ } catch (const std::exception& exception) {
+ return Status::MemoryAllocFailed("allocate file range read buffer
failed: {}",
+ exception.what());
+ }
+ return Status::OK();
+}
+
+bool FileRangeRead::_is_terminal(State state) {
+ return state == State::READY || state == State::FAILED || state ==
State::CANCELLED;
+}
+
+Status FileRangeRead::wait(bool* ready_on_entry) {
+ std::unique_lock lock(_mutex);
+ if (ready_on_entry != nullptr) {
+ *ready_on_entry = _state == State::READY;
+ }
+ _cv.wait(lock, [this]() { return _is_terminal(_state); });
+ return _status;
+}
+
+void FileRangeRead::request_cancel() {
+ bool publish_cancelled = false;
+ {
+ std::lock_guard lock(_mutex);
+ if (_is_terminal(_state)) {
+ return;
+ }
+ _cancel_requested = true;
+ publish_cancelled = _state == State::QUEUED;
+ if (publish_cancelled) {
+ _state = State::CANCELLED;
+ _status = Status::Cancelled("asynchronous file range read
cancelled");
+ }
+ }
+ if (publish_cancelled) {
+ _cv.notify_all();
+ }
+}
+
+FileRangeRead::State FileRangeRead::state() const {
+ std::lock_guard lock(_mutex);
+ return _state;
+}
+
+Slice FileRangeRead::data() const {
+ return slice(0, _range.size);
+}
+
+Slice FileRangeRead::slice(size_t buffer_offset, size_t size) const {
+ std::lock_guard lock(_mutex);
+ DORIS_CHECK(_state == State::READY);
+ DORIS_CHECK(buffer_offset <= _range.size);
+ DORIS_CHECK(size <= _range.size - buffer_offset);
+ return {_data + buffer_offset, size};
+}
+
+FileRangeReadStats FileRangeRead::stats() const {
+ std::lock_guard lock(_mutex);
+ return _stats;
+}
+
+bool FileRangeRead::_mark_running() {
+ std::lock_guard lock(_mutex);
+ if (_state == State::CANCELLED) {
+ return false;
+ }
+ DORIS_CHECK(_state == State::QUEUED);
+ _state = State::RUNNING;
+ return true;
+}
+
+bool FileRangeRead::_is_cancel_requested() const {
+ std::lock_guard lock(_mutex);
+ return _cancel_requested;
+}
+
+void FileRangeRead::_publish_ready(FileRangeReadStats stats) {
+ _publish_from_running(State::READY, Status::OK(), std::move(stats));
+}
+
+void FileRangeRead::_publish_failed(Status status, FileRangeReadStats stats) {
+ DORIS_CHECK(!status.ok());
+ _publish_from_running(State::FAILED, std::move(status), std::move(stats));
+}
+
+void FileRangeRead::_publish_submit_failure(Status status) {
+ DORIS_CHECK(!status.ok());
+ {
+ std::lock_guard lock(_mutex);
+ if (_state == State::CANCELLED) {
+ return;
+ }
+ DORIS_CHECK(_state == State::QUEUED);
+ _state = State::FAILED;
+ _status = std::move(status);
+ }
+ _cv.notify_all();
+}
+
+void FileRangeRead::_publish_cancelled(FileRangeReadStats stats) {
+ _publish_from_running(State::CANCELLED,
+ Status::Cancelled("asynchronous file range read
cancelled"),
+ std::move(stats));
+}
+
+void FileRangeRead::_publish_from_running(State state, Status status,
FileRangeReadStats stats) {
+ {
+ std::lock_guard lock(_mutex);
+ DORIS_CHECK(_state == State::RUNNING);
+ DORIS_CHECK(_is_terminal(state));
+ if (_cancel_requested) {
+ state = State::CANCELLED;
+ status = Status::Cancelled("asynchronous file range read
cancelled");
+ }
+ _state = state;
+ _status = std::move(status);
+ _stats = std::move(stats);
+ // Publish accounting before waking a consumer that may immediately
close its scanner.
+ if (_statistics != nullptr) {
+ auto& outcome = state == State::READY ?
_statistics->completed_ranges
+ : state == State::FAILED ?
_statistics->failed_ranges
+ :
_statistics->cancelled_ranges;
+ COUNTER_UPDATE(&outcome, 1);
+ COUNTER_UPDATE(&_statistics->queue_wait_time,
_stats.queue_wait_ns);
+ COUNTER_UPDATE(&_statistics->read_time, _stats.read_ns);
+ COUNTER_UPDATE(&_statistics->source_bytes,
static_cast<int64_t>(_stats.bytes_read));
+ COUNTER_UPDATE(&_statistics->successful_source_bytes,
+
static_cast<int64_t>(_stats.successful_source_bytes));
+ COUNTER_UPDATE(&_statistics->remote_bytes,
_stats.file_cache.bytes_read_from_remote);
+ COUNTER_UPDATE(&_statistics->local_bytes,
_stats.file_cache.bytes_read_from_local);
+ COUNTER_UPDATE(&_statistics->remote_requests,
_stats.file_cache.num_remote_io_total);
+ }
+ auto& metrics = read_ahead_bvars();
+ metrics.source_bytes << static_cast<int64_t>(_stats.bytes_read);
+ metrics.successful_source_bytes <<
static_cast<int64_t>(_stats.successful_source_bytes);
+ metrics.remote_bytes << _stats.file_cache.bytes_read_from_remote;
+ metrics.remote_requests << _stats.file_cache.num_remote_io_total;
+ if (state == State::FAILED) {
+ read_ahead_bvars().failed_ranges << 1;
+ }
+ }
+ _cv.notify_all();
+}
+
+FileRangeReadIOContext FileRangeReadIOContext::from_caller(const IOContext&
source) {
+ FileRangeReadIOContext result;
+ result.io_context = source;
+ if (source.query_id != nullptr) {
+ result.query_id = std::make_shared<TUniqueId>(*source.query_id);
+ }
+ result.io_context.file_cache_stats = nullptr;
+ result.io_context.file_reader_stats = nullptr;
+ result.io_context.remote_scan_cache_write_limiter = nullptr;
+ result.io_context.condition_cache_filtered_rows = 0;
+ result.io_context.predicate_filtered_rows = 0;
+ result.io_context.cache_write_mode_override = CacheWriteMode::NO_WRITE;
+ result.io_context.query_id = result.query_id.get();
+ return result;
+}
+
+FileRangeReadScheduler::FileRangeReadScheduler(FileRangeReadSchedulerOptions
options,
+ ThreadPool* executor)
+ : _options(options),
+
_global_budget(std::make_shared<FileRangeReadBudget>(options.max_bytes_per_be)),
+
_mem_tracker(MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::CACHE,
+
"FileRangeReadScheduler")),
+ _executor(executor) {
+ DORIS_CHECK(_executor != nullptr);
+ read_ahead_bvars().resident_bytes << 0;
+}
+
+FileRangeReadScheduler::~FileRangeReadScheduler() {
+ shutdown();
+}
+
+Status FileRangeReadSchedulerOptions::validate() const {
+ if (max_bytes_per_query == 0) {
+ return Status::InvalidArgument("file range read query byte limit must
be positive");
+ }
+ if (max_bytes_per_be == 0) {
+ return Status::InvalidArgument("file range read BE byte limit must be
positive");
+ }
+ return Status::OK();
+}
+
+Status FileRangeReadScheduler::create(const FileRangeReadSchedulerOptions&
options,
+ ThreadPool* executor,
+ std::unique_ptr<FileRangeReadScheduler>*
output) {
+ DORIS_CHECK(output != nullptr);
+ DORIS_CHECK(executor != nullptr);
+ RETURN_IF_ERROR(options.validate());
+ output->reset(new FileRangeReadScheduler(options, executor));
+ return Status::OK();
+}
+
+std::shared_ptr<FileRangeReadContext> FileRangeReadScheduler::create_context()
const {
+ return
std::make_shared<FileRangeReadContext>(_options.max_bytes_per_query);
+}
+
+FileRangeReadSubmitResult FileRangeReadScheduler::try_submit(
+ const std::vector<FileRange>& ranges, FileReaderSPtr reader,
+ FileRangeReadIOContext io_context, const
std::shared_ptr<FileRangeReadContext>& context) {
+ const auto reject = [&](FileRangeReadRejectReason reason, Status status =
Status::OK()) {
+ read_ahead_bvars().rejected_batches << 1;
+ if (io_context.statistics != nullptr) {
+ COUNTER_UPDATE(&io_context.statistics->rejected_batches, 1);
+ if (reason == FileRangeReadRejectReason::QUERY_BYTE_LIMIT) {
+
COUNTER_UPDATE(&io_context.statistics->query_budget_rejected_batches, 1);
+ } else if (reason == FileRangeReadRejectReason::GLOBAL_BYTE_LIMIT)
{
+
COUNTER_UPDATE(&io_context.statistics->be_budget_rejected_batches, 1);
+ }
+ }
+ return FileRangeReadSubmitResult::rejected(reason, std::move(status));
+ };
+ size_t total_bytes = 0;
+ Status validation_status = _validate_request(ranges, reader, context,
&total_bytes);
+ if (!validation_status.ok()) {
+ return reject(FileRangeReadRejectReason::INVALID_REQUEST,
std::move(validation_status));
+ }
+ if (!accepting()) {
+ return reject(FileRangeReadRejectReason::SHUTTING_DOWN,
+ Status::Cancelled("file range read scheduler is shutting
down"));
+ }
+ if (context->cancelled()) {
+ return reject(FileRangeReadRejectReason::QUERY_CANCELLED,
+ Status::Cancelled("query cancelled before file range
submission"));
+ }
+
+ std::vector<FileRangeReadReservation> reservations;
+ reservations.reserve(ranges.size());
+ const auto reserve_reason = _reserve_batch(context, ranges, total_bytes,
&reservations);
+ if (reserve_reason != FileRangeReadRejectReason::NONE) {
+ return reject(reserve_reason);
+ }
+
+ std::vector<std::shared_ptr<FileRangeRead>> reads;
+ reads.reserve(ranges.size());
+ std::vector<ReadTask> tasks;
+ tasks.reserve(ranges.size());
+ for (size_t index = 0; index < ranges.size(); ++index) {
+ std::shared_ptr<FileRangeRead> read;
+ Status status =
+ FileRangeRead::create(ranges[index], _mem_tracker,
std::move(reservations[index]),
+ &read, io_context.statistics);
+ if (!status.ok()) {
+ return reject(FileRangeReadRejectReason::ALLOCATION_FAILED,
std::move(status));
+ }
+ tasks.push_back(ReadTask {
+ .reader = reader, .read = read, .context = context,
.io_context = io_context});
+ reads.emplace_back(std::move(read));
+ }
+ context->_register_reads(reads);
+
+ FileRangeReadRejectReason submit_reason = FileRangeReadRejectReason::NONE;
+ Status submit_status = _submit_tasks(std::move(tasks), &submit_reason);
+ if (!submit_status.ok()) {
+ return reject(submit_reason, std::move(submit_status));
+ }
+ return {.reads = std::move(reads),
+ .reject_reason = FileRangeReadRejectReason::NONE,
+ .status = Status::OK()};
+}
+
+Status FileRangeReadScheduler::_validate_request(
+ const std::vector<FileRange>& ranges, const FileReaderSPtr& reader,
+ const std::shared_ptr<FileRangeReadContext>& context, size_t*
total_bytes) const {
+ DORIS_CHECK(total_bytes != nullptr);
+ if (reader == nullptr) {
+ return Status::InvalidArgument("file range reader is null");
+ }
+ if (context == nullptr) {
+ return Status::InvalidArgument("file range read context is null");
+ }
+ if (ranges.empty()) {
+ return Status::InvalidArgument("file range read batch is empty");
+ }
+
+ *total_bytes = 0;
+ for (const auto& range : ranges) {
+ if (range.size == 0) {
+ return Status::InvalidArgument("file range at offset {} is empty",
range.offset);
+ }
+ if (add_overflows(range.offset, range.size)) {
+ return Status::InvalidArgument("file range at offset {} with size
{} overflows",
+ range.offset, range.size);
+ }
+ if (range.end() > reader->size()) {
+ return Status::InvalidArgument("file range [{}, {}) exceeds file
size {}", range.offset,
+ range.end(), reader->size());
+ }
+ if (add_overflows(*total_bytes, range.size)) {
+ return Status::InvalidArgument("file range read batch byte size
overflows");
+ }
+ *total_bytes += range.size;
+ }
+ return Status::OK();
+}
+
+FileRangeReadRejectReason FileRangeReadScheduler::_reserve_batch(
+ const std::shared_ptr<FileRangeReadContext>& context, const
std::vector<FileRange>& ranges,
+ size_t total_bytes, std::vector<FileRangeReadReservation>*
reservations) {
+ DORIS_CHECK(reservations != nullptr);
+ DORIS_CHECK(reservations->capacity() >= ranges.size());
+ auto& query_budget = context->_budget;
+ std::scoped_lock lock(query_budget._mutex, _global_budget->_mutex);
+
+ if (total_bytes > query_budget._max_bytes - query_budget._resident_bytes) {
+ return FileRangeReadRejectReason::QUERY_BYTE_LIMIT;
+ }
+ if (total_bytes > _global_budget->_max_bytes -
_global_budget->_resident_bytes) {
+ return FileRangeReadRejectReason::GLOBAL_BYTE_LIMIT;
+ }
+
+ query_budget._reserve_unlocked(total_bytes);
+ _global_budget->_reserve_unlocked(total_bytes);
+ read_ahead_bvars().resident_bytes << static_cast<int64_t>(total_bytes);
+ for (const auto& range : ranges) {
+ reservations->push_back(FileRangeReadReservation(context,
_global_budget, range.size));
+ }
+ return FileRangeReadRejectReason::NONE;
+}
+
+Status FileRangeReadScheduler::_submit_tasks(std::vector<ReadTask> tasks,
+ FileRangeReadRejectReason*
reject_reason) {
+ DORIS_CHECK(!tasks.empty());
+ DORIS_CHECK(reject_reason != nullptr);
+ std::lock_guard lock(_mutex);
+ if (!_accepting) {
+ *reject_reason = FileRangeReadRejectReason::SHUTTING_DOWN;
+ return Status::Cancelled("file range read scheduler is shutting down");
+ }
+
+ size_t submitted_tasks = 0;
+ Status first_failure;
+ for (auto& task : tasks) {
+ auto read = task.read;
+ DORIS_CHECK(_inflight_reads.emplace(read).second);
+ read_ahead_bvars().inflight_ranges << 1;
+ task.submit_ns = MonotonicNanos();
+ Status status = _executor->submit_func([this, task = std::move(task),
read]() mutable {
+ Defer task_finished {[this, read]() { _task_finished(read); }};
+ _run_task(std::move(task));
+ });
+ if (status.ok()) {
+ ++submitted_tasks;
+ if (read->_statistics != nullptr) {
+ COUNTER_UPDATE(&read->_statistics->submitted_ranges, 1);
+ COUNTER_UPDATE(&read->_statistics->submitted_bytes,
+ static_cast<int64_t>(read->range().size));
+ }
+ continue;
+ }
+ DORIS_CHECK(_inflight_reads.erase(read) == 1);
+ read_ahead_bvars().inflight_ranges << -1;
+ if (first_failure.ok()) {
+ first_failure = status;
+ }
+ if (read->_statistics != nullptr) {
+ COUNTER_UPDATE(&read->_statistics->submit_failed_ranges, 1);
+ }
+ read->_publish_submit_failure(std::move(status));
+ }
+
+ if (submitted_tasks == 0) {
+ DORIS_CHECK(!first_failure.ok());
+ *reject_reason = FileRangeReadRejectReason::EXECUTOR_REJECTED;
+ return first_failure;
+ }
+ return Status::OK();
+}
+
+void FileRangeReadScheduler::_task_finished(const
std::shared_ptr<FileRangeRead>& read) {
+ std::lock_guard lock(_mutex);
+ DORIS_CHECK(_inflight_reads.erase(read) == 1);
+ read_ahead_bvars().inflight_ranges << -1;
+ if (_inflight_reads.empty()) {
+ _tasks_cv.notify_all();
+ }
+}
+
+void FileRangeReadScheduler::_run_task(ReadTask task) const {
+ if (!task.read->_mark_running()) {
+ // Count queued cancellation only for tasks accepted by the executor.
A handle may be
+ // cancelled before submission, including a submission the executor
later rejects.
+ if (task.read->_statistics != nullptr) {
+ COUNTER_UPDATE(&task.read->_statistics->cancelled_ranges, 1);
+ }
+ return;
+ }
+
+ FileRangeReadStats stats;
+ stats.queue_wait_ns = MonotonicNanos() - task.submit_ns;
+ if (task.context->cancelled() || !accepting()) {
+ task.read->_publish_cancelled(std::move(stats));
+ return;
+ }
+ task.io_context.io_context.file_cache_stats = &stats.file_cache;
+ task.io_context.io_context.file_reader_stats = &stats.file_reader;
+ task.io_context.io_context.read_trace_source =
FileReadTraceSource::READ_AHEAD;
+ task.io_context.io_context.read_trace_id = task.read->trace_id();
+ Status status;
+ const int64_t read_start_ns = MonotonicNanos();
+ try {
+ status = task.reader->read_at(task.read->range().offset,
Review Comment:
[P2] Preserve workload-group context for scheduled query reads. The scanner
previously reached S3FileReader with SCOPED_ATTACH_TASK(ctx->state()), but
these reads now run in a plain shared-prefetch-pool task with no retained
ResourceContext or task attachment. S3FileReader's LIMIT_REMOTE_SCAN_IO only
obtains the workload-group throttle and updates its I/O bytes when
thread_context()->is_attach_task() is true, so admitted read-ahead GETs bypass
remote_read_bytes_per_second and disappear from group I/O accounting. These are
the query's required page reads. Retain the appropriate resource/workload-group
owner and attach it at the executor callback before source I/O.
--
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]
