csun5285 commented on code in PR #67468:
URL: https://github.com/apache/doris/pull/67468#discussion_r3923354232
##########
be/src/storage/segment/vertical_segment_writer.cpp:
##########
@@ -368,113 +407,175 @@ Status
VerticalSegmentWriter::_finalize_column_writer_and_update_meta(size_t cid
return Status::OK();
}
-Status VerticalSegmentWriter::batch_block(const Block* block, size_t row_pos,
size_t num_rows) {
- // input width is checked by the transform chain's ValidateStage
- DCHECK(block->columns() == _tablet_schema->num_columns())
- << "block columns = " << block->dump_structure()
- << ", tablet_schema columns = " <<
_tablet_schema->dump_structure();
- _batched_blocks.emplace_back(block, row_pos, num_rows);
+Status VerticalSegmentWriter::append_block(const Block* block, size_t row_pos,
size_t num_rows) {
+ if (block->columns() < _column_writers.size()) {
+ return Status::InternalError(
+ "block->columns() < _column_writers.size(), block->columns()="
+
+ std::to_string(block->columns()) +
+ ", _column_writers.size()=" +
std::to_string(_column_writers.size()) +
+ ", _tablet_schema->dump_structure()=" +
_tablet_schema->dump_structure());
+ }
+ CHECK(block->columns() >= _column_writers.size())
+ << ", block->columns()=" << block->columns()
+ << ", _column_writers.size()=" << _column_writers.size()
+ << ", _tablet_schema->dump_structure()=" <<
_tablet_schema->dump_structure();
+ _olap_data_convertor->set_source_content(block, row_pos, num_rows);
+
+ // convert column data from engine format to storage layer format
+ std::vector<IOlapColumnDataAccessor*> key_columns;
+ IOlapColumnDataAccessor* seq_column = nullptr;
+ // keyed by cluster key unique id
+ std::map<uint32_t, IOlapColumnDataAccessor*> cluster_key_columns;
+ for (size_t id = 0; id < _column_writers.size(); ++id) {
+ // olap data convertor alway start from id = 0
+ auto converted_result = _olap_data_convertor->convert_column_data(id);
+ if (!converted_result.first.ok()) {
+ return converted_result.first;
+ }
+ auto cid = _column_ids[id];
+ if (_has_key && cid < _tablet_schema->num_key_columns()) {
+ key_columns.push_back(converted_result.second);
+ } else if (_has_key && _tablet_schema->has_sequence_col() &&
+ cid == _tablet_schema->sequence_col_idx()) {
+ seq_column = converted_result.second;
+ }
+ if (_has_key) {
+ _collect_cluster_key_column(cid, converted_result.second,
&cluster_key_columns);
+ }
+
RETURN_IF_ERROR(_column_writers[id]->append(converted_result.second->get_nullmap(),
+
converted_result.second->get_data(), num_rows));
+ }
+ if (_opts.write_type == DataWriteType::TYPE_COMPACTION) {
+ RETURN_IF_ERROR(
+ _variant_stats_calculator->calculate_variant_stats(block,
row_pos, num_rows));
+ }
+
+ // value groups carry no key columns
+ if (_has_key) {
+ RETURN_IF_ERROR(
+ _generate_key_index(key_columns, seq_column, num_rows,
cluster_key_columns));
+ }
+
+ _num_rows_written += num_rows;
+ _olap_data_convertor->clear_source_content();
return Status::OK();
}
-Status VerticalSegmentWriter::write_batch() {
- // Blocks arrive fully transformed: validated, partial-update rows filled,
- // variants parsed, and the derived (row-store) column decided by the
chain.
- // Its generator is pumped here in bounded batches.
+Status VerticalSegmentWriter::write_block(const Block* block, size_t row_pos,
size_t num_rows) {
+ RETURN_IF_ERROR(_open_group(_all_column_ids(), true));
+ // Each column writer is created right before its column is written, so
+ // only one column's page buffers are alive at a time. The derived
+ // (row-store) column goes first, pumped from its generator in batches.
if (_derived_column.second) {
const auto& [cid, generator] = _derived_column;
- RETURN_IF_ERROR(_create_column_writer(cid,
_tablet_schema->column(cid), _tablet_schema));
- for (auto& data : _batched_blocks) {
- RETURN_IF_ERROR(_append_generated_column(*generator, *data.block,
data.row_pos,
- data.num_rows, cid));
- }
+ RETURN_IF_ERROR(_create_column_writer(cid, cid, _tablet_schema));
+ RETURN_IF_ERROR(_append_generated_column(*generator, *block, row_pos,
num_rows, cid));
RETURN_IF_ERROR(_check_column_writer_disk_capacity(cid));
RETURN_IF_ERROR(_finalize_column_writer_and_update_meta(cid));
}
std::vector<IOlapColumnDataAccessor*> key_columns;
IOlapColumnDataAccessor* seq_column = nullptr;
- // the key is cluster key column unique id
- std::map<uint32_t, IOlapColumnDataAccessor*> cid_to_column;
+ // keyed by cluster key unique id
+ std::map<uint32_t, IOlapColumnDataAccessor*> cluster_key_columns;
for (uint32_t cid = 0; cid < _tablet_schema->num_columns(); ++cid) {
if (_derived_column.second && _derived_column.first == cid) {
continue;
}
- RETURN_IF_ERROR(_create_column_writer(cid,
_tablet_schema->column(cid), _tablet_schema));
- for (auto& data : _batched_blocks) {
-
RETURN_IF_ERROR(_olap_data_convertor->set_source_content_with_specifid_columns(
- data.block, data.row_pos, data.num_rows,
std::vector<uint32_t> {cid}));
-
- // convert column data from engine format to storage layer format
- auto [status, column] =
_olap_data_convertor->convert_column_data(cid);
- if (!status.ok()) {
- return status;
- }
- if (cid < _tablet_schema->num_key_columns()) {
- key_columns.push_back(column);
- }
- if (_tablet_schema->has_sequence_col() && cid ==
_tablet_schema->sequence_col_idx()) {
- seq_column = column;
- }
- auto column_unique_id = _tablet_schema->column(cid).unique_id();
- if (_is_mow_with_cluster_key() &&
- std::find(_tablet_schema->cluster_key_uids().begin(),
- _tablet_schema->cluster_key_uids().end(),
- column_unique_id) !=
_tablet_schema->cluster_key_uids().end()) {
- cid_to_column[column_unique_id] = column;
- }
-
RETURN_IF_ERROR(_column_writers[cid]->append(column->get_nullmap(),
column->get_data(),
- data.num_rows));
- _olap_data_convertor->clear_source_content();
+ RETURN_IF_ERROR(_create_column_writer(cid, cid, _tablet_schema));
+
RETURN_IF_ERROR(_olap_data_convertor->set_source_content_with_specifid_columns(
+ block, row_pos, num_rows, std::vector<uint32_t> {cid}));
+
+ // convert column data from engine format to storage layer format
+ auto [status, column] = _olap_data_convertor->convert_column_data(cid);
+ if (!status.ok()) {
+ return status;
+ }
+ if (cid < _tablet_schema->num_key_columns()) {
+ key_columns.push_back(column);
}
+ if (_tablet_schema->has_sequence_col() && cid ==
_tablet_schema->sequence_col_idx()) {
+ seq_column = column;
+ }
+ _collect_cluster_key_column(cid, column, &cluster_key_columns);
+ RETURN_IF_ERROR(
+ _column_writers[cid]->append(column->get_nullmap(),
column->get_data(), num_rows));
+ _olap_data_convertor->clear_source_content();
RETURN_IF_ERROR(_check_column_writer_disk_capacity(cid));
RETURN_IF_ERROR(_finalize_column_writer_and_update_meta(cid));
}
-
- for (auto& data : _batched_blocks) {
- _olap_data_convertor->set_source_content(data.block, data.row_pos,
data.num_rows);
- RETURN_IF_ERROR(_generate_key_index(data, key_columns, seq_column,
cid_to_column));
- _olap_data_convertor->clear_source_content();
- _num_rows_written += data.num_rows;
+ _columns_data_flushed = true;
+
+ // The accessors converted above keep their data, but the loop cleared the
+ // source columns the key encoder reads the null maps through, so put them
+ // back for the key pass.
+ _olap_data_convertor->set_source_content(block, row_pos, num_rows);
+ RETURN_IF_ERROR(_generate_key_index(key_columns, seq_column, num_rows,
cluster_key_columns));
+ _olap_data_convertor->clear_source_content();
+ _num_rows_written += num_rows;
+ // the group's rows are all in, so the cluster key primary keys can go
into the index now
+ RETURN_IF_ERROR(_flush_primary_keys());
Review Comment:
这是之前就存在的问题，已解决。
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
