This is an automated email from the ASF dual-hosted git repository.
csun5285 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 211965dcb37 [opt](point query) Reuse segments for point query row data
lookup (#67649)
211965dcb37 is described below
commit 211965dcb37bba6842a0f477046838d2432d8bd9
Author: zyp-V <[email protected]>
AuthorDate: Mon Oct 26 10:09:22 2026 +0800
[opt](point query) Reuse segments for point query row data lookup (#67649)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: Row-store point queries load rowset segments to locate
a primary key, then load and search those segments again to read the row
data. Return the matched segment from key lookup and retain it in the
per-row read context so row-data lookup can use it directly. Keep the
existing lookup interface for other callers and preserve the current
branch's IOContext. This removes the second segment-loader call and
segment scan from the row data stage; no latency or throughput benchmark
was run.
### Release note
Reduce repeated segment lookups in row-store point queries.
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [x] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---------
Co-authored-by: zhangyipeng.0818 <[email protected]>
---
be/src/service/point_query_executor.cpp | 12 +-
be/src/service/point_query_executor.h | 2 +
be/src/storage/segment/segment_loader.cpp | 2 +
be/src/storage/tablet/base_tablet.cpp | 37 ++-
be/src/storage/tablet/base_tablet.h | 15 +-
be/test/service/point_query_exector_test.cpp | 351 +++++++++++++++++++++++++
be/test/storage/segment/segment_cache_test.cpp | 103 +++++++-
7 files changed, 509 insertions(+), 13 deletions(-)
diff --git a/be/src/service/point_query_executor.cpp
b/be/src/service/point_query_executor.cpp
index 560fb3ca1e6..5fc85aebe8c 100644
--- a/be/src/service/point_query_executor.cpp
+++ b/be/src/service/point_query_executor.cpp
@@ -506,10 +506,10 @@ Status PointQueryExecutor::_lookup_row_key() {
}
// Get rowlocation and rowset, ctx._rowset_ptr will acquire wrap this
ptr
auto rowset_ptr = std::make_unique<RowsetSharedPtr>();
- st = (_tablet->lookup_row_key(_row_read_ctxs[i]._primary_key, nullptr,
false,
- specified_rowsets, &location, INT32_MAX
/*rethink?*/,
- segment_caches, rowset_ptr.get(), false,
nullptr,
- &_profile_metrics.read_stats, nullptr,
&io_ctx));
+ st = (_tablet->lookup_row_key(
+ _row_read_ctxs[i]._primary_key, nullptr, false,
specified_rowsets, &location,
+ INT32_MAX /*rethink?*/, segment_caches, rowset_ptr.get(),
false, nullptr,
+ &_profile_metrics.read_stats, nullptr, &io_ctx,
&_row_read_ctxs[i]._segment));
if (st.is<ErrorCode::KEY_NOT_FOUND>()) {
continue;
}
@@ -554,8 +554,8 @@ Status PointQueryExecutor::_lookup_row_data() {
io_ctx.remote_scan_cache_write_limiter =
_remote_scan_cache_write_limiter.get();
RETURN_IF_ERROR(_tablet->lookup_row_data(
_row_read_ctxs[i]._primary_key,
_row_read_ctxs[i]._row_location.value(),
- *(_row_read_ctxs[i]._rowset_ptr),
_profile_metrics.read_stats, value,
- use_row_cache, &io_ctx));
+ _row_read_ctxs[i]._segment,
*(_row_read_ctxs[i]._rowset_ptr),
+ _profile_metrics.read_stats, value, use_row_cache,
&io_ctx));
// serialize value to block, currently only jsonb row format
RETURN_IF_ERROR(JsonbSerializeUtil::jsonb_to_columns(
_reusable->get_data_type_serdes(), value.data(),
value.size(),
diff --git a/be/src/service/point_query_executor.h
b/be/src/service/point_query_executor.h
index 91a8632b00a..285b77c6cd2 100644
--- a/be/src/service/point_query_executor.h
+++ b/be/src/service/point_query_executor.h
@@ -335,6 +335,8 @@ private:
// rowset will be aquired during read
// and released after used
std::unique_ptr<RowsetSharedPtr, decltype(&release_rowset)>
_rowset_ptr;
+ // Reuse the segment loaded during key lookup; release it before its
acquired rowset.
+ segment_v2::SegmentSharedPtr _segment;
};
PTabletKeyLookupResponse* _response = nullptr;
diff --git a/be/src/storage/segment/segment_loader.cpp
b/be/src/storage/segment/segment_loader.cpp
index c7f4a1a6d3b..8585267a420 100644
--- a/be/src/storage/segment/segment_loader.cpp
+++ b/be/src/storage/segment/segment_loader.cpp
@@ -21,6 +21,7 @@
#include "common/config.h"
#include "common/status.h"
+#include "cpp/sync_point.h"
#include "storage/olap_define.h"
#include "storage/rowset/beta_rowset.h"
#include "storage/rowset/rowset_meta.h"
@@ -118,6 +119,7 @@ Status SegmentLoader::load_segments(const
BetaRowsetSharedPtr& rowset,
bool need_load_pk_index_and_bf,
OlapReaderStatistics* index_load_stats,
const io::IOContext* io_ctx) {
+ TEST_SYNC_POINT_CALLBACK("SegmentLoader::load_segments", rowset.get());
if (cache_handle->is_inited()) {
return Status::OK();
}
diff --git a/be/src/storage/tablet/base_tablet.cpp
b/be/src/storage/tablet/base_tablet.cpp
index eba76fe573b..ebb55462a1a 100644
--- a/be/src/storage/tablet/base_tablet.cpp
+++ b/be/src/storage/tablet/base_tablet.cpp
@@ -484,6 +484,26 @@ Status BaseTablet::lookup_row_data(const Slice&
encoded_key, const RowLocation&
RowsetSharedPtr input_rowset,
OlapReaderStatistics& stats,
std::string& values, bool write_to_cache,
const io::IOContext* io_ctx) {
+ return _lookup_row_data(encoded_key, row_location, nullptr,
std::move(input_rowset), stats,
+ values, write_to_cache, io_ctx);
+}
+
+Status BaseTablet::lookup_row_data(const Slice& encoded_key, const
RowLocation& row_location,
+ const segment_v2::SegmentSharedPtr& segment,
+ RowsetSharedPtr input_rowset,
OlapReaderStatistics& stats,
+ std::string& values, bool write_to_cache,
+ const io::IOContext* io_ctx) {
+ DCHECK(segment != nullptr);
+ DCHECK_EQ(segment->id(), row_location.segment_id);
+ return _lookup_row_data(encoded_key, row_location, &segment,
std::move(input_rowset), stats,
+ values, write_to_cache, io_ctx);
+}
+
+Status BaseTablet::_lookup_row_data(const Slice& encoded_key, const
RowLocation& row_location,
+ const segment_v2::SegmentSharedPtr*
segment,
+ RowsetSharedPtr input_rowset,
OlapReaderStatistics& stats,
+ std::string& values, bool write_to_cache,
+ const io::IOContext* io_ctx) const {
MonotonicStopWatch watch;
size_t row_size = 1;
watch.start();
@@ -498,9 +518,14 @@ Status BaseTablet::lookup_row_data(const Slice&
encoded_key, const RowLocation&
SegmentCacheHandle segment_cache_handle;
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,
- io_ctx));
+ if (segment != nullptr) {
+ RETURN_IF_ERROR(_init_segment_column_iterator(rowset, *segment,
column, &column_iterator,
+ stats, io_ctx));
+ } else {
+ RETURN_IF_ERROR(_get_segment_column_iterator(rowset,
row_location.segment_id, column,
+ &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)};
@@ -523,7 +548,8 @@ Status BaseTablet::lookup_row_key(const Slice& encoded_key,
TabletSchema* latest
std::vector<std::unique_ptr<SegmentCacheHandle>>& segment_caches,
RowsetSharedPtr* rowset, bool with_rowid,
std::string* encoded_seq_value,
OlapReaderStatistics* stats,
- DeleteBitmapPtr delete_bitmap, const
io::IOContext* io_ctx) {
+ DeleteBitmapPtr delete_bitmap, const
io::IOContext* io_ctx,
+ segment_v2::SegmentSharedPtr* segment) {
SCOPED_BVAR_LATENCY(g_tablet_lookup_rowkey_latency);
size_t seq_col_length = 0;
// use the latest tablet schema to decide if the tablet has sequence
column currently
@@ -608,6 +634,9 @@ Status BaseTablet::lookup_row_key(const Slice& encoded_key,
TabletSchema* latest
}
TEST_SYNC_POINT_CALLBACK("BaseTablet::lookup_row_key:found", this,
rs.get(),
with_seq_col, s.code());
+ if (segment) {
+ *segment = segments[id];
+ }
// find it and return
return s;
}
diff --git a/be/src/storage/tablet/base_tablet.h
b/be/src/storage/tablet/base_tablet.h
index e570fc79944..c18247ea81f 100644
--- a/be/src/storage/tablet/base_tablet.h
+++ b/be/src/storage/tablet/base_tablet.h
@@ -182,6 +182,13 @@ public:
Status lookup_row_data(const Slice& encoded_key, const RowLocation&
row_location,
RowsetSharedPtr rowset, OlapReaderStatistics&
stats, std::string& values,
bool write_to_cache = false, const io::IOContext*
io_ctx = nullptr);
+
+ // Reuse the segment returned by lookup_row_key. The caller must keep the
segment and
+ // its acquired rowset alive until this method returns.
+ Status lookup_row_data(const Slice& encoded_key, const RowLocation&
row_location,
+ const segment_v2::SegmentSharedPtr& segment,
RowsetSharedPtr rowset,
+ OlapReaderStatistics& stats, std::string& values,
+ 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.
@@ -193,7 +200,8 @@ public:
std::string* encoded_seq_value = nullptr,
OlapReaderStatistics* stats = nullptr,
DeleteBitmapPtr tablet_delete_bitmap = nullptr,
- const io::IOContext* io_ctx = nullptr);
+ const io::IOContext* io_ctx = nullptr,
+ segment_v2::SegmentSharedPtr* segment = nullptr);
// calc delete bitmap when flush memtable, use a fake version to calc
// For example, cur max version is 5, and we use version 6 to calc but
@@ -371,6 +379,11 @@ public:
const
CaptureRowsetOps& options);
protected:
+ Status _lookup_row_data(const Slice& encoded_key, const RowLocation&
row_location,
+ const segment_v2::SegmentSharedPtr* segment,
RowsetSharedPtr rowset,
+ OlapReaderStatistics& stats, std::string& values,
bool write_to_cache,
+ const io::IOContext* io_ctx) const;
+
// Protected by the tablet header lock; discovery resumes after this
version.
Version _changelog_ttl_scan_cursor {-1, -1};
diff --git a/be/test/service/point_query_exector_test.cpp
b/be/test/service/point_query_exector_test.cpp
index 74bd7606e39..3e2269b2ae6 100644
--- a/be/test/service/point_query_exector_test.cpp
+++ b/be/test/service/point_query_exector_test.cpp
@@ -21,15 +21,31 @@
#include <gen_cpp/internal_service.pb.h>
#include <gtest/gtest.h>
+#include <atomic>
+
+#include "cloud/cloud_storage_engine.h"
+#include "cloud/cloud_tablet.h"
+#include "common/config.h"
#include "common/object_pool.h"
#include "core/block/block.h"
+#include "core/column/column_nullable.h"
+#include "core/field.h"
+#include "cpp/sync_point.h"
#include "exprs/vexpr.h"
+#include "io/fs/local_file_system.h"
#include "runtime/descriptor_helper.h"
#include "runtime/descriptors.h"
#include "runtime/exec_env.h"
+#include "runtime/memory/cache_manager.h"
#include "runtime/runtime_state.h"
#include "service/point_query_executor.h"
+#include "storage/options.h"
+#include "storage/rowset/beta_rowset.h"
+#include "storage/rowset/beta_rowset_writer.h"
+#include "storage/segment/segment_loader.h"
+#include "storage/storage_engine.h"
#include "storage/tablet/tablet_schema.h"
+#include "util/thrift_util.h"
namespace doris {
@@ -419,4 +435,339 @@ TEST_F(LookupConnectionCacheTest, PQTestEntryLifetime) {
EXPECT_EQ(entry.use_count(), 2) << "Cache should maintain sole ownership
after scope exit";
}
+// Exercise the executor's key/data stages with real MoW segments. The
parameters cover
+// local contiguous IDs and cloud segment-list IDs.
+class PointQuerySegmentReuseTest : public testing::TestWithParam<bool> {
+protected:
+ void SetUp() override {
+ _saved_disable_row_cache = config::disable_storage_row_cache;
+ _saved_disable_segment_cache = config::disable_segment_cache;
+ config::disable_storage_row_cache = true;
+ config::disable_segment_cache = false;
+ auto* env = ExecEnv::GetInstance();
+ _saved_cache_manager = env->get_cache_manager();
+ _cache_manager.reset(CacheManager::create_global_instance());
+ env->set_cache_manager(_cache_manager.get());
+ _saved_row_cache = env->get_row_cache();
+ _saved_segment_loader = env->segment_loader();
+ _row_cache.reset(new RowCache(1024 * 1024, 1));
+ _segment_loader = std::make_unique<SegmentLoader>(64 * 1024 * 1024,
1024);
+ env->_row_cache = _row_cache.get();
+ env->set_segment_loader(_segment_loader.get());
+ EngineOptions options;
+ options.backend_uid = UniqueId::gen_uid();
+ _engine = std::make_unique<StorageEngine>(options);
+
ASSERT_TRUE(io::global_local_filesystem()->delete_directory(kTestDir).ok());
+
ASSERT_TRUE(io::global_local_filesystem()->create_directory(kTestDir).ok());
+ }
+
+ void TearDown() override {
+ auto* sync_point = SyncPoint::get_instance();
+ sync_point->clear_call_back("SegmentLoader::load_segments");
+ sync_point->disable_processing();
+ evict_segments();
+ _tablet.reset();
+ _rowsets.clear();
+ _cloud_engine.reset();
+ _engine.reset();
+ auto* env = ExecEnv::GetInstance();
+ env->_row_cache = _saved_row_cache;
+ env->set_segment_loader(_saved_segment_loader);
+ _row_cache.reset();
+ _segment_loader.reset();
+ env->set_cache_manager(_saved_cache_manager);
+ _cache_manager.reset();
+ config::disable_storage_row_cache = _saved_disable_row_cache;
+ config::disable_segment_cache = _saved_disable_segment_cache;
+
EXPECT_TRUE(io::global_local_filesystem()->delete_directory(kTestDir).ok());
+ }
+
+ Status create_tablet() {
+ TabletMetaPB meta_pb;
+ meta_pb.set_tablet_id(987654);
+ meta_pb.set_enable_unique_key_merge_on_write(true);
+ meta_pb.set_tablet_state(PB_RUNNING);
+ auto* schema_pb = meta_pb.mutable_schema();
+ schema_pb->set_keys_type(UNIQUE_KEYS);
+ schema_pb->set_num_short_key_columns(1);
+ schema_pb->set_num_rows_per_row_block(1024);
+ schema_pb->set_next_column_unique_id(5);
+ schema_pb->set_delete_sign_idx(3);
+ schema_pb->set_store_row_column(true);
+ auto add_column = [&](int uid, const std::string& name, const
std::string& type, int length,
+ bool nullable) {
+ auto* column = schema_pb->add_column();
+ column->set_unique_id(uid);
+ column->set_name(name);
+ column->set_type(type);
+ column->set_length(length);
+ column->set_index_length(type == "STRING" ? 4 : length);
+ column->set_is_key(uid == 0);
+ column->set_is_nullable(nullable);
+ column->set_aggregation("NONE");
+ };
+ add_column(0, "k", "INT", 4, false);
+ add_column(1, "v1", "INT", 4, false);
+ add_column(2, "v2", "INT", 4, true);
+ add_column(3, DELETE_SIGN, "TINYINT", 1, false);
+ add_column(4, BeConsts::ROW_STORE_COL, "STRING", 2147483643, false);
+ auto meta = std::make_shared<TabletMeta>();
+ meta->init_from_pb(meta_pb);
+ if (GetParam()) {
+ _cloud_engine = std::make_unique<CloudStorageEngine>(EngineOptions
{});
+ _tablet = std::make_shared<CloudTablet>(*_cloud_engine, meta);
+ } else {
+ _tablet = std::make_shared<Tablet>(*_engine, meta, nullptr);
+ }
+ RETURN_IF_ERROR(write_rowset(1, {{10, 11}, {20}, {30}}));
+ RETURN_IF_ERROR(write_rowset(2, {{40}}));
+ // Key 20 still has a primary-index entry but is invisible through the
delete bitmap.
+ const auto& rowset = _rowsets.front();
+ meta->delete_bitmap().add({rowset->rowset_id(),
rowset->segment(1).id(), 1}, 0);
+ auto* sync_point = SyncPoint::get_instance();
+ sync_point->set_call_back("SegmentLoader::load_segments", [&](auto&&
args) {
+ if (try_any_cast<BetaRowset*>(args[0])->rowset_meta()->tablet_id()
==
+ _tablet->tablet_id()) {
+ _segment_load_calls.fetch_add(1, std::memory_order_relaxed);
+ }
+ });
+ sync_point->enable_processing();
+ return Status::OK();
+ }
+
+ Status write_rowset(int64_t version, const
std::vector<std::vector<int32_t>>& keys) {
+ RowsetWriterContext context;
+ context.rowset_id = _engine->next_rowset_id();
+ context.tablet_id = _tablet->tablet_id();
+ context.tablet_schema = _tablet->tablet_schema();
+ context.tablet_path = kTestDir;
+ context.rowset_state = VISIBLE;
+ context.version = {version, version};
+ context.enable_unique_key_merge_on_write = true;
+ context.enable_segcompaction = false;
+ context.write_type = DataWriteType::TYPE_DIRECT;
+ BetaRowsetWriter writer(*_engine);
+ RETURN_IF_ERROR(writer.init(context));
+ for (const auto& segment_keys : keys) {
+ Block block = context.tablet_schema->create_storage_block();
+ {
+ auto guard = block.mutate_columns_scoped();
+ auto& columns = guard.mutable_columns();
+ for (int32_t key : segment_keys) {
+ columns[0]->insert(Field::create_field<TYPE_INT>(key));
+ columns[1]->insert(Field::create_field<TYPE_INT>(key *
10));
+ if (key == 11) {
+ columns[2]->insert_default();
+ } else {
+ columns[2]->insert(Field::create_field<TYPE_INT>(key *
100));
+ }
+ columns[3]->insert_default();
+ columns[4]->insert_default();
+ }
+ }
+ RETURN_IF_ERROR(writer.flush_single_block(&block));
+ }
+ RowsetSharedPtr rowset;
+ RETURN_IF_ERROR(writer.build(rowset));
+ if (GetParam()) {
+ // Model a cloud segment-list rowset using local files, without a
remote service.
+ // Each rowset can use the same physical IDs; its rowset ID
distinguishes the files.
+ std::vector<int64_t> segment_ids;
+ for (auto segment : rowset->segments()) {
+ auto path = DORIS_TRY(segment.path());
+ int64_t physical_id = 7 + segment.pos() * 13;
+ RETURN_IF_ERROR(io::global_local_filesystem()->rename(
+ path, std::string(kTestDir) + "/" +
rowset->rowset_id().to_string() + "_" +
+ std::to_string(physical_id) + ".dat"));
+ segment_ids.push_back(physical_id);
+ }
+ rowset->rowset_meta()->set_segment_ids(segment_ids);
+ }
+ {
+ std::unique_lock lock(_tablet->get_header_lock());
+ _tablet->_rs_version_map.emplace(rowset->version(), rowset);
+ }
+ _rowsets.push_back(std::move(rowset));
+ return Status::OK();
+ }
+
+ Status prepare_executor(PointQueryExecutor* executor, const
std::vector<int32_t>& keys) {
+ TDescriptorTableBuilder descriptor_builder;
+ TTupleDescriptorBuilder tuple_builder;
+ for (int pos = 0; pos < 4; ++pos) {
+ const auto& column = _tablet->tablet_schema()->column(pos);
+ auto slot = TSlotDescriptorBuilder()
+ .type(pos == 3 ? TYPE_TINYINT : TYPE_INT)
+ .column_name(column.name())
+ .column_pos(pos)
+ .nullable(column.is_nullable())
+ .build();
+ slot.__set_col_unique_id(column.unique_id());
+ tuple_builder.add_slot(slot);
+ }
+ tuple_builder.build(&descriptor_builder);
+ auto descriptor = descriptor_builder.desc_tbl();
+ std::vector<TExpr> output_exprs;
+ for (int pos = 0; pos < 3; ++pos) {
+ TExprNode node;
+ node.__set_node_type(TExprNodeType::SLOT_REF);
+ node.__set_type(descriptor.slotDescriptors[pos].slotType);
+
node.__set_is_nullable(_tablet->tablet_schema()->column(pos).is_nullable());
+ node.__set_num_children(0);
+ TSlotRef slot_ref;
+ slot_ref.__set_slot_id(descriptor.slotDescriptors[pos].id);
+ slot_ref.__set_tuple_id(0);
+ node.__set_slot_ref(slot_ref);
+ TExpr expr;
+ expr.nodes.push_back(node);
+ output_exprs.push_back(expr);
+ }
+ TQueryOptions options;
+ executor->_tablet = _tablet;
+ executor->_reusable = std::make_shared<Reusable>();
+ RETURN_IF_ERROR(executor->_reusable->init(descriptor, output_exprs,
options,
+ *_tablet->tablet_schema(),
1));
+ PTabletKeyLookupRequest request;
+ ThriftSerializer serializer(false, 128);
+ for (int32_t key : keys) {
+ TExprNode literal;
+ literal.__set_node_type(TExprNodeType::INT_LITERAL);
+ literal.__set_type(descriptor.slotDescriptors[0].slotType);
+ literal.__set_is_nullable(false);
+ literal.__set_num_children(0);
+ TIntLiteral int_literal;
+ int_literal.__set_value(key);
+ literal.__set_int_literal(int_literal);
+ RETURN_IF_ERROR(serializer.serialize(
+ &literal,
request.add_key_tuples()->add_key_column_literals()));
+ }
+ RETURN_IF_ERROR(executor->_init_keys(&request));
+ executor->_result_block = executor->_reusable->get_block();
+ return Status::OK();
+ }
+
+ void evict_segments() {
+ for (const auto& rowset : _rowsets) {
+ SegmentLoader::instance()->erase_segments(*rowset->rowset_meta());
+ }
+ }
+
+ static void check_rows(const Block& block) {
+ ASSERT_EQ(block.rows(), 4);
+ const std::vector<int32_t> expected_keys {30, 10, 40, 11};
+ const auto& v2 = assert_cast<const
ColumnNullable&>(*block.get_by_position(2).column);
+ for (size_t row = 0; row < expected_keys.size(); ++row) {
+ EXPECT_EQ(block.get_by_position(0).column->get_int(row),
expected_keys[row]);
+ EXPECT_EQ(block.get_by_position(1).column->get_int(row),
expected_keys[row] * 10);
+ EXPECT_EQ(v2.is_null_at(row), expected_keys[row] == 11);
+ if (expected_keys[row] != 11) {
+ EXPECT_EQ(v2.get_nested_column().get_int(row),
expected_keys[row] * 100);
+ }
+ EXPECT_EQ(block.get_by_position(3).column->get_int(row), 0);
+ }
+ }
+
+ static constexpr const char* kTestDir = "./point_query_segment_reuse_test";
+ std::unique_ptr<StorageEngine> _engine;
+ std::unique_ptr<CloudStorageEngine> _cloud_engine;
+ BaseTabletSPtr _tablet;
+ std::vector<RowsetSharedPtr> _rowsets;
+ std::atomic<int> _segment_load_calls = 0;
+ std::unique_ptr<RowCache> _row_cache;
+ std::unique_ptr<SegmentLoader> _segment_loader;
+ std::unique_ptr<CacheManager> _cache_manager;
+ CacheManager* _saved_cache_manager = nullptr;
+ RowCache* _saved_row_cache = nullptr;
+ SegmentLoader* _saved_segment_loader = nullptr;
+ bool _saved_disable_row_cache = true;
+ bool _saved_disable_segment_cache = false;
+};
+
+// NOLINTNEXTLINE(readability-function-cognitive-complexity): gtest assertions
expand to branches; keep the key/data lifetime checks together.
+TEST_P(PointQuerySegmentReuseTest, ReuseAfterKeyLookupAndCacheEviction) {
+ ASSERT_EQ(Status::OK(), create_tablet());
+ std::vector<std::weak_ptr<segment_v2::Segment>> retained_segments;
+ {
+ PointQueryExecutor executor;
+ ASSERT_EQ(Status::OK(), prepare_executor(&executor, {30, 10, 40, 11,
99, 20}));
+ ASSERT_EQ(Status::OK(), executor._lookup_row_key());
+ ASSERT_EQ(executor._row_hits, 4);
+ ASSERT_EQ(_segment_load_calls.load(), 2);
+ const auto& contexts = executor._row_read_ctxs;
+ for (size_t row = 0; row < 4; ++row) {
+ const auto& context = contexts[row];
+ ASSERT_TRUE(context._row_location.has_value());
+ ASSERT_NE(context._segment, nullptr);
+ ASSERT_NE(context._rowset_ptr, nullptr);
+ EXPECT_EQ(context._segment->id(),
context._row_location->segment_id);
+ EXPECT_EQ((*context._rowset_ptr)->rowset_id(),
context._row_location->rowset_id);
+ retained_segments.emplace_back(context._segment);
+ }
+ EXPECT_EQ(contexts[0]._segment->id(), _rowsets[0]->segment(2).id());
+ EXPECT_EQ(contexts[1]._segment.get(), contexts[3]._segment.get());
+ EXPECT_NE(contexts[1]._segment.get(), contexts[2]._segment.get());
+ EXPECT_EQ(contexts[3]._row_location->row_id, 1);
+ for (size_t row : {4, 5}) {
+ EXPECT_FALSE(contexts[row]._row_location.has_value());
+ EXPECT_EQ(contexts[row]._segment, nullptr);
+ EXPECT_EQ(contexts[row]._rowset_ptr, nullptr);
+ }
+ ASSERT_GT(_segment_loader->_segment_cache->get_element_count(), 0);
+ evict_segments();
+ ASSERT_EQ(_segment_loader->_segment_cache->get_element_count(), 0);
+ for (const auto& segment : retained_segments) {
+ EXPECT_FALSE(segment.expired());
+ }
+ ASSERT_EQ(Status::OK(), executor._lookup_row_data());
+ check_rows(*executor._result_block);
+ // This fails if the executor switches back to the overload that
reloads segments.
+ EXPECT_EQ(_segment_load_calls.load(), 2);
+ EXPECT_EQ(_rowsets[0]->_refs_by_reader.load(), 3);
+ EXPECT_EQ(_rowsets[1]->_refs_by_reader.load(), 1);
+ }
+ for (const auto& segment : retained_segments) {
+ EXPECT_TRUE(segment.expired());
+ }
+ for (const auto& rowset : _rowsets) {
+ EXPECT_EQ(rowset->_refs_by_reader.load(), 0);
+ }
+}
+
+// NOLINTNEXTLINE(readability-function-cognitive-complexity): compare cold and
cached execution using the same request and expected result.
+TEST_P(PointQuerySegmentReuseTest, LookupUpAndRowCacheHit) {
+ ASSERT_EQ(Status::OK(), create_tablet());
+ config::disable_storage_row_cache = false;
+ // MySQL text rows: length-prefixed cells, with 0xfb for SQL NULL.
+ const std::vector<std::string> expected_rows {"\00230\003300\0043000",
"\00210\003100\0041000",
+ "\00240\003400\0044000",
"\00211\003110\373"};
+ for (bool cached : {false, true}) {
+ PointQueryExecutor executor;
+ PTabletKeyLookupResponse response;
+ // Only existing keys: the warm request must not need an index lookup
for a missing key.
+ ASSERT_EQ(Status::OK(), prepare_executor(&executor, {30, 10, 40, 11}));
+ executor._response = &response;
+ _segment_load_calls = 0;
+ ASSERT_EQ(Status::OK(), executor.lookup_up());
+ EXPECT_FALSE(response.empty_batch());
+ TResultBatch batch;
+ auto length = static_cast<uint32_t>(response.row_batch().size());
+ ASSERT_TRUE(deserialize_thrift_msg(
+ reinterpret_cast<const
uint8_t*>(response.row_batch().data()), &length,
+ false, &batch)
+ .ok());
+ EXPECT_EQ(batch.rows, expected_rows);
+ EXPECT_EQ(_segment_load_calls.load(), cached ? 0 : 2);
+ EXPECT_EQ(executor._profile_metrics.row_cache_hits, cached ? 4 : 0);
+ if (cached) {
+ for (const auto& context : executor._row_read_ctxs) {
+ EXPECT_EQ(context._segment, nullptr);
+ EXPECT_EQ(context._rowset_ptr, nullptr);
+ }
+ }
+ evict_segments();
+ }
+}
+
+INSTANTIATE_TEST_SUITE_P(SegmentLayouts, PointQuerySegmentReuseTest,
testing::Bool());
+
} // namespace doris
diff --git a/be/test/storage/segment/segment_cache_test.cpp
b/be/test/storage/segment/segment_cache_test.cpp
index 720aea1c0aa..19463fe2ecf 100644
--- a/be/test/storage/segment/segment_cache_test.cpp
+++ b/be/test/storage/segment/segment_cache_test.cpp
@@ -24,6 +24,7 @@
#include <stdlib.h>
#include <unistd.h>
+#include <atomic>
#include <iostream>
#include <map>
#include <memory>
@@ -37,7 +38,9 @@
#include "core/block/column_with_type_and_name.h"
#include "core/column/column.h"
#include "core/data_type/define_primitive_type.h"
+#include "core/field.h"
#include "core/value/vdatetime_value.h"
+#include "cpp/sync_point.h"
#include "exprs/function/cast/cast_to_date_or_datetime_impl.hpp"
#include "gtest/gtest_pred_impl.h"
#include "io/fs/local_file_system.h"
@@ -50,6 +53,7 @@
#include "storage/iterators.h"
#include "storage/olap_define.h"
#include "storage/options.h"
+#include "storage/row_cursor.h"
#include "storage/rowset/beta_rowset.h"
#include "storage/schema.h"
#include "storage/segment/segment.h"
@@ -60,6 +64,7 @@
#include "storage/tablet_info.h"
#include "storage/task/engine_publish_version_task.h"
#include "storage/txn/txn_manager.h"
+#include "util/defer_op.h"
namespace doris {
class OlapMeta;
@@ -189,6 +194,12 @@ static TDescriptorTable
create_descriptor_tablet_with_sequence_col() {
.column_pos(4)
.nullable(false)
.build());
+ tuple_builder.add_slot(TSlotDescriptorBuilder()
+ .type(TYPE_STRING)
+ .column_name(BeConsts::ROW_STORE_COL)
+ .column_pos(5)
+ .nullable(false)
+ .build());
tuple_builder.build(&dtb);
return dtb.desc_tbl();
@@ -219,6 +230,7 @@ static void generate_data(Block* block, int8_t k1, int16_t
k2, int32_t seq) {
int32_t c5 = seq;
columns[4]->insert_data((const char*)&c5, sizeof(c5));
+ columns[5]->insert_default();
block->set_columns(std::move(columns));
}
@@ -239,7 +251,17 @@ TEST_F(SegmentCacheTest, vec_sequence_col) {
profile = std::make_unique<RuntimeProfile>("CreateTablet");
TCreateTabletReq request;
sleep(20);
- create_tablet_request_with_sequence_col(55555, 270068377, &request);
+ create_tablet_request_with_sequence_col(55555, 270068377, &request, true);
+ request.tablet_schema.__set_store_row_column(true);
+ TColumn row_store_column;
+ row_store_column.column_name = BeConsts::ROW_STORE_COL;
+ row_store_column.column_type.type = TPrimitiveType::STRING;
+ row_store_column.__set_is_key(false);
+ row_store_column.__set_is_allow_null(false);
+ row_store_column.__set_aggregation_type(TAggregationType::REPLACE);
+ row_store_column.__set_default_value("");
+ row_store_column.__set_visible(false);
+ request.tablet_schema.columns.push_back(row_store_column);
Status res = engine_ref->create_tablet(request, profile.get());
ASSERT_TRUE(res.ok());
@@ -325,7 +347,6 @@ TEST_F(SegmentCacheTest, vec_sequence_col) {
ASSERT_EQ(1, tablet->num_rows());
std::vector<segment_v2::SegmentSharedPtr> segments;
- SegmentCacheHandle handle;
BetaRowsetSharedPtr rowset_ptr =
std::dynamic_pointer_cast<BetaRowset>(rowset);
std::mutex lock;
@@ -333,6 +354,7 @@ TEST_F(SegmentCacheTest, vec_sequence_col) {
// Use lock to make sure only this segment can be loaded by
SegmentLoader during this test.
// SegmeentLoader is singleton. Without this lock, multiple segments
will be loaded. Result will be wrong.
std::lock_guard<std::mutex> l(lock);
+ SegmentCacheHandle handle;
// load segments first
int64_t start_size = SegmentLoader::instance()->cache_mem_usage();
res = SegmentLoader::instance()->load_segments(rowset_ptr, &handle,
true, true);
@@ -351,6 +373,83 @@ TEST_F(SegmentCacheTest, vec_sequence_col) {
segment_ptr->meta_mem_usage());
}
+ std::atomic<int> segment_load_calls = 0;
+ auto* sync_point = SyncPoint::get_instance();
+ sync_point->set_call_back("SegmentLoader::load_segments", [&](auto&& args)
{
+ if (try_any_cast<BetaRowset*>(args[0]) == rowset_ptr.get()) {
+ segment_load_calls.fetch_add(1, std::memory_order_relaxed);
+ }
+ });
+ sync_point->enable_processing();
+ Defer clear_sync_point([&]() {
+ sync_point->clear_call_back("SegmentLoader::load_segments");
+ sync_point->disable_processing();
+ });
+
+ RowCursor key_cursor;
+ ASSERT_TRUE(key_cursor
+ .init_scan_key(tablet->tablet_schema(),
+
{Field::create_field<TYPE_TINYINT>(int8_t {123}),
+
Field::create_field<TYPE_SMALLINT>(int16_t {456})})
+ .ok());
+ std::string encoded_key;
+ key_cursor.encode_key_with_padding<true>(&encoded_key,
+
tablet->tablet_schema()->num_key_columns(), true);
+
+ std::vector<RowsetSharedPtr> specified_rowsets {rowset};
+ std::vector<std::unique_ptr<SegmentCacheHandle>>
segment_caches(specified_rowsets.size());
+ RowLocation row_location;
+ RowsetSharedPtr found_rowset;
+ segment_v2::SegmentSharedPtr found_segment;
+ OlapReaderStatistics lookup_stats;
+ res = tablet->lookup_row_key(encoded_key, nullptr, false,
specified_rowsets, &row_location,
+ version.second, segment_caches,
&found_rowset, false, nullptr,
+ &lookup_stats, nullptr, nullptr,
&found_segment);
+ ASSERT_TRUE(res.ok()) << res;
+ ASSERT_EQ(rowset.get(), found_rowset.get());
+ ASSERT_NE(nullptr, segment_caches[0]);
+ ASSERT_EQ(segment_caches[0]->get_segments().front().get(),
found_segment.get());
+ ASSERT_EQ(found_segment->id(), row_location.segment_id);
+
+ const int load_calls_after_key_lookup =
segment_load_calls.load(std::memory_order_relaxed);
+ ASSERT_EQ(1, load_calls_after_key_lookup);
+ std::string row_data;
+ res = tablet->lookup_row_data(encoded_key, row_location, found_segment,
found_rowset,
+ lookup_stats, row_data, false);
+ ASSERT_TRUE(res.ok()) << res;
+ ASSERT_FALSE(row_data.empty());
+ EXPECT_EQ(load_calls_after_key_lookup,
segment_load_calls.load(std::memory_order_relaxed));
+
+ std::string legacy_row_data;
+ res = tablet->lookup_row_data(encoded_key, row_location, found_rowset,
lookup_stats,
+ legacy_row_data, false);
+ ASSERT_TRUE(res.ok()) << res;
+ EXPECT_EQ(row_data, legacy_row_data);
+ EXPECT_EQ(load_calls_after_key_lookup + 1,
segment_load_calls.load(std::memory_order_relaxed));
+
+ // The executor drops its key-lookup cache handles before reading row
data. Even if
+ // the cache entry is then evicted, the returned shared pointer must keep
the segment alive.
+ segment_caches.clear();
+ SegmentLoader::instance()->erase_segments(*rowset->rowset_meta());
+ std::string uncached_row_data;
+ res = tablet->lookup_row_data(encoded_key, row_location, found_segment,
found_rowset,
+ lookup_stats, uncached_row_data, false);
+ ASSERT_TRUE(res.ok()) << res;
+ EXPECT_EQ(row_data, uncached_row_data);
+ EXPECT_EQ(load_calls_after_key_lookup + 1,
segment_load_calls.load(std::memory_order_relaxed));
+
+ // A deleted key must not return a segment for the row-data stage.
+ auto delete_bitmap = std::make_shared<DeleteBitmap>(tablet->tablet_id());
+ delete_bitmap->add({row_location.rowset_id, row_location.segment_id,
version.second},
+ row_location.row_id);
+ segment_caches.resize(specified_rowsets.size());
+ segment_v2::SegmentSharedPtr deleted_segment;
+ res = tablet->lookup_row_key(encoded_key, nullptr, false,
specified_rowsets, &row_location,
+ version.second, segment_caches, nullptr,
false, nullptr,
+ &lookup_stats, delete_bitmap, nullptr,
&deleted_segment);
+ EXPECT_TRUE(res.is<ErrorCode::KEY_NOT_FOUND>()) << res;
+ EXPECT_EQ(nullptr, deleted_segment);
+
res = ((BetaRowset*)rowset.get())->load_segments(&segments);
ASSERT_TRUE(res.ok());
ASSERT_EQ(1, rowset->num_segments());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
