diff --git a/be/src/service/point_query_executor.cpp b/be/src/service/point_query_executor.cpp index 560fb3ca1e6d0e..5fc85aebe8c756 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(); - 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()) { 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 91a8632b00ac9f..285b77c6cd2f27 100644 --- a/be/src/service/point_query_executor.h +++ b/be/src/service/point_query_executor.h @@ -335,6 +335,8 @@ class PointQueryExecutor { // rowset will be aquired during read // and released after used std::unique_ptr _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 c7f4a1a6d3b2ef..8585267a4206dc 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 eba76fe573b28f..ebb55462a1ad76 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 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 rowids {static_cast(row_location.row_id)}; @@ -523,7 +548,8 @@ Status BaseTablet::lookup_row_key(const Slice& encoded_key, TabletSchema* latest std::vector>& 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 e570fc79944496..c18247ea81f58f 100644 --- a/be/src/storage/tablet/base_tablet.h +++ b/be/src/storage/tablet/base_tablet.h @@ -182,6 +182,13 @@ class BaseTablet : public std::enable_shared_from_this { 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 @@ class BaseTablet : public std::enable_shared_from_this { 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 @@ class BaseTablet : public std::enable_shared_from_this { 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 74bd7606e3905e..3e2269b2ae60da 100644 --- a/be/test/service/point_query_exector_test.cpp +++ b/be/test/service/point_query_exector_test.cpp @@ -21,15 +21,31 @@ #include #include +#include + +#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 { +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(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(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(); + meta->init_from_pb(meta_pb); + if (GetParam()) { + _cloud_engine = std::make_unique(EngineOptions {}); + _tablet = std::make_shared(*_cloud_engine, meta); + } else { + _tablet = std::make_shared(*_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(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>& 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(key)); + columns[1]->insert(Field::create_field(key * 10)); + if (key == 11) { + columns[2]->insert_default(); + } else { + columns[2]->insert(Field::create_field(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 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& 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 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(); + 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 expected_keys {30, 10, 40, 11}; + const auto& v2 = assert_cast(*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 _engine; + std::unique_ptr _cloud_engine; + BaseTabletSPtr _tablet; + std::vector _rowsets; + std::atomic _segment_load_calls = 0; + std::unique_ptr _row_cache; + std::unique_ptr _segment_loader; + std::unique_ptr _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> 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 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(response.row_batch().size()); + ASSERT_TRUE(deserialize_thrift_msg( + reinterpret_cast(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 720aea1c0aa110..19463fe2ecf160 100644 --- a/be/test/storage/segment/segment_cache_test.cpp +++ b/be/test/storage/segment/segment_cache_test.cpp @@ -24,6 +24,7 @@ #include #include +#include #include #include #include @@ -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("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 segments; - SegmentCacheHandle handle; BetaRowsetSharedPtr rowset_ptr = std::dynamic_pointer_cast(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 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 segment_load_calls = 0; + auto* sync_point = SyncPoint::get_instance(); + sync_point->set_call_back("SegmentLoader::load_segments", [&](auto&& args) { + if (try_any_cast(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(int8_t {123}), + Field::create_field(int16_t {456})}) + .ok()); + std::string encoded_key; + key_cursor.encode_key_with_padding(&encoded_key, + tablet->tablet_schema()->num_key_columns(), true); + + std::vector specified_rowsets {rowset}; + std::vector> 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(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()) << res; + EXPECT_EQ(nullptr, deleted_segment); + res = ((BetaRowset*)rowset.get())->load_segments(&segments); ASSERT_TRUE(res.ok()); ASSERT_EQ(1, rowset->num_segments());