Skip to content
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

Jump to bottom

[fix](partial update) partial update should not read old fileds from rows with delete sign #36210

Merged
merged 6 commits into from
Jun 14, 2024
Merged

[fix](partial update) partial update should not read old fileds from rows with delete sign #36210

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
21837b1
fix with case
zhannngchen Jun 12, 2024
0b52b33
update case and output
zhannngchen Jun 13, 2024
1135904
remove useless debug log
zhannngchen Jun 13, 2024
b7e57db
reformat code
zhannngchen Jun 13, 2024
8821fd7
resolve memtracker check fail
zhannngchen Jun 13, 2024
a6e3956
fix compile
zhannngchen Jun 13, 2024
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
58 changes: 53 additions & 5 deletions be/src/olap/base_tablet.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -798,8 +798,8 @@ Status BaseTablet::calc_segment_delete_bitmap(RowsetSharedPtr rowset,
auto partial_update_info = rowset_writer->get_partial_update_info();
DCHECK(partial_update_info);
RETURN_IF_ERROR(generate_new_block_for_partial_update(
rowset_schema, partial_update_info->missing_cids, partial_update_info->update_cids,
read_plan_ori, read_plan_update, rsid_to_rowset, &block));
rowset_schema, partial_update_info.get(), read_plan_ori, read_plan_update,
rsid_to_rowset, &block));
RETURN_IF_ERROR(sort_block(block, ordered_block));
RETURN_IF_ERROR(rowset_writer->flush_single_block(&ordered_block));
if (new_generated_rows != rowset_writer->num_rows()) {
Expand Down Expand Up @@ -928,9 +928,8 @@ Status BaseTablet::fetch_value_by_rowids(RowsetSharedPtr input_rowset, uint32_t
}

Status BaseTablet::generate_new_block_for_partial_update(
TabletSchemaSPtr rowset_schema, const std::vector<uint32>& missing_cids,
const std::vector<uint32>& update_cids, const PartialUpdateReadPlan& read_plan_ori,
const PartialUpdateReadPlan& read_plan_update,
TabletSchemaSPtr rowset_schema, const PartialUpdateInfo* partial_update_info,
const PartialUpdateReadPlan& read_plan_ori, const PartialUpdateReadPlan& read_plan_update,
const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
vectorized::Block* output_block) {
// do partial update related works
Expand All @@ -940,6 +939,8 @@ Status BaseTablet::generate_new_block_for_partial_update(
// 4. mark current keys deleted
CHECK(output_block);
auto full_mutable_columns = output_block->mutate_columns();
const auto& missing_cids = partial_update_info->missing_cids;
const auto& update_cids = partial_update_info->update_cids;
auto old_block = rowset_schema->create_block_by_cids(missing_cids);
auto update_block = rowset_schema->create_block_by_cids(update_cids);

Expand All @@ -951,10 +952,57 @@ Status BaseTablet::generate_new_block_for_partial_update(
RETURN_IF_ERROR(read_columns_by_plan(rowset_schema, update_cids, read_plan_update,
rsid_to_rowset, update_block, &read_index_update));

const vectorized::Int8* delete_sign_column_data = nullptr;
if (const vectorized::ColumnWithTypeAndName* delete_sign_column =
old_block.try_get_by_name(DELETE_SIGN);
delete_sign_column != nullptr) {
auto& delete_sign_col =
reinterpret_cast<const vectorized::ColumnInt8&>(*(delete_sign_column->column));
delete_sign_column_data = delete_sign_col.get_data().data();
}

// build default value block
auto default_value_block = old_block.clone_empty();
auto mutable_default_value_columns = default_value_block.mutate_columns();
if (delete_sign_column_data != nullptr) {
for (auto i = 0; i < missing_cids.size(); ++i) {
const auto& column = rowset_schema->column(missing_cids[i]);
if (column.has_default_value()) {
const auto& default_value = partial_update_info->default_values[i];
vectorized::ReadBuffer rb(const_cast<char*>(default_value.c_str()),
default_value.size());
RETURN_IF_ERROR(old_block.get_by_position(i).type->from_string(
rb, mutable_default_value_columns[i].get()));
}
}
}

// build full block
CHECK(read_index_old.size() == read_index_update.size());

for (auto i = 0; i < missing_cids.size(); ++i) {
const auto& rs_column = rowset_schema->column(missing_cids[i]);
for (auto idx = 0; idx < read_index_old.size(); ++idx) {
// if the conflict update is a delete sign, which means that the key is
// not exist now, we should not read old values from the deleted data,
// and should use default value instead.
// NOTE: since now we are in the publishing phase, all data is commited
// before, even the `strict_mode` is true (which requires partial update
// load job can't insert new keys), this "new" key MUST be written into
// the new generated segment file.
if (delete_sign_column_data != nullptr &&
delete_sign_column_data[read_index_old[idx]] != 0) {
auto& mutable_column = full_mutable_columns[missing_cids[i]];
if (rs_column.has_default_value()) {
mutable_column->insert_from(*mutable_default_value_columns[i].get(), 0);
} else if (rs_column.is_nullable()) {
assert_cast<vectorized::ColumnNullable*>(mutable_column.get())
->insert_null_elements(1);
} else {
mutable_column->insert_default();
}
continue;
}
full_mutable_columns[missing_cids[i]]->insert_from(
*old_block.get_columns_with_type_and_name()[i].column.get(),
read_index_old[idx]);
Expand Down
4 changes: 2 additions & 2 deletions be/src/olap/base_tablet.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -185,8 +185,8 @@ class BaseTablet {
std::vector<RowsetSharedPtr>* rowsets = nullptr);

static Status generate_new_block_for_partial_update(
TabletSchemaSPtr rowset_schema, const std::vector<uint32>& missing_cids,
const std::vector<uint32>& update_cids, const PartialUpdateReadPlan& read_plan_ori,
TabletSchemaSPtr rowset_schema, const PartialUpdateInfo* partial_update_info,
const PartialUpdateReadPlan& read_plan_ori,
const PartialUpdateReadPlan& read_plan_update,
const std::map<RowsetId, RowsetSharedPtr>& rsid_to_rowset,
vectorized::Block* output_block);
Expand Down
41 changes: 41 additions & 0 deletions be/src/olap/partial_update_info.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ struct PartialUpdateInfo {
const std::string& auto_increment_column) {
is_partial_update = partial_update;
partial_update_input_columns = partial_update_cols;

this->timestamp_ms = timestamp_ms;
this->timezone = timezone;
missing_cids.clear();
Expand All @@ -50,8 +51,45 @@ struct PartialUpdateInfo {
this->is_strict_mode = is_strict_mode;
is_input_columns_contains_auto_inc_column =
is_partial_update && partial_update_input_columns.contains(auto_increment_column);
_generate_default_values_for_missing_cids(tablet_schema);
}

private:
void _generate_default_values_for_missing_cids(const TabletSchema& tablet_schema) {
for (auto i = 0; i < missing_cids.size(); ++i) {
auto cur_cid = missing_cids[i];
const auto& column = tablet_schema.column(cur_cid);
if (column.has_default_value()) {
std::string default_value;
if (UNLIKELY(tablet_schema.column(cur_cid).type() ==
FieldType::OLAP_FIELD_TYPE_DATETIMEV2 &&
to_lower(tablet_schema.column(cur_cid).default_value())
.find(to_lower("CURRENT_TIMESTAMP")) !=
std::string::npos)) {
DateV2Value<DateTimeV2ValueType> dtv;
dtv.from_unixtime(timestamp_ms / 1000, timezone);
default_value = dtv.debug_string();
} else if (UNLIKELY(tablet_schema.column(cur_cid).type() ==
FieldType::OLAP_FIELD_TYPE_DATEV2 &&
to_lower(tablet_schema.column(cur_cid).default_value())
.find(to_lower("CURRENT_DATE")) !=
std::string::npos)) {
DateV2Value<DateV2ValueType> dv;
dv.from_unixtime(timestamp_ms / 1000, timezone);
default_value = dv.debug_string();
} else {
default_value = tablet_schema.column(cur_cid).default_value();
}
default_values.emplace_back(default_value);
} else {
// place an empty string here
default_values.emplace_back();
}
}
CHECK_EQ(missing_cids.size(), default_values.size());
}

public:
bool is_partial_update {false};
std::set<std::string> partial_update_input_columns;
std::vector<uint32_t> missing_cids;
Expand All @@ -64,5 +102,8 @@ struct PartialUpdateInfo {
std::string timezone;
bool is_input_columns_contains_auto_inc_column = false;
bool is_schema_contains_auto_inc_column = false;

// default values for missing cids
std::vector<std::string> default_values;
};
} // namespace doris
27 changes: 3 additions & 24 deletions be/src/olap/rowset/segment_v2/segment_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -784,7 +784,7 @@ Status SegmentWriter::fill_missing_columns(vectorized::MutableColumns& mutable_f
const vectorized::Int8* delete_sign_column_data = nullptr;
if (const vectorized::ColumnWithTypeAndName* delete_sign_column =
old_value_block.try_get_by_name(DELETE_SIGN);
delete_sign_column != nullptr && _tablet_schema->has_sequence_col()) {
delete_sign_column != nullptr) {
auto& delete_sign_col =
reinterpret_cast<const vectorized::ColumnInt8&>(*(delete_sign_column->column));
delete_sign_column_data = delete_sign_col.get_data().data();
Expand All @@ -794,29 +794,8 @@ Status SegmentWriter::fill_missing_columns(vectorized::MutableColumns& mutable_f
for (auto i = 0; i < cids_missing.size(); ++i) {
const auto& column = _tablet_schema->column(cids_missing[i]);
if (column.has_default_value()) {
std::string default_value;
if (UNLIKELY(_tablet_schema->column(cids_missing[i]).type() ==
FieldType::OLAP_FIELD_TYPE_DATETIMEV2 &&
to_lower(_tablet_schema->column(cids_missing[i]).default_value())
.find(to_lower("CURRENT_TIMESTAMP")) !=
std::string::npos)) {
DateV2Value<DateTimeV2ValueType> dtv;
dtv.from_unixtime(_opts.rowset_ctx->partial_update_info->timestamp_ms / 1000,
_opts.rowset_ctx->partial_update_info->timezone);
default_value = dtv.debug_string();
} else if (UNLIKELY(
_tablet_schema->column(cids_missing[i]).type() ==
FieldType::OLAP_FIELD_TYPE_DATEV2 &&
to_lower(_tablet_schema->column(cids_missing[i]).default_value())
.find(to_lower("CURRENT_DATE")) !=
std::string::npos)) {
DateV2Value<DateV2ValueType> dv;
dv.from_unixtime(_opts.rowset_ctx->partial_update_info->timestamp_ms / 1000,
_opts.rowset_ctx->partial_update_info->timezone);
default_value = dv.debug_string();
} else {
default_value = _tablet_schema->column(cids_missing[i]).default_value();
}
const auto& default_value =
_opts.rowset_ctx->partial_update_info->default_values[i];
vectorized::ReadBuffer rb(const_cast<char*>(default_value.c_str()),
default_value.size());
RETURN_IF_ERROR(old_value_block.get_by_position(i).type->from_string(
Expand Down
27 changes: 3 additions & 24 deletions be/src/olap/rowset/segment_v2/vertical_segment_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -608,7 +608,7 @@ Status VerticalSegmentWriter::_fill_missing_columns(
const vectorized::Int8* delete_sign_column_data = nullptr;
if (const vectorized::ColumnWithTypeAndName* delete_sign_column =
old_value_block.try_get_by_name(DELETE_SIGN);
delete_sign_column != nullptr && _tablet_schema->has_sequence_col()) {
delete_sign_column != nullptr) {
auto& delete_sign_col =
reinterpret_cast<const vectorized::ColumnInt8&>(*(delete_sign_column->column));
delete_sign_column_data = delete_sign_col.get_data().data();
Expand All @@ -618,29 +618,8 @@ Status VerticalSegmentWriter::_fill_missing_columns(
for (auto i = 0; i < missing_cids.size(); ++i) {
const auto& column = _tablet_schema->column(missing_cids[i]);
if (column.has_default_value()) {
std::string default_value;
if (UNLIKELY(_tablet_schema->column(missing_cids[i]).type() ==
FieldType::OLAP_FIELD_TYPE_DATETIMEV2 &&
to_lower(_tablet_schema->column(missing_cids[i]).default_value())
.find(to_lower("CURRENT_TIMESTAMP")) !=
std::string::npos)) {
DateV2Value<DateTimeV2ValueType> dtv;
dtv.from_unixtime(_opts.rowset_ctx->partial_update_info->timestamp_ms / 1000,
_opts.rowset_ctx->partial_update_info->timezone);
default_value = dtv.debug_string();
} else if (UNLIKELY(
_tablet_schema->column(missing_cids[i]).type() ==
FieldType::OLAP_FIELD_TYPE_DATEV2 &&
to_lower(_tablet_schema->column(missing_cids[i]).default_value())
.find(to_lower("CURRENT_DATE")) !=
std::string::npos)) {
DateV2Value<DateV2ValueType> dv;
dv.from_unixtime(_opts.rowset_ctx->partial_update_info->timestamp_ms / 1000,
_opts.rowset_ctx->partial_update_info->timezone);
default_value = dv.debug_string();
} else {
default_value = _tablet_schema->column(missing_cids[i]).default_value();
}
const auto& default_value =
_opts.rowset_ctx->partial_update_info->default_values[i];
vectorized::ReadBuffer rb(const_cast<char*>(default_value.c_str()),
default_value.size());
RETURN_IF_ERROR(old_value_block.get_by_position(i).type->from_string(
Expand Down
5 changes: 5 additions & 0 deletions ...-test/data/unique_with_mow_p0/partial_update/partial_update_parallel_with_delete_sign.csv
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
1,10,1
2,20,0
3,30,1
4,40,0
5,50,1
82 changes: 80 additions & 2 deletions regression-test/data/unique_with_mow_p0/partial_update/test_partial_update_delete_sign.out
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
2 2 2 2 2
4 4 4 4 4

-- !with_delete_sign --
-- !1 --
1 \N \N \N \N 1
1 1 1 1 1 0
2 2 2 2 2 0
Expand All @@ -21,12 +21,51 @@
5 5 5 5 5 0
6 \N \N \N \N 1

-- !2 --
1 \N \N \N \N 1
2 2 2 2 2 0
3 \N \N \N \N 1
4 4 4 4 4 0
5 \N \N \N \N 1
6 \N \N \N \N 1

-- !sql --
1 1 1 1 1
2 2 2 2 2
3 3 3 3 3
4 4 4 4 4
5 5 5 5 5

-- !after_delete --
2 2 2 2 2
4 4 4 4 4

-- !1 --
1 1 1 1 1 0
1 1 1 1 1 1
2 2 2 2 2 0
3 3 3 3 3 0
3 3 3 3 3 1
4 4 4 4 4 0
5 5 5 5 5 0
5 5 5 5 5 1
6 \N \N \N \N 1

-- !2 --
1 1 1 1 1 1
2 2 2 2 2 0
3 3 3 3 3 1
4 4 4 4 4 0
5 5 5 5 5 1
6 \N \N \N \N 1

-- !1 --
1 1 1

-- !2 --

-- !3 --
1 2 \N

-- !1 --
1 1 1 1
Expand All @@ -47,7 +86,7 @@
2 2 2 2 2
4 4 4 4 4

-- !with_delete_sign --
-- !1 --
1 \N \N \N \N 1
1 1 1 1 1 0
2 2 2 2 2 0
Expand All @@ -58,12 +97,51 @@
5 5 5 5 5 0
6 \N \N \N \N 1

-- !2 --
1 \N \N \N \N 1
2 2 2 2 2 0
3 \N \N \N \N 1
4 4 4 4 4 0
5 \N \N \N \N 1
6 \N \N \N \N 1

-- !sql --
1 1 1 1 1
2 2 2 2 2
3 3 3 3 3
4 4 4 4 4
5 5 5 5 5

-- !after_delete --
2 2 2 2 2
4 4 4 4 4

-- !1 --
1 1 1 1 1 0
1 1 1 1 1 1
2 2 2 2 2 0
3 3 3 3 3 0
3 3 3 3 3 1
4 4 4 4 4 0
5 5 5 5 5 0
5 5 5 5 5 1
6 \N \N \N \N 1

-- !2 --
1 1 1 1 1 1
2 2 2 2 2 0
3 3 3 3 3 1
4 4 4 4 4 0
5 5 5 5 5 1
6 \N \N \N \N 1

-- !1 --
1 1 1

-- !2 --

-- !3 --
1 2 \N

-- !1 --
1 1 1 1
Expand Down
19 changes: 19 additions & 0 deletions .../data/unique_with_mow_p0/partial_update/test_partial_update_delete_sign_with_conflict.out
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
-- This file is automatically generated. You should know what you did if you want to edit this
-- !sql --
1 1 1 1 1
2 2 2 2 2
3 3 3 3 3
4 4 4 4 4
5 5 5 5 5

-- !sql --
2 20 \N \N foo
4 40 \N \N foo

-- !sql --
1 100 10 \N foo
2 20 20 \N foo
3 100 30 \N foo
4 40 40 \N foo
5 100 50 \N foo

Loading
Loading