mirror of
https://github.com/apache/doris.git
synced 2026-11-04 18:29:59 +00:00
[feature](be) Add cache-aware file range planning
### What problem does this PR solve? Issue Number: #67611 (tracking) Related PR: #67292, #67579, #67612 Problem Summary: Add cache-aware range planning and the shared read-ahead statistics contract. Coalesce sorted requested ranges, optionally complete high-coverage boundary blocks, and map exact inputs back to final buffers. Keep planning timers for column initialization, exact-row batches, physical-page advancement, window discard and extension in the shared contract; collection remains with the later consumers. ### Release note Internal range planning and shared read-ahead Profile/bvar definitions. ### Check List (For Author) - Test: Unit Test — component coverage is included with this change. The integrated source snapshot d5c427fe337 passed 198 ASAN tests from 11 suites via run-be-ut.sh -j100. This individual intermediate commit was not built separately; target-branch validation is recorded on the PR. Repository clang-format v16, build hygiene and diff checks passed on the integrated tree. No cluster or clang-tidy run. - Behavior changed: Yes — introduces the opt-in component; this synchronization preserves existing defaults and IO/writeback semantics. - Does this need documentation: Yes — component scope and current behavior are tracked in #67611 and the related PR descriptions.
This commit is contained in:
bobhan1 committed
1 parent
7654cf77cd
commit
90cf5b6ce6
6 files changed
+905
No files matched your search
+258
@@ -0,0 +1,258 @@
|
||||
// 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/file_range_planner.h"
|
||||
|
||||
#include <algorithm>
|
||||
#include <cmath>
|
||||
#include <map>
|
||||
#include <utility>
|
||||
|
||||
#include "common/logging.h"
|
||||
#include "io/fs/read_ahead_metrics.h"
|
||||
|
||||
namespace doris::io {
|
||||
|
||||
Status FileRangePlanOptions::validate() const {
|
||||
RETURN_IF_ERROR(coalesce_options.validate());
|
||||
if (cache_block_size == 0) {
|
||||
return Status::InvalidArgument("file range plan cache block size must be positive");
|
||||
}
|
||||
if (!std::isfinite(block_fill_min_coverage) || block_fill_min_coverage <= 0.0 ||
|
||||
block_fill_min_coverage > 1.0) {
|
||||
return Status::InvalidArgument(
|
||||
"file range plan block fill coverage must be finite and in (0, 1]");
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
namespace {
|
||||
|
||||
struct BlockCandidate {
|
||||
FileRange range;
|
||||
size_t missing_base_bytes {0};
|
||||
};
|
||||
|
||||
Status validate_input_ranges(const std::vector<FileRange>& input_ranges, size_t file_size) {
|
||||
size_t previous_end = 0;
|
||||
for (const auto& range : input_ranges) {
|
||||
if (range.size == 0) {
|
||||
return Status::InvalidArgument("input file range at offset {} is empty", range.offset);
|
||||
}
|
||||
if (range.offset > file_size || range.size > file_size - range.offset) {
|
||||
return Status::InvalidArgument(
|
||||
"input file range at offset {} with size {} exceeds file size {}", range.offset,
|
||||
range.size, file_size);
|
||||
}
|
||||
if (range.offset < previous_end) {
|
||||
return Status::InvalidArgument(
|
||||
"input file ranges must be sorted and disjoint: offset {} follows end {}",
|
||||
range.offset, previous_end);
|
||||
}
|
||||
previous_end = range.end();
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
FileRange block_range(size_t offset, size_t file_size, size_t block_size) {
|
||||
const size_t block_offset = offset / block_size * block_size;
|
||||
return {.offset = block_offset, .size = std::min(block_size, file_size - block_offset)};
|
||||
}
|
||||
|
||||
size_t intersection_bytes(const FileRange& left, const FileRange& right) {
|
||||
const size_t begin = std::max(left.offset, right.offset);
|
||||
const size_t end = std::min(left.end(), right.end());
|
||||
return begin < end ? end - begin : 0;
|
||||
}
|
||||
|
||||
// Count coverage in `target` with a monotonic cursor over sorted, disjoint ranges. The caller must
|
||||
// visit targets in ascending order.
|
||||
size_t covered_bytes(const std::vector<FileRange>& ranges, const FileRange& target,
|
||||
size_t* cursor) {
|
||||
while (*cursor < ranges.size() && ranges[*cursor].end() <= target.offset) {
|
||||
++*cursor;
|
||||
}
|
||||
|
||||
size_t result = 0;
|
||||
for (size_t index = *cursor; index < ranges.size() && ranges[index].offset < target.end();
|
||||
++index) {
|
||||
result += intersection_bytes(ranges[index], target);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
void append_boundary_block(std::vector<FileRange>* blocks, const FileRange& range, size_t file_size,
|
||||
size_t block_size) {
|
||||
const FileRange block = block_range(range.offset, file_size, block_size);
|
||||
if (!blocks->empty() && blocks->back().offset == block.offset) {
|
||||
DORIS_CHECK(blocks->back() == block);
|
||||
return;
|
||||
}
|
||||
DORIS_CHECK(blocks->empty() || blocks->back().end() <= block.offset);
|
||||
blocks->push_back(block);
|
||||
}
|
||||
|
||||
std::vector<BlockCandidate> find_block_candidates(const std::vector<FileRange>& input_ranges,
|
||||
const std::vector<FileRange>& base_ranges,
|
||||
size_t file_size,
|
||||
const FileRangePlanOptions& options) {
|
||||
// Interior blocks are already covered by a base read, so only its two boundary blocks can add
|
||||
// bytes. This bounds the candidate count to twice the number of base ranges.
|
||||
std::vector<FileRange> boundary_blocks;
|
||||
boundary_blocks.reserve(base_ranges.size() * 2);
|
||||
for (const auto& base : base_ranges) {
|
||||
append_boundary_block(&boundary_blocks, base, file_size, options.cache_block_size);
|
||||
append_boundary_block(&boundary_blocks, {.offset = base.end() - 1, .size = 1}, file_size,
|
||||
options.cache_block_size);
|
||||
}
|
||||
|
||||
std::vector<BlockCandidate> candidates;
|
||||
candidates.reserve(boundary_blocks.size());
|
||||
size_t input_cursor = 0;
|
||||
size_t base_cursor = 0;
|
||||
for (const auto& block : boundary_blocks) {
|
||||
const size_t input_bytes = covered_bytes(input_ranges, block, &input_cursor);
|
||||
const double coverage = static_cast<double>(input_bytes) / static_cast<double>(block.size);
|
||||
if (coverage < options.block_fill_min_coverage) {
|
||||
continue;
|
||||
}
|
||||
const size_t base_bytes = covered_bytes(base_ranges, block, &base_cursor);
|
||||
DORIS_CHECK(base_bytes <= block.size);
|
||||
if (base_bytes < block.size) {
|
||||
candidates.push_back({.range = block, .missing_base_bytes = block.size - base_bytes});
|
||||
}
|
||||
}
|
||||
std::ranges::sort(candidates, [](const BlockCandidate& left, const BlockCandidate& right) {
|
||||
return std::pair {left.missing_base_bytes, left.range.offset} <
|
||||
std::pair {right.missing_base_bytes, right.range.offset};
|
||||
});
|
||||
return candidates;
|
||||
}
|
||||
|
||||
// Treat each base range as one component, then try boundary-block fills from the fewest added bytes
|
||||
// to the most. An accepted fill joins only the components it overlaps and still obeys the physical
|
||||
// range-size limit.
|
||||
std::vector<FileRange> choose_components(const std::vector<FileRange>& input_ranges,
|
||||
const std::vector<FileRange>& base_ranges,
|
||||
size_t file_size, const FileRangePlanOptions& options) {
|
||||
std::map<size_t, FileRange> components;
|
||||
for (const auto& base : base_ranges) {
|
||||
components.emplace(base.offset, base);
|
||||
}
|
||||
|
||||
// Try the cheapest fills first. The interval map updates only components touched by a
|
||||
// candidate instead of rebuilding every component after each attempt.
|
||||
const auto candidates = find_block_candidates(input_ranges, base_ranges, file_size, options);
|
||||
for (const auto& candidate : candidates) {
|
||||
auto first = components.lower_bound(candidate.range.offset);
|
||||
if (first != components.begin()) {
|
||||
auto previous = first;
|
||||
--previous;
|
||||
if (previous->second.end() > candidate.range.offset) {
|
||||
first = previous;
|
||||
}
|
||||
}
|
||||
DORIS_CHECK(first != components.end());
|
||||
DORIS_CHECK(first->second.offset < candidate.range.end());
|
||||
|
||||
size_t merged_offset = std::min(candidate.range.offset, first->second.offset);
|
||||
size_t merged_end = candidate.range.end();
|
||||
auto last = first;
|
||||
while (last != components.end() && last->second.offset < candidate.range.end()) {
|
||||
DORIS_CHECK(last->second.end() > candidate.range.offset);
|
||||
merged_end = std::max(merged_end, last->second.end());
|
||||
++last;
|
||||
}
|
||||
if (merged_end - merged_offset > options.coalesce_options.max_range_bytes) {
|
||||
continue;
|
||||
}
|
||||
|
||||
components.erase(first, last);
|
||||
components.emplace(merged_offset,
|
||||
FileRange {.offset = merged_offset, .size = merged_end - merged_offset});
|
||||
}
|
||||
|
||||
std::vector<FileRange> result;
|
||||
result.reserve(components.size());
|
||||
for (const auto& [offset, component] : components) {
|
||||
DORIS_CHECK(offset == component.offset);
|
||||
result.push_back(component);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
// Map every original page range back to the containing physical read with one monotonic scan.
|
||||
FileRangePlan build_final_plan(const std::vector<FileRange>& input_ranges,
|
||||
std::vector<FileRange> components) {
|
||||
FileRangePlan result {.ranges = std::move(components), .input_locations = {}};
|
||||
result.input_locations.reserve(input_ranges.size());
|
||||
size_t range_index = 0;
|
||||
for (const auto& input : input_ranges) {
|
||||
while (range_index < result.ranges.size() &&
|
||||
result.ranges[range_index].end() <= input.offset) {
|
||||
++range_index;
|
||||
}
|
||||
DORIS_CHECK(range_index < result.ranges.size());
|
||||
const auto& range = result.ranges[range_index];
|
||||
DORIS_CHECK(range.offset <= input.offset);
|
||||
DORIS_CHECK(input.end() <= range.end());
|
||||
result.input_locations.push_back(
|
||||
{.range_index = range_index, .buffer_offset = input.offset - range.offset});
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
Status FileRangePlanner::plan(const std::vector<FileRange>& input_ranges, size_t file_size,
|
||||
const FileRangePlanOptions& options, FileRangePlan* const output_plan,
|
||||
ReadAheadStatistics* statistics) {
|
||||
DORIS_CHECK(output_plan != nullptr);
|
||||
RETURN_IF_ERROR(options.validate());
|
||||
RETURN_IF_ERROR(validate_input_ranges(input_ranges, file_size));
|
||||
|
||||
const auto base_ranges = FileRangeCoalescer::coalesce(input_ranges, options.coalesce_options);
|
||||
auto components = choose_components(input_ranges, base_ranges, file_size, options);
|
||||
*output_plan = build_final_plan(input_ranges, std::move(components));
|
||||
if (statistics != nullptr) {
|
||||
size_t input_bytes = 0;
|
||||
size_t coalesced_bytes = 0;
|
||||
size_t planned_bytes = 0;
|
||||
for (const auto& range : input_ranges) {
|
||||
input_bytes += range.size;
|
||||
}
|
||||
for (const auto& range : base_ranges) {
|
||||
coalesced_bytes += range.size;
|
||||
}
|
||||
for (const auto& range : output_plan->ranges) {
|
||||
planned_bytes += range.size;
|
||||
}
|
||||
COUNTER_UPDATE(&statistics->input_pages, static_cast<int64_t>(input_ranges.size()));
|
||||
COUNTER_UPDATE(&statistics->input_bytes, static_cast<int64_t>(input_bytes));
|
||||
COUNTER_UPDATE(&statistics->coalesced_ranges, static_cast<int64_t>(base_ranges.size()));
|
||||
COUNTER_UPDATE(&statistics->coalesced_bytes, static_cast<int64_t>(coalesced_bytes));
|
||||
COUNTER_UPDATE(&statistics->block_fill_bytes,
|
||||
static_cast<int64_t>(planned_bytes - coalesced_bytes));
|
||||
COUNTER_UPDATE(&statistics->planned_ranges,
|
||||
static_cast<int64_t>(output_plan->ranges.size()));
|
||||
COUNTER_UPDATE(&statistics->planned_bytes, static_cast<int64_t>(planned_bytes));
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
} // namespace doris::io
|
||||
Vendored
+72
@@ -0,0 +1,72 @@
|
||||
// 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 <cstddef>
|
||||
#include <vector>
|
||||
|
||||
#include "common/status.h"
|
||||
#include "io/fs/file_range_coalescer.h"
|
||||
|
||||
namespace doris::io {
|
||||
|
||||
struct ReadAheadStatistics;
|
||||
|
||||
struct FileRangePlanOptions {
|
||||
/// Limits for the initial range coalescing pass.
|
||||
FileRangeCoalesceOptions coalesce_options;
|
||||
/// File Cache block size used when considering boundary-block completion.
|
||||
size_t cache_block_size {1};
|
||||
/// Minimum input-byte coverage required before completing a boundary block.
|
||||
double block_fill_min_coverage {1.0};
|
||||
|
||||
Status validate() const;
|
||||
};
|
||||
|
||||
/// The location of one input range in its final read buffer.
|
||||
struct FileRangeLocation {
|
||||
/// Index into FileRangePlan::ranges.
|
||||
size_t range_index {0};
|
||||
/// Offset of the original input's first byte within that physical range buffer.
|
||||
size_t buffer_offset {0};
|
||||
|
||||
bool operator==(const FileRangeLocation&) const = default;
|
||||
};
|
||||
|
||||
struct FileRangePlan {
|
||||
/// Physical ranges to read after coalescing and optional cache-block completion.
|
||||
std::vector<FileRange> ranges;
|
||||
/// One location per input in caller-supplied order; every input is contained by that range.
|
||||
std::vector<FileRangeLocation> input_locations;
|
||||
|
||||
bool operator==(const FileRangePlan&) const = default;
|
||||
};
|
||||
|
||||
/// Builds a read-only plan without allocating buffers or performing file IO.
|
||||
class FileRangePlanner {
|
||||
public:
|
||||
/// `input_ranges` must be sorted by offset and mutually disjoint. The planner first coalesces
|
||||
/// them, then completes high-coverage cache blocks when the resulting physical ranges remain
|
||||
/// within `max_range_bytes`. One input larger than that limit remains a standalone read. The
|
||||
/// physical EOF block uses its valid file bytes as the coverage denominator.
|
||||
static Status plan(const std::vector<FileRange>& input_ranges, size_t file_size,
|
||||
const FileRangePlanOptions& options, FileRangePlan* output_plan,
|
||||
ReadAheadStatistics* statistics = nullptr);
|
||||
};
|
||||
|
||||
} // namespace doris::io
|
||||
@@ -0,0 +1,97 @@
|
||||
// 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/read_ahead_metrics.h"
|
||||
|
||||
namespace doris::io {
|
||||
|
||||
void ReadAheadStatistics::update_profile(RuntimeProfile* profile) {
|
||||
DORIS_CHECK(profile != nullptr);
|
||||
ADD_LABEL_COUNTER(profile, "PageReadAhead");
|
||||
for (const char* group :
|
||||
{"ReadAheadPlanning", "ReadAheadIO", "ReadAheadConsume", "ReadAheadWriteback"}) {
|
||||
ADD_CHILD_COUNTER(profile, group, TUnit::NONE, "PageReadAhead");
|
||||
}
|
||||
const auto update = [&](const char* name, const char* parent,
|
||||
RuntimeProfile::Counter& counter) {
|
||||
const int64_t delta = counter.value();
|
||||
COUNTER_UPDATE(ADD_CHILD_COUNTER(profile, name, counter.type(), parent), delta);
|
||||
// Only the reporting thread drains counters. Subtract this snapshot so concurrent
|
||||
// increments remain available for the next report.
|
||||
COUNTER_UPDATE(&counter, -delta);
|
||||
};
|
||||
update("ReadAheadPlanCalls", "ReadAheadPlanning", plan_calls);
|
||||
update("ReadAheadPlanTime", "ReadAheadPlanning", plan_time);
|
||||
update("ReadAheadColumnPlanTime", "ReadAheadPlanning", column_plan_time);
|
||||
update("ReadAheadColumnInitTime", "ReadAheadColumnPlanTime", column_init_time);
|
||||
update("ReadAheadPageAdvanceTime", "ReadAheadColumnPlanTime", page_advance_time);
|
||||
update("ReadAheadWindowDiscardTime", "ReadAheadColumnPlanTime", window_discard_time);
|
||||
update("ReadAheadCurrentBatchPlanTime", "ReadAheadColumnPlanTime", current_batch_plan_time);
|
||||
update("ReadAheadWindowExtendTime", "ReadAheadColumnPlanTime", window_extend_time);
|
||||
update("ReadAheadCandidatePages", "ReadAheadPlanning", candidate_pages);
|
||||
update("ReadAheadCandidateBytes", "ReadAheadPlanning", candidate_bytes);
|
||||
update("ReadAheadPageCacheHits", "ReadAheadPlanning", page_cache_hits);
|
||||
update("ReadAheadPageCacheHitBytes", "ReadAheadPlanning", page_cache_hit_bytes);
|
||||
update("ReadAheadInputPages", "ReadAheadPlanning", input_pages);
|
||||
update("ReadAheadInputBytes", "ReadAheadPlanning", input_bytes);
|
||||
update("ReadAheadCoalescedRanges", "ReadAheadPlanning", coalesced_ranges);
|
||||
update("ReadAheadCoalescedBytes", "ReadAheadPlanning", coalesced_bytes);
|
||||
update("ReadAheadBlockFillBytes", "ReadAheadPlanning", block_fill_bytes);
|
||||
update("ReadAheadPlannedRanges", "ReadAheadPlanning", planned_ranges);
|
||||
update("ReadAheadPlannedBytes", "ReadAheadPlanning", planned_bytes);
|
||||
update("ReadAheadSubmittedRanges", "ReadAheadIO", submitted_ranges);
|
||||
update("ReadAheadSubmittedBytes", "ReadAheadIO", submitted_bytes);
|
||||
update("ReadAheadRejectedBatches", "ReadAheadIO", rejected_batches);
|
||||
update("ReadAheadQueryBudgetRejectedBatches", "ReadAheadIO", query_budget_rejected_batches);
|
||||
update("ReadAheadBEBudgetRejectedBatches", "ReadAheadIO", be_budget_rejected_batches);
|
||||
update("ReadAheadSubmitFailedRanges", "ReadAheadIO", submit_failed_ranges);
|
||||
update("ReadAheadCompletedRanges", "ReadAheadIO", completed_ranges);
|
||||
update("ReadAheadFailedRanges", "ReadAheadIO", failed_ranges);
|
||||
update("ReadAheadCancelledRanges", "ReadAheadIO", cancelled_ranges);
|
||||
update("ReadAheadQueueWaitTime", "ReadAheadIO", queue_wait_time);
|
||||
update("ReadAheadReadTime", "ReadAheadIO", read_time);
|
||||
update("ReadAheadSourceBytes", "ReadAheadIO", source_bytes);
|
||||
update("ReadAheadSuccessfulSourceBytes", "ReadAheadIO", successful_source_bytes);
|
||||
update("ReadAheadRemoteBytes", "ReadAheadIO", remote_bytes);
|
||||
update("ReadAheadLocalBytes", "ReadAheadIO", local_bytes);
|
||||
update("ReadAheadRemoteRequests", "ReadAheadIO", remote_requests);
|
||||
update("ReadAheadReadyPageHits", "ReadAheadConsume", ready_page_hits);
|
||||
update("ReadAheadWaitPageHits", "ReadAheadConsume", wait_page_hits);
|
||||
update("ReadAheadWaitTime", "ReadAheadConsume", wait_time);
|
||||
update("ReadAheadFallbackPages", "ReadAheadConsume", fallback_pages);
|
||||
update("ReadAheadFallbackBytes", "ReadAheadConsume", fallback_bytes);
|
||||
update("ReadAheadFallbackTime", "ReadAheadConsume", fallback_time);
|
||||
update("ReadAheadConsumedPageBytes", "ReadAheadConsume", consumed_page_bytes);
|
||||
update("ReadAheadConsumedRangeBytes", "ReadAheadConsume", consumed_range_bytes);
|
||||
update("ReadAheadReleasedPages", "ReadAheadConsume", released_pages);
|
||||
update("ReadAheadWritebackTime", "ReadAheadWriteback", writeback_time);
|
||||
update("ReadAheadCompleteBlocksSubmitted", "ReadAheadWriteback", complete_blocks_submitted);
|
||||
update("ReadAheadCompleteBlockBytes", "ReadAheadWriteback", complete_block_bytes);
|
||||
update("ReadAheadPartialBlocksQueued", "ReadAheadWriteback", partial_blocks_queued);
|
||||
update("ReadAheadPartialBlocksMerged", "ReadAheadWriteback", partial_blocks_merged);
|
||||
update("ReadAheadPartialFragmentBytes", "ReadAheadWriteback", partial_fragment_bytes);
|
||||
update("ReadAheadWritebackDeduplicatedBlocks", "ReadAheadWriteback",
|
||||
writeback_deduplicated_blocks);
|
||||
update("ReadAheadWritebackRejectedBlocks", "ReadAheadWriteback", writeback_rejected_blocks);
|
||||
}
|
||||
|
||||
ReadAheadBvars& read_ahead_bvars() {
|
||||
static ReadAheadBvars metrics;
|
||||
return metrics;
|
||||
}
|
||||
|
||||
} // namespace doris::io
|
||||
@@ -0,0 +1,124 @@
|
||||
// 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 <bvar/reducer.h>
|
||||
|
||||
#include <cstdint>
|
||||
|
||||
#include "runtime/runtime_profile.h"
|
||||
|
||||
namespace doris::io {
|
||||
|
||||
/// Shared by the reader and its asynchronous range handles. Counters are owned here rather than
|
||||
/// by the caller's RuntimeProfile, so outstanding tasks can finish after the scanner is destroyed.
|
||||
struct ReadAheadStatistics {
|
||||
RuntimeProfile::Counter plan_calls {TUnit::UNIT};
|
||||
/// Range planning and submission in apply_plans(), after physical-column planning.
|
||||
RuntimeProfile::Counter plan_time {TUnit::TIME_NS};
|
||||
/// Physical-column preparation, including initialization, validation and empty plans.
|
||||
RuntimeProfile::Counter column_plan_time {TUnit::TIME_NS};
|
||||
RuntimeProfile::Counter column_init_time {TUnit::TIME_NS};
|
||||
/// Page-index driven window maintenance after scan initialization, excluding apply_plans().
|
||||
RuntimeProfile::Counter page_advance_time {TUnit::TIME_NS};
|
||||
/// Window maintenance sub-timers can overlap initialization and page advancement.
|
||||
RuntimeProfile::Counter window_discard_time {TUnit::TIME_NS};
|
||||
/// Exact-rowid planning for point reads and dependent nested data, including extensions.
|
||||
RuntimeProfile::Counter current_batch_plan_time {TUnit::TIME_NS};
|
||||
/// Shared sub-timer of initialization, page advancement and exact-rowid planning.
|
||||
RuntimeProfile::Counter window_extend_time {TUnit::TIME_NS};
|
||||
RuntimeProfile::Counter candidate_pages {TUnit::UNIT};
|
||||
RuntimeProfile::Counter candidate_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter page_cache_hits {TUnit::UNIT};
|
||||
RuntimeProfile::Counter page_cache_hit_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter input_pages {TUnit::UNIT};
|
||||
RuntimeProfile::Counter input_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter coalesced_ranges {TUnit::UNIT};
|
||||
RuntimeProfile::Counter coalesced_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter block_fill_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter planned_ranges {TUnit::UNIT};
|
||||
RuntimeProfile::Counter planned_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter submitted_ranges {TUnit::UNIT};
|
||||
RuntimeProfile::Counter submitted_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter rejected_batches {TUnit::UNIT};
|
||||
RuntimeProfile::Counter query_budget_rejected_batches {TUnit::UNIT};
|
||||
RuntimeProfile::Counter be_budget_rejected_batches {TUnit::UNIT};
|
||||
/// Executor rejection, including the rejected part of an otherwise accepted batch.
|
||||
RuntimeProfile::Counter submit_failed_ranges {TUnit::UNIT};
|
||||
RuntimeProfile::Counter completed_ranges {TUnit::UNIT};
|
||||
RuntimeProfile::Counter failed_ranges {TUnit::UNIT};
|
||||
RuntimeProfile::Counter cancelled_ranges {TUnit::UNIT};
|
||||
RuntimeProfile::Counter queue_wait_time {TUnit::TIME_NS};
|
||||
RuntimeProfile::Counter read_time {TUnit::TIME_NS};
|
||||
RuntimeProfile::Counter source_bytes {TUnit::BYTES};
|
||||
/// Successful full source reads, including buffers cancelled after the read returned.
|
||||
RuntimeProfile::Counter successful_source_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter remote_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter local_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter remote_requests {TUnit::UNIT};
|
||||
RuntimeProfile::Counter ready_page_hits {TUnit::UNIT};
|
||||
RuntimeProfile::Counter wait_page_hits {TUnit::UNIT};
|
||||
RuntimeProfile::Counter wait_time {TUnit::TIME_NS};
|
||||
/// Synchronous data-page read attempts and requested bytes; a retry counts again.
|
||||
RuntimeProfile::Counter fallback_pages {TUnit::UNIT};
|
||||
RuntimeProfile::Counter fallback_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter fallback_time {TUnit::TIME_NS};
|
||||
RuntimeProfile::Counter consumed_page_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter consumed_range_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter released_pages {TUnit::UNIT};
|
||||
RuntimeProfile::Counter writeback_time {TUnit::TIME_NS};
|
||||
RuntimeProfile::Counter complete_blocks_submitted {TUnit::UNIT};
|
||||
RuntimeProfile::Counter complete_block_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter partial_blocks_queued {TUnit::UNIT};
|
||||
RuntimeProfile::Counter partial_blocks_merged {TUnit::UNIT};
|
||||
RuntimeProfile::Counter partial_fragment_bytes {TUnit::BYTES};
|
||||
RuntimeProfile::Counter writeback_deduplicated_blocks {TUnit::UNIT};
|
||||
RuntimeProfile::Counter writeback_rejected_blocks {TUnit::UNIT};
|
||||
|
||||
/// Transfer accumulated deltas into PageReadAhead counters on the reporting thread. Repeated
|
||||
/// calls add only new work, including when multiple scanners share the destination profile.
|
||||
void update_profile(RuntimeProfile* profile);
|
||||
};
|
||||
|
||||
/// BE-wide gauges and cumulative totals, without query/file/cache-path labels. Gauge updates
|
||||
/// follow the same ownership transitions as the scheduler and hole-fill queue.
|
||||
struct ReadAheadBvars {
|
||||
bvar::Adder<int64_t> resident_bytes {"doris_read_ahead_resident_bytes"};
|
||||
bvar::Adder<int64_t> inflight_ranges {"doris_read_ahead_inflight_ranges"};
|
||||
bvar::Adder<int64_t> rejected_batches {"doris_read_ahead_rejected_batches"};
|
||||
bvar::Adder<int64_t> failed_ranges {"doris_read_ahead_failed_ranges"};
|
||||
// Updated on IO completion or page consumption, independently of Profile reporting.
|
||||
bvar::Adder<int64_t> source_bytes {"doris_read_ahead_source_bytes"};
|
||||
bvar::Adder<int64_t> successful_source_bytes {"doris_read_ahead_successful_source_bytes"};
|
||||
bvar::Adder<int64_t> remote_bytes {"doris_read_ahead_remote_bytes"};
|
||||
bvar::Adder<int64_t> remote_requests {"doris_read_ahead_remote_requests"};
|
||||
bvar::Adder<int64_t> consumed_page_bytes {"doris_read_ahead_consumed_page_bytes"};
|
||||
bvar::Adder<int64_t> hole_fill_pending_bytes {"doris_hole_fill_pending_bytes"};
|
||||
bvar::Adder<int64_t> hole_fill_active_blocks {"doris_hole_fill_active_blocks"};
|
||||
bvar::Adder<int64_t> hole_fill_remote_requests {"doris_hole_fill_remote_requests"};
|
||||
bvar::Adder<int64_t> hole_fill_remote_bytes {"doris_hole_fill_remote_bytes"};
|
||||
bvar::Adder<int64_t> hole_fill_remote_read_time_ns {"doris_hole_fill_remote_read_time_ns"};
|
||||
bvar::Adder<int64_t> hole_fill_failed_blocks {"doris_hole_fill_failed_blocks"};
|
||||
bvar::Adder<int64_t> hole_fill_write_submitted_blocks {
|
||||
"doris_hole_fill_write_submitted_blocks"};
|
||||
bvar::Adder<int64_t> hole_fill_dropped_blocks {"doris_hole_fill_dropped_blocks"};
|
||||
};
|
||||
|
||||
ReadAheadBvars& read_ahead_bvars();
|
||||
|
||||
} // namespace doris::io
|
||||
+222
@@ -0,0 +1,222 @@
|
||||
// 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/file_range_planner.h"
|
||||
|
||||
#include <gtest/gtest.h>
|
||||
|
||||
#include <limits>
|
||||
#include <vector>
|
||||
|
||||
#include "io/fs/read_ahead_metrics.h"
|
||||
|
||||
namespace doris::io {
|
||||
namespace {
|
||||
|
||||
FileRange file_range(size_t offset, size_t size) {
|
||||
return {.offset = offset, .size = size};
|
||||
}
|
||||
|
||||
FileRangePlanOptions options(size_t max_gap_bytes, size_t max_range_bytes,
|
||||
double max_read_amplification_ratio, size_t cache_block_size,
|
||||
double block_fill_min_coverage) {
|
||||
return {.coalesce_options = {.max_gap_bytes = max_gap_bytes,
|
||||
.max_range_bytes = max_range_bytes,
|
||||
.max_read_amplification_ratio = max_read_amplification_ratio},
|
||||
.cache_block_size = cache_block_size,
|
||||
.block_fill_min_coverage = block_fill_min_coverage};
|
||||
}
|
||||
|
||||
FileRangePlan plan(const std::vector<FileRange>& ranges, size_t file_size,
|
||||
const FileRangePlanOptions& plan_options) {
|
||||
FileRangePlan result;
|
||||
EXPECT_TRUE(FileRangePlanner::plan(ranges, file_size, plan_options, &result).ok());
|
||||
return result;
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, EmptyInputClearsPlan) {
|
||||
FileRangePlan result {.ranges = {file_range(0, 1)}, .input_locations = {{.range_index = 0}}};
|
||||
ASSERT_TRUE(FileRangePlanner::plan({}, 0, options(0, 1, 1.0, 1, 1.0), &result).ok());
|
||||
EXPECT_EQ(result, FileRangePlan {});
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, CoalescesAndMapsSortedInputs) {
|
||||
const auto result = plan({file_range(0, 100), file_range(120, 40), file_range(200, 100)}, 512,
|
||||
options(20, 256, 2.0, 512, 1.0));
|
||||
|
||||
ASSERT_EQ(result.ranges.size(), 2);
|
||||
EXPECT_EQ(result.ranges[0], file_range(0, 160));
|
||||
EXPECT_EQ(result.ranges[1], file_range(200, 100));
|
||||
EXPECT_EQ(result.input_locations,
|
||||
(std::vector<FileRangeLocation> {{.range_index = 0, .buffer_offset = 0},
|
||||
{.range_index = 0, .buffer_offset = 120},
|
||||
{.range_index = 1, .buffer_offset = 0}}));
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, CompletesBlockAtCoverageBoundary) {
|
||||
const auto result = plan({file_range(0, 512)}, 1024, options(64, 2048, 2.0, 1024, 0.5));
|
||||
|
||||
ASSERT_EQ(result.ranges.size(), 1);
|
||||
EXPECT_EQ(result.ranges[0], file_range(0, 1024));
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, LeavesBlockPartialBelowCoverageBoundary) {
|
||||
const auto result = plan({file_range(0, 511)}, 1024, options(64, 2048, 2.0, 1024, 0.5));
|
||||
|
||||
ASSERT_EQ(result.ranges.size(), 1);
|
||||
EXPECT_EQ(result.ranges[0], file_range(0, 511));
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, CompletesBlockAfterCoalescingInputGap) {
|
||||
const auto result = plan({file_range(0, 300), file_range(350, 250)}, 1024,
|
||||
options(64, 2048, 2.0, 1024, 0.5));
|
||||
|
||||
ASSERT_EQ(result.ranges.size(), 1);
|
||||
EXPECT_EQ(result.ranges[0], file_range(0, 1024));
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, SeparatesCoalescingAndBlockFillBytes) {
|
||||
ReadAheadStatistics statistics;
|
||||
FileRangePlan result;
|
||||
ASSERT_TRUE(FileRangePlanner::plan({file_range(0, 300), file_range(350, 250)}, 1024,
|
||||
options(64, 2048, 2.0, 1024, 0.5), &result, &statistics)
|
||||
.ok());
|
||||
EXPECT_EQ(statistics.input_pages.value(), 2);
|
||||
EXPECT_EQ(statistics.input_bytes.value(), 550);
|
||||
EXPECT_EQ(statistics.coalesced_ranges.value(), 1);
|
||||
EXPECT_EQ(statistics.coalesced_bytes.value(), 600);
|
||||
EXPECT_EQ(statistics.block_fill_bytes.value(), 424);
|
||||
EXPECT_EQ(statistics.planned_ranges.value(), 1);
|
||||
EXPECT_EQ(statistics.planned_bytes.value(), 1024);
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, CompletedBlockConnectsSeparateBaseRanges) {
|
||||
const auto result = plan({file_range(0, 400), file_range(500, 200)}, 1024,
|
||||
options(64, 1024, 2.0, 1024, 0.5));
|
||||
|
||||
ASSERT_EQ(result.ranges.size(), 1);
|
||||
EXPECT_EQ(result.ranges[0], file_range(0, 1024));
|
||||
EXPECT_EQ(result.input_locations[0].range_index, 0);
|
||||
EXPECT_EQ(result.input_locations[1].range_index, 0);
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, UsesValidEofBlockSizeForCoverage) {
|
||||
const auto result = plan({file_range(1100, 140)}, 1280, options(64, 2048, 2.0, 1024, 0.5));
|
||||
|
||||
ASSERT_EQ(result.ranges.size(), 1);
|
||||
EXPECT_EQ(result.ranges[0], file_range(1024, 256));
|
||||
EXPECT_EQ(result.input_locations[0],
|
||||
(FileRangeLocation {.range_index = 0, .buffer_offset = 76}));
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, ChoosesSmallerBlockFillWhenBothEdgesDoNotFit) {
|
||||
const auto result = plan({file_range(70, 170)}, 300, options(0, 250, 1.0, 100, 0.3));
|
||||
|
||||
ASSERT_EQ(result.ranges.size(), 1);
|
||||
EXPECT_EQ(result.ranges[0], file_range(70, 230));
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, RejectsBlockFillThatWouldExceedFinalRangeLimit) {
|
||||
const auto result = plan({file_range(0, 700), file_range(900, 400)}, 2000,
|
||||
options(64, 1200, 2.0, 1000, 0.5));
|
||||
|
||||
ASSERT_EQ(result.ranges.size(), 2);
|
||||
EXPECT_EQ(result.ranges[0], file_range(0, 700));
|
||||
EXPECT_EQ(result.ranges[1], file_range(900, 400));
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, KeepsInputLargerThanCoalesceLimit) {
|
||||
const auto result =
|
||||
plan({file_range(0, 300), file_range(320, 32)}, 512, options(64, 256, 2.0, 512, 1.0));
|
||||
|
||||
ASSERT_EQ(result.ranges.size(), 2);
|
||||
EXPECT_EQ(result.ranges[0], file_range(0, 300));
|
||||
EXPECT_EQ(result.ranges[1], file_range(320, 32));
|
||||
EXPECT_EQ(result.input_locations[0],
|
||||
(FileRangeLocation {.range_index = 0, .buffer_offset = 0}));
|
||||
EXPECT_EQ(result.input_locations[1],
|
||||
(FileRangeLocation {.range_index = 1, .buffer_offset = 0}));
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, OversizedInputDoesNotPreventOtherBlockFill) {
|
||||
const auto result =
|
||||
plan({file_range(0, 300), file_range(400, 80)}, 512, options(64, 256, 2.0, 100, 0.8));
|
||||
|
||||
ASSERT_EQ(result.ranges.size(), 2);
|
||||
EXPECT_EQ(result.ranges[0], file_range(0, 300));
|
||||
EXPECT_EQ(result.ranges[1], file_range(400, 100));
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, KeepsNaturallyCompleteBlock) {
|
||||
const auto result = plan({file_range(1024, 1024)}, 3072, options(64, 2048, 2.0, 1024, 1.0));
|
||||
|
||||
ASSERT_EQ(result.ranges.size(), 1);
|
||||
EXPECT_EQ(result.ranges[0], file_range(1024, 1024));
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, AlignsCompleteBlocksWithoutOverflow) {
|
||||
const size_t file_size = std::numeric_limits<size_t>::max();
|
||||
const auto result = plan({file_range(file_size - 1, 1)}, file_size, options(0, 1, 1.0, 4, 1.0));
|
||||
|
||||
ASSERT_EQ(result.ranges.size(), 1);
|
||||
EXPECT_EQ(result.ranges[0], file_range(file_size - 1, 1));
|
||||
}
|
||||
|
||||
TEST(FileRangePlannerTest, PreservesOutputOnInvalidOptionsOrInput) {
|
||||
const FileRangePlan original {.ranges = {file_range(7, 9)},
|
||||
.input_locations = {{.range_index = 0}}};
|
||||
auto result = original;
|
||||
|
||||
EXPECT_FALSE(FileRangePlanner::plan({}, 0, options(0, 1, 1.0, 0, 0.5), &result).ok());
|
||||
EXPECT_EQ(result, original);
|
||||
|
||||
EXPECT_FALSE(FileRangePlanner::plan({}, 0, options(0, 1, 1.0, 1, 0.0), &result).ok());
|
||||
EXPECT_EQ(result, original);
|
||||
|
||||
EXPECT_FALSE(
|
||||
FileRangePlanner::plan(
|
||||
{}, 0, options(0, 1, 1.0, 1, std::numeric_limits<double>::quiet_NaN()), &result)
|
||||
.ok());
|
||||
EXPECT_EQ(result, original);
|
||||
|
||||
EXPECT_FALSE(FileRangePlanner::plan({}, 0, options(0, 0, 1.0, 1, 0.5), &result).ok());
|
||||
EXPECT_EQ(result, original);
|
||||
|
||||
EXPECT_FALSE(
|
||||
FileRangePlanner::plan({file_range(5, 0)}, 10, options(0, 10, 1.0, 10, 0.5), &result)
|
||||
.ok());
|
||||
EXPECT_EQ(result, original);
|
||||
|
||||
EXPECT_FALSE(
|
||||
FileRangePlanner::plan({file_range(9, 2)}, 10, options(0, 10, 1.0, 10, 0.5), &result)
|
||||
.ok());
|
||||
EXPECT_EQ(result, original);
|
||||
|
||||
EXPECT_FALSE(FileRangePlanner::plan({file_range(0, 80), file_range(40, 80)}, 120,
|
||||
options(0, 100, 1.0, 100, 1.0), &result)
|
||||
.ok());
|
||||
EXPECT_EQ(result, original);
|
||||
|
||||
EXPECT_FALSE(FileRangePlanner::plan({file_range(std::numeric_limits<size_t>::max() - 1, 2)},
|
||||
std::numeric_limits<size_t>::max(),
|
||||
options(0, 2, 1.0, 1, 1.0), &result)
|
||||
.ok());
|
||||
EXPECT_EQ(result, original);
|
||||
}
|
||||
|
||||
} // namespace
|
||||
} // namespace doris::io
|
||||
@@ -0,0 +1,132 @@
|
||||
// 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/read_ahead_metrics.h"
|
||||
|
||||
#include <gtest/gtest.h>
|
||||
|
||||
#include <memory>
|
||||
#include <thread>
|
||||
|
||||
namespace doris::io {
|
||||
|
||||
TEST(ReadAheadMetricsTest, ReportsOnlyNewDeltasAndAggregatesReaders) {
|
||||
ReadAheadStatistics first;
|
||||
ReadAheadStatistics second;
|
||||
RuntimeProfile profile("SegmentIterator");
|
||||
|
||||
COUNTER_UPDATE(&first.input_pages, 2);
|
||||
COUNTER_UPDATE(&first.input_bytes, 48);
|
||||
COUNTER_UPDATE(&first.plan_time, 100);
|
||||
COUNTER_UPDATE(&first.column_plan_time, 1000);
|
||||
COUNTER_UPDATE(&first.column_init_time, 200);
|
||||
COUNTER_UPDATE(&first.page_advance_time, 80);
|
||||
COUNTER_UPDATE(&first.window_discard_time, 100);
|
||||
COUNTER_UPDATE(&first.current_batch_plan_time, 600);
|
||||
COUNTER_UPDATE(&first.window_extend_time, 50);
|
||||
COUNTER_UPDATE(&first.candidate_bytes, 60);
|
||||
COUNTER_UPDATE(&first.page_cache_hit_bytes, 12);
|
||||
COUNTER_UPDATE(&first.query_budget_rejected_batches, 1);
|
||||
COUNTER_UPDATE(&first.be_budget_rejected_batches, 2);
|
||||
COUNTER_UPDATE(&first.submit_failed_ranges, 3);
|
||||
COUNTER_UPDATE(&first.successful_source_bytes, 40);
|
||||
COUNTER_UPDATE(&first.fallback_pages, 1);
|
||||
COUNTER_UPDATE(&first.fallback_bytes, 8);
|
||||
COUNTER_UPDATE(&first.fallback_time, 200);
|
||||
first.update_profile(&profile);
|
||||
first.update_profile(&profile);
|
||||
COUNTER_UPDATE(&first.input_bytes, 12);
|
||||
COUNTER_UPDATE(&second.input_bytes, 20);
|
||||
COUNTER_UPDATE(&second.column_plan_time, 300);
|
||||
first.update_profile(&profile);
|
||||
second.update_profile(&profile);
|
||||
|
||||
EXPECT_EQ(profile.get_counter("PageReadAhead")->type(), TUnit::NONE);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadInputPages")->value(), 2);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadInputBytes")->value(), 80);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadInputBytes")->type(), TUnit::BYTES);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadPlanTime")->value(), 100);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadPlanTime")->type(), TUnit::TIME_NS);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadColumnPlanTime")->value(), 1300);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadColumnInitTime")->value(), 200);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadPageAdvanceTime")->value(), 80);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadWindowDiscardTime")->value(), 100);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadCurrentBatchPlanTime")->value(), 600);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadWindowExtendTime")->value(), 50);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadCandidateBytes")->value(), 60);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadPageCacheHitBytes")->value(), 12);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadQueryBudgetRejectedBatches")->value(), 1);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadBEBudgetRejectedBatches")->value(), 2);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadSubmitFailedRanges")->value(), 3);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadSuccessfulSourceBytes")->value(), 40);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadFallbackPages")->value(), 1);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadFallbackBytes")->value(), 8);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadFallbackTime")->value(), 200);
|
||||
EXPECT_EQ(first.input_bytes.value(), 0);
|
||||
EXPECT_EQ(second.input_bytes.value(), 0);
|
||||
EXPECT_EQ(first.column_plan_time.value(), 0);
|
||||
EXPECT_EQ(first.column_init_time.value(), 0);
|
||||
EXPECT_EQ(first.page_advance_time.value(), 0);
|
||||
EXPECT_EQ(first.window_discard_time.value(), 0);
|
||||
EXPECT_EQ(first.current_batch_plan_time.value(), 0);
|
||||
EXPECT_EQ(first.window_extend_time.value(), 0);
|
||||
EXPECT_EQ(second.column_plan_time.value(), 0);
|
||||
|
||||
TRuntimeProfileTree tree;
|
||||
profile.to_thrift(&tree);
|
||||
const auto& children = tree.nodes.at(0).child_counters_map;
|
||||
EXPECT_EQ(children.at("ReadAheadPlanning").count("ReadAheadColumnPlanTime"), 1);
|
||||
EXPECT_EQ(children.at("ReadAheadColumnPlanTime").count("ReadAheadColumnInitTime"), 1);
|
||||
EXPECT_EQ(children.at("ReadAheadColumnPlanTime").count("ReadAheadWindowDiscardTime"), 1);
|
||||
EXPECT_EQ(children.at("ReadAheadColumnPlanTime").count("ReadAheadCurrentBatchPlanTime"), 1);
|
||||
EXPECT_EQ(children.at("ReadAheadColumnPlanTime").count("ReadAheadPageAdvanceTime"), 1);
|
||||
EXPECT_EQ(children.at("ReadAheadColumnPlanTime").count("ReadAheadWindowExtendTime"), 1);
|
||||
}
|
||||
|
||||
TEST(ReadAheadMetricsTest, PreservesConcurrentUpdatesDuringReporting) {
|
||||
auto statistics = std::make_shared<ReadAheadStatistics>();
|
||||
RuntimeProfile profile("SegmentIterator");
|
||||
constexpr int64_t updates = 50000;
|
||||
std::thread worker([statistics]() {
|
||||
for (int64_t index = 0; index < updates; ++index) {
|
||||
COUNTER_UPDATE(&statistics->remote_bytes, 1);
|
||||
}
|
||||
});
|
||||
for (int report = 0; report < 100; ++report) {
|
||||
statistics->update_profile(&profile);
|
||||
}
|
||||
worker.join();
|
||||
statistics->update_profile(&profile);
|
||||
EXPECT_EQ(profile.get_counter("ReadAheadRemoteBytes")->value(), updates);
|
||||
EXPECT_EQ(statistics->remote_bytes.value(), 0);
|
||||
}
|
||||
|
||||
TEST(ReadAheadMetricsTest, CountersOutliveReportedProfile) {
|
||||
auto statistics = std::make_shared<ReadAheadStatistics>();
|
||||
{
|
||||
RuntimeProfile profile("ClosedScanner");
|
||||
COUNTER_UPDATE(&statistics->completed_ranges, 1);
|
||||
statistics->update_profile(&profile);
|
||||
}
|
||||
std::thread worker([statistics]() { COUNTER_UPDATE(&statistics->completed_ranges, 1); });
|
||||
worker.join();
|
||||
RuntimeProfile next_report("RemainingSnapshot");
|
||||
statistics->update_profile(&next_report);
|
||||
EXPECT_EQ(next_report.get_counter("ReadAheadCompletedRanges")->value(), 1);
|
||||
}
|
||||
|
||||
} // namespace doris::io
|
||||
Reference in new issue
Block a user