This is an automated email from the ASF dual-hosted git repository.
bobhan1 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 f45888fe98b [improvement](topn) Add option to skip file cache writes
in topn lazy materialization (#65021)
f45888fe98b is described below
commit f45888fe98b45ae171f77021d408fa8e1842a285
Author: bobhan1 <[email protected]>
AuthorDate: Tue Jul 7 09:54:49 2026 +0800
[improvement](topn) Add option to skip file cache writes in topn lazy
materialization (#65021)
## Summary
This PR adds an opt-in session variable for TopN lazy materialization
phase-2 file-cache miss handling.
- add `enable_topn_lazy_mat_phase2_no_write_file_cache`
- keep the option disabled by default, so existing behavior is unchanged
- when the option is enabled, phase-2 lazy materialization uses
already-downloaded file-cache blocks for full local hits and reads
remote data directly on cache miss without writing the file cache
- expose aggregate and per-BE phase-2 rows, segments, and file-cache
counters in the `MaterializeNode` profile
- accumulate per-BE stats across multiple phase-2 fetch calls
- cover both row-store and column-store fetch paths
## Validation
- `./build.sh --be --fe --cloud -j100`
- before packaging, verified no `output/fe/conf/fe_custom.conf`, no
`output/be/conf/be_custom.conf`, empty `output/fe/doris-meta`, and empty
`output/be/storage`
- for docker regression image, verified FE/BE debug-point config and BE
Java support config:
- `output/fe/conf/fe.conf`: `enable_debug_points=true`
- `output/be/conf/be.conf`: `enable_debug_points=true`,
`enable_java_support=false`
- `docker build -f docker/runtime/doris-compose/Dockerfile -t
bh-cluster-2 .`
- `env -u HTTP_PROXY -u HTTPS_PROXY -u http_proxy -u https_proxy -u
ALL_PROXY -u all_proxy ./run-regression-test.sh --run -d
cloud_p0/cache/topn_lazy_file_cache -s
test_topn_lazy_mat_phase2_no_write_file_cache -g docker -runMode=cloud
-dockerSuiteParallel 1`
- `All suites success`
- `Test 1 suites, failed 0 suites, fatal 0 scripts, skipped 0 scripts`
- `./run-be-ut.sh --run
--filter=MaterializationSharedStateTest.*:BlockFileCacheTest.get_downloaded_blocks_if_fully_covered_is_read_only:BlockFileCacheTest.cached_remote_file_reader_remote_only_on_miss:BlockFileCacheTest.fd_cache_remove:BlockFileCacheTest.fd_cache_evict
-j100`
- `PASSED 9 tests`
---
be/src/exec/operator/materialization_opertor.cpp | 209 ++++++++++++++-
be/src/exec/operator/materialization_opertor.h | 27 +-
be/src/exec/rowid_fetcher.cpp | 74 +++++-
be/src/exec/rowid_fetcher.h | 19 +-
be/src/io/cache/block_file_cache.cpp | 91 +++++++
be/src/io/cache/block_file_cache.h | 8 +
be/src/io/cache/cached_remote_file_reader.cpp | 93 +++++++
be/src/io/cache/cached_remote_file_reader.h | 17 ++
be/src/io/io_common.h | 6 +
.../bloom_filter/bloom_filter_index_reader.cpp | 18 +-
.../index/bloom_filter/bloom_filter_index_reader.h | 19 +-
be/src/storage/index/indexed_column_reader.cpp | 23 +-
be/src/storage/index/indexed_column_reader.h | 18 +-
be/src/storage/index/ordinal_page_index.cpp | 15 +-
be/src/storage/index/ordinal_page_index.h | 10 +-
be/src/storage/index/zone_map/zone_map_index.cpp | 14 +-
be/src/storage/index/zone_map/zone_map_index.h | 6 +-
be/src/storage/segment/column_reader.cpp | 11 +-
be/src/storage/segment/segment.cpp | 7 +-
be/src/storage/tablet/base_tablet.cpp | 18 +-
be/src/storage/tablet/base_tablet.h | 2 +-
.../operator/materialization_shared_state_test.cpp | 86 +++---
be/test/io/cache/block_file_cache_test.cpp | 134 ++++++++++
.../java/org/apache/doris/qe/SessionVariable.java | 12 +
gensrc/proto/internal_service.proto | 14 +
gensrc/thrift/PaloInternalService.thrift | 1 +
...topn_lazy_mat_phase2_no_write_file_cache.groovy | 294 +++++++++++++++++++++
27 files changed, 1136 insertions(+), 110 deletions(-)
diff --git a/be/src/exec/operator/materialization_opertor.cpp
b/be/src/exec/operator/materialization_opertor.cpp
index 7ceeae0f261..1a475af0bcc 100644
--- a/be/src/exec/operator/materialization_opertor.cpp
+++ b/be/src/exec/operator/materialization_opertor.cpp
@@ -21,8 +21,12 @@
#include <fmt/format.h>
#include <gen_cpp/internal_service.pb.h>
+#include <set>
+#include <sstream>
#include <utility>
+#include "cloud/config.h"
+#include "common/config.h"
#include "common/status.h"
#include "core/block/block.h"
#include "core/column/column.h"
@@ -31,9 +35,107 @@
#include "exec/scan/file_scanner.h"
#include "util/brpc_client_cache.h"
#include "util/brpc_closure.h"
+#include "util/pretty_printer.h"
namespace doris {
+namespace {
+
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND =
+ "TopNLazyMaterializationSecondPhasePerBackend";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_ROWS_READ =
+ "TopNLazyMaterializationSecondPhasePerBackendRowsRead";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_SEGMENTS_READ =
+ "TopNLazyMaterializationSecondPhasePerBackendSegmentsRead";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_LOCAL_IO_COUNT =
+ "TopNLazyMaterializationSecondPhasePerBackendLocalIOCount";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_LOCAL_IO_BYTES =
+ "TopNLazyMaterializationSecondPhasePerBackendLocalIOBytes";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_REMOTE_IO_COUNT =
+ "TopNLazyMaterializationSecondPhasePerBackendRemoteIOCount";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_REMOTE_IO_BYTES =
+ "TopNLazyMaterializationSecondPhasePerBackendRemoteIOBytes";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_SKIP_CACHE_IO_COUNT =
+ "TopNLazyMaterializationSecondPhasePerBackendSkipCacheIOCount";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_WRITE_CACHE_BYTES =
+ "TopNLazyMaterializationSecondPhasePerBackendWriteCacheBytes";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_LOCAL_IO_TIME =
+ "TopNLazyMaterializationSecondPhasePerBackendLocalIOTime";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_REMOTE_IO_TIME =
+ "TopNLazyMaterializationSecondPhasePerBackendRemoteIOTime";
+constexpr const char* TOPN_LAZY_MAT_PHASE2_PER_BACKEND_WRITE_CACHE_IO_TIME =
+ "TopNLazyMaterializationSecondPhasePerBackendWriteCacheIOTime";
+
+void update_counter(RuntimeProfile* profile, const std::string& name,
TUnit::type unit,
+ int64_t value) {
+ COUNTER_UPDATE(ADD_COUNTER_WITH_LEVEL(profile, name, unit, 2), value);
+}
+
+void update_topn_lazy_materialization_profile(RuntimeProfile* profile,
+ const
PTopNLazyMaterializationFileCacheStats& stats) {
+ if (profile == nullptr) {
+ return;
+ }
+ update_counter(profile,
RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOCount,
+ TUnit::UNIT, stats.local_io_count());
+ update_counter(profile,
RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOBytes,
+ TUnit::BYTES, stats.local_io_bytes());
+ update_counter(profile,
RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOCount,
+ TUnit::UNIT, stats.remote_io_count());
+ update_counter(profile,
RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOBytes,
+ TUnit::BYTES, stats.remote_io_bytes());
+ update_counter(profile,
RowIdStorageReader::TopNLazyMaterializationSecondPhaseSkipCacheIOCount,
+ TUnit::UNIT, stats.skip_cache_io_count());
+ update_counter(profile,
RowIdStorageReader::TopNLazyMaterializationSecondPhaseWriteCacheBytes,
+ TUnit::BYTES, stats.write_cache_bytes());
+ update_counter(profile,
RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOTime,
+ TUnit::TIME_NS, stats.local_io_time());
+ update_counter(profile,
RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOTime,
+ TUnit::TIME_NS, stats.remote_io_time());
+ update_counter(profile,
RowIdStorageReader::TopNLazyMaterializationSecondPhaseWriteCacheIOTime,
+ TUnit::TIME_NS, stats.write_cache_io_time());
+}
+
+int64_t count_request_rows(const PMultiGetRequestV2& request) {
+ int64_t rows = 0;
+ for (const auto& request_block_desc : request.request_block_descs()) {
+ rows += request_block_desc.row_id_size();
+ }
+ return rows;
+}
+
+int64_t count_request_segments(const PMultiGetRequestV2& request) {
+ std::set<uint32_t> file_ids;
+ for (const auto& request_block_desc : request.request_block_descs()) {
+ DCHECK_EQ(request_block_desc.file_id_size(),
request_block_desc.row_id_size());
+ for (const auto file_id : request_block_desc.file_id()) {
+ file_ids.insert(file_id);
+ }
+ }
+ return file_ids.size();
+}
+
+template <typename AppendValue>
+std::string format_array(size_t size, AppendValue append_value) {
+ std::stringstream values;
+ values << "[";
+ for (size_t i = 0; i < size; ++i) {
+ append_value(values, i);
+ values << ", ";
+ }
+ values << "]";
+ return values.str();
+}
+
+template <typename GetValue>
+std::string format_counter_array(size_t size, TUnit::type unit, GetValue
get_value) {
+ return format_array(size, [&](std::stringstream& values, size_t i) {
+ values << PrettyPrinter::print(static_cast<int64_t>(get_value(i)),
unit);
+ });
+}
+
+} // namespace
+
void MaterializationSharedState::get_block(Block* block) {
for (int i = 0, j = 0, rowid_to_block_loc = rowid_locs[j]; i <
origin_block.columns(); i++) {
if (i != rowid_to_block_loc) {
@@ -53,6 +155,98 @@ void MaterializationSharedState::get_block(Block* block) {
origin_block.clear();
}
+void MaterializationSharedState::_update_topn_lazy_materialization_profile(
+ RuntimeProfile* profile) {
+ DORIS_CHECK(profile != nullptr);
+ for (const auto& [backend_id, rpc_struct] : rpc_struct_map) {
+ const int64_t rows_read = count_request_rows(rpc_struct.request);
+ const int64_t segments_read =
count_request_segments(rpc_struct.request);
+ update_counter(profile,
RowIdStorageReader::TopNLazyMaterializationSecondPhaseRowsRead,
+ TUnit::UNIT, rows_read);
+ update_counter(profile,
RowIdStorageReader::TopNLazyMaterializationSecondPhaseSegmentsRead,
+ TUnit::UNIT, segments_read);
+
+ auto& stats = _topn_lazy_materialization_backend_stats[backend_id];
+ if (stats.backend.empty()) {
+ stats.backend = rpc_struct.backend_address.empty() ?
fmt::format("id={}", backend_id)
+ :
rpc_struct.backend_address;
+ }
+ stats.rows_read += rows_read;
+ stats.segments_read += segments_read;
+ if
(!rpc_struct.response.has_topn_lazy_materialization_file_cache_stats()) {
+ continue;
+ }
+
+ const auto& file_cache_stats =
+
rpc_struct.response.topn_lazy_materialization_file_cache_stats();
+ update_topn_lazy_materialization_profile(profile, file_cache_stats);
+ stats.local_io_count += file_cache_stats.local_io_count();
+ stats.local_io_bytes += file_cache_stats.local_io_bytes();
+ stats.remote_io_count += file_cache_stats.remote_io_count();
+ stats.remote_io_bytes += file_cache_stats.remote_io_bytes();
+ stats.skip_cache_io_count += file_cache_stats.skip_cache_io_count();
+ stats.write_cache_bytes += file_cache_stats.write_cache_bytes();
+ stats.local_io_time += file_cache_stats.local_io_time();
+ stats.remote_io_time += file_cache_stats.remote_io_time();
+ stats.write_cache_io_time += file_cache_stats.write_cache_io_time();
+ }
+
+ std::vector<const TopNLazyMaterializationBackendStats*> stats;
+ stats.reserve(_topn_lazy_materialization_backend_stats.size());
+ for (const auto& [_, backend_stats] :
_topn_lazy_materialization_backend_stats) {
+ stats.push_back(&backend_stats);
+ }
+
+ const size_t size = stats.size();
+ profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND,
+ format_array(size, [&](std::stringstream& values,
size_t i) {
+ values << stats[i]->backend;
+ }));
+ profile->add_info_string(
+ TOPN_LAZY_MAT_PHASE2_PER_BACKEND_ROWS_READ,
+ format_counter_array(size, TUnit::UNIT, [&](size_t i) { return
stats[i]->rows_read; }));
+ profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_SEGMENTS_READ,
+ format_counter_array(size, TUnit::UNIT,
[&](size_t i) {
+ return stats[i]->segments_read;
+ }));
+ profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_LOCAL_IO_COUNT,
+ format_counter_array(size, TUnit::UNIT,
[&](size_t i) {
+ return stats[i]->local_io_count;
+ }));
+ profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_LOCAL_IO_BYTES,
+ format_counter_array(size, TUnit::BYTES,
[&](size_t i) {
+ return stats[i]->local_io_bytes;
+ }));
+ profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_REMOTE_IO_COUNT,
+ format_counter_array(size, TUnit::UNIT,
[&](size_t i) {
+ return stats[i]->remote_io_count;
+ }));
+ profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_REMOTE_IO_BYTES,
+ format_counter_array(size, TUnit::BYTES,
[&](size_t i) {
+ return stats[i]->remote_io_bytes;
+ }));
+
profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_SKIP_CACHE_IO_COUNT,
+ format_counter_array(size, TUnit::UNIT,
[&](size_t i) {
+ return stats[i]->skip_cache_io_count;
+ }));
+
profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_WRITE_CACHE_BYTES,
+ format_counter_array(size, TUnit::BYTES,
[&](size_t i) {
+ return stats[i]->write_cache_bytes;
+ }));
+ profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_LOCAL_IO_TIME,
+ format_counter_array(size, TUnit::TIME_NS,
[&](size_t i) {
+ return stats[i]->local_io_time;
+ }));
+ profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_REMOTE_IO_TIME,
+ format_counter_array(size, TUnit::TIME_NS,
[&](size_t i) {
+ return stats[i]->remote_io_time;
+ }));
+
profile->add_info_string(TOPN_LAZY_MAT_PHASE2_PER_BACKEND_WRITE_CACHE_IO_TIME,
+ format_counter_array(size, TUnit::TIME_NS,
[&](size_t i) {
+ return stats[i]->write_cache_io_time;
+ }));
+}
+
// Merges RPC responses from multiple BEs into `response_blocks` in the
original row order.
//
// After parallel multiget_data_v2 RPCs complete, each BE's response contains
a partial block
@@ -63,7 +257,9 @@ void MaterializationSharedState::get_block(Block* block) {
// rpc_struct_map[backend_id].response (per-BE partial blocks, unordered
across BEs)
// + block_order_results[i][j] (maps each output row → its source
backend_id)
// → response_blocks[i] (final merged result in original
TopN row order)
-Status MaterializationSharedState::merge_multi_response() {
+Status MaterializationSharedState::merge_multi_response(RuntimeProfile*
profile) {
+ _update_topn_lazy_materialization_profile(profile);
+
// Outer loop: iterate over each relation (i.e., each rowid column /
table).
// A query with lazy materialization on 2 tables would have
block_order_results.size() == 2,
// each with its own set of response_blocks and RPC request_block_descs.
@@ -265,6 +461,9 @@ Status MaterializationSharedState::init_multi_requests(
// Initialize the base struct of PMultiGetRequestV2
multi_get_request.set_be_exec_version(state->be_exec_version());
multi_get_request.set_wg_id(state->get_query_ctx()->workload_group()->id());
+ multi_get_request.set_file_cache_remote_only_on_miss(
+ config::is_cloud_mode() &&
+
state->query_options().enable_topn_lazy_mat_phase2_no_write_file_cache);
auto* query_id = multi_get_request.mutable_query_id();
query_id->set_hi(state->query_id().hi);
query_id->set_lo(state->query_id().lo);
@@ -315,7 +514,10 @@ Status MaterializationSharedState::init_multi_requests(
FetchRpcStruct {.stub = std::move(client),
.cntl =
std::make_unique<brpc::Controller>(),
.request = multi_get_request,
- .response =
PMultiGetResponseV2()});
+ .response =
PMultiGetResponseV2(),
+ .backend_address = fmt::format(
+ "id={} {}:{}",
node_info.id, node_info.host,
+
node_info.async_internal_port)});
}
return Status::OK();
@@ -427,7 +629,8 @@ Status MaterializationOperator::push(RuntimeState* state,
Block* in_block, bool
if (local_state._materialization_state.need_merge_block) {
SCOPED_TIMER(local_state._merge_response_timer);
-
RETURN_IF_ERROR(local_state._materialization_state.merge_multi_response());
+
RETURN_IF_ERROR(local_state._materialization_state.merge_multi_response(
+ local_state.operator_profile()));
local_state._max_rows_per_backend_counter->set(
(int64_t)local_state._materialization_state._max_rows_per_backend);
}
diff --git a/be/src/exec/operator/materialization_opertor.h
b/be/src/exec/operator/materialization_opertor.h
index 5cf2bf1ee9e..889d94a11ac 100644
--- a/be/src/exec/operator/materialization_opertor.h
+++ b/be/src/exec/operator/materialization_opertor.h
@@ -19,6 +19,10 @@
#include <stdint.h>
+#include <map>
+#include <string>
+#include <unordered_map>
+
#include "common/status.h"
#include "exec/operator/operator.h"
@@ -32,6 +36,7 @@ struct FetchRpcStruct {
std::unique_ptr<brpc::Controller> cntl;
PMultiGetRequestV2 request;
PMultiGetResponseV2 response;
+ std::string backend_address;
};
struct MaterializationSharedState {
@@ -41,11 +46,27 @@ public:
Status init_multi_requests(const TMaterializationNode& tnode,
RuntimeState* state);
Status create_muiltget_result(const Columns& columns, bool eos);
- Status merge_multi_response();
+ Status merge_multi_response(RuntimeProfile* profile);
void get_block(Block* block);
private:
void _update_profile_info(int64_t backend_id, RuntimeProfile*
response_profile);
+ void _update_topn_lazy_materialization_profile(RuntimeProfile* profile);
+
+ struct TopNLazyMaterializationBackendStats {
+ std::string backend;
+ int64_t rows_read = 0;
+ int64_t segments_read = 0;
+ int64_t local_io_count = 0;
+ int64_t local_io_bytes = 0;
+ int64_t remote_io_count = 0;
+ int64_t remote_io_bytes = 0;
+ int64_t skip_cache_io_count = 0;
+ int64_t write_cache_bytes = 0;
+ int64_t local_io_time = 0;
+ int64_t remote_io_time = 0;
+ int64_t write_cache_io_time = 0;
+ };
public:
bool rpc_struct_inited = false;
@@ -68,6 +89,10 @@ public:
uint32_t _max_rows_per_backend = 0;
// Store the number of rows processed by each backend
std::unordered_map<int64_t, uint32_t> _backend_rows_count; // backend_id
=> rows_count
+
+private:
+ // backend id => accumulated TopN phase-2 profile stats.
+ std::map<int64_t, TopNLazyMaterializationBackendStats>
_topn_lazy_materialization_backend_stats;
};
class MaterializationLocalState final : public
PipelineXLocalState<FakeSharedState> {
diff --git a/be/src/exec/rowid_fetcher.cpp b/be/src/exec/rowid_fetcher.cpp
index c2317a103e5..1c2a6f1729a 100644
--- a/be/src/exec/rowid_fetcher.cpp
+++ b/be/src/exec/rowid_fetcher.cpp
@@ -53,6 +53,7 @@
#include "exec/scan/file_scanner.h"
#include "format/orc/vorc_reader.h"
#include "format/parquet/vparquet_reader.h"
+#include "io/io_common.h"
#include "runtime/descriptors.h"
#include "runtime/exec_env.h" // ExecEnv
#include "runtime/fragment_mgr.h" // FragmentMgr
@@ -73,6 +74,23 @@
namespace doris {
+namespace {
+
+void set_topn_lazy_materialization_file_cache_stats(
+ const io::FileCacheStatistics& stats,
PTopNLazyMaterializationFileCacheStats* pstats) {
+ pstats->set_local_io_count(stats.num_local_io_total);
+ pstats->set_local_io_bytes(stats.bytes_read_from_local);
+ pstats->set_remote_io_count(stats.num_remote_io_total);
+ pstats->set_remote_io_bytes(stats.bytes_read_from_remote);
+ pstats->set_skip_cache_io_count(stats.num_skip_cache_io_total);
+ pstats->set_write_cache_bytes(stats.bytes_write_into_cache);
+ pstats->set_local_io_time(stats.local_io_timer);
+ pstats->set_remote_io_time(stats.remote_io_timer);
+ pstats->set_write_cache_io_time(stats.write_cache_io_timer);
+}
+
+} // namespace
+
Status RowIDFetcher::init() {
DorisNodesInfo nodes_info;
nodes_info.setNodes(_fetch_option.t_fetch_opt.nodes_info);
@@ -548,6 +566,11 @@ Status RowIdStorageReader::read_by_rowids(const
PMultiGetRequestV2& request,
int64_t external_get_block_avg_ms = 0;
size_t external_scan_range_cnt = 0;
+ const auto file_cache_miss_policy =
+ request.file_cache_remote_only_on_miss()
+ ? io::FileCacheMissPolicy::REMOTE_ONLY_ON_MISS
+ : io::FileCacheMissPolicy::READ_THROUGH_AND_WRITE_BACK;
+
// Add counters for different file mapping types
std::unordered_map<FileMappingType, int64_t> file_type_counts;
@@ -589,7 +612,7 @@ Status RowIdStorageReader::read_by_rowids(const
PMultiGetRequestV2& request,
RETURN_IF_ERROR(read_batch_doris_format_row(
request_block_desc, id_file_map, slots,
tquery_id, result_blocks[i],
stats, &acquire_tablet_ms, &acquire_rowsets_ms,
- &acquire_segments_ms, &lookup_row_data_ms));
+ &acquire_segments_ms, &lookup_row_data_ms,
file_cache_miss_policy));
} else {
RETURN_IF_ERROR(read_batch_external_row(
request.wg_id(), request_block_desc,
id_file_map, slots,
@@ -637,6 +660,9 @@ Status RowIdStorageReader::read_by_rowids(const
PMultiGetRequestV2& request,
acquire_rowsets_ms, acquire_segments_ms,
lookup_row_data_ms,
file_type_stats, external_init_reader_avg_ms,
external_get_block_avg_ms,
external_scan_range_cnt);
+ set_topn_lazy_materialization_file_cache_stats(
+ stats.file_cache_stats,
+
response->mutable_topn_lazy_materialization_file_cache_stats());
}
return Status::OK();
@@ -646,7 +672,8 @@ Status RowIdStorageReader::read_batch_doris_format_row(
const PRequestBlockDesc& request_block_desc,
std::shared_ptr<IdFileMap> id_file_map,
std::vector<SlotDescriptor>& slots, const TUniqueId& query_id, Block&
result_block,
OlapReaderStatistics& stats, int64_t* acquire_tablet_ms, int64_t*
acquire_rowsets_ms,
- int64_t* acquire_segments_ms, int64_t* lookup_row_data_ms) {
+ int64_t* acquire_segments_ms, int64_t* lookup_row_data_ms,
+ io::FileCacheMissPolicy file_cache_miss_policy) {
if (result_block.is_empty_column()) [[likely]] {
result_block = Block(slots, request_block_desc.row_id_size());
}
@@ -724,11 +751,11 @@ Status RowIdStorageReader::read_batch_doris_format_row(
}
scan_blocks[batch_idx] = Block(slots, row_ids.size());
- RETURN_IF_ERROR(read_doris_format_row(id_file_map,
scan_batch.file_mapping, row_ids, slots,
- full_read_schema,
row_store_read_struct, stats,
- acquire_tablet_ms,
acquire_rowsets_ms,
- acquire_segments_ms,
lookup_row_data_ms, seg_map,
- iterator_map,
scan_blocks[batch_idx]));
+ RETURN_IF_ERROR(read_doris_format_row(
+ id_file_map, scan_batch.file_mapping, row_ids, slots,
full_read_schema,
+ row_store_read_struct, stats, acquire_tablet_ms,
acquire_rowsets_ms,
+ acquire_segments_ms, lookup_row_data_ms, seg_map, iterator_map,
+ file_cache_miss_policy, scan_blocks[batch_idx]));
}
scatter_scan_blocks_to_result_block(row_id_block_idx, scan_blocks,
result_block);
@@ -740,6 +767,28 @@ const std::string
RowIdStorageReader::ScannersRunningTimeProfile = "ScannersRunn
const std::string RowIdStorageReader::InitReaderAvgTimeProfile =
"InitReaderAvgTime";
const std::string RowIdStorageReader::GetBlockAvgTimeProfile =
"GetBlockAvgTime";
const std::string RowIdStorageReader::FileReadLinesProfile = "FileReadLines";
+const std::string
RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOCount =
+ "TopNLazyMaterializationSecondPhaseLocalIOCount";
+const std::string
RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOBytes =
+ "TopNLazyMaterializationSecondPhaseLocalIOBytes";
+const std::string
RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOCount =
+ "TopNLazyMaterializationSecondPhaseRemoteIOCount";
+const std::string
RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOBytes =
+ "TopNLazyMaterializationSecondPhaseRemoteIOBytes";
+const std::string
RowIdStorageReader::TopNLazyMaterializationSecondPhaseSkipCacheIOCount =
+ "TopNLazyMaterializationSecondPhaseSkipCacheIOCount";
+const std::string
RowIdStorageReader::TopNLazyMaterializationSecondPhaseWriteCacheBytes =
+ "TopNLazyMaterializationSecondPhaseWriteCacheBytes";
+const std::string
RowIdStorageReader::TopNLazyMaterializationSecondPhaseLocalIOTime =
+ "TopNLazyMaterializationSecondPhaseLocalIOTime";
+const std::string
RowIdStorageReader::TopNLazyMaterializationSecondPhaseRemoteIOTime =
+ "TopNLazyMaterializationSecondPhaseRemoteIOTime";
+const std::string
RowIdStorageReader::TopNLazyMaterializationSecondPhaseWriteCacheIOTime =
+ "TopNLazyMaterializationSecondPhaseWriteCacheIOTime";
+const std::string
RowIdStorageReader::TopNLazyMaterializationSecondPhaseRowsRead =
+ "TopNLazyMaterializationSecondPhaseRowsRead";
+const std::string
RowIdStorageReader::TopNLazyMaterializationSecondPhaseSegmentsRead =
+ "TopNLazyMaterializationSecondPhaseSegmentsRead";
Status RowIdStorageReader::read_external_row_from_file_mapping(
size_t idx, const std::multimap<segment_v2::rowid_t, size_t>& row_ids,
@@ -1034,7 +1083,7 @@ Status RowIdStorageReader::read_doris_format_row(
int64_t* acquire_tablet_ms, int64_t* acquire_rowsets_ms, int64_t*
acquire_segments_ms,
int64_t* lookup_row_data_ms, std::unordered_map<SegKey, SegItem,
HashOfSegKey>& seg_map,
std::unordered_map<IteratorKey, IteratorItem, HashOfIteratorKey>&
iterator_map,
- Block& result_block) {
+ io::FileCacheMissPolicy file_cache_miss_policy, Block& result_block) {
auto [tablet_id, rowset_id, segment_id] =
file_mapping->get_doris_format_info();
SegKey seg_key {.tablet_id = tablet_id, .rowset_id = rowset_id,
.segment_id = segment_id};
@@ -1104,13 +1153,18 @@ Status RowIdStorageReader::read_doris_format_row(
}
auto result_columns_guard = result_block.mutate_columns_scoped();
MutableColumns& result_columns =
result_columns_guard.mutable_columns();
+ io::IOContext io_ctx;
+ io_ctx.reader_type = ReaderType::READER_QUERY;
+ io_ctx.file_cache_stats = &stats.file_cache_stats;
+ io_ctx.file_cache_miss_policy = file_cache_miss_policy;
for (auto row_id : row_ids) {
RowLocation loc(rowset_id, segment->id(),
cast_set<uint32_t>(row_id));
row_store_read_struct.row_store_buffer.clear();
RETURN_IF_ERROR(scope_timer_run(
[&]() {
return tablet->lookup_row_data({}, loc, rowset, stats,
-
row_store_read_struct.row_store_buffer);
+
row_store_read_struct.row_store_buffer,
+ false, &io_ctx);
},
lookup_row_data_ms));
@@ -1133,6 +1187,8 @@ Status RowIdStorageReader::read_doris_format_row(
iterator_map[iterator_key].segment = segment;
iterator_item.storage_read_options.stats = &stats;
iterator_item.storage_read_options.io_ctx.reader_type =
ReaderType::READER_QUERY;
+
iterator_item.storage_read_options.io_ctx.file_cache_miss_policy =
+ file_cache_miss_policy;
}
set_slot_access_paths(slots[x], full_read_schema,
iterator_item.storage_read_options);
RETURN_IF_ERROR(segment->seek_and_read_by_rowid(
diff --git a/be/src/exec/rowid_fetcher.h b/be/src/exec/rowid_fetcher.h
index 76411036669..790f9cf17e7 100644
--- a/be/src/exec/rowid_fetcher.h
+++ b/be/src/exec/rowid_fetcher.h
@@ -38,6 +38,9 @@ namespace doris {
class DorisNodesInfo;
class RuntimeState;
class TupleDescriptor;
+namespace io {
+enum class FileCacheMissPolicy : uint8_t;
+}
struct FileMapping;
struct SegKey;
@@ -97,6 +100,17 @@ public:
static const std::string InitReaderAvgTimeProfile;
static const std::string GetBlockAvgTimeProfile;
static const std::string FileReadLinesProfile;
+ static const std::string TopNLazyMaterializationSecondPhaseLocalIOCount;
+ static const std::string TopNLazyMaterializationSecondPhaseLocalIOBytes;
+ static const std::string TopNLazyMaterializationSecondPhaseRemoteIOCount;
+ static const std::string TopNLazyMaterializationSecondPhaseRemoteIOBytes;
+ static const std::string
TopNLazyMaterializationSecondPhaseSkipCacheIOCount;
+ static const std::string TopNLazyMaterializationSecondPhaseWriteCacheBytes;
+ static const std::string TopNLazyMaterializationSecondPhaseLocalIOTime;
+ static const std::string TopNLazyMaterializationSecondPhaseRemoteIOTime;
+ static const std::string
TopNLazyMaterializationSecondPhaseWriteCacheIOTime;
+ static const std::string TopNLazyMaterializationSecondPhaseRowsRead;
+ static const std::string TopNLazyMaterializationSecondPhaseSegmentsRead;
static Status read_by_rowids(const PMultiGetRequest& request,
PMultiGetResponse* response);
static Status read_by_rowids(const PMultiGetRequestV2& request,
PMultiGetResponseV2* response);
@@ -112,13 +126,14 @@ private:
int64_t* acquire_tablet_ms, int64_t* acquire_rowsets_ms, int64_t*
acquire_segments_ms,
int64_t* lookup_row_data_ms, std::unordered_map<SegKey, SegItem,
HashOfSegKey>& seg_map,
std::unordered_map<IteratorKey, IteratorItem, HashOfIteratorKey>&
iterator_map,
- Block& result_block);
+ io::FileCacheMissPolicy file_cache_miss_policy, Block&
result_block);
static Status read_batch_doris_format_row(
const PRequestBlockDesc& request_block_desc,
std::shared_ptr<IdFileMap> id_file_map,
std::vector<SlotDescriptor>& slots, const TUniqueId& query_id,
Block& result_block,
OlapReaderStatistics& stats, int64_t* acquire_tablet_ms, int64_t*
acquire_rowsets_ms,
- int64_t* acquire_segments_ms, int64_t* lookup_row_data_ms);
+ int64_t* acquire_segments_ms, int64_t* lookup_row_data_ms,
+ io::FileCacheMissPolicy file_cache_miss_policy);
static Status read_batch_external_row(
const uint64_t workload_group_id, const PRequestBlockDesc&
request_block_desc,
diff --git a/be/src/io/cache/block_file_cache.cpp
b/be/src/io/cache/block_file_cache.cpp
index 4f24eab0a58..df8bfb12b70 100644
--- a/be/src/io/cache/block_file_cache.cpp
+++ b/be/src/io/cache/block_file_cache.cpp
@@ -730,6 +730,97 @@ void
BlockFileCache::add_need_update_lru_block(FileBlockSPtr block) {
}
}
+Status BlockFileCache::get_downloaded_blocks_if_fully_covered(const
UInt128Wrapper& hash,
+ size_t offset,
size_t size,
+ const
CacheContext& context,
+ FileBlocks*
blocks,
+ bool*
fully_covered) {
+ DCHECK(blocks != nullptr);
+ DCHECK(fully_covered != nullptr);
+ blocks->clear();
+ *fully_covered = false;
+ if (size == 0) {
+ *fully_covered = true;
+ return Status::OK();
+ }
+
+ FileBlock::Range range(offset, offset + size - 1);
+ std::lock_guard cache_lock(_mutex);
+ auto it = _files.find(hash);
+ if (it == _files.end()) {
+ if (_async_open_done) {
+ return Status::OK();
+ }
+ FileCacheKey key;
+ key.hash = hash;
+ key.meta.type = context.cache_type;
+ key.meta.expiration_time = context.expiration_time;
+ key.meta.tablet_id = context.tablet_id;
+ _storage->load_blocks_directly_unlocked(this, key, cache_lock);
+
+ it = _files.find(hash);
+ if (it == _files.end()) {
+ return Status::OK();
+ }
+ }
+
+ auto& file_blocks = it->second;
+ if (file_blocks.empty()) {
+ LOG(WARNING) << "file_blocks is empty for hash=" << hash.to_string()
+ << " cache type=" << context.cache_type
+ << " cache expiration time=" << context.expiration_time
+ << " cache range=" << range.left << " " << range.right
+ << " query id=" << context.query_id;
+ DCHECK(false);
+ _files.erase(hash);
+ return Status::OK();
+ }
+
+ std::vector<FileBlockCell*> covered_cells;
+ auto block_it = file_blocks.lower_bound(range.left);
+ if (block_it == file_blocks.end() ||
block_it->second.file_block->range().left > range.left) {
+ if (block_it == file_blocks.begin()) {
+ return Status::OK();
+ }
+ --block_it;
+ }
+
+ size_t current_pos = range.left;
+ while (current_pos <= range.right) {
+ if (block_it == file_blocks.end()) {
+ return Status::OK();
+ }
+
+ auto& cell = block_it->second;
+ const auto& block_range = cell.file_block->range();
+ if (block_range.right < current_pos) {
+ ++block_it;
+ continue;
+ }
+ if (block_range.left > current_pos ||
+ cell.file_block->state() != FileBlock::State::DOWNLOADED) {
+ return Status::OK();
+ }
+
+ covered_cells.push_back(&cell);
+ if (range.right <= block_range.right) {
+ *fully_covered = true;
+ break;
+ }
+ current_pos = block_range.right + 1;
+ ++block_it;
+ }
+
+ if (!*fully_covered) {
+ return Status::OK();
+ }
+ for (const auto* cell : covered_cells) {
+ use_cell(*cell, blocks, need_to_move(cell->file_block->cache_type(),
context.cache_type),
+ cache_lock);
+ }
+ return Status::OK();
+}
+
std::string BlockFileCache::clear_file_cache_async() {
return clear_file_cache_impl(false);
}
diff --git a/be/src/io/cache/block_file_cache.h
b/be/src/io/cache/block_file_cache.h
index 1a19df5eb4a..9a01d07d136 100644
--- a/be/src/io/cache/block_file_cache.h
+++ b/be/src/io/cache/block_file_cache.h
@@ -237,6 +237,14 @@ public:
FileBlocksHolder get_or_set(const UInt128Wrapper& hash, size_t offset,
size_t size,
CacheContext& context);
+ /**
+ * Return existing downloaded blocks only if they fully cover [offset,
offset + size).
+ * This lookup is read-only: it does not reserve cache space or create
EMPTY blocks.
+ */
+ Status get_downloaded_blocks_if_fully_covered(const UInt128Wrapper& hash,
size_t offset,
+ size_t size, const
CacheContext& context,
+ FileBlocks* blocks, bool*
fully_covered);
+
/**
* record blocks read directly by CachedRemoteFileReader
*/
diff --git a/be/src/io/cache/cached_remote_file_reader.cpp
b/be/src/io/cache/cached_remote_file_reader.cpp
index 907a6e7df10..8868969b233 100644
--- a/be/src/io/cache/cached_remote_file_reader.cpp
+++ b/be/src/io/cache/cached_remote_file_reader.cpp
@@ -1080,6 +1080,93 @@ Status
CachedRemoteFileReader::_read_from_indirect_cache(size_t offset, Slice re
return Status::OK();
}
+Status CachedRemoteFileReader::_read_remote_only_on_cache_miss(
+ size_t offset, Slice result, size_t bytes_req, bool is_dryrun, size_t*
bytes_read,
+ ReadStatistics& stats, SourceReadBreakdown& source_read_breakdown,
+ const IOContext* io_ctx) {
+ auto read_remote = [&]() -> Status {
+ stats.hit_cache = false;
+ stats.from_peer_cache = false;
+ stats.skip_cache = true;
+ s3_read_counter << 1;
+ if (is_dryrun) [[unlikely]] {
+ *bytes_read = bytes_req;
+ g_read_cache_indirect_bytes << 0;
+ g_read_cache_indirect_total_bytes << bytes_req;
+ return Status::OK();
+ }
+
+ size_t remote_bytes_read = bytes_req;
+ SCOPED_RAW_TIMER(&stats.remote_read_timer);
+ RETURN_IF_ERROR(_remote_file_reader->read_at(offset,
Slice(result.data, bytes_req),
+ &remote_bytes_read,
io_ctx));
+ *bytes_read = remote_bytes_read;
+ DCHECK_EQ(*bytes_read, bytes_req);
+ source_read_breakdown.remote_bytes += remote_bytes_read;
+ g_read_cache_indirect_bytes << remote_bytes_read;
+ g_read_cache_indirect_total_bytes << remote_bytes_read;
+ return Status::OK();
+ };
+
+ g_read_cache_indirect_num << 1;
+ CacheContext cache_context(io_ctx);
+ cache_context.stats = &stats;
+ cache_context.tablet_id = _tablet_id;
+ FileBlocks file_blocks;
+ bool fully_covered = false;
+ {
+ SCOPED_RAW_TIMER(&stats.get_timer);
+ RETURN_IF_ERROR(_cache->get_downloaded_blocks_if_fully_covered(
+ _cache_hash, offset, bytes_req, cache_context, &file_blocks,
&fully_covered));
+ }
+ if (!fully_covered) {
+ return read_remote();
+ }
+
+ size_t local_read_bytes = 0;
+ size_t current_offset = offset;
+ size_t end_offset = offset + bytes_req - 1;
+ for (auto& block : file_blocks) {
+ if (current_offset > end_offset) {
+ break;
+ }
+ const auto& block_range = block->range();
+ if (block_range.right < current_offset) {
+ continue;
+ }
+
+ size_t read_left = std::max(current_offset, block_range.left);
+ size_t read_right = std::min(end_offset, block_range.right);
+ size_t read_size = read_right - read_left + 1;
+ if (is_dryrun) [[unlikely]] {
+ g_skip_local_cache_io_sum_bytes << read_size;
+ } else {
+ SCOPED_RAW_TIMER(&stats.local_read_timer);
+ Status st = block->read(Slice(result.data + (read_left - offset),
read_size),
+ read_left - block_range.left);
+ if (!st.ok()) {
+ if (st.is<ErrorCode::NOT_FOUND>()) {
+ _cache->remove_if_cached_async(_cache_hash);
+ }
+ LOG_EVERY_N(WARNING, 100)
+ << "Read data failed from file cache in
remote-only-on-miss path. "
+ << "Fallback to remote. err=" << st.msg()
+ << ", block state=" << block->state();
+ return read_remote();
+ }
+ source_read_breakdown.local_bytes += read_size;
+ local_read_bytes += read_size;
+ }
+ current_offset = read_right + 1;
+ }
+
+ *bytes_read = bytes_req;
+ stats.hit_cache = true;
+ g_read_cache_indirect_bytes << local_read_bytes;
+ g_read_cache_indirect_total_bytes << bytes_req;
+ return Status::OK();
+}
+
Status CachedRemoteFileReader::read_at_impl(size_t offset, Slice result,
size_t* bytes_read,
const IOContext* io_ctx) {
SCOPED_CONCURRENCY_COUNT(ConcurrencyStatsManager::instance().cached_remote_reader_read_at);
@@ -1142,6 +1229,12 @@ Status CachedRemoteFileReader::read_at_impl(size_t
offset, Slice result, size_t*
}
}};
+ if (io_ctx->file_cache_miss_policy ==
FileCacheMissPolicy::REMOTE_ONLY_ON_MISS) {
+ read_st = _read_remote_only_on_cache_miss(offset, result, bytes_req,
is_dryrun, bytes_read,
+ stats,
source_read_breakdown, io_ctx);
+ return read_st;
+ }
+
size_t already_read = 0;
if (_try_read_from_cached_files_directly(offset, result, bytes_req,
is_dryrun, stats,
source_read_breakdown,
already_read, bytes_read)) {
diff --git a/be/src/io/cache/cached_remote_file_reader.h
b/be/src/io/cache/cached_remote_file_reader.h
index 562d79bb9f1..5c562c82513 100644
--- a/be/src/io/cache/cached_remote_file_reader.h
+++ b/be/src/io/cache/cached_remote_file_reader.h
@@ -228,6 +228,23 @@ private:
SourceReadBreakdown&
source_read_breakdown,
const IOContext* io_ctx);
+ /// Read local cache only when downloaded blocks fully cover the request;
otherwise read remote
+ /// data directly without writing file cache.
+ /// @param[in] offset Original request offset.
+ /// @param[out] result Destination buffer for the original request.
+ /// @param[in] bytes_req Original request size.
+ /// @param[in] is_dryrun True if local cache IO should be skipped.
+ /// @param[out] bytes_read Total bytes read for the request.
+ /// @param[in,out] stats Read statistics updated for local or remote work.
+ /// @param[in,out] source_read_breakdown Source bytes used by profile
metrics.
+ /// @param[in] io_ctx IO context passed to cache lookup and remote read.
+ /// @return OK on success; otherwise an error from cache lookup, cache
read, or remote read.
+ Status _read_remote_only_on_cache_miss(size_t offset, Slice result, size_t
bytes_req,
+ bool is_dryrun, size_t* bytes_read,
+ ReadStatistics& stats,
+ SourceReadBreakdown&
source_read_breakdown,
+ const IOContext* io_ctx);
+
/// Fall back to S3: clear peer_result, allocate buffer, and read from
remote storage.
/// @param[in] empty_start Start offset of the contiguous remote-read
range.
/// @param[in] span_size Size of the remote-read range in bytes.
diff --git a/be/src/io/io_common.h b/be/src/io/io_common.h
index afb5f9e8862..e686c23a313 100644
--- a/be/src/io/io_common.h
+++ b/be/src/io/io_common.h
@@ -40,6 +40,11 @@ enum class ReaderType : uint8_t {
namespace io {
+enum class FileCacheMissPolicy : uint8_t {
+ READ_THROUGH_AND_WRITE_BACK = 0,
+ REMOTE_ONLY_ON_MISS = 1,
+};
+
struct FileReaderStats {
size_t read_calls = 0;
size_t read_bytes = 0;
@@ -188,6 +193,7 @@ struct IOContext {
int64_t predicate_filtered_rows = 0;
// if true, bypass peer read / peer-vs-S3 race and read directly from
remote storage
bool bypass_peer_read {false};
+ FileCacheMissPolicy file_cache_miss_policy =
FileCacheMissPolicy::READ_THROUGH_AND_WRITE_BACK;
};
} // namespace io
diff --git a/be/src/storage/index/bloom_filter/bloom_filter_index_reader.cpp
b/be/src/storage/index/bloom_filter/bloom_filter_index_reader.cpp
index b8c1c9b3744..2d0a9a0ab8b 100644
--- a/be/src/storage/index/bloom_filter/bloom_filter_index_reader.cpp
+++ b/be/src/storage/index/bloom_filter/bloom_filter_index_reader.cpp
@@ -32,10 +32,11 @@
namespace doris::segment_v2 {
Status BloomFilterIndexReader::load(bool use_page_cache, bool kept_in_memory,
- OlapReaderStatistics* index_load_stats) {
+ OlapReaderStatistics* index_load_stats,
+ const io::IOContext* io_ctx) {
// TODO yyq: implement a new once flag to avoid status construct.
- return _load_once.call([this, use_page_cache, kept_in_memory,
index_load_stats] {
- return _load(use_page_cache, kept_in_memory, index_load_stats);
+ return _load_once.call([this, use_page_cache, kept_in_memory,
index_load_stats, io_ctx] {
+ return _load(use_page_cache, kept_in_memory, index_load_stats, io_ctx);
});
}
@@ -45,21 +46,24 @@ int64_t BloomFilterIndexReader::get_metadata_size() const {
}
Status BloomFilterIndexReader::_load(bool use_page_cache, bool kept_in_memory,
- OlapReaderStatistics* index_load_stats) {
+ OlapReaderStatistics* index_load_stats,
+ const io::IOContext* io_ctx) {
const IndexedColumnMetaPB& bf_index_meta =
_bloom_filter_index_meta->bloom_filter();
_bloom_filter_reader = std::make_unique<IndexedColumnReader>(_file_reader,
bf_index_meta);
- RETURN_IF_ERROR(_bloom_filter_reader->load(use_page_cache, kept_in_memory,
index_load_stats));
+ RETURN_IF_ERROR(
+ _bloom_filter_reader->load(use_page_cache, kept_in_memory,
index_load_stats, io_ctx));
update_metadata_size();
return Status::OK();
}
Status
BloomFilterIndexReader::new_iterator(std::unique_ptr<BloomFilterIndexIterator>*
iterator,
- OlapReaderStatistics*
index_load_stats) {
+ OlapReaderStatistics*
index_load_stats,
+ const io::IOContext* io_ctx) {
DBUG_EXECUTE_IF("BloomFilterIndexReader::new_iterator.fail", {
return Status::InternalError("new_iterator for bloom filter index
failed");
});
- *iterator = std::make_unique<BloomFilterIndexIterator>(this,
index_load_stats);
+ *iterator = std::make_unique<BloomFilterIndexIterator>(this,
index_load_stats, io_ctx);
return Status::OK();
}
diff --git a/be/src/storage/index/bloom_filter/bloom_filter_index_reader.h
b/be/src/storage/index/bloom_filter/bloom_filter_index_reader.h
index dc0b78f16c3..43958651744 100644
--- a/be/src/storage/index/bloom_filter/bloom_filter_index_reader.h
+++ b/be/src/storage/index/bloom_filter/bloom_filter_index_reader.h
@@ -31,6 +31,9 @@
#include "util/once.h"
namespace doris {
+namespace io {
+struct IOContext;
+}
namespace segment_v2 {
@@ -46,19 +49,21 @@ public:
_bloom_filter_index_meta.reset(new
BloomFilterIndexPB(bloom_filter_index_meta));
}
- Status load(bool use_page_cache, bool kept_in_memory,
- OlapReaderStatistics* bf_index_load_stats);
+ Status load(bool use_page_cache, bool kept_in_memory,
OlapReaderStatistics* bf_index_load_stats,
+ const io::IOContext* io_ctx = nullptr);
BloomFilterAlgorithmPB algorithm() { return
_bloom_filter_index_meta->algorithm(); }
// create a new column iterator.
Status new_iterator(std::unique_ptr<BloomFilterIndexIterator>* iterator,
- OlapReaderStatistics* index_load_stats);
+ OlapReaderStatistics* index_load_stats,
+ const io::IOContext* io_ctx = nullptr);
FieldType type() const { return FieldType::OLAP_FIELD_TYPE_VARCHAR; }
private:
- Status _load(bool use_page_cache, bool kept_in_memory,
OlapReaderStatistics* index_load_stats);
+ Status _load(bool use_page_cache, bool kept_in_memory,
OlapReaderStatistics* index_load_stats,
+ const io::IOContext* io_ctx);
int64_t get_metadata_size() const override;
@@ -73,8 +78,10 @@ private:
class BloomFilterIndexIterator {
public:
- explicit BloomFilterIndexIterator(BloomFilterIndexReader* reader,
OlapReaderStatistics* stats)
- : _reader(reader),
_bloom_filter_iter(reader->_bloom_filter_reader.get(), stats) {}
+ explicit BloomFilterIndexIterator(BloomFilterIndexReader* reader,
OlapReaderStatistics* stats,
+ const io::IOContext* io_ctx)
+ : _reader(reader),
+ _bloom_filter_iter(reader->_bloom_filter_reader.get(), stats,
io_ctx) {}
// Read bloom filter at the given ordinal into `bf`.
Status read_bloom_filter(rowid_t ordinal, std::unique_ptr<BloomFilter>*
bf);
diff --git a/be/src/storage/index/indexed_column_reader.cpp
b/be/src/storage/index/indexed_column_reader.cpp
index d30c2c82dbc..29510f21452 100644
--- a/be/src/storage/index/indexed_column_reader.cpp
+++ b/be/src/storage/index/indexed_column_reader.cpp
@@ -60,7 +60,8 @@ int64_t IndexedColumnReader::get_metadata_size() const {
}
Status IndexedColumnReader::load(bool use_page_cache, bool kept_in_memory,
- OlapReaderStatistics* index_load_stats) {
+ OlapReaderStatistics* index_load_stats,
+ const io::IOContext* io_ctx) {
_use_page_cache = use_page_cache;
_kept_in_memory = kept_in_memory;
@@ -78,7 +79,7 @@ Status IndexedColumnReader::load(bool use_page_cache, bool
kept_in_memory,
} else {
RETURN_IF_ERROR(load_index_page(_meta.ordinal_index_meta().root_page(),
&_ordinal_index_page_handle,
- _ordinal_index_reader.get(),
index_load_stats));
+ _ordinal_index_reader.get(),
index_load_stats, io_ctx));
_has_index_page = true;
}
}
@@ -90,7 +91,7 @@ Status IndexedColumnReader::load(bool use_page_cache, bool
kept_in_memory,
} else {
RETURN_IF_ERROR(load_index_page(_meta.value_index_meta().root_page(),
&_value_index_page_handle,
_value_index_reader.get(),
- index_load_stats));
+ index_load_stats, io_ctx));
_has_index_page = true;
}
}
@@ -102,13 +103,14 @@ Status IndexedColumnReader::load(bool use_page_cache,
bool kept_in_memory,
Status IndexedColumnReader::load_index_page(const PagePointerPB& pp,
PageHandle* handle,
IndexPageReader* reader,
- OlapReaderStatistics*
index_load_stats) {
+ OlapReaderStatistics*
index_load_stats,
+ const io::IOContext* io_ctx) {
Slice body;
PageFooterPB footer;
BlockCompressionCodec* local_compress_codec;
RETURN_IF_ERROR(get_block_compression_codec(_meta.compression(),
&local_compress_codec));
RETURN_IF_ERROR(read_page(PagePointer(pp), handle, &body, &footer,
INDEX_PAGE,
- local_compress_codec, false, index_load_stats));
+ local_compress_codec, false, index_load_stats,
io_ctx));
RETURN_IF_ERROR(reader->parse(body, footer.index_page_footer()));
_mem_size += body.get_size();
return Status::OK();
@@ -117,11 +119,14 @@ Status IndexedColumnReader::load_index_page(const
PagePointerPB& pp, PageHandle*
Status IndexedColumnReader::read_page(const PagePointer& pp, PageHandle*
handle, Slice* body,
PageFooterPB* footer, PageTypePB type,
BlockCompressionCodec* codec, bool
pre_decode,
- OlapReaderStatistics* stats) const {
+ OlapReaderStatistics* stats,
+ const io::IOContext* io_ctx) const {
OlapReaderStatistics tmp_stats;
OlapReaderStatistics* stats_ptr = stats != nullptr ? stats : &tmp_stats;
- PageReadOptions opts(io::IOContext {.is_index_data = true,
- .file_cache_stats =
&stats_ptr->file_cache_stats});
+ io::IOContext page_io_ctx = io_ctx != nullptr ? *io_ctx : io::IOContext {};
+ page_io_ctx.is_index_data = true;
+ page_io_ctx.file_cache_stats = &stats_ptr->file_cache_stats;
+ PageReadOptions opts(page_io_ctx);
opts.use_page_cache = _use_page_cache;
opts.kept_in_memory = _kept_in_memory;
opts.pre_decode = pre_decode;
@@ -158,7 +163,7 @@ Status IndexedColumnIterator::_read_data_page(const
PagePointer& pp) {
Slice body;
PageFooterPB footer;
RETURN_IF_ERROR(_reader->read_page(pp, &handle, &body, &footer, DATA_PAGE,
_compress_codec,
- true, _stats));
+ true, _stats, _io_ctx));
// parse data page
// note that page_index is not used in IndexedColumnIterator, so we pass 0
PageDecoderOptions opts;
diff --git a/be/src/storage/index/indexed_column_reader.h
b/be/src/storage/index/indexed_column_reader.h
index 1cea4641595..b1562409582 100644
--- a/be/src/storage/index/indexed_column_reader.h
+++ b/be/src/storage/index/indexed_column_reader.h
@@ -40,6 +40,9 @@ namespace doris {
class KeyCoder;
class BlockCompressionCodec;
+namespace io {
+struct IOContext;
+}
namespace segment_v2 {
@@ -57,12 +60,14 @@ public:
~IndexedColumnReader() override;
Status load(bool use_page_cache, bool kept_in_memory,
- OlapReaderStatistics* index_load_stats = nullptr);
+ OlapReaderStatistics* index_load_stats = nullptr,
+ const io::IOContext* io_ctx = nullptr);
// read a page specified by `pp' from `file' into `handle'
Status read_page(const PagePointer& pp, PageHandle* handle, Slice* body,
PageFooterPB* footer,
PageTypePB type, BlockCompressionCodec* codec, bool
pre_decode,
- OlapReaderStatistics* stats = nullptr) const;
+ OlapReaderStatistics* stats = nullptr,
+ const io::IOContext* io_ctx = nullptr) const;
int64_t num_values() const { return _num_values; }
const EncodingInfo* encoding_info() const { return _encoding_info; }
@@ -76,7 +81,7 @@ public:
private:
Status load_index_page(const PagePointerPB& pp, PageHandle* handle,
IndexPageReader* reader,
- OlapReaderStatistics* index_load_stats);
+ OlapReaderStatistics* index_load_stats, const
io::IOContext* io_ctx);
int64_t get_metadata_size() const override;
@@ -108,11 +113,13 @@ private:
class IndexedColumnIterator {
public:
explicit IndexedColumnIterator(const IndexedColumnReader* reader,
- OlapReaderStatistics* stats = nullptr)
+ OlapReaderStatistics* stats = nullptr,
+ const io::IOContext* io_ctx = nullptr)
: _reader(reader),
_ordinal_iter(reader->_ordinal_index_reader.get()),
_value_iter(reader->_value_index_reader.get()),
- _stats(stats) {}
+ _stats(stats),
+ _io_ctx(io_ctx) {}
// Seek to the given ordinal entry. Entry 0 is the first entry.
// Return Status::Error<ENTRY_NOT_FOUND> if provided seek point is past
the end.
@@ -162,6 +169,7 @@ private:
// iterator owned compress codec, should NOT be shared by threads,
initialized before used
BlockCompressionCodec* _compress_codec = nullptr;
OlapReaderStatistics* _stats = nullptr;
+ const io::IOContext* _io_ctx = nullptr;
};
} // namespace segment_v2
diff --git a/be/src/storage/index/ordinal_page_index.cpp
b/be/src/storage/index/ordinal_page_index.cpp
index 054f30f67f2..688657f5755 100644
--- a/be/src/storage/index/ordinal_page_index.cpp
+++ b/be/src/storage/index/ordinal_page_index.cpp
@@ -71,16 +71,17 @@ Status OrdinalIndexWriter::finish(io::FileWriter*
file_writer, ColumnIndexMetaPB
}
Status OrdinalIndexReader::load(bool use_page_cache, bool kept_in_memory,
- OlapReaderStatistics* index_load_stats) {
+ OlapReaderStatistics* index_load_stats,
+ const io::IOContext* io_ctx) {
// TODO yyq: implement a new once flag to avoid status construct.
- return _load_once.call([this, use_page_cache, kept_in_memory,
index_load_stats] {
- return _load(use_page_cache, kept_in_memory, std::move(_meta_pb),
index_load_stats);
+ return _load_once.call([this, use_page_cache, kept_in_memory,
index_load_stats, io_ctx] {
+ return _load(use_page_cache, kept_in_memory, std::move(_meta_pb),
index_load_stats, io_ctx);
});
}
Status OrdinalIndexReader::_load(bool use_page_cache, bool kept_in_memory,
std::unique_ptr<OrdinalIndexPB> index_meta,
- OlapReaderStatistics* stats) {
+ OlapReaderStatistics* stats, const
io::IOContext* io_ctx) {
if (index_meta->root_page().is_root_data_page()) {
// only one data page, no index page
_num_pages = 1;
@@ -92,8 +93,10 @@ Status OrdinalIndexReader::_load(bool use_page_cache, bool
kept_in_memory,
// need to read index page
OlapReaderStatistics tmp_stats;
OlapReaderStatistics* stats_ptr = stats != nullptr ? stats : &tmp_stats;
- PageReadOptions opts(io::IOContext {.is_index_data = true,
- .file_cache_stats =
&stats_ptr->file_cache_stats});
+ io::IOContext page_io_ctx = io_ctx != nullptr ? *io_ctx : io::IOContext {};
+ page_io_ctx.is_index_data = true;
+ page_io_ctx.file_cache_stats = &stats_ptr->file_cache_stats;
+ PageReadOptions opts(page_io_ctx);
opts.use_page_cache = use_page_cache;
opts.kept_in_memory = kept_in_memory;
opts.type = INDEX_PAGE;
diff --git a/be/src/storage/index/ordinal_page_index.h
b/be/src/storage/index/ordinal_page_index.h
index f17d80c20ed..87bc4dc3177 100644
--- a/be/src/storage/index/ordinal_page_index.h
+++ b/be/src/storage/index/ordinal_page_index.h
@@ -36,7 +36,8 @@ namespace doris {
namespace io {
class FileWriter;
-}
+struct IOContext;
+} // namespace io
namespace segment_v2 {
class ColumnIndexMetaPB;
@@ -75,7 +76,8 @@ public:
virtual ~OrdinalIndexReader();
// load and parse the index page into memory
- Status load(bool use_page_cache, bool kept_in_memory,
OlapReaderStatistics* index_load_stats);
+ Status load(bool use_page_cache, bool kept_in_memory,
OlapReaderStatistics* index_load_stats,
+ const io::IOContext* io_ctx = nullptr);
// the returned iter points to the largest element which is less than
`ordinal`,
// or points to the first element if all elements are greater than
`ordinal`,
@@ -94,8 +96,8 @@ public:
private:
Status _load(bool use_page_cache, bool kept_in_memory,
- std::unique_ptr<OrdinalIndexPB> index_meta,
- OlapReaderStatistics* index_load_stats);
+ std::unique_ptr<OrdinalIndexPB> index_meta,
OlapReaderStatistics* index_load_stats,
+ const io::IOContext* io_ctx);
private:
friend class OrdinalPageIndexIterator;
diff --git a/be/src/storage/index/zone_map/zone_map_index.cpp
b/be/src/storage/index/zone_map/zone_map_index.cpp
index b12de9d03b6..23e2e5a4a0f 100644
--- a/be/src/storage/index/zone_map/zone_map_index.cpp
+++ b/be/src/storage/index/zone_map/zone_map_index.cpp
@@ -297,20 +297,22 @@ Status
TypedZoneMapIndexWriter<Type>::finish(io::FileWriter* file_writer,
}
Status ZoneMapIndexReader::load(bool use_page_cache, bool kept_in_memory,
- OlapReaderStatistics* index_load_stats) {
+ OlapReaderStatistics* index_load_stats,
+ const io::IOContext* io_ctx) {
// TODO yyq: implement a new once flag to avoid status construct.
- return _load_once.call([this, use_page_cache, kept_in_memory,
index_load_stats] {
+ return _load_once.call([this, use_page_cache, kept_in_memory,
index_load_stats, io_ctx] {
return _load(use_page_cache, kept_in_memory,
std::move(_page_zone_maps_meta),
- index_load_stats);
+ index_load_stats, io_ctx);
});
}
Status ZoneMapIndexReader::_load(bool use_page_cache, bool kept_in_memory,
std::unique_ptr<IndexedColumnMetaPB>
page_zone_maps_meta,
- OlapReaderStatistics* index_load_stats) {
+ OlapReaderStatistics* index_load_stats,
+ const io::IOContext* io_ctx) {
IndexedColumnReader reader(_file_reader, *page_zone_maps_meta);
- RETURN_IF_ERROR(reader.load(use_page_cache, kept_in_memory,
index_load_stats));
- IndexedColumnIterator iter(&reader, index_load_stats);
+ RETURN_IF_ERROR(reader.load(use_page_cache, kept_in_memory,
index_load_stats, io_ctx));
+ IndexedColumnIterator iter(&reader, index_load_stats, io_ctx);
_page_zone_maps.resize(reader.num_values());
diff --git a/be/src/storage/index/zone_map/zone_map_index.h
b/be/src/storage/index/zone_map/zone_map_index.h
index 96d9a7a26e5..80c6928f60b 100644
--- a/be/src/storage/index/zone_map/zone_map_index.h
+++ b/be/src/storage/index/zone_map/zone_map_index.h
@@ -38,6 +38,7 @@
namespace doris {
namespace io {
class FileWriter;
+struct IOContext;
} // namespace io
namespace segment_v2 {
@@ -182,7 +183,8 @@ public:
// load all page zone maps into memory
Status load(bool use_page_cache, bool kept_in_memory,
- OlapReaderStatistics* index_load_stats = nullptr);
+ OlapReaderStatistics* index_load_stats = nullptr,
+ const io::IOContext* io_ctx = nullptr);
const std::vector<ZoneMapPB>& page_zone_maps() const { return
_page_zone_maps; }
@@ -190,7 +192,7 @@ public:
private:
Status _load(bool use_page_cache, bool kept_in_memory,
std::unique_ptr<IndexedColumnMetaPB>,
- OlapReaderStatistics* index_load_stats);
+ OlapReaderStatistics* index_load_stats, const io::IOContext*
io_ctx);
int64_t get_metadata_size() const override;
diff --git a/be/src/storage/segment/column_reader.cpp
b/be/src/storage/segment/column_reader.cpp
index d0900561fa2..f15e7462e6c 100644
--- a/be/src/storage/segment/column_reader.cpp
+++ b/be/src/storage/segment/column_reader.cpp
@@ -609,7 +609,8 @@ Status ColumnReader::get_row_ranges_by_bloom_filter(const
AndBlockColumnPredicat
_load_bloom_filter_index(_use_index_page_cache,
_opts.kept_in_memory, iter_opts));
RowRanges bf_row_ranges;
std::unique_ptr<BloomFilterIndexIterator> bf_iter;
- RETURN_IF_ERROR(_bloom_filter_index->new_iterator(&bf_iter,
iter_opts.stats));
+ RETURN_IF_ERROR(
+ _bloom_filter_index->new_iterator(&bf_iter, iter_opts.stats,
&iter_opts.io_ctx));
size_t range_size = row_ranges->range_size();
// get covered page ids
std::set<uint32_t> page_ids;
@@ -641,13 +642,14 @@ Status ColumnReader::_load_ordinal_index(bool
use_page_cache, bool kept_in_memor
if (!_ordinal_index) {
return Status::InternalError("ordinal_index not inited");
}
- return _ordinal_index->load(use_page_cache, kept_in_memory,
iter_opts.stats);
+ return _ordinal_index->load(use_page_cache, kept_in_memory,
iter_opts.stats, &iter_opts.io_ctx);
}
Status ColumnReader::_load_zone_map_index(bool use_page_cache, bool
kept_in_memory,
const ColumnIteratorOptions&
iter_opts) {
if (_zone_map_index != nullptr) {
- return _zone_map_index->load(use_page_cache, kept_in_memory,
iter_opts.stats);
+ return _zone_map_index->load(use_page_cache, kept_in_memory,
iter_opts.stats,
+ &iter_opts.io_ctx);
}
return Status::OK();
}
@@ -766,7 +768,8 @@ bool ColumnReader::has_bloom_filter_index(bool ngram) const
{
Status ColumnReader::_load_bloom_filter_index(bool use_page_cache, bool
kept_in_memory,
const ColumnIteratorOptions&
iter_opts) {
if (_bloom_filter_index != nullptr) {
- return _bloom_filter_index->load(use_page_cache, kept_in_memory,
iter_opts.stats);
+ return _bloom_filter_index->load(use_page_cache, kept_in_memory,
iter_opts.stats,
+ &iter_opts.io_ctx);
}
return Status::OK();
}
diff --git a/be/src/storage/segment/segment.cpp
b/be/src/storage/segment/segment.cpp
index c7124e708ee..251452babe5 100644
--- a/be/src/storage/segment/segment.cpp
+++ b/be/src/storage/segment/segment.cpp
@@ -1145,13 +1145,14 @@ Status Segment::seek_and_read_by_rowid(const
TabletSchema& schema, SlotDescripto
DORIS_CHECK(std::adjacent_find(row_ids.begin(), row_ids.end()) ==
row_ids.end());
// ColumnIterator::seek_and_read expects monotonically increasing row_ids
without
// duplicates for correct ordinal scanning. Enforce this contract at the
entry point.
+ auto io_ctx = storage_read_options.io_ctx;
+ io_ctx.reader_type = ReaderType::READER_QUERY;
+ io_ctx.file_cache_stats = &storage_read_options.stats->file_cache_stats;
segment_v2::ColumnIteratorOptions opt {
.use_page_cache = !config::disable_storage_page_cache,
.file_reader = file_reader().get(),
.stats = storage_read_options.stats,
- .io_ctx = io::IOContext {.reader_type = ReaderType::READER_QUERY,
- .file_cache_stats =
-
&storage_read_options.stats->file_cache_stats},
+ .io_ctx = io_ctx,
};
if (!slot->column_paths().empty()) {
diff --git a/be/src/storage/tablet/base_tablet.cpp
b/be/src/storage/tablet/base_tablet.cpp
index 09f73ec43d2..3cd8747ee79 100644
--- a/be/src/storage/tablet/base_tablet.cpp
+++ b/be/src/storage/tablet/base_tablet.cpp
@@ -81,7 +81,8 @@ Status _get_segment_column_iterator(const
BetaRowsetSharedPtr& rowset, uint32_t
const TabletColumn& target_column,
SegmentCacheHandle* segment_cache_handle,
std::unique_ptr<segment_v2::ColumnIterator>* column_iterator,
- OlapReaderStatistics* stats) {
+ OlapReaderStatistics* stats,
+ const io::IOContext* input_io_ctx =
nullptr) {
RETURN_IF_ERROR(SegmentLoader::instance()->load_segments(rowset,
segment_cache_handle, true));
// find segment
auto it = std::find_if(
@@ -95,13 +96,18 @@ Status _get_segment_column_iterator(const
BetaRowsetSharedPtr& rowset, uint32_t
segment_v2::SegmentSharedPtr segment = *it;
StorageReadOptions opts;
opts.stats = stats;
+ if (input_io_ctx != nullptr) {
+ opts.io_ctx = *input_io_ctx;
+ }
RETURN_IF_ERROR(segment->new_column_iterator(target_column,
column_iterator, &opts));
+ auto io_ctx = opts.io_ctx;
+ io_ctx.reader_type = ReaderType::READER_QUERY;
+ io_ctx.file_cache_stats = &stats->file_cache_stats;
segment_v2::ColumnIteratorOptions opt {
.use_page_cache = !config::disable_storage_page_cache,
.file_reader = segment->file_reader().get(),
.stats = stats,
- .io_ctx = io::IOContext {.reader_type = ReaderType::READER_QUERY,
- .file_cache_stats =
&stats->file_cache_stats},
+ .io_ctx = io_ctx,
};
RETURN_IF_ERROR((*column_iterator)->init(opt));
return Status::OK();
@@ -428,7 +434,8 @@ std::vector<RowsetSharedPtr> BaseTablet::get_rowset_by_ids(
Status BaseTablet::lookup_row_data(const Slice& encoded_key, const
RowLocation& row_location,
RowsetSharedPtr input_rowset,
OlapReaderStatistics& stats,
- std::string& values, bool write_to_cache) {
+ std::string& values, bool write_to_cache,
+ const io::IOContext* io_ctx) {
MonotonicStopWatch watch;
size_t row_size = 1;
watch.start();
@@ -444,7 +451,8 @@ Status BaseTablet::lookup_row_data(const Slice&
encoded_key, const RowLocation&
std::unique_ptr<segment_v2::ColumnIterator> column_iterator;
const auto& column =
*DORIS_TRY(tablet_schema->column(BeConsts::ROW_STORE_COL));
RETURN_IF_ERROR(_get_segment_column_iterator(rowset,
row_location.segment_id, column,
- &segment_cache_handle,
&column_iterator, &stats));
+ &segment_cache_handle,
&column_iterator, &stats,
+ io_ctx));
// get and parse tuple row
MutableColumnPtr column_ptr = ColumnString::create();
std::vector<segment_v2::rowid_t> rowids
{static_cast<segment_v2::rowid_t>(row_location.row_id)};
diff --git a/be/src/storage/tablet/base_tablet.h
b/be/src/storage/tablet/base_tablet.h
index a835e2465ee..c26e944b01c 100644
--- a/be/src/storage/tablet/base_tablet.h
+++ b/be/src/storage/tablet/base_tablet.h
@@ -170,7 +170,7 @@ public:
// Lookup a row with TupleDescriptor and fill Block
Status lookup_row_data(const Slice& encoded_key, const RowLocation&
row_location,
RowsetSharedPtr rowset, OlapReaderStatistics&
stats, std::string& values,
- bool write_to_cache = false);
+ bool write_to_cache = false, const io::IOContext*
io_ctx = nullptr);
// Lookup the row location of `encoded_key`, the function sets
`row_location` on success.
// NOTE: the method only works in unique key model with primary key index,
you will got a
// not supported error in other data model.
diff --git a/be/test/exec/operator/materialization_shared_state_test.cpp
b/be/test/exec/operator/materialization_shared_state_test.cpp
index f03a150496e..b1950482a70 100644
--- a/be/test/exec/operator/materialization_shared_state_test.cpp
+++ b/be/test/exec/operator/materialization_shared_state_test.cpp
@@ -23,9 +23,19 @@
#include "core/field.h"
#include "exec/operator/materialization_opertor.h"
#include "exec/pipeline/dependency.h"
+#include "runtime/runtime_profile.h"
namespace doris {
+namespace {
+
+void add_request_row(PRequestBlockDesc* request_block_desc, uint32_t row_id,
uint32_t file_id) {
+ request_block_desc->add_row_id(row_id);
+ request_block_desc->add_file_id(file_id);
+}
+
+} // namespace
+
class MaterializationSharedStateTest : public testing::Test {
protected:
void SetUp() override {
@@ -107,12 +117,10 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponse) {
// 2. Setup response blocks from multiple backends
// Backend 1's response
{
- _shared_state->rpc_struct_map[_backend_id1]
- .request.mutable_request_block_descs(0)
- ->add_row_id(0);
- _shared_state->rpc_struct_map[_backend_id1]
- .request.mutable_request_block_descs(0)
- ->add_row_id(1);
+ auto* request_block_desc =
+
_shared_state->rpc_struct_map[_backend_id1].request.mutable_request_block_descs(0);
+ add_request_row(request_block_desc, 0, 1);
+ add_request_row(request_block_desc, 1, 1);
Block resp_block1;
auto resp_value_col1 = _int_type->create_column();
auto* value_col_data1 =
reinterpret_cast<ColumnInt32*>(resp_value_col1.get());
@@ -137,9 +145,9 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponse) {
// Backend 2's response
{
- _shared_state->rpc_struct_map[_backend_id2]
- .request.mutable_request_block_descs(0)
- ->add_row_id(2);
+ add_request_row(
+
_shared_state->rpc_struct_map[_backend_id2].request.mutable_request_block_descs(0),
+ 2, 2);
Block resp_block2;
auto resp_value_col2 = _int_type->create_column();
auto* value_col_data2 =
reinterpret_cast<ColumnInt32*>(resp_value_col2.get());
@@ -166,7 +174,8 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponse) {
// 4. Test merging responses
Block result_block;
- Status st = _shared_state->merge_multi_response();
+ RuntimeProfile profile("MaterializationSharedStateTest");
+ Status st = _shared_state->merge_multi_response(&profile);
_shared_state->get_block(&result_block);
EXPECT_TRUE(st.ok());
@@ -219,9 +228,9 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponseMultiBlocks) {
// 2. Setup response blocks from multiple backends for first rowid
{
- _shared_state->rpc_struct_map[_backend_id1]
- .request.mutable_request_block_descs(0)
- ->add_row_id(0);
+ add_request_row(
+
_shared_state->rpc_struct_map[_backend_id1].request.mutable_request_block_descs(0),
+ 0, 1);
Block resp_block1;
auto resp_value_col1 = _int_type->create_column();
auto* value_col_data1 =
reinterpret_cast<ColumnInt32*>(resp_value_col1.get());
@@ -244,9 +253,9 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponseMultiBlocks) {
// Backend 2's response for first rowid
{
- _shared_state->rpc_struct_map[_backend_id2]
- .request.mutable_request_block_descs(0)
- ->add_row_id(0);
+ add_request_row(
+
_shared_state->rpc_struct_map[_backend_id2].request.mutable_request_block_descs(0),
+ 0, 2);
Block resp_block2;
auto resp_value_col2 = _int_type->create_column();
auto* value_col_data2 =
reinterpret_cast<ColumnInt32*>(resp_value_col2.get());
@@ -270,9 +279,9 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponseMultiBlocks) {
_shared_state->rpc_struct_map[_backend_id1].request.add_request_block_descs();
_shared_state->rpc_struct_map[_backend_id2].request.add_request_block_descs();
{
- _shared_state->rpc_struct_map[_backend_id1]
- .request.mutable_request_block_descs(1)
- ->add_row_id(0);
+ add_request_row(
+
_shared_state->rpc_struct_map[_backend_id1].request.mutable_request_block_descs(1),
+ 0, 3);
Block resp_block1;
auto resp_value_col1 = _int_type->create_column();
auto* value_col_data1 =
reinterpret_cast<ColumnInt32*>(resp_value_col1.get());
@@ -292,9 +301,9 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponseMultiBlocks) {
}
{
- _shared_state->rpc_struct_map[_backend_id2]
- .request.mutable_request_block_descs(1)
- ->add_row_id(0);
+ add_request_row(
+
_shared_state->rpc_struct_map[_backend_id2].request.mutable_request_block_descs(1),
+ 0, 4);
Block resp_block2;
auto resp_value_col2 = _int_type->create_column();
auto* value_col_data2 =
reinterpret_cast<ColumnInt32*>(resp_value_col2.get());
@@ -320,7 +329,8 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponseMultiBlocks) {
// 4. Test merging responses
Block result_block;
- Status st = _shared_state->merge_multi_response();
+ RuntimeProfile profile("MaterializationSharedStateTest");
+ Status st = _shared_state->merge_multi_response(&profile);
EXPECT_TRUE(st.ok());
_shared_state->get_block(&result_block);
@@ -361,9 +371,9 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponseBackendNotFound) {
// --- BE_1: valid response with 1 row ---
{
- _shared_state->rpc_struct_map[_backend_id1]
- .request.mutable_request_block_descs(0)
- ->add_row_id(0);
+ add_request_row(
+
_shared_state->rpc_struct_map[_backend_id1].request.mutable_request_block_descs(0),
+ 0, 1);
Block resp_block;
auto col = _int_type->create_column();
reinterpret_cast<ColumnInt32*>(col.get())->insert(
@@ -388,9 +398,9 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponseBackendNotFound) {
// After deserialization this produces a Block with 0 columns,
// so is_empty_column() == true and it won't be inserted into block_maps.
{
- _shared_state->rpc_struct_map[_backend_id2]
- .request.mutable_request_block_descs(0)
- ->add_row_id(0);
+ add_request_row(
+
_shared_state->rpc_struct_map[_backend_id2].request.mutable_request_block_descs(0),
+ 0, 2);
PMultiGetResponseV2 response;
response.add_blocks(); // empty PMultiGetBlockV2, no mutable_block()
data
_shared_state->rpc_struct_map[_backend_id2].response =
std::move(response);
@@ -410,7 +420,8 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponseBackendNotFound) {
_shared_state->rowid_locs = {0};
// merge_multi_response() should return InternalError
- Status st = _shared_state->merge_multi_response();
+ RuntimeProfile profile("MaterializationSharedStateTest");
+ Status st = _shared_state->merge_multi_response(&profile);
ASSERT_FALSE(st.ok());
ASSERT_TRUE(st.is<ErrorCode::INTERNAL_ERROR>());
ASSERT_TRUE(st.to_string().find("not match request row id count") !=
std::string::npos)
@@ -444,9 +455,9 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponseStaleBlockMaps) {
// --- Build BE_1's response: blocks[0]=1 row (INT), blocks[1]=empty ---
{
- _shared_state->rpc_struct_map[_backend_id1]
- .request.mutable_request_block_descs(0)
- ->add_row_id(0);
+ add_request_row(
+
_shared_state->rpc_struct_map[_backend_id1].request.mutable_request_block_descs(0),
+ 0, 1);
PMultiGetResponseV2 response;
// blocks[0]: 1 row of INT for relation 0
@@ -481,9 +492,9 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponseStaleBlockMaps) {
reinterpret_cast<ColumnString*>(col.get())->insert_data("Alice", 5);
rel1_block.insert({make_nullable(std::move(col)),
make_nullable(_string_type), "name"});
- _shared_state->rpc_struct_map[_backend_id2]
- .request.mutable_request_block_descs(1)
- ->add_row_id(0);
+ add_request_row(
+
_shared_state->rpc_struct_map[_backend_id2].request.mutable_request_block_descs(1),
+ 0, 2);
auto* pb1 = response.add_blocks()->mutable_block();
size_t us = 0, cs = 0;
int64_t ct = 0;
@@ -509,7 +520,8 @@ TEST_F(MaterializationSharedStateTest,
TestMergeMultiResponseStaleBlockMaps) {
_shared_state->rowid_locs = {0, 1};
// merge should succeed — each relation only references the BE that has
data
- Status st = _shared_state->merge_multi_response();
+ RuntimeProfile profile("MaterializationSharedStateTest");
+ Status st = _shared_state->merge_multi_response(&profile);
ASSERT_TRUE(st.ok()) << "merge_multi_response failed: " << st.to_string();
// Verify results
diff --git a/be/test/io/cache/block_file_cache_test.cpp
b/be/test/io/cache/block_file_cache_test.cpp
index 18c41953764..974cbe1e502 100644
--- a/be/test/io/cache/block_file_cache_test.cpp
+++ b/be/test/io/cache/block_file_cache_test.cpp
@@ -387,6 +387,140 @@ void complete_into_memory(const io::FileBlocksHolder&
holder) {
}
}
+TEST_F(BlockFileCacheTest,
get_downloaded_blocks_if_fully_covered_is_read_only) {
+ const std::string local_cache_path =
+ (caches_dir / "remote_only_on_miss_helper_cache" / "").string();
+ if (fs::exists(local_cache_path)) {
+ fs::remove_all(local_cache_path);
+ }
+ fs::create_directories(local_cache_path);
+
+ io::BlockFileCache mgr(local_cache_path,
cached_remote_reader_cache_settings());
+ ASSERT_TRUE(mgr.initialize().ok());
+
+ ReadStatistics read_stats;
+ io::CacheContext context;
+ context.stats = &read_stats;
+ context.cache_type = io::FileCacheType::NORMAL;
+ auto key = io::BlockFileCache::hash("remote_only_on_miss_helper_key");
+ const size_t block_size =
cached_remote_reader_cache_settings().max_file_block_size;
+
+ auto assert_not_fully_covered = [&](size_t offset, size_t size, size_t
expected_blocks_num) {
+ io::FileBlocks blocks;
+ bool fully_covered = true;
+ ASSERT_TRUE(mgr.get_downloaded_blocks_if_fully_covered(key, offset,
size, context, &blocks,
+ &fully_covered)
+ .ok());
+ EXPECT_FALSE(fully_covered);
+ EXPECT_TRUE(blocks.empty());
+ EXPECT_EQ(mgr.get_file_blocks_num(io::FileCacheType::NORMAL),
expected_blocks_num);
+ };
+
+ auto assert_fully_covered = [&](size_t offset, size_t size, size_t
expected_blocks_size) {
+ io::FileBlocks blocks;
+ bool fully_covered = false;
+ ASSERT_TRUE(mgr.get_downloaded_blocks_if_fully_covered(key, offset,
size, context, &blocks,
+ &fully_covered)
+ .ok());
+ EXPECT_TRUE(fully_covered);
+ EXPECT_EQ(blocks.size(), expected_blocks_size);
+ };
+
+ assert_not_fully_covered(0, 10, 0);
+
+ complete_into_memory(mgr.get_or_set(key, 0, block_size, context));
+ ASSERT_EQ(mgr.get_file_blocks_num(io::FileCacheType::NORMAL), 1);
+ assert_fully_covered(2, 5, 1);
+ assert_not_fully_covered(block_size - 4, 8, 1);
+
+ complete_into_memory(mgr.get_or_set(key, block_size, block_size, context));
+ ASSERT_EQ(mgr.get_file_blocks_num(io::FileCacheType::NORMAL), 2);
+ assert_fully_covered(block_size - 4, 8, 2);
+
+ complete_into_memory(mgr.get_or_set(key, 2 * block_size, block_size,
context));
+ ASSERT_EQ(mgr.get_file_blocks_num(io::FileCacheType::NORMAL), 3);
+ assert_fully_covered(block_size - 4, block_size + 8, 3);
+
+ const size_t partial_block_size = 123 * 1024 + 17;
+ complete_into_memory(mgr.get_or_set(key, 3 * block_size,
partial_block_size, context));
+ ASSERT_EQ(mgr.get_file_blocks_num(io::FileCacheType::NORMAL), 4);
+ assert_fully_covered(3 * block_size + 7, 31, 1);
+ assert_fully_covered(3 * block_size - 4, 16, 2);
+ assert_fully_covered(0, 3 * block_size + partial_block_size, 4);
+ assert_not_fully_covered(3 * block_size + partial_block_size - 4, 8, 4);
+ assert_not_fully_covered(0, 3 * block_size + partial_block_size + 1, 4);
+}
+
+TEST_F(BlockFileCacheTest, cached_remote_file_reader_remote_only_on_miss) {
+ const std::string local_cache_path =
+ (caches_dir / "remote_only_on_miss_reader_cache" / "").string();
+ BlockFileCache* cache = nullptr;
+ ASSERT_TRUE(create_cached_remote_reader_cache(local_cache_path,
&cache).ok());
+
+ {
+ const auto remote_file = caches_dir /
"remote_only_on_miss_reader_file";
+ {
+ std::ofstream ofs(remote_file, std::ios::binary | std::ios::trunc);
+ ASSERT_TRUE(ofs.is_open());
+ for (int i = 0; i < 2; ++i) {
+ std::string data(1_mb, static_cast<char>('a' + i));
+ ofs.write(data.data(), data.size());
+ }
+ }
+
+ FileReaderSPtr local_reader;
+ ASSERT_TRUE(global_local_filesystem()->open_file(remote_file,
&local_reader).ok());
+ io::FileReaderOptions opts;
+ opts.cache_type = io::FileCachePolicy::FILE_BLOCK_CACHE;
+ opts.is_doris_table = true;
+ opts.tablet_id = 10086;
+ auto reader = std::make_shared<CachedRemoteFileReader>(local_reader,
opts);
+
+ auto key = io::BlockFileCache::hash(remote_file.filename().string());
+ Defer fd_cache_cleanup {
+ [key] {
io::FDCache::instance()->remove_file_reader(std::make_pair(key, 0)); }};
+ cache->remove_if_cached(key);
+
+ std::string buffer(4_kb, '\0');
+ size_t bytes_read = 0;
+ io::IOContext io_ctx;
+ io::FileCacheStatistics stats;
+ io_ctx.file_cache_stats = &stats;
+ io_ctx.file_cache_miss_policy =
io::FileCacheMissPolicy::REMOTE_ONLY_ON_MISS;
+
+ ASSERT_TRUE(reader->read_at(123, Slice(buffer.data(), buffer.size()),
&bytes_read, &io_ctx)
+ .ok());
+ EXPECT_EQ(bytes_read, buffer.size());
+ EXPECT_EQ(stats.num_remote_io_total, 1);
+ EXPECT_EQ(stats.bytes_read_from_remote, buffer.size());
+ EXPECT_EQ(stats.num_skip_cache_io_total, 1);
+ EXPECT_EQ(stats.bytes_write_into_cache, 0);
+ EXPECT_TRUE(cache->get_blocks_by_key(key).empty());
+
+ io::FileCacheStatistics write_back_stats;
+ io_ctx.file_cache_stats = &write_back_stats;
+ io_ctx.file_cache_miss_policy =
io::FileCacheMissPolicy::READ_THROUGH_AND_WRITE_BACK;
+ ASSERT_TRUE(reader->read_at(123, Slice(buffer.data(), buffer.size()),
&bytes_read, &io_ctx)
+ .ok());
+ EXPECT_EQ(bytes_read, buffer.size());
+ EXPECT_GT(write_back_stats.bytes_write_into_cache, 0);
+ EXPECT_FALSE(cache->get_blocks_by_key(key).empty());
+
+ io::FileCacheStatistics full_hit_stats;
+ io_ctx.file_cache_stats = &full_hit_stats;
+ io_ctx.file_cache_miss_policy =
io::FileCacheMissPolicy::REMOTE_ONLY_ON_MISS;
+ ASSERT_TRUE(reader->read_at(123, Slice(buffer.data(), buffer.size()),
&bytes_read, &io_ctx)
+ .ok());
+ EXPECT_EQ(bytes_read, buffer.size());
+ EXPECT_EQ(full_hit_stats.num_local_io_total, 1);
+ EXPECT_EQ(full_hit_stats.bytes_read_from_local, buffer.size());
+ EXPECT_EQ(full_hit_stats.num_remote_io_total, 0);
+ EXPECT_EQ(full_hit_stats.bytes_write_into_cache, 0);
+ }
+
+ cleanup_cached_remote_reader_cache(local_cache_path);
+}
+
void test_file_cache(io::FileCacheType cache_type) {
TUniqueId query_id;
query_id.hi = 1;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index 749726bc9ca..3e613bc0a31 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -514,6 +514,9 @@ public class SessionVariable implements Serializable,
Writable {
public static final String DISABLE_FILE_CACHE = "disable_file_cache";
+ public static final String ENABLE_TOPN_LAZY_MAT_PHASE2_NO_WRITE_FILE_CACHE
+ = "enable_topn_lazy_mat_phase2_no_write_file_cache";
+
public static final String FILE_CACHE_QUERY_LIMIT_PERCENT =
"file_cache_query_limit_percent";
public static final String FILE_CACHE_BASE_PATH = "file_cache_base_path";
@@ -2308,6 +2311,14 @@ public class SessionVariable implements Serializable,
Writable {
@VarAttrDef.VarAttr(name = DISABLE_FILE_CACHE, needForward = true)
public boolean disableFileCache = false;
+ @VarAttrDef.VarAttr(name =
ENABLE_TOPN_LAZY_MAT_PHASE2_NO_WRITE_FILE_CACHE, needForward = true,
+ description = {
+ "开启后，TopN 延迟物化第二阶段读取在 file cache miss 时直接读远端且不写回 file
cache。",
+ "When enabled, TopN lazy materialization phase-2 reads go
remote-only on "
+ + "file-cache miss and do not write the missed
range back to file cache."
+ })
+ public boolean enableTopnLazyMatPhase2NoWriteFileCache = false;
+
// Whether enable block file cache. Only take effect when BE config item
enable_file_cache is true.
@VarAttrDef.VarAttr(name = ENABLE_FILE_CACHE, needForward = true,
description = {
"是否启用 file cache。该变量只有在 be.conf 中 enable_file_cache=true 时才有效，"
@@ -5685,6 +5696,7 @@ public class SessionVariable implements Serializable,
Writable {
tResult.setParallelScanMinRowsPerScanner(parallelScanMinRowsPerScanner);
tResult.setOptimizeIndexScanParallelism(optimizeIndexScanParallelism);
tResult.setDisableFileCache(disableFileCache);
+
tResult.setEnableTopnLazyMatPhase2NoWriteFileCache(enableTopnLazyMatPhase2NoWriteFileCache);
tResult.setEnablePreferCachedRowset(getEnablePreferCachedRowset());
tResult.setQueryFreshnessToleranceMs(getQueryFreshnessToleranceMs());
diff --git a/gensrc/proto/internal_service.proto
b/gensrc/proto/internal_service.proto
index 2ca659b283c..ec60a1b2a3a 100644
--- a/gensrc/proto/internal_service.proto
+++ b/gensrc/proto/internal_service.proto
@@ -822,6 +822,18 @@ message PRequestBlockDesc {
repeated uint32 column_idxs = 7;
}
+message PTopNLazyMaterializationFileCacheStats {
+ optional int64 local_io_count = 1;
+ optional int64 local_io_bytes = 2;
+ optional int64 remote_io_count = 3;
+ optional int64 remote_io_bytes = 4;
+ optional int64 skip_cache_io_count = 5;
+ optional int64 write_cache_bytes = 6;
+ optional int64 local_io_time = 7;
+ optional int64 remote_io_time = 8;
+ optional int64 write_cache_io_time = 9;
+}
+
message PMultiGetRequestV2 {
repeated PRequestBlockDesc request_block_descs = 1;
@@ -830,6 +842,7 @@ message PMultiGetRequestV2 {
optional PUniqueId query_id = 3;
optional bool gc_id_map = 4;
optional uint64 wg_id = 5;
+ optional bool file_cache_remote_only_on_miss = 6;
};
message PMultiGetBlockV2 {
@@ -846,6 +859,7 @@ message PMultiGetBlockV2 {
message PMultiGetResponseV2 {
optional PStatus status = 1;
repeated PMultiGetBlockV2 blocks = 2;
+ optional PTopNLazyMaterializationFileCacheStats
topn_lazy_materialization_file_cache_stats = 3;
};
message PFetchColIdsRequest {
diff --git a/gensrc/thrift/PaloInternalService.thrift
b/gensrc/thrift/PaloInternalService.thrift
index 97b9ef86b99..49416576064 100644
--- a/gensrc/thrift/PaloInternalService.thrift
+++ b/gensrc/thrift/PaloInternalService.thrift
@@ -513,6 +513,7 @@ struct TQueryOptions {
1000: optional bool disable_file_cache = false
1001: optional i32 file_cache_query_limit_percent = -1
1002: optional bool enable_file_scanner_v2 = false
+ 1003: optional bool enable_topn_lazy_mat_phase2_no_write_file_cache = false
}
diff --git
a/regression-test/suites/cloud_p0/cache/topn_lazy_file_cache/test_topn_lazy_mat_phase2_no_write_file_cache.groovy
b/regression-test/suites/cloud_p0/cache/topn_lazy_file_cache/test_topn_lazy_mat_phase2_no_write_file_cache.groovy
new file mode 100644
index 00000000000..2e3ea83265c
--- /dev/null
+++
b/regression-test/suites/cloud_p0/cache/topn_lazy_file_cache/test_topn_lazy_mat_phase2_no_write_file_cache.groovy
@@ -0,0 +1,294 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.Http
+
+import java.util.regex.Pattern
+
+suite("test_topn_lazy_mat_phase2_no_write_file_cache", "docker") {
+ def options = new ClusterOptions()
+ options.feNum = 1
+ options.beNum = 1
+ options.msNum = 1
+ options.cloudMode = true
+ options.beConfigs += [
+ "enable_file_cache=true",
+ "disable_storage_page_cache=true",
+ "enable_java_support=false",
+ "enable_evict_file_cache_in_advance=false",
+ "file_cache_enter_disk_resource_limit_mode_percent=99",
+ "file_cache_each_block_size=4096",
+
"file_cache_path=[{\"path\":\"/opt/apache-doris/be/storage/file_cache\","
+ + "\"total_size\":104857600,\"query_limit\":104857600}]"
+ ]
+
+ docker(options) {
+ def clusters = sql "SHOW CLUSTERS"
+ assert !clusters.isEmpty()
+ def computeGroup = clusters[0][0]
+ sql "use @${computeGroup}"
+
+ def backends = sql "SHOW BACKENDS"
+ assert backends.size() == 1
+ def beHost = backends[0][1]
+ def beHttpPort = backends[0][4]
+ def clearFileCache = {
+ def result =
Http.GET("http://${beHost}:${beHttpPort}/api/file_cache?op=clear&sync=true";,
true)
+ logger.info("clear file cache result: ${result}")
+ }
+
+ sql "set enable_profile = true"
+ sql "set profile_level = 2"
+ sql "set enable_sql_cache = false"
+ sql "set enable_query_cache = false"
+ sql "set enable_page_cache = false"
+ sql "set topn_lazy_materialization_threshold = 1024"
+
+ def metricValue = { String profileText, String metricName ->
+ def matcher = profileText =~
+
/(?m)^\s*-\s*${Pattern.quote(metricName)}:\s*(?:sum\s+)?([0-9]+(?:\.[0-9]+)?).*$/
+ assert matcher.find() : "missing metric ${metricName} in
profile:\n${profileText}"
+ return new BigDecimal(matcher.group(1))
+ }
+
+ def metricValues = { String profileText, String metricName ->
+ def matcher = profileText =~
+
/(?m)^\s*-\s*${Pattern.quote(metricName)}:\s*(?:sum\s+)?([0-9]+(?:\.[0-9]+)?).*$/
+ def values = []
+ while (matcher.find()) {
+ values.add(new BigDecimal(matcher.group(1)))
+ }
+ assert !values.isEmpty() : "missing metric ${metricName} in
profile:\n${profileText}"
+ return values
+ }
+
+ def numericValues = { String text ->
+ def matcher = text =~ /([0-9]+(?:\.[0-9]+)?)/
+ def values = []
+ while (matcher.find()) {
+ values.add(new BigDecimal(matcher.group(1)))
+ }
+ return values
+ }
+
+ def metricLineValues = { String profileText, String metricName ->
+ def matcher = profileText =~
/(?m)^\s*-\s*${Pattern.quote(metricName)}:\s*(.*)$/
+ def values = []
+ while (matcher.find()) {
+ values.add(matcher.group(1).trim())
+ }
+ return values
+ }
+
+ def topnSecondPhaseMetricNames = [
+ "TopNLazyMaterializationSecondPhaseLocalIOCount",
+ "TopNLazyMaterializationSecondPhaseLocalIOBytes",
+ "TopNLazyMaterializationSecondPhaseRemoteIOCount",
+ "TopNLazyMaterializationSecondPhaseRemoteIOBytes",
+ "TopNLazyMaterializationSecondPhaseSkipCacheIOCount",
+ "TopNLazyMaterializationSecondPhaseWriteCacheBytes",
+ "TopNLazyMaterializationSecondPhaseLocalIOTime",
+ "TopNLazyMaterializationSecondPhaseRemoteIOTime",
+ "TopNLazyMaterializationSecondPhaseWriteCacheIOTime",
+ "TopNLazyMaterializationSecondPhaseRowsRead",
+ "TopNLazyMaterializationSecondPhaseSegmentsRead",
+ "TopNLazyMaterializationSecondPhasePerBackend",
+ "TopNLazyMaterializationSecondPhasePerBackendRowsRead",
+ "TopNLazyMaterializationSecondPhasePerBackendSegmentsRead",
+ "TopNLazyMaterializationSecondPhasePerBackendLocalIOCount",
+ "TopNLazyMaterializationSecondPhasePerBackendLocalIOBytes",
+ "TopNLazyMaterializationSecondPhasePerBackendRemoteIOCount",
+ "TopNLazyMaterializationSecondPhasePerBackendRemoteIOBytes",
+ "TopNLazyMaterializationSecondPhasePerBackendSkipCacheIOCount",
+ "TopNLazyMaterializationSecondPhasePerBackendWriteCacheBytes",
+ "TopNLazyMaterializationSecondPhasePerBackendLocalIOTime",
+ "TopNLazyMaterializationSecondPhasePerBackendRemoteIOTime",
+ "TopNLazyMaterializationSecondPhasePerBackendWriteCacheIOTime"
+ ]
+
+ def logTopnSecondPhaseMetrics = { String name, String profileText ->
+ def metrics = topnSecondPhaseMetricNames.collectEntries {
metricName ->
+ def values = metricLineValues(profileText, metricName)
+ assert !values.isEmpty() : "missing metric ${metricName} in
profile:\n${profileText}"
+ [(metricName): values]
+ }
+ logger.info("${name} TopN lazy materialization second phase
metrics: ${metrics}")
+ logger.info("${name} CachedPagesNum values:
${metricLineValues(profileText, 'CachedPagesNum')}")
+ }
+
+ def runProfileQuery = { String name, String query, Closure checker ->
+ profile(name) {
+ run {
+ sql "/* ${name} */ ${query}"
+ sleep(1000)
+ }
+ check { profileString, exception ->
+ if (exception != null) {
+ logger.error("Profile failed, profile
result:\n${profileString}", exception)
+ throw exception
+ }
+ assert
profileString.contains("TopNLazyMaterializationSecondPhase") :
+ "missing TopN lazy materialization
profile:\n${profileString}"
+ assert profileString.contains("Is Cached: No")
+ || profileString.contains("Is Cached: No") :
+ "query should not hit SQL/query
cache:\n${profileString}"
+ logTopnSecondPhaseMetrics(name, profileString)
+ checker(profileString)
+ }
+ }
+ }
+
+ def assertRemoteOnlyMiss = { String profileText ->
+ assert metricValue(profileText,
"TopNLazyMaterializationSecondPhaseRemoteIOCount") > 0
+ assert metricValue(profileText,
"TopNLazyMaterializationSecondPhaseRemoteIOBytes") > 0
+ assert metricValue(profileText,
"TopNLazyMaterializationSecondPhaseSkipCacheIOCount") > 0
+ assert metricValue(profileText,
"TopNLazyMaterializationSecondPhaseWriteCacheBytes")
+ .compareTo(BigDecimal.ZERO) == 0
+ }
+
+ def assertPerBackendRowsMatchAggregate = { String profileText ->
+ def aggregateRows = metricValues(profileText,
+ "TopNLazyMaterializationSecondPhaseRowsRead")
+ .inject(BigDecimal.ZERO) { sum, value -> sum + value }
+ def perBackendRows = metricLineValues(profileText,
+ "TopNLazyMaterializationSecondPhasePerBackendRowsRead")
+ .collectMany { line -> numericValues(line) }
+ assert !perBackendRows.isEmpty() :
+ "missing per-backend rows-read values:\n${profileText}"
+ def perBackendRowsSum = perBackendRows.inject(BigDecimal.ZERO) {
sum, value -> sum + value }
+ assert aggregateRows.compareTo(perBackendRowsSum) == 0 :
+ "per-backend rows-read sum ${perBackendRowsSum} should
equal aggregate ${aggregateRows}:\n${profileText}"
+ }
+
+ def assertLocalFullHit = { String profileText ->
+ assert metricValue(profileText,
"TopNLazyMaterializationSecondPhaseLocalIOCount") > 0
+ assert metricValue(profileText,
"TopNLazyMaterializationSecondPhaseLocalIOBytes") > 0
+ assert metricValue(profileText,
"TopNLazyMaterializationSecondPhaseRemoteIOCount")
+ .compareTo(BigDecimal.ZERO) == 0
+ assert metricValue(profileText,
"TopNLazyMaterializationSecondPhaseRemoteIOBytes")
+ .compareTo(BigDecimal.ZERO) == 0
+ assert metricValue(profileText,
"TopNLazyMaterializationSecondPhaseWriteCacheBytes")
+ .compareTo(BigDecimal.ZERO) == 0
+ }
+
+ sql "DROP TABLE IF EXISTS topn_lazy_remote_only_no_row_store"
+ sql """
+ CREATE TABLE topn_lazy_remote_only_no_row_store (
+ k INT NOT NULL,
+ sort_key INT NOT NULL,
+ payload VARCHAR(4096) NOT NULL,
+ pad VARCHAR(4096) NOT NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "disable_auto_compaction" = "true",
+ "store_row_column" = "false"
+ )
+ """
+ sql """
+ INSERT INTO topn_lazy_remote_only_no_row_store
+ SELECT number,
+ 4096 - number,
+ repeat(cast(number as string), 128),
+ repeat('x', 256)
+ FROM numbers("number" = "4096")
+ """
+
+ def noRowStoreQuery =
+ "SELECT k, payload, pad FROM
topn_lazy_remote_only_no_row_store ORDER BY sort_key LIMIT 16"
+ explain {
+ sql noRowStoreQuery
+ contains("MaterializeNode")
+ }
+
+ clearFileCache()
+ sql "set enable_topn_lazy_mat_phase2_no_write_file_cache = true"
+ runProfileQuery("topn_lazy_remote_only_no_row_store_remote_only_miss",
+ noRowStoreQuery, assertRemoteOnlyMiss)
+
+ clearFileCache()
+ sql "set enable_topn_lazy_mat_phase2_no_write_file_cache = false"
+ sql "/* topn_lazy_remote_only_no_row_store_local_full_hit_warm */
${noRowStoreQuery}"
+ sleep(1000)
+
+ sql "set enable_topn_lazy_mat_phase2_no_write_file_cache = true"
+ runProfileQuery("topn_lazy_remote_only_no_row_store_local_full_hit",
+ noRowStoreQuery, assertLocalFullHit)
+
+ clearFileCache()
+ sql "set enable_topn_lazy_mat_phase2_no_write_file_cache = true"
+ sql "set batch_size = 1"
+ try {
+
runProfileQuery("topn_lazy_remote_only_no_row_store_multi_fetch_accumulate",
+ noRowStoreQuery, { profileText ->
+ assertRemoteOnlyMiss(profileText)
+ assertPerBackendRowsMatchAggregate(profileText)
+ })
+ } finally {
+ sql "set batch_size = 4062"
+ }
+
+ sql "DROP TABLE IF EXISTS topn_lazy_remote_only_row_store"
+ sql """
+ CREATE TABLE topn_lazy_remote_only_row_store (
+ k INT NOT NULL,
+ sort_key INT NOT NULL,
+ payload VARCHAR(4096) NOT NULL,
+ pad VARCHAR(4096) NOT NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "disable_auto_compaction" = "true",
+ "store_row_column" = "true"
+ )
+ """
+ sql """
+ INSERT INTO topn_lazy_remote_only_row_store
+ SELECT number,
+ 4096 - number,
+ repeat(cast(number as string), 128),
+ repeat('x', 256)
+ FROM numbers("number" = "4096")
+ """
+
+ def rowStoreQuery =
+ "SELECT k, payload, pad FROM topn_lazy_remote_only_row_store
ORDER BY sort_key LIMIT 16"
+ explain {
+ sql rowStoreQuery
+ contains("MaterializeNode")
+ }
+
+ clearFileCache()
+ sql "set enable_topn_lazy_mat_phase2_no_write_file_cache = true"
+ runProfileQuery("topn_lazy_remote_only_row_store_remote_only_miss",
+ rowStoreQuery, assertRemoteOnlyMiss)
+
+ clearFileCache()
+ sql "set enable_topn_lazy_mat_phase2_no_write_file_cache = false"
+ sql "/* topn_lazy_remote_only_row_store_local_full_hit_warm */
${rowStoreQuery}"
+ sleep(1000)
+
+ sql "set enable_topn_lazy_mat_phase2_no_write_file_cache = true"
+ runProfileQuery("topn_lazy_remote_only_row_store_local_full_hit",
+ rowStoreQuery, assertLocalFullHit)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
