This is an automated email from the ASF dual-hosted git repository.
gavinchou 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 7b7cea2e98d [opt](partial update) Select publish-conflict read
strategy by update mode (#67295)
7b7cea2e98d is described below
commit 7b7cea2e98d68f1019490c199645c16b120b4436
Author: foxtail463 <[email protected]>
AuthorDate: Wed Sep 16 20:18:14 2026 +0800
[opt](partial update) Select publish-conflict read strategy by update mode
(#67295)
# Problem Summary:
FixedReadPlan previously preferred the row store whenever a complete
row-store column was available, regardless of the caller’s update mode.
This caused fixed partial-update conflict reconstruction to materialize
and decode the complete JSONB row when only the updated columns were
required. However, globally switching to physical-column reads would
break UPSERT semantics for values such as raw Variant data.
Physical-column reads also repeated segment setup for each requested
column.
# Solution:
Introduce an explicit read strategy selected according to the update
mode. Fixed partial updates read their current projection from physical
columns, while UPSERT and historical, flexible, sequence, and row-binlog
paths retain row-store-first behavior. Batch physical-column reads so
each planned segment is loaded once and column iterators are created and
released sequentially.
# benchmark:
A RELEASE warm-cache microbenchmark compared row-store and
physical-column reads on an 8,192-row segment with 100 mixed STRING/INT
value columns.
For 8,192 conflict rows, physical-column reads improved latency by
30.21x, 7.59x, and 5.59x for 2-, 50-, and 100-column projections, while
reducing tracked peak memory by 96.7%, 52.9%, and 35.8%, respectively.
For sparse 64-row conflicts, physical columns were 5.10x faster for 2
columns, but slower for wide 50- and 100-column projections because
per-column setup dominated. This is a focused read-path microbenchmark,
not an end-to-end publish throughput test.
Co-authored-by: yangtao555 <[email protected]>
---
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 | 116 ++++++++---
be/src/storage/tablet/base_tablet.h | 6 +
.../storage/mow/historical_row_fetcher_test.cpp | 138 +++++++++++++
.../storage/transform/variant_rowstore_test.cpp | 227 +++++++++++++++++++++
7 files changed, 490 insertions(+), 46 deletions(-)
diff --git a/be/src/storage/mow/historical_row_fetcher.cpp
b/be/src/storage/mow/historical_row_fetcher.cpp
index 974099b8c83..4b66468a0fa 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 535021aa93b..c8fd16811b0 100644
--- a/be/src/storage/partial_update_info.cpp
+++ b/be/src/storage/partial_update_info.cpp
@@ -356,8 +356,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<uint32_t> cids_to_read,
const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset, Block&
block,
- std::map<uint32_t, uint32_t>* read_index, bool
force_read_old_delete_signs,
- const signed char* __restrict cur_delete_signs) const {
+ std::map<uint32_t, uint32_t>* 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) {
@@ -366,10 +366,11 @@ Status FixedReadPlan::read_columns_by_plan(
block.swap(tablet_schema.create_storage_block(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<Block::ScopedMutableColumns> 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();
}
@@ -386,7 +387,7 @@ Status FixedReadPlan::read_columns_by_plan(
rids.emplace_back(rid);
(*read_index)[static_cast<uint32_t>(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()) {
@@ -395,16 +396,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;
}
}
}
@@ -462,7 +459,8 @@ Status FixedReadPlan::fill_missing_columns(
// segment pos to write -> rowid to read in old_value_block
std::map<uint32_t, uint32_t> 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);
@@ -1158,8 +1156,9 @@ Status BlockAggregator::fill_sequence_column(Block*
block, size_t num_rows,
auto seq_col_block = _tablet_schema.create_storage_block(cids);
auto tmp_block = _tablet_schema.create_storage_block(cids);
std::map<uint32_t, uint32_t> 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 4b8ab99840f..f5341c4f559 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 @@ public:
std::vector<uint32_t> cids_to_read,
const std::map<RowsetId, RowsetSharedPtr>&
rsid_to_rowset,
Block& block, std::map<uint32_t, uint32_t>*
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<RowsetId, RowsetSharedPtr>&
rsid_to_rowset,
diff --git a/be/src/storage/tablet/base_tablet.cpp
b/be/src/storage/tablet/base_tablet.cpp
index 2084b3abbb3..fcdd19416c1 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<size_t> 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<segment_v2::ColumnIterator>* 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<segment_v2::ColumnIterator>* 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<segment_v2::ColumnIterator>* 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;
@@ -865,9 +885,9 @@ Status
BaseTablet::calc_segment_delete_bitmap(RowsetSharedPtr rowset,
std::map<RowsetId, RowsetSharedPtr> rsid_to_row_binlog {
{row_binlog_rowset->rowset_id(), row_binlog_rowset}};
std::map<uint32_t, uint32_t> 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<uint32_t> sort_perm;
@@ -974,6 +994,7 @@ Status
BaseTablet::fetch_value_through_row_column(RowsetSharedPtr input_rowset,
BetaRowsetSharedPtr rowset =
std::static_pointer_cast<BetaRowset>(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<segment_v2::ColumnIterator> column_iterator;
@@ -1003,6 +1024,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<uint32_t>& rowids,
+ const std::vector<uint32_t>& 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<BetaRowset>(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<segment_v2::ColumnIterator> 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<uint32_t>& rowids,
const TabletColumn& tablet_column,
MutableColumnPtr& dst) {
@@ -1083,10 +1135,22 @@ 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<uint32_t, uint32_t> 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
+ // Row-store and physical Variant columns are not always
representation-equivalent. A typed
+ // Variant path can be coerced by the column writer after RowStoreFill
(for example, string
+ // "001" becomes integer 1), so rebuilding a conflicting row from physical
Variant columns
+ // would make its row-store value depend on whether a publish conflict
occurred. Keep the
+ // row-store path for projections containing Variant; fixed updates of
ordinary columns can use
+ // the narrower physical-column read.
+ const bool update_contains_variant = std::ranges::any_of(update_cids,
[&](uint32_t cid) {
+ return rowset_schema->column(cid).is_variant_type();
+ });
+ const auto update_read_strategy =
+ partial_update_info->is_fixed_partial_update() &&
!update_contains_variant
+ ? 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) {
@@ -1104,9 +1168,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<uint32_t, uint32_t> 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);
@@ -1245,8 +1309,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
@@ -1261,7 +1326,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<uint32_t, uint32_t> 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 0bf620b7e15..769f9bb28e2 100644
--- a/be/src/storage/tablet/base_tablet.h
+++ b/be/src/storage/tablet/base_tablet.h
@@ -251,6 +251,12 @@ public:
const std::vector<uint32_t>&
rowids,
const std::vector<uint32_t>&
cids, Block& block);
+ static Status fetch_values_by_rowids(RowsetSharedPtr input_rowset,
+ const TabletSchema& tablet_schema,
uint32_t segid,
+ const std::vector<uint32_t>& rowids,
+ const std::vector<uint32_t>& cids,
+ MutableColumns& dst_columns);
+
static Status fetch_value_by_rowids(RowsetSharedPtr input_rowset, uint32_t
segid,
const std::vector<uint32_t>& 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 3a931cbe333..3e81bced3ea 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 <memory>
#include <vector>
+#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<RowsetId, RowsetSharedPtr> 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<uint32_t> cids {0, 1};
+ auto column_store_block = schema->create_storage_block(cids);
+ auto row_store_block = schema->create_storage_block(cids);
+ std::map<uint32_t, uint32_t> column_store_read_index;
+ std::map<uint32_t, uint32_t> 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<PartialUpdateInfo>();
+ 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<RowsetId, RowsetSharedPtr> 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<BetaRowset*>(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<BetaRowset*>(args[0]);
+ if (rowset->rowset_id() == current_rowset->rowset_id()) {
+ ++current_batch_reads;
+ current_batch_column_count =
+ try_any_cast<const
std::vector<uint32_t>*>(args[1])->size();
+ }
+ },
+ &batch_read_guard);
+ sync_point->set_call_back(
+ "BaseTablet::fetch_value_through_row_column",
+ [&](auto&& args) {
+ auto* rowset = try_any_cast<BetaRowset*>(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_storage_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 48f95dc8c1b..2750ac5af0e 100644
--- a/be/test/storage/transform/variant_rowstore_test.cpp
+++ b/be/test/storage/transform/variant_rowstore_test.cpp
@@ -21,6 +21,7 @@
#include <gtest/gtest.h>
#include <limits>
+#include <numeric>
#include <string>
#include <string_view>
#include <unordered_map>
@@ -35,8 +36,10 @@
#include "core/data_type_serde/data_type_variant_v2_serde.h"
#include "core/field.h"
#include "core/string_buffer.hpp"
+#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"
@@ -113,6 +116,35 @@ protected:
return schema;
}
+ TabletSchemaSPtr create_typed_variant_pu_row_store_schema() {
+ TabletSchemaPB pb;
+ create_variant_pu_schema()->to_schema_pb(&pb);
+ pb.set_store_row_column(true);
+ ColumnPB* row_store = pb.add_column();
+ row_store->set_unique_id(10);
+ row_store->set_name(BeConsts::ROW_STORE_COL);
+ row_store->set_type("STRING");
+ row_store->set_is_key(false);
+ row_store->set_length(2147483643);
+ row_store->set_index_length(4);
+ row_store->set_is_nullable(false);
+ row_store->set_aggregation("NONE");
+
+ auto schema = std::make_shared<TabletSchema>();
+ schema->init_from_pb(pb);
+
+ ColumnPB typed_path_pb;
+ typed_path_pb.set_unique_id(-1);
+ typed_path_pb.set_name("a");
+ typed_path_pb.set_type("INT");
+ typed_path_pb.set_is_nullable(true);
+ typed_path_pb.set_pattern_type(PatternTypePB::MATCH_NAME);
+ TabletColumn typed_path;
+ typed_path.init_from_pb(typed_path_pb);
+ schema->mutable_column_by_uid(1).add_sub_column(typed_path);
+ return schema;
+ }
+
// Inserts one root-scalar JSON object string into a block's variant
column.
static void insert_variant_json(Block& block, size_t variant_pos,
std::string_view json) {
auto* variant = assert_cast<ColumnVariantV2*>(
@@ -500,6 +532,201 @@ TEST_F(VariantRowStoreTest, RowStorePreservesVariantV2) {
EXPECT_EQ(stored_variant.find(R"("flag":1)"), std::string::npos) <<
stored_variant;
}
+// UPSERT publish-conflict rewrites read the complete current row. Keep that
path on row-store
+// instead of opening every physical column, while fixed updates remain free
to choose a narrow
+// physical projection.
+TEST_F(VariantRowStoreTest, UpsertPublishConflictUsesRowStore) {
+ 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<const char*>(&key), sizeof(key));
+ insert_variant_json(block, 1, R"({"flag":true})");
+ block.get_by_position(2).column->assert_mutable()->insert_data(
+ reinterpret_cast<const char*>(&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":true)"),
std::string::npos);
+ const auto& persisted_row_store =
+ assert_cast<const
ColumnString&>(*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<PartialUpdateInfo>();
+ 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<RowsetId, RowsetSharedPtr> 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<BetaRowset*>(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<BetaRowset*>(args[0]);
+ if (rowset->rowset_id() == current_rowset->rowset_id()) {
+ ++batch_column_reads;
+ }
+ },
+ &batch_read_guard);
+ sync_point->enable_processing();
+
+ auto rebuilt = schema->create_storage_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<std::string_view> {"Validate",
"RowStoreFill"}));
+ 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":true)"),
std::string::npos);
+ const auto& rebuilt_row_store =
+ assert_cast<const
ColumnString&>(*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;
+}
+
+// Fixed updates normally read their narrow current projection from physical
columns during
+// publish-conflict reconstruction. Variant is the exception: typed paths are
coerced by the
+// column writer after RowStoreFill, so the physical value can no longer
reproduce the row-store
+// representation written by the original update.
+TEST_F(VariantRowStoreTest, FixedPublishConflictPreservesTypedVariantRowStore)
{
+ auto schema = create_typed_variant_pu_row_store_schema();
+ TabletSharedPtr tablet;
+ auto current_rowset = write_rowset_block(
+ schema, 8301, 3,
+ [&](Block& block) {
+ int32_t key = 1;
+ int8_t delete_sign = 0;
+ int32_t vv = 100;
+ block.get_by_position(0).column->assert_mutable()->insert_data(
+ reinterpret_cast<const char*>(&key), sizeof(key));
+ insert_variant_json(block, 1, R"({"a":"001"})");
+ block.get_by_position(2).column->assert_mutable()->insert_data(
+ reinterpret_cast<const char*>(&delete_sign),
sizeof(delete_sign));
+ block.get_by_position(3).column->assert_mutable()->insert_data(
+ reinterpret_cast<const char*>(&vv), sizeof(vv));
+
block.get_by_position(4).column->assert_mutable()->insert_default();
+ },
+ &tablet);
+ TabletSharedPtr unused_historical_tablet;
+ auto historical_rowset = write_rowset_block(
+ schema, 8302, 2,
+ [&](Block& block) {
+ int32_t key = 1;
+ int8_t delete_sign = 0;
+ int32_t vv = 7;
+ block.get_by_position(0).column->assert_mutable()->insert_data(
+ reinterpret_cast<const char*>(&key), sizeof(key));
+ insert_variant_json(block, 1, R"({"a":9})");
+ block.get_by_position(2).column->assert_mutable()->insert_data(
+ reinterpret_cast<const char*>(&delete_sign),
sizeof(delete_sign));
+ block.get_by_position(3).column->assert_mutable()->insert_data(
+ reinterpret_cast<const char*>(&vv), sizeof(vv));
+
block.get_by_position(4).column->assert_mutable()->insert_default();
+ },
+ &unused_historical_tablet);
+
+ Block persisted;
+ ASSERT_TRUE(read_rowset(current_rowset, schema, &persisted).ok());
+ EXPECT_NE(variant_row_json(persisted, 1, 0).find(R"("a":1)"),
std::string::npos);
+ const auto& persisted_row_store =
+ assert_cast<const
ColumnString&>(*persisted.get_by_position(4).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"("a":"001")"),
+ std::string::npos);
+
+ auto partial_update_info = std::make_shared<PartialUpdateInfo>();
+ 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<RowsetId, RowsetSharedPtr> rowsets {
+ {current_rowset->rowset_id(), current_rowset},
+ {historical_rowset->rowset_id(), historical_rowset}};
+
+ auto rebuilt = schema->create_storage_block();
+ ASSERT_TRUE(BaseTablet::generate_new_block_for_partial_update(
+ schema, partial_update_info.get(),
read_plan_historical, read_plan_update,
+ rowsets, &rebuilt)
+ .ok());
+ EXPECT_NE(variant_row_json(rebuilt, 1, 0).find(R"("a":"001")"),
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);
+ 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());
+
+ const auto& rebuilt_row_store =
+ assert_cast<const
ColumnString&>(*rebuilt.get_by_position(4).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"("a":"001")"), std::string::npos) <<
stored_variant;
+ EXPECT_EQ(stored_variant.find(R"("a":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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
