Skip to content

[opt](partial update) Select publish-conflict read strategy by update… #67295

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
foxtail463 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[opt](partial update) Select publish-conflict read strategy by update… #67295

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions be/src/storage/mow/historical_row_fetcher.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
35 changes: 17 additions & 18 deletions be/src/storage/partial_update_info.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<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) {
Expand All @@ -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<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();
}
Expand All @@ -357,7 +358,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()) {
Expand All @@ -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;
}
}
}
Expand Down Expand Up @@ -433,7 +430,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);
Expand Down Expand Up @@ -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<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;
Expand Down
9 changes: 8 additions & 1 deletion be/src/storage/partial_update_info.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,14 +127,21 @@ 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);
Status 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,
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,
Expand Down
109 changes: 84 additions & 25 deletions be/src/storage/tablet/base_tablet.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand All @@ -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;
Expand Down Expand Up @@ -866,9 +886,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;
Expand Down Expand Up @@ -976,6 +996,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;
Expand Down Expand Up @@ -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<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) {
Expand Down Expand Up @@ -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<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
// 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) {
Expand All @@ -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<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);
Expand Down Expand Up @@ -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
Expand All @@ -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<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 =
Expand Down
6 changes: 6 additions & 0 deletions be/src/storage/tablet/base_tablet.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -251,6 +251,12 @@ class BaseTablet : public std::enable_shared_from_this<BaseTablet> {
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);
Expand Down
Loading