From 8794bf62c3b215d607ed7cd677d7417339bbaef0 Mon Sep 17 00:00:00 2001 From: yangtao555 Date: Fri, 28 Aug 2026 18:58:23 +0800 Subject: [PATCH] [opt](partial update) Select publish-conflict read strategy by update mode --- be/src/storage/mow/historical_row_fetcher.cpp | 5 +- be/src/storage/partial_update_info.cpp | 35 +++-- be/src/storage/partial_update_info.h | 9 +- be/src/storage/tablet/base_tablet.cpp | 109 ++++++++++---- be/src/storage/tablet/base_tablet.h | 6 + .../mow/historical_row_fetcher_test.cpp | 138 ++++++++++++++++++ .../transform/variant_rowstore_test.cpp | 105 +++++++++++++ 7 files changed, 361 insertions(+), 46 deletions(-) diff --git a/be/src/storage/mow/historical_row_fetcher.cpp b/be/src/storage/mow/historical_row_fetcher.cpp index 974099b8c831ff..4b66468a0facac 100644 --- a/be/src/storage/mow/historical_row_fetcher.cpp +++ b/be/src/storage/mow/historical_row_fetcher.cpp @@ -72,8 +72,9 @@ Status HistoricalRowFetcher::read_columns(const TabletSchema& tablet_schema, bool force_read_old_delete_signs, const signed char* __restrict cur_delete_signs) const { return _fixed_plan.read_columns_by_plan(tablet_schema, std::move(cids_to_read), _rsid_to_rowset, - dst_block, read_index, force_read_old_delete_signs, - cur_delete_signs); + dst_block, read_index, + FixedReadPlan::ReadStrategy::PREFER_ROW_STORE, + force_read_old_delete_signs, cur_delete_signs); } } // namespace doris diff --git a/be/src/storage/partial_update_info.cpp b/be/src/storage/partial_update_info.cpp index 7aef98adefd0d1..b26e0f2b671742 100644 --- a/be/src/storage/partial_update_info.cpp +++ b/be/src/storage/partial_update_info.cpp @@ -327,8 +327,8 @@ void FixedReadPlan::prepare_to_read(const RowLocation& row_location, size_t pos) Status FixedReadPlan::read_columns_by_plan( const TabletSchema& tablet_schema, std::vector cids_to_read, const std::map& rsid_to_rowset, Block& block, - std::map* read_index, bool force_read_old_delete_signs, - const signed char* __restrict cur_delete_signs) const { + std::map* read_index, ReadStrategy read_strategy, + bool force_read_old_delete_signs, const signed char* __restrict cur_delete_signs) const { if (force_read_old_delete_signs) { // always read delete sign column from historical data if (block.get_position_by_name(DELETE_SIGN) == -1) { @@ -337,10 +337,11 @@ Status FixedReadPlan::read_columns_by_plan( block.swap(tablet_schema.create_block_by_cids(cids_to_read)); } } - bool has_row_column = tablet_schema.has_row_store_for_all_columns(); + const bool use_row_store = read_strategy == ReadStrategy::PREFER_ROW_STORE && + tablet_schema.has_row_store_for_all_columns(); std::optional mutable_columns_guard; MutableColumns* mutable_columns = nullptr; - if (!has_row_column) { + if (!use_row_store) { mutable_columns_guard.emplace(block); mutable_columns = &mutable_columns_guard->mutable_columns(); } @@ -357,7 +358,7 @@ Status FixedReadPlan::read_columns_by_plan( rids.emplace_back(rid); (*read_index)[static_cast(pos)] = read_idx++; } - if (has_row_column) { + if (use_row_store) { auto st = BaseTablet::fetch_value_through_row_column( rowset_iter->second, tablet_schema, segment_id, rids, cids_to_read, block); if (!st.ok()) { @@ -366,16 +367,12 @@ Status FixedReadPlan::read_columns_by_plan( } continue; } - for (size_t cid = 0; cid < mutable_columns->size(); ++cid) { - TabletColumn tablet_column = tablet_schema.column(cids_to_read[cid]); - auto st = doris::BaseTablet::fetch_value_by_rowids(rowset_iter->second, segment_id, - rids, tablet_column, - (*mutable_columns)[cid]); - // set read value to output block - if (!st.ok()) { - LOG(WARNING) << "failed to fetch value"; - return st; - } + auto st = BaseTablet::fetch_values_by_rowids(rowset_iter->second, tablet_schema, + segment_id, rids, cids_to_read, + *mutable_columns); + if (!st.ok()) { + LOG(WARNING) << "failed to fetch values by rowids"; + return st; } } } @@ -433,7 +430,8 @@ Status FixedReadPlan::fill_missing_columns( // segment pos to write -> rowid to read in old_value_block std::map read_index; RETURN_IF_ERROR(read_columns_by_plan(tablet_schema, missing_cids, rsid_to_rowset, - old_value_block, &read_index, true, nullptr)); + old_value_block, &read_index, + ReadStrategy::PREFER_ROW_STORE, true, nullptr)); const auto* old_delete_sign_column_data = BaseTablet::get_delete_sign_column_data(old_value_block); @@ -1085,8 +1083,9 @@ Status BlockAggregator::fill_sequence_column(Block* block, size_t num_rows, auto seq_col_block = _tablet_schema.create_block_by_cids(cids); auto tmp_block = _tablet_schema.create_block_by_cids(cids); std::map read_index; - RETURN_IF_ERROR(read_plan.read_columns_by_plan(_tablet_schema, cids, _fetcher.pinned_rowsets(), - seq_col_block, &read_index, false)); + RETURN_IF_ERROR(read_plan.read_columns_by_plan( + _tablet_schema, cids, _fetcher.pinned_rowsets(), seq_col_block, &read_index, + FixedReadPlan::ReadStrategy::PREFER_ROW_STORE, false)); auto new_seq_col_ptr = tmp_block.get_by_position(0).column->assert_mutable(); const auto& old_seq_col_ptr = *seq_col_block.get_by_position(0).column; diff --git a/be/src/storage/partial_update_info.h b/be/src/storage/partial_update_info.h index 01b4bb3f30221a..250885f1d14e4c 100644 --- a/be/src/storage/partial_update_info.h +++ b/be/src/storage/partial_update_info.h @@ -127,6 +127,13 @@ struct RidAndPos { class FixedReadPlan { public: + enum class ReadStrategy { + // Use the full row-store column when available; otherwise read physical columns. + PREFER_ROW_STORE, + // Read only the requested physical columns, even when a full row-store column exists. + COLUMN_STORE, + }; + bool empty() const; void clear() { plan.clear(); } void prepare_to_read(const RowLocation& row_location, size_t pos); @@ -134,7 +141,7 @@ class FixedReadPlan { std::vector cids_to_read, const std::map& rsid_to_rowset, Block& block, std::map* read_index, - bool force_read_old_delete_signs, + ReadStrategy read_strategy, bool force_read_old_delete_signs, const signed char* __restrict cur_delete_signs = nullptr) const; Status fill_missing_columns(const segment_v2::HistoricalRowRetrieverContext& historical_context, const std::map& rsid_to_rowset, diff --git a/be/src/storage/tablet/base_tablet.cpp b/be/src/storage/tablet/base_tablet.cpp index 3c861801f8a353..0e57b669df6b15 100644 --- a/be/src/storage/tablet/base_tablet.cpp +++ b/be/src/storage/tablet/base_tablet.cpp @@ -79,24 +79,31 @@ bvar::LatencyRecorder g_tablet_update_delete_bitmap_latency("doris_pk", "update_ static bvar::Adder g_total_tablet_num("doris_total_tablet_num"); -Status _get_segment_column_iterator(const BetaRowsetSharedPtr& rowset, uint32_t segid, - const TabletColumn& target_column, - SegmentCacheHandle* segment_cache_handle, - std::unique_ptr* column_iterator, - OlapReaderStatistics* stats, - const io::IOContext* input_io_ctx = nullptr) { +Status _load_segment(const BetaRowsetSharedPtr& rowset, uint32_t segid, + SegmentCacheHandle* segment_cache_handle, + segment_v2::SegmentSharedPtr* segment, OlapReaderStatistics* stats, + const io::IOContext* input_io_ctx = nullptr) { RETURN_IF_ERROR(SegmentLoader::instance()->load_segments(rowset, segment_cache_handle, true, false, stats, input_io_ctx)); - // find segment - auto it = std::find_if( - segment_cache_handle->get_segments().begin(), - segment_cache_handle->get_segments().end(), - [&segid](const segment_v2::SegmentSharedPtr& seg) { return seg->id() == segid; }); + auto it = std::find_if(segment_cache_handle->get_segments().begin(), + segment_cache_handle->get_segments().end(), + [segid](const segment_v2::SegmentSharedPtr& candidate) { + return candidate->id() == segid; + }); if (it == segment_cache_handle->get_segments().end()) { - return Status::NotFound(fmt::format("rowset {} 's segemnt not found, seg_id {}", + return Status::NotFound(fmt::format("rowset {}'s segment not found, seg_id {}", rowset->rowset_id().to_string(), segid)); } - segment_v2::SegmentSharedPtr segment = *it; + *segment = *it; + TEST_SYNC_POINT_CALLBACK("BaseTablet::_load_segment", rowset.get(), &segid); + return Status::OK(); +} + +Status _init_segment_column_iterator(const segment_v2::SegmentSharedPtr& segment, + const TabletColumn& target_column, + std::unique_ptr* column_iterator, + OlapReaderStatistics* stats, + const io::IOContext* input_io_ctx = nullptr) { StorageReadOptions opts; opts.stats = stats; if (input_io_ctx != nullptr) { @@ -116,6 +123,19 @@ Status _get_segment_column_iterator(const BetaRowsetSharedPtr& rowset, uint32_t return Status::OK(); } +Status _get_segment_column_iterator(const BetaRowsetSharedPtr& rowset, uint32_t segid, + const TabletColumn& target_column, + SegmentCacheHandle* segment_cache_handle, + std::unique_ptr* column_iterator, + OlapReaderStatistics* stats, + const io::IOContext* input_io_ctx = nullptr) { + segment_v2::SegmentSharedPtr segment; + RETURN_IF_ERROR( + _load_segment(rowset, segid, segment_cache_handle, &segment, stats, input_io_ctx)); + return _init_segment_column_iterator(segment, target_column, column_iterator, stats, + input_io_ctx); +} + } // namespace extern MetricPrototype METRIC_query_scan_bytes; @@ -866,9 +886,9 @@ Status BaseTablet::calc_segment_delete_bitmap(RowsetSharedPtr rowset, std::map rsid_to_row_binlog { {row_binlog_rowset->rowset_id(), row_binlog_rowset}}; std::map read_index; - RETURN_IF_ERROR(read_plan_lsn.read_columns_by_plan(*row_binlog_schema, lsn_cids, - rsid_to_row_binlog, lsn_block, - &read_index, false)); + RETURN_IF_ERROR(read_plan_lsn.read_columns_by_plan( + *row_binlog_schema, lsn_cids, rsid_to_row_binlog, lsn_block, &read_index, + FixedReadPlan::ReadStrategy::PREFER_ROW_STORE, false)); } std::vector sort_perm; @@ -976,6 +996,7 @@ Status BaseTablet::fetch_value_through_row_column(RowsetSharedPtr input_rowset, BetaRowsetSharedPtr rowset = std::static_pointer_cast(input_rowset); CHECK(rowset); + TEST_SYNC_POINT_CALLBACK("BaseTablet::fetch_value_through_row_column", rowset.get()); CHECK(tablet_schema.has_row_store_for_all_columns()); SegmentCacheHandle segment_cache_handle; std::unique_ptr column_iterator; @@ -1005,6 +1026,37 @@ Status BaseTablet::fetch_value_through_row_column(RowsetSharedPtr input_rowset, return Status::OK(); } +Status BaseTablet::fetch_values_by_rowids(RowsetSharedPtr input_rowset, + const TabletSchema& tablet_schema, uint32_t segid, + const std::vector& rowids, + const std::vector& cids, + MutableColumns& dst_columns) { + MonotonicStopWatch watch; + watch.start(); + Defer _defer([&]() { + LOG_EVERY_N(INFO, 500) << "fetch_values_by_rowids, cost(us):" << watch.elapsed_time() / 1000 + << ", row_batch_size:" << rowids.size() + << ", column_count:" << cids.size(); + }); + + BetaRowsetSharedPtr rowset = std::static_pointer_cast(input_rowset); + CHECK(rowset); + TEST_SYNC_POINT_CALLBACK("BaseTablet::fetch_values_by_rowids", rowset.get(), &cids); + CHECK_EQ(cids.size(), dst_columns.size()); + SegmentCacheHandle segment_cache_handle; + OlapReaderStatistics stats; + segment_v2::SegmentSharedPtr segment; + RETURN_IF_ERROR(_load_segment(rowset, segid, &segment_cache_handle, &segment, &stats)); + for (size_t i = 0; i < cids.size(); ++i) { + std::unique_ptr column_iterator; + RETURN_IF_ERROR(_init_segment_column_iterator(segment, tablet_schema.column(cids[i]), + &column_iterator, &stats)); + RETURN_IF_ERROR( + column_iterator->read_by_rowids(rowids.data(), rowids.size(), dst_columns[i])); + } + return Status::OK(); +} + Status BaseTablet::fetch_value_by_rowids(RowsetSharedPtr input_rowset, uint32_t segid, const std::vector& rowids, const TabletColumn& tablet_column, MutableColumnPtr& dst) { @@ -1085,10 +1137,15 @@ Status BaseTablet::generate_new_block_for_partial_update( // rowid in the final block(start from 0, increase continuously) -> rowid to read in update_block std::map read_index_update; - // read current rowset first, if a row in the current rowset has delete sign mark - // we don't need to read values from old block + // Fixed partial updates only need their explicit update projection, so reading the full + // row-store JSONB adds an unnecessary full-row allocation. UPSERT rewrites keep the row-store + // path because it preserves the pre-VariantParse row representation. + const auto update_read_strategy = partial_update_info->is_fixed_partial_update() + ? FixedReadPlan::ReadStrategy::COLUMN_STORE + : FixedReadPlan::ReadStrategy::PREFER_ROW_STORE; RETURN_IF_ERROR(read_plan_update.read_columns_by_plan( - *rowset_schema, update_cids, rsid_to_rowset, update_block, &read_index_update, false)); + *rowset_schema, update_cids, rsid_to_rowset, update_block, &read_index_update, + update_read_strategy, false)); size_t update_rows = read_index_update.size(); for (auto i = 0; i < update_cids.size(); ++i) { for (auto idx = 0; idx < update_rows; ++idx) { @@ -1106,9 +1163,9 @@ Status BaseTablet::generate_new_block_for_partial_update( // rowid in the final block(start from 0, increase, may not continuous becasue we skip to read some rows) -> rowid to read in old_block std::map read_index_old; - RETURN_IF_ERROR(read_plan_ori.read_columns_by_plan(*rowset_schema, missing_cids, rsid_to_rowset, - old_block, &read_index_old, true, - new_block_delete_signs)); + RETURN_IF_ERROR(read_plan_ori.read_columns_by_plan( + *rowset_schema, missing_cids, rsid_to_rowset, old_block, &read_index_old, + FixedReadPlan::ReadStrategy::PREFER_ROW_STORE, true, new_block_delete_signs)); size_t old_rows = read_index_old.size(); const auto* __restrict old_block_delete_signs = get_delete_sign_column_data(old_block, old_rows); @@ -1247,8 +1304,9 @@ Status BaseTablet::generate_new_block_for_flexible_partial_update( // 1. read the current rowset first, if a row in the current rowset has delete sign mark // we don't need to read values from old block for that row - RETURN_IF_ERROR(read_plan_update.read_columns_by_plan(*rowset_schema, all_cids, rsid_to_rowset, - update_block, &read_index_update, true)); + RETURN_IF_ERROR(read_plan_update.read_columns_by_plan( + *rowset_schema, all_cids, rsid_to_rowset, update_block, &read_index_update, + FixedReadPlan::ReadStrategy::PREFER_ROW_STORE, true)); size_t update_rows = read_index_update.size(); // TODO(bobhan1): add the delete sign optimazation here @@ -1263,7 +1321,8 @@ Status BaseTablet::generate_new_block_for_flexible_partial_update( // rowid in the final block(start from 0, increase, may not continuous becasue we skip to read some rows) -> rowid to read in old_block std::map read_index_old; RETURN_IF_ERROR(read_plan_ori.read_columns_by_plan( - *rowset_schema, non_sort_key_cids, rsid_to_rowset, old_block, &read_index_old, true)); + *rowset_schema, non_sort_key_cids, rsid_to_rowset, old_block, &read_index_old, + FixedReadPlan::ReadStrategy::PREFER_ROW_STORE, true)); size_t old_rows = read_index_old.size(); DCHECK(update_rows == old_rows); const auto* __restrict old_block_delete_signs = diff --git a/be/src/storage/tablet/base_tablet.h b/be/src/storage/tablet/base_tablet.h index 290722a05d453f..b01b18bf109a19 100644 --- a/be/src/storage/tablet/base_tablet.h +++ b/be/src/storage/tablet/base_tablet.h @@ -251,6 +251,12 @@ class BaseTablet : public std::enable_shared_from_this { const std::vector& rowids, const std::vector& cids, Block& block); + static Status fetch_values_by_rowids(RowsetSharedPtr input_rowset, + const TabletSchema& tablet_schema, uint32_t segid, + const std::vector& rowids, + const std::vector& cids, + MutableColumns& dst_columns); + static Status fetch_value_by_rowids(RowsetSharedPtr input_rowset, uint32_t segid, const std::vector& rowids, const TabletColumn& tablet_column, MutableColumnPtr& dst); diff --git a/be/test/storage/mow/historical_row_fetcher_test.cpp b/be/test/storage/mow/historical_row_fetcher_test.cpp index 2b8d1c7da9ec8c..be8e7d03d54ab3 100644 --- a/be/test/storage/mow/historical_row_fetcher_test.cpp +++ b/be/test/storage/mow/historical_row_fetcher_test.cpp @@ -23,6 +23,7 @@ #include #include +#include "cpp/sync_point.h" #include "storage/mow/mow_transform_test_base.h" #include "storage/partial_update_info.h" @@ -80,6 +81,143 @@ TEST_F(HistoricalRowFetcherTest, ReadColumnsReturnsThePlannedRows) { EXPECT_EQ(read_int(old_values, 0, read_index[1]), 11); // dst 1 <- row 0 } +// A full row-store schema can still read a narrow projection directly from physical columns. +// Both sources must preserve planned row order and delete-sign semantics. +TEST_F(HistoricalRowFetcherTest, FixedPlanColumnStoreReadMatchesRowStore) { + auto schema = create_row_store_schema(); + TabletSharedPtr tablet; + auto rowset = write_rowset(schema, 5002, 2, {{1, 11, 0, 0}, {2, 22, 0, 1}}, &tablet); + std::map rowsets {{rowset->rowset_id(), rowset}}; + + FixedReadPlan read_plan; + read_plan.prepare_to_read(RowLocation {rowset->rowset_id(), 0, 1}, /*dst_pos=*/0); + read_plan.prepare_to_read(RowLocation {rowset->rowset_id(), 0, 0}, /*dst_pos=*/1); + + const std::vector cids {0, 1}; + auto column_store_block = schema->create_block_by_cids(cids); + auto row_store_block = schema->create_block_by_cids(cids); + std::map column_store_read_index; + std::map row_store_read_index; + + ASSERT_TRUE(read_plan + .read_columns_by_plan(*schema, cids, rowsets, column_store_block, + &column_store_read_index, + FixedReadPlan::ReadStrategy::COLUMN_STORE, + /*force_read_old_delete_signs=*/true) + .ok()); + ASSERT_TRUE(read_plan + .read_columns_by_plan(*schema, cids, rowsets, row_store_block, + &row_store_read_index, + FixedReadPlan::ReadStrategy::PREFER_ROW_STORE, + /*force_read_old_delete_signs=*/true) + .ok()); + + EXPECT_EQ(column_store_read_index, row_store_read_index); + EXPECT_EQ(column_store_block.dump_data(), row_store_block.dump_data()); + ASSERT_EQ(column_store_block.rows(), 2); + EXPECT_EQ(read_int(column_store_block, 0, 0), 2); + EXPECT_EQ(read_int(column_store_block, 0, 1), 1); + EXPECT_EQ(read_int(column_store_block, 1, 0), 22); + EXPECT_EQ(read_int(column_store_block, 1, 1), 11); + EXPECT_EQ(read_tinyint(column_store_block, 2, 0), 1); + EXPECT_EQ(read_tinyint(column_store_block, 2, 1), 0); +} + +// The production publish-conflict rebuild must read the current rowset's update projection from +// physical columns, while the historical missing projection still uses the row store. The batch +// physical read must load/locate each planned segment once, not once per update column. +TEST_F(HistoricalRowFetcherTest, PublishConflictUsesBatchedColumnStoreForCurrentProjection) { + auto schema = create_row_store_schema(/*has_seq=*/true); + TabletSharedPtr tablet; + auto current_rowset = write_rowset(schema, 5003, 3, {{1, 101, 5, 0}}, &tablet); + auto historical_rowset = write_rowset(schema, 5004, 2, {{1, 11, 3, 0}}, &tablet); + + auto partial_update_info = std::make_shared(); + ASSERT_TRUE(partial_update_info + ->init(kTabletId, /*txn_id=*/1, *schema, + UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS, + PartialUpdateNewRowPolicyPB::APPEND, {"k", "v"}, + /*is_strict_mode=*/false, /*timestamp_ms=*/0, + /*nano_seconds=*/0, "UTC", "") + .ok()); + + FixedReadPlan read_plan_update; + read_plan_update.prepare_to_read( + RowLocation {current_rowset->rowset_id(), /*segment_id=*/0, /*row_id=*/0}, + /*dst_pos=*/0); + FixedReadPlan read_plan_historical; + read_plan_historical.prepare_to_read( + RowLocation {historical_rowset->rowset_id(), /*segment_id=*/0, /*row_id=*/0}, + /*dst_pos=*/0); + std::map rowsets { + {current_rowset->rowset_id(), current_rowset}, + {historical_rowset->rowset_id(), historical_rowset}}; + + int current_segment_loads = 0; + int historical_segment_loads = 0; + int current_batch_reads = 0; + int current_row_store_reads = 0; + int historical_row_store_reads = 0; + size_t current_batch_column_count = 0; + auto* sync_point = SyncPoint::get_instance(); + SyncPoint::CallbackGuard load_segment_guard; + SyncPoint::CallbackGuard batch_read_guard; + SyncPoint::CallbackGuard row_store_read_guard; + sync_point->set_call_back( + "BaseTablet::_load_segment", + [&](auto&& args) { + auto* rowset = try_any_cast(args[0]); + if (rowset->rowset_id() == current_rowset->rowset_id()) { + ++current_segment_loads; + } else if (rowset->rowset_id() == historical_rowset->rowset_id()) { + ++historical_segment_loads; + } + }, + &load_segment_guard); + sync_point->set_call_back( + "BaseTablet::fetch_values_by_rowids", + [&](auto&& args) { + auto* rowset = try_any_cast(args[0]); + if (rowset->rowset_id() == current_rowset->rowset_id()) { + ++current_batch_reads; + current_batch_column_count = + try_any_cast*>(args[1])->size(); + } + }, + &batch_read_guard); + sync_point->set_call_back( + "BaseTablet::fetch_value_through_row_column", + [&](auto&& args) { + auto* rowset = try_any_cast(args[0]); + if (rowset->rowset_id() == current_rowset->rowset_id()) { + ++current_row_store_reads; + } else if (rowset->rowset_id() == historical_rowset->rowset_id()) { + ++historical_row_store_reads; + } + }, + &row_store_read_guard); + sync_point->enable_processing(); + + auto output_block = schema->create_block(); + auto st = BaseTablet::generate_new_block_for_partial_update( + schema, partial_update_info.get(), read_plan_historical, read_plan_update, rowsets, + &output_block); + sync_point->disable_processing(); + + ASSERT_TRUE(st.ok()) << st; + ASSERT_EQ(output_block.rows(), 1); + EXPECT_EQ(read_int(output_block, 0, 0), 1); + EXPECT_EQ(read_int(output_block, 1, 0), 101); + EXPECT_EQ(read_int(output_block, 2, 0), 3); + EXPECT_EQ(read_tinyint(output_block, 3, 0), 0); + EXPECT_EQ(current_batch_reads, 1); + EXPECT_EQ(current_batch_column_count, 2); + EXPECT_EQ(current_row_store_reads, 0); + EXPECT_EQ(historical_row_store_reads, 1); + EXPECT_EQ(current_segment_loads, 1); + EXPECT_EQ(historical_segment_loads, 1); +} + // The fixed partial update fill: rows flagged for a historical read take the old value, rows // flagged use-default take the column default. TEST_F(HistoricalRowFetcherTest, FillMissingColumnsMixesHistoryAndDefaults) { diff --git a/be/test/storage/transform/variant_rowstore_test.cpp b/be/test/storage/transform/variant_rowstore_test.cpp index 94da32c6b767a0..c7dfd458d78cc0 100644 --- a/be/test/storage/transform/variant_rowstore_test.cpp +++ b/be/test/storage/transform/variant_rowstore_test.cpp @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -33,8 +34,10 @@ #include "core/column/column_variant.h" #include "core/data_type_serde/data_type_serde.h" #include "core/field.h" +#include "cpp/sync_point.h" #include "storage/mow/mow_transform_test_base.h" #include "storage/rowset/rowset_writer_context.h" +#include "storage/tablet/base_tablet.h" #include "storage/transform/block_transform.h" #include "testutil/variant_util.h" #include "util/jsonb/serialize.h" @@ -510,6 +513,108 @@ TEST_F(VariantRowStoreTest, RowStoreSnapshotsVariantBeforeParse) { EXPECT_EQ(stored_variant.find(R"("flag":1)"), std::string::npos) << stored_variant; } +// UPSERT publish-conflict rewrites share generate_new_block_for_partial_update with fixed partial +// updates, but must keep reading the current row through row-store. Direct writes snapshot raw +// Variant JSON before VariantParse, so switching this path to physical columns would silently turn +// JSON boolean `true` into the normalized physical representation `1` in the rebuilt row store. +TEST_F(VariantRowStoreTest, UpsertPublishConflictPreservesRawVariantRowStore) { + auto schema = create_variant_row_store_schema(); + TabletSharedPtr tablet; + auto current_rowset = write_rowset_block( + schema, 8201, 2, + [&](Block& block) { + int32_t key = 1; + int8_t delete_sign = 0; + block.get_by_position(0).column->assert_mutable()->insert_data( + reinterpret_cast(&key), sizeof(key)); + insert_variant_json(block, 1, R"({"flag":true})"); + block.get_by_position(2).column->assert_mutable()->insert_data( + reinterpret_cast(&delete_sign), sizeof(delete_sign)); + block.get_by_position(3).column->assert_mutable()->insert_default(); + }, + &tablet); + + Block persisted; + ASSERT_TRUE(read_rowset(current_rowset, schema, &persisted).ok()); + EXPECT_NE(variant_row_json(persisted, 1, 0).find(R"("flag":1)"), std::string::npos); + const auto& persisted_row_store = + assert_cast(*persisted.get_by_position(3).column); + Block decoded_before_rewrite = + decode_row_store_cell(schema, persisted_row_store.get_data_at(0)); + EXPECT_NE(variant_row_json(decoded_before_rewrite, 1, 0).find(R"("flag":true)"), + std::string::npos); + + auto partial_update_info = std::make_shared(); + ASSERT_TRUE(partial_update_info + ->init(kTabletId, /*txn_id=*/1, *schema, UniqueKeyUpdateModePB::UPSERT, + PartialUpdateNewRowPolicyPB::APPEND, {}, /*is_strict_mode=*/false, + /*timestamp_ms=*/0, /*nano_seconds=*/0, "UTC", "") + .ok()); + partial_update_info->update_cids.resize(schema->num_columns()); + std::iota(partial_update_info->update_cids.begin(), partial_update_info->update_cids.end(), 0); + + FixedReadPlan read_plan_update; + read_plan_update.prepare_to_read( + RowLocation {current_rowset->rowset_id(), /*segment_id=*/0, /*row_id=*/0}, + /*dst_pos=*/0); + FixedReadPlan empty_historical_plan; + std::map rowsets {{current_rowset->rowset_id(), current_rowset}}; + + int row_store_reads = 0; + int batch_column_reads = 0; + auto* sync_point = SyncPoint::get_instance(); + SyncPoint::CallbackGuard row_store_read_guard; + SyncPoint::CallbackGuard batch_read_guard; + sync_point->set_call_back( + "BaseTablet::fetch_value_through_row_column", + [&](auto&& args) { + auto* rowset = try_any_cast(args[0]); + if (rowset->rowset_id() == current_rowset->rowset_id()) { + ++row_store_reads; + } + }, + &row_store_read_guard); + sync_point->set_call_back( + "BaseTablet::fetch_values_by_rowids", + [&](auto&& args) { + auto* rowset = try_any_cast(args[0]); + if (rowset->rowset_id() == current_rowset->rowset_id()) { + ++batch_column_reads; + } + }, + &batch_read_guard); + sync_point->enable_processing(); + + auto rebuilt = schema->create_block(); + auto rebuild_status = BaseTablet::generate_new_block_for_partial_update( + schema, partial_update_info.get(), empty_historical_plan, read_plan_update, rowsets, + &rebuilt); + sync_point->disable_processing(); + + ASSERT_TRUE(rebuild_status.ok()) << rebuild_status; + EXPECT_EQ(row_store_reads, 1); + EXPECT_EQ(batch_column_reads, 0); + EXPECT_NE(variant_row_json(rebuilt, 1, 0).find(R"("flag":true)"), std::string::npos); + + RowsetWriterContext transient_context = direct_rwc(schema); + transient_context.partial_update_info = partial_update_info; + transient_context.is_transient_rowset_writer = true; + auto chain = build_transform_chain(transient_context); + EXPECT_EQ(chain.stage_names(), + (std::vector {"Validate", "RowStoreFill", "VariantParse"})); + auto transform_context = exec_ctx(schema, &transient_context); + ASSERT_TRUE(chain.apply(transform_context, &rebuilt).ok()); + ASSERT_TRUE(materialize_derived_columns(transform_context.derived_column, &rebuilt).ok()); + + EXPECT_NE(variant_row_json(rebuilt, 1, 0).find(R"("flag":1)"), std::string::npos); + const auto& rebuilt_row_store = + assert_cast(*rebuilt.get_by_position(3).column); + Block decoded_after_rewrite = decode_row_store_cell(schema, rebuilt_row_store.get_data_at(0)); + const std::string stored_variant = variant_row_json(decoded_after_rewrite, 1, 0); + EXPECT_NE(stored_variant.find(R"("flag":true)"), std::string::npos) << stored_variant; + EXPECT_EQ(stored_variant.find(R"("flag":1)"), std::string::npos) << stored_variant; +} + // Drive the registered generator directly as the vertical writer does -- a // fresh clone_empty() dst per batch, max_bytes huge, batch_rows = 2. Over 5 // rows this yields 2,2,1 and walks pos 0->2->4->5, and the concatenation