Skip to content

Commit 1fbfb81

Browse files
bobhan1bobhan1
authored
[branch-2.1] Picks "[Fix](partial update) Persist partial_update_info in RocksDB in case of BE restart after a partial update has commited #38331" (#39035)
picks #38331 and #39066
1 parent 4668ebd commit 1fbfb81

13 files changed

+586
-77
lines changed

‎be/src/olap/partial_update_info.cpp

+155
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
#include "olap/partial_update_info.h"
19+
20+
#include <gen_cpp/olap_file.pb.h>
21+
22+
#include "olap/tablet_schema.h"
23+
24+
namespace doris {
25+
26+
void PartialUpdateInfo::init(const TabletSchema& tablet_schema, bool partial_update,
27+
const std::set<string>& partial_update_cols, bool is_strict_mode,
28+
int64_t timestamp_ms, const std::string& timezone,
29+
const std::string& auto_increment_column, int64_t cur_max_version) {
30+
is_partial_update = partial_update;
31+
partial_update_input_columns = partial_update_cols;
32+
max_version_in_flush_phase = cur_max_version;
33+
this->timestamp_ms = timestamp_ms;
34+
this->timezone = timezone;
35+
missing_cids.clear();
36+
update_cids.clear();
37+
for (auto i = 0; i < tablet_schema.num_columns(); ++i) {
38+
auto tablet_column = tablet_schema.column(i);
39+
if (!partial_update_input_columns.contains(tablet_column.name())) {
40+
missing_cids.emplace_back(i);
41+
if (!tablet_column.has_default_value() && !tablet_column.is_nullable() &&
42+
tablet_schema.auto_increment_column() != tablet_column.name()) {
43+
can_insert_new_rows_in_partial_update = false;
44+
}
45+
} else {
46+
update_cids.emplace_back(i);
47+
}
48+
if (auto_increment_column == tablet_column.name()) {
49+
is_schema_contains_auto_inc_column = true;
50+
}
51+
}
52+
this->is_strict_mode = is_strict_mode;
53+
is_input_columns_contains_auto_inc_column =
54+
is_partial_update && partial_update_input_columns.contains(auto_increment_column);
55+
_generate_default_values_for_missing_cids(tablet_schema);
56+
}
57+
58+
void PartialUpdateInfo::to_pb(PartialUpdateInfoPB* partial_update_info_pb) const {
59+
partial_update_info_pb->set_is_partial_update(is_partial_update);
60+
partial_update_info_pb->set_max_version_in_flush_phase(max_version_in_flush_phase);
61+
for (const auto& col : partial_update_input_columns) {
62+
partial_update_info_pb->add_partial_update_input_columns(col);
63+
}
64+
for (auto cid : missing_cids) {
65+
partial_update_info_pb->add_missing_cids(cid);
66+
}
67+
for (auto cid : update_cids) {
68+
partial_update_info_pb->add_update_cids(cid);
69+
}
70+
partial_update_info_pb->set_can_insert_new_rows_in_partial_update(
71+
can_insert_new_rows_in_partial_update);
72+
partial_update_info_pb->set_is_strict_mode(is_strict_mode);
73+
partial_update_info_pb->set_timestamp_ms(timestamp_ms);
74+
partial_update_info_pb->set_timezone(timezone);
75+
partial_update_info_pb->set_is_input_columns_contains_auto_inc_column(
76+
is_input_columns_contains_auto_inc_column);
77+
partial_update_info_pb->set_is_schema_contains_auto_inc_column(
78+
is_schema_contains_auto_inc_column);
79+
for (const auto& value : default_values) {
80+
partial_update_info_pb->add_default_values(value);
81+
}
82+
}
83+
84+
void PartialUpdateInfo::from_pb(PartialUpdateInfoPB* partial_update_info_pb) {
85+
is_partial_update = partial_update_info_pb->is_partial_update();
86+
max_version_in_flush_phase = partial_update_info_pb->has_max_version_in_flush_phase()
87+
? partial_update_info_pb->max_version_in_flush_phase()
88+
: -1;
89+
partial_update_input_columns.clear();
90+
for (const auto& col : partial_update_info_pb->partial_update_input_columns()) {
91+
partial_update_input_columns.insert(col);
92+
}
93+
missing_cids.clear();
94+
for (auto cid : partial_update_info_pb->missing_cids()) {
95+
missing_cids.push_back(cid);
96+
}
97+
update_cids.clear();
98+
for (auto cid : partial_update_info_pb->update_cids()) {
99+
update_cids.push_back(cid);
100+
}
101+
can_insert_new_rows_in_partial_update =
102+
partial_update_info_pb->can_insert_new_rows_in_partial_update();
103+
is_strict_mode = partial_update_info_pb->is_strict_mode();
104+
timestamp_ms = partial_update_info_pb->timestamp_ms();
105+
timezone = partial_update_info_pb->timezone();
106+
is_input_columns_contains_auto_inc_column =
107+
partial_update_info_pb->is_input_columns_contains_auto_inc_column();
108+
is_schema_contains_auto_inc_column =
109+
partial_update_info_pb->is_schema_contains_auto_inc_column();
110+
default_values.clear();
111+
for (const auto& value : partial_update_info_pb->default_values()) {
112+
default_values.push_back(value);
113+
}
114+
}
115+
116+
std::string PartialUpdateInfo::summary() const {
117+
return fmt::format(
118+
"update_cids={}, missing_cids={}, is_strict_mode={}, max_version_in_flush_phase={}",
119+
update_cids.size(), missing_cids.size(), is_strict_mode, max_version_in_flush_phase);
120+
}
121+
122+
void PartialUpdateInfo::_generate_default_values_for_missing_cids(
123+
const TabletSchema& tablet_schema) {
124+
for (unsigned int cur_cid : missing_cids) {
125+
const auto& column = tablet_schema.column(cur_cid);
126+
if (column.has_default_value()) {
127+
std::string default_value;
128+
if (UNLIKELY(tablet_schema.column(cur_cid).type() ==
129+
FieldType::OLAP_FIELD_TYPE_DATETIMEV2 &&
130+
to_lower(tablet_schema.column(cur_cid).default_value())
131+
.find(to_lower("CURRENT_TIMESTAMP")) !=
132+
std::string::npos)) {
133+
DateV2Value<DateTimeV2ValueType> dtv;
134+
dtv.from_unixtime(timestamp_ms / 1000, timezone);
135+
default_value = dtv.debug_string();
136+
} else if (UNLIKELY(tablet_schema.column(cur_cid).type() ==
137+
FieldType::OLAP_FIELD_TYPE_DATEV2 &&
138+
to_lower(tablet_schema.column(cur_cid).default_value())
139+
.find(to_lower("CURRENT_DATE")) !=
140+
std::string::npos)) {
141+
DateV2Value<DateV2ValueType> dv;
142+
dv.from_unixtime(timestamp_ms / 1000, timezone);
143+
default_value = dv.debug_string();
144+
} else {
145+
default_value = tablet_schema.column(cur_cid).default_value();
146+
}
147+
default_values.emplace_back(default_value);
148+
} else {
149+
// place an empty string here
150+
default_values.emplace_back();
151+
}
152+
}
153+
CHECK_EQ(missing_cids.size(), default_values.size());
154+
}
155+
} // namespace doris

‎be/src/olap/partial_update_info.h

+12-64
Original file line numberDiff line numberDiff line change
@@ -16,78 +16,26 @@
1616
// under the License.
1717

1818
#pragma once
19-
20-
#include "olap/tablet_schema.h"
19+
#include <cstdint>
20+
#include <set>
21+
#include <string>
22+
#include <vector>
2123

2224
namespace doris {
25+
class TabletSchema;
26+
class PartialUpdateInfoPB;
2327

2428
struct PartialUpdateInfo {
2529
void init(const TabletSchema& tablet_schema, bool partial_update,
26-
const std::set<string>& partial_update_cols, bool is_strict_mode,
30+
const std::set<std::string>& partial_update_cols, bool is_strict_mode,
2731
int64_t timestamp_ms, const std::string& timezone,
28-
const std::string& auto_increment_column, int64_t cur_max_version = -1) {
29-
is_partial_update = partial_update;
30-
partial_update_input_columns = partial_update_cols;
31-
max_version_in_flush_phase = cur_max_version;
32-
this->timestamp_ms = timestamp_ms;
33-
this->timezone = timezone;
34-
missing_cids.clear();
35-
update_cids.clear();
36-
for (auto i = 0; i < tablet_schema.num_columns(); ++i) {
37-
auto tablet_column = tablet_schema.column(i);
38-
if (!partial_update_input_columns.contains(tablet_column.name())) {
39-
missing_cids.emplace_back(i);
40-
if (!tablet_column.has_default_value() && !tablet_column.is_nullable() &&
41-
tablet_schema.auto_increment_column() != tablet_column.name()) {
42-
can_insert_new_rows_in_partial_update = false;
43-
}
44-
} else {
45-
update_cids.emplace_back(i);
46-
}
47-
if (auto_increment_column == tablet_column.name()) {
48-
is_schema_contains_auto_inc_column = true;
49-
}
50-
}
51-
this->is_strict_mode = is_strict_mode;
52-
is_input_columns_contains_auto_inc_column =
53-
is_partial_update && partial_update_input_columns.contains(auto_increment_column);
54-
_generate_default_values_for_missing_cids(tablet_schema);
55-
}
32+
const std::string& auto_increment_column, int64_t cur_max_version = -1);
33+
void to_pb(PartialUpdateInfoPB* partial_update_info) const;
34+
void from_pb(PartialUpdateInfoPB* partial_update_info);
35+
std::string summary() const;
5636

5737
private:
58-
void _generate_default_values_for_missing_cids(const TabletSchema& tablet_schema) {
59-
for (auto i = 0; i < missing_cids.size(); ++i) {
60-
auto cur_cid = missing_cids[i];
61-
const auto& column = tablet_schema.column(cur_cid);
62-
if (column.has_default_value()) {
63-
std::string default_value;
64-
if (UNLIKELY(tablet_schema.column(cur_cid).type() ==
65-
FieldType::OLAP_FIELD_TYPE_DATETIMEV2 &&
66-
to_lower(tablet_schema.column(cur_cid).default_value())
67-
.find(to_lower("CURRENT_TIMESTAMP")) !=
68-
std::string::npos)) {
69-
DateV2Value<DateTimeV2ValueType> dtv;
70-
dtv.from_unixtime(timestamp_ms / 1000, timezone);
71-
default_value = dtv.debug_string();
72-
} else if (UNLIKELY(tablet_schema.column(cur_cid).type() ==
73-
FieldType::OLAP_FIELD_TYPE_DATEV2 &&
74-
to_lower(tablet_schema.column(cur_cid).default_value())
75-
.find(to_lower("CURRENT_DATE")) !=
76-
std::string::npos)) {
77-
DateV2Value<DateV2ValueType> dv;
78-
dv.from_unixtime(timestamp_ms / 1000, timezone);
79-
default_value = dv.debug_string();
80-
} else {
81-
default_value = tablet_schema.column(cur_cid).default_value();
82-
}
83-
default_values.emplace_back(default_value);
84-
} else {
85-
// place an empty string here
86-
default_values.emplace_back();
87-
}
88-
}
89-
CHECK_EQ(missing_cids.size(), default_values.size());
90-
}
38+
void _generate_default_values_for_missing_cids(const TabletSchema& tablet_schema);
9139

9240
public:
9341
bool is_partial_update {false};

‎be/src/olap/rowset/rowset_meta_manager.cpp

+94
Original file line numberDiff line numberDiff line change
@@ -535,4 +535,98 @@ Status RowsetMetaManager::load_json_rowset_meta(OlapMeta* meta,
535535
return status;
536536
}
537537

538+
Status RowsetMetaManager::save_partial_update_info(
539+
OlapMeta* meta, int64_t tablet_id, int64_t partition_id, int64_t txn_id,
540+
const PartialUpdateInfoPB& partial_update_info_pb) {
541+
std::string key =
542+
fmt::format("{}{}_{}_{}", PARTIAL_UPDATE_INFO_PREFIX, tablet_id, partition_id, txn_id);
543+
std::string value;
544+
if (!partial_update_info_pb.SerializeToString(&value)) {
545+
return Status::Error<SERIALIZE_PROTOBUF_ERROR>(
546+
"serialize partial update info failed. key={}", key);
547+
}
548+
VLOG_NOTICE << "save partial update info, key=" << key << ", value_size=" << value.size();
549+
return meta->put(META_COLUMN_FAMILY_INDEX, key, value);
550+
}
551+
552+
Status RowsetMetaManager::try_get_partial_update_info(OlapMeta* meta, int64_t tablet_id,
553+
int64_t partition_id, int64_t txn_id,
554+
PartialUpdateInfoPB* partial_update_info_pb) {
555+
std::string key =
556+
fmt::format("{}{}_{}_{}", PARTIAL_UPDATE_INFO_PREFIX, tablet_id, partition_id, txn_id);
557+
std::string value;
558+
Status status = meta->get(META_COLUMN_FAMILY_INDEX, key, &value);
559+
if (status.is<META_KEY_NOT_FOUND>()) {
560+
return status;
561+
}
562+
if (!status.ok()) {
563+
LOG_WARNING("failed to get partial update info. tablet_id={}, partition_id={}, txn_id={}",
564+
tablet_id, partition_id, txn_id);
565+
return status;
566+
}
567+
if (!partial_update_info_pb->ParseFromString(value)) {
568+
return Status::Error<ErrorCode::PARSE_PROTOBUF_ERROR>(
569+
"fail to parse partial update info content to protobuf object. tablet_id={}, "
570+
"partition_id={}, txn_id={}",
571+
tablet_id, partition_id, txn_id);
572+
}
573+
return Status::OK();
574+
}
575+
576+
Status RowsetMetaManager::traverse_partial_update_info(
577+
OlapMeta* meta,
578+
std::function<bool(int64_t, int64_t, int64_t, std::string_view)> const& func) {
579+
auto traverse_partial_update_info_func = [&func](const std::string& key,
580+
const std::string& value) -> bool {
581+
std::vector<std::string> parts;
582+
// key format: pui_{tablet_id}_{partition_id}_{txn_id}
583+
RETURN_IF_ERROR(split_string(key, '_', &parts));
584+
if (parts.size() != 4) {
585+
LOG_WARNING("invalid rowset key={}, splitted size={}", key, parts.size());
586+
return true;
587+
}
588+
int64_t tablet_id = std::stoll(parts[1]);
589+
int64_t partition_id = std::stoll(parts[2]);
590+
int64_t txn_id = std::stoll(parts[3]);
591+
return func(tablet_id, partition_id, txn_id, value);
592+
};
593+
return meta->iterate(META_COLUMN_FAMILY_INDEX, PARTIAL_UPDATE_INFO_PREFIX,
594+
traverse_partial_update_info_func);
595+
}
596+
597+
Status RowsetMetaManager::remove_partial_update_info(OlapMeta* meta, int64_t tablet_id,
598+
int64_t partition_id, int64_t txn_id) {
599+
std::string key =
600+
fmt::format("{}{}_{}_{}", PARTIAL_UPDATE_INFO_PREFIX, tablet_id, partition_id, txn_id);
601+
Status res = meta->remove(META_COLUMN_FAMILY_INDEX, key);
602+
VLOG_NOTICE << "remove partial update info, key=" << key;
603+
return res;
604+
}
605+
606+
Status RowsetMetaManager::remove_partial_update_infos(
607+
OlapMeta* meta, const std::vector<std::tuple<int64_t, int64_t, int64_t>>& keys) {
608+
std::vector<std::string> remove_keys;
609+
for (auto [tablet_id, partition_id, txn_id] : keys) {
610+
remove_keys.push_back(fmt::format("{}{}_{}_{}", PARTIAL_UPDATE_INFO_PREFIX, tablet_id,
611+
partition_id, txn_id));
612+
}
613+
Status res = meta->remove(META_COLUMN_FAMILY_INDEX, remove_keys);
614+
VLOG_NOTICE << "remove partial update info, remove_keys.size()=" << remove_keys.size();
615+
return res;
616+
}
617+
618+
Status RowsetMetaManager::remove_tablet_related_partial_update_info(OlapMeta* meta,
619+
int64_t tablet_id) {
620+
std::string prefix = fmt::format("{}{}", PARTIAL_UPDATE_INFO_PREFIX, tablet_id);
621+
std::vector<std::string> remove_keys;
622+
auto get_remove_keys_func = [&](const std::string& key, const std::string& value) -> bool {
623+
remove_keys.emplace_back(key);
624+
return true;
625+
};
626+
VLOG_NOTICE << "remove tablet related partial update info, tablet_id: " << tablet_id
627+
<< " removed keys size: " << remove_keys.size();
628+
RETURN_IF_ERROR(meta->iterate(META_COLUMN_FAMILY_INDEX, prefix, get_remove_keys_func));
629+
return meta->remove(META_COLUMN_FAMILY_INDEX, remove_keys);
630+
}
631+
538632
} // namespace doris

‎be/src/olap/rowset/rowset_meta_manager.h

+21
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@
1818
#ifndef DORIS_BE_SRC_OLAP_ROWSET_ROWSET_META_MANAGER_H
1919
#define DORIS_BE_SRC_OLAP_ROWSET_ROWSET_META_MANAGER_H
2020

21+
#include <gen_cpp/olap_file.pb.h>
22+
2123
#include <cstdint>
2224
#include <functional>
2325
#include <string>
@@ -32,11 +34,15 @@
3234
namespace doris {
3335
class OlapMeta;
3436
class RowsetMetaPB;
37+
class PartialUpdateInfoPB;
3538
} // namespace doris
3639

3740
namespace doris {
3841
namespace {
3942
const std::string ROWSET_PREFIX = "rst_";
43+
44+
const std::string PARTIAL_UPDATE_INFO_PREFIX = "pui_";
45+
4046
} // namespace
4147

4248
// Helper class for managing rowset meta of one root path.
@@ -80,6 +86,21 @@ class RowsetMetaManager {
8086

8187
static Status load_json_rowset_meta(OlapMeta* meta, const std::string& rowset_meta_path);
8288

89+
static Status save_partial_update_info(OlapMeta* meta, int64_t tablet_id, int64_t partition_id,
90+
int64_t txn_id,
91+
const PartialUpdateInfoPB& partial_update_info_pb);
92+
static Status try_get_partial_update_info(OlapMeta* meta, int64_t tablet_id,
93+
int64_t partition_id, int64_t txn_id,
94+
PartialUpdateInfoPB* partial_update_info_pb);
95+
static Status traverse_partial_update_info(
96+
OlapMeta* meta,
97+
std::function<bool(int64_t, int64_t, int64_t, std::string_view)> const& func);
98+
static Status remove_partial_update_info(OlapMeta* meta, int64_t tablet_id,
99+
int64_t partition_id, int64_t txn_id);
100+
static Status remove_partial_update_infos(
101+
OlapMeta* meta, const std::vector<std::tuple<int64_t, int64_t, int64_t>>& keys);
102+
static Status remove_tablet_related_partial_update_info(OlapMeta* meta, int64_t tablet_id);
103+
83104
private:
84105
static Status _save(OlapMeta* meta, TabletUid tablet_uid, const RowsetId& rowset_id,
85106
const RowsetMetaPB& rowset_meta_pb);

0 commit comments

Comments
 (0)