github-actions[bot] commented on code in PR #22504:
URL: https://github.com/apache/doris/pull/22504#discussion_r1369609704
##########
be/src/olap/tablet.cpp:
##########
@@ -3170,108 +3210,274 @@ Status Tablet::generate_new_block_for_partial_update(
const std::vector<uint32>& update_cids, const PartialUpdateReadPlan&
read_plan_ori,
const PartialUpdateReadPlan& read_plan_update,
const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
+ std::shared_ptr<std::map<uint32_t, std::vector<uint32_t>>>
indicator_maps,
vectorized::Block* output_block) {
// do partial update related works
// 1. read columns by read plan
// 2. generate new block
// 3. write a new segment and modify rowset meta
// 4. mark current keys deleted
CHECK(output_block);
- auto full_mutable_columns = output_block->mutate_columns();
- auto old_block = rowset_schema->create_block_by_cids(missing_cids);
- auto update_block = rowset_schema->create_block_by_cids(update_cids);
-
- std::map<uint32_t, uint32_t> read_index_old;
- RETURN_IF_ERROR(read_columns_by_plan(rowset_schema, missing_cids,
read_plan_ori, rsid_to_rowset,
- old_block, &read_index_old));
-
- std::map<uint32_t, uint32_t> read_index_update;
- RETURN_IF_ERROR(read_columns_by_plan(rowset_schema, update_cids,
read_plan_update,
- rsid_to_rowset, update_block,
&read_index_update));
+ if (!indicator_maps) {
+ auto full_mutable_columns = output_block->mutate_columns();
+ auto old_block = rowset_schema->create_block_by_cids(missing_cids);
+ auto update_block = rowset_schema->create_block_by_cids(update_cids);
+
+ std::map<uint32_t, uint32_t> read_index_old;
+ RETURN_IF_ERROR(read_columns_by_plan(rowset_schema, rsid_to_rowset,
read_plan_ori,
+ &missing_cids, &old_block,
&read_index_old));
+
+ std::map<uint32_t, uint32_t> read_index_update;
+ RETURN_IF_ERROR(read_columns_by_plan(rowset_schema, rsid_to_rowset,
read_plan_update,
+ &update_cids, &update_block,
&read_index_update));
+
+ // build full block
+ CHECK(read_index_old.size() == read_index_update.size());
+ for (auto i = 0; i < missing_cids.size(); ++i) {
+ for (auto idx = 0; idx < read_index_old.size(); ++idx) {
+ full_mutable_columns[missing_cids[i]]->insert_from(
+
*old_block.get_columns_with_type_and_name()[i].column.get(),
+ read_index_old[idx]);
+ }
+ }
+ for (auto i = 0; i < update_cids.size(); ++i) {
+ for (auto idx = 0; idx < read_index_update.size(); ++idx) {
+ full_mutable_columns[update_cids[i]]->insert_from(
+
*update_block.get_columns_with_type_and_name()[i].column.get(),
+ read_index_update[idx]);
+ }
+ }
+ VLOG_DEBUG << "full block when publish: " << output_block->dump_data();
+ } else {
+ std::unordered_set<uint32_t> cids_point_read;
+ for (const auto& [_, cids] : *indicator_maps) {
+ for (uint32_t cid : cids) {
+ cids_point_read.insert(cid);
+ }
+ }
+ std::vector<uint32_t> point_read_cids;
+ point_read_cids.reserve(cids_point_read.size());
+ for (uint32_t cid : cids_point_read) {
+ point_read_cids.emplace_back(cid);
+ }
+ auto full_mutable_columns = output_block->mutate_columns();
+ auto old_full_read_block =
rowset_schema->create_block_by_cids(missing_cids);
+ auto point_read_block =
rowset_schema->create_block_by_cids(point_read_cids);
+
+ std::map<uint32_t, uint32_t> full_read_index_old;
+ std::map<uint32_t, std::map<uint32_t, uint32_t>> point_read_index_old;
+ RETURN_IF_ERROR(read_columns_by_plan(rowset_schema, rsid_to_rowset,
read_plan_ori,
+ &missing_cids, &point_read_cids,
&old_full_read_block,
+ &point_read_block,
&full_read_index_old,
+ &point_read_index_old));
+
+ auto update_block = rowset_schema->create_block_by_cids(update_cids);
+ std::map<uint32_t, uint32_t> read_index_update;
+ RETURN_IF_ERROR(read_columns_by_plan(rowset_schema, rsid_to_rowset,
read_plan_update,
+ &update_cids, &update_block,
&read_index_update));
+ // construct the full block
+ CHECK(full_read_index_old.size() == read_index_update.size());
+
+ for (size_t i = 0; i < missing_cids.size(); ++i) {
+ for (auto idx = 0; idx < full_read_index_old.size(); ++idx) {
+ full_mutable_columns[missing_cids[i]]->insert_from(
+ *old_full_read_block.get_by_position(i).column.get(),
+ full_read_index_old[idx]);
+ }
+ }
- // build full block
- CHECK(read_index_old.size() == read_index_update.size());
- for (auto i = 0; i < missing_cids.size(); ++i) {
- for (auto idx = 0; idx < read_index_old.size(); ++idx) {
- full_mutable_columns[missing_cids[i]]->insert_from(
-
*old_block.get_columns_with_type_and_name()[i].column.get(),
- read_index_old[idx]);
+ for (size_t i = 0; i < update_cids.size(); i++) {
+ uint32_t cid = update_cids[i];
+ if (!cids_point_read.contains(cid)) {
+ for (auto idx = 0; idx < full_read_index_old.size(); ++idx) {
+ full_mutable_columns[cid]->insert_from(
+ *update_block.get_by_position(i).column.get(),
read_index_update[idx]);
+ }
+ }
}
- }
- for (auto i = 0; i < update_cids.size(); ++i) {
- for (auto idx = 0; idx < read_index_update.size(); ++idx) {
- full_mutable_columns[update_cids[i]]->insert_from(
-
*update_block.get_columns_with_type_and_name()[i].column.get(),
- read_index_update[idx]);
+
+ for (size_t i = 0; i < point_read_cids.size(); i++) {
+ uint32_t cid = point_read_cids[i];
+ for (uint32_t idx = 0; i < full_read_index_old.size(); i++) {
+ if (point_read_index_old[cid].contains(idx)) {
+ full_mutable_columns[cid]->insert_from(
+ *point_read_block.get_by_position(i).column.get(),
+ point_read_index_old[cid][idx]);
+ } else {
+ full_mutable_columns[cid]->insert_from(
+ *update_block.get_by_position(i).column.get(),
idx);
+ }
+ }
}
}
- VLOG_DEBUG << "full block when publish: " << output_block->dump_data();
return Status::OK();
}
-
-// read columns by read plan
-// read_index: ori_pos-> block_idx
Status Tablet::read_columns_by_plan(TabletSchemaSPtr tablet_schema,
- const std::vector<uint32_t> cids_to_read,
- const PartialUpdateReadPlan& read_plan,
const std::map<RowsetId, RowsetSharedPtr>&
rsid_to_rowset,
- vectorized::Block& block,
- std::map<uint32_t, uint32_t>* read_index) {
- bool has_row_column = tablet_schema->store_row_column();
- auto mutable_columns = block.mutate_columns();
- size_t read_idx = 0;
- for (auto rs_it : read_plan) {
- for (auto seg_it : rs_it.second) {
- auto rowset_iter = rsid_to_rowset.find(rs_it.first);
- CHECK(rowset_iter != rsid_to_rowset.end());
- std::vector<uint32_t> rids;
- for (auto id_and_pos : seg_it.second) {
- rids.emplace_back(id_and_pos.rid);
- (*read_index)[id_and_pos.pos] = read_idx++;
- }
- if (has_row_column) {
- auto st = fetch_value_through_row_column(rowset_iter->second,
seg_it.first, rids,
- cids_to_read, block);
+ const PartialUpdateReadPlan& read_plan,
+ const std::vector<uint32_t>*
cids_full_read,
+ vectorized::Block* block_full_read,
+ std::map<uint32_t, uint32_t>*
full_read_index) {
+ auto full_read_columns = block_full_read->mutate_columns();
+ uint32_t read_idx1 = 0;
+
+ if (std::holds_alternative<RowStoreReadPlan>(read_plan)) {
+ const auto& row_store_read_plan =
std::get<RowStoreReadPlan>(read_plan);
+ for (const auto& [rowset_id, segment_read_infos] :
row_store_read_plan) {
+ auto rowset = rsid_to_rowset.at(rowset_id);
+ CHECK(rowset);
+ for (const auto& [segment_id, rows_info] : segment_read_infos) {
+ std::vector<uint32_t> rids;
+ for (const auto& [id_and_pos, cids] : rows_info) {
+ // set read index for missing columns
+ rids.emplace_back(id_and_pos.rid);
+ (*full_read_index)[id_and_pos.pos] = read_idx1++;
+ }
+
+ auto st = fetch_value_through_row_column(rowset, segment_id,
rids, rows_info,
+ cids_full_read,
nullptr, block_full_read,
+ nullptr, false);
if (!st.ok()) {
LOG(WARNING) << "failed to fetch value through row column";
return st;
}
- continue;
}
- for (size_t cid = 0; cid < mutable_columns.size(); ++cid) {
- TabletColumn tablet_column =
tablet_schema->column(cids_to_read[cid]);
- auto st = fetch_value_by_rowids(rowset_iter->second,
seg_it.first, rids,
- tablet_column,
mutable_columns[cid]);
- // set read value to output block
+ }
+ } else {
+ const auto& column_store_read_plan =
std::get<ColumnStoreReadPlan>(read_plan);
+ for (const auto& [rowset_id, segment_read_infos] :
column_store_read_plan) {
+ auto rowset = rsid_to_rowset.at(rowset_id);
+ CHECK(rowset);
+
+ for (const auto& [segment_id, columns_info] : segment_read_infos) {
+ std::vector<uint32_t> rids;
+ for (auto [rid, pos] : columns_info.missing_column_rows) {
+ rids.emplace_back(rid);
+ // set read index for missing columns
+ (*full_read_index)[pos] = read_idx1++;
+ }
+
+ // read values for missing columns
+ for (size_t i = 0; i < cids_full_read->size(); ++i) {
+ TabletColumn tablet_column =
tablet_schema->column(cids_full_read->at(i));
+ auto st = fetch_value_by_rowids(rowset, segment_id, rids,
tablet_column,
+ full_read_columns[i]);
+ if (!st.ok()) {
+ LOG(WARNING) << "failed to fetch value by rowids";
+ return st;
+ }
+ }
+ }
+ }
+ }
+ return Status::OK();
+}
+
+Status Tablet::read_columns_by_plan(
+ TabletSchemaSPtr tablet_schema, const std::map<RowsetId,
RowsetSharedPtr>& rsid_to_rowset,
Review Comment:
warning: method 'read_columns_by_plan' can be made static
[readability-convert-member-functions-to-static]
be/src/olap/tablet.cpp:3375:
```diff
- }
+ }static
```
##########
be/src/olap/tablet.cpp:
##########
@@ -2719,15 +2719,16 @@ Status Tablet::_get_segment_column_iterator(
}
// fetch value by row column
-Status Tablet::fetch_value_through_row_column(RowsetSharedPtr input_rowset,
uint32_t segid,
- const std::vector<uint32_t>&
rowids,
- const std::vector<uint32_t>&
cids,
- vectorized::Block& block) {
+Status Tablet::fetch_value_through_row_column(
+ RowsetSharedPtr input_rowset, uint32_t segid, const
std::vector<uint32_t>& rids,
Review Comment:
warning: method 'fetch_value_through_row_column' can be made static
[readability-convert-member-functions-to-static]
be/src/olap/tablet.cpp:2721:
```diff
- }
+ }static
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
