database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[enhancement](recover) support skipping missing version in select by session variable (#25654)

Browse Source
This commit is contained in:
xy720
2023-11-02 20:01:51 +08:00
committed by GitHub
parent 89cf828f03
commit a5ef90dacc
20 changed files with 147 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
be/src/olap/base_tablet.h
View File

@ -76,7 +76,8 @@ public:
std::unique_ptr<RowsetWriter>* rowset_writer) = 0;
virtual Status capture_rs_readers(const Version& spec_version,
std::vector<RowSetSplits>* rs_splits) const = 0;
std::vector<RowSetSplits>* rs_splits,
bool skip_missing_version) const = 0;
virtual size_t tablet_footprint() = 0;

2
be/src/olap/schema_change.cpp
View File

@ -1023,7 +1023,7 @@ Status SchemaChangeHandler::_get_versions_to_be_changed(
*max_rowset = rowset;
RETURN_IF_ERROR(base_tablet->capture_consistent_versions(Version(0, rowset->version().second),
versions_to_be_changed));
versions_to_be_changed, false, false));
return Status::OK();
}

23
be/src/olap/tablet.cpp
View File

@ -738,7 +738,7 @@ void Tablet::delete_expired_stale_rowset() {
Version test_version = Version(0, lastest_delta->end_version());
stale_version_path_map[*path_id_iter] = version_path;
Status status = capture_consistent_versions(test_version, nullptr);
Status status = capture_consistent_versions(test_version, nullptr, false, false);
// 1. When there is no consistent versions, we must reconstruct the tracker.
if (!status.ok()) {
// 2. fetch missing version after delete
@ -859,7 +859,8 @@ bool Tablet::_reconstruct_version_tracker_if_necessary() {
}
Status Tablet::capture_consistent_versions(const Version& spec_version,
std::vector<Version>* version_path, bool quiet) const {
std::vector<Version>* version_path,
bool skip_missing_version, bool quiet) const {
Status status =
_timestamped_version_tracker.capture_consistent_versions(spec_version, version_path);
if (!status.ok() && !quiet) {
@ -878,6 +879,10 @@ Status Tablet::capture_consistent_versions(const Version& spec_version,
LOG(WARNING) << "status:" << status << ", tablet:" << tablet_id()
<< ", missed version for version:" << spec_version;
_print_missed_versions(missed_versions);
if (skip_missing_version) {
LOG(WARNING) << "force skipping missing version for tablet:" << tablet_id();
return Status::OK();
}
}
}
}
@ -886,7 +891,7 @@ Status Tablet::capture_consistent_versions(const Version& spec_version,
Status Tablet::check_version_integrity(const Version& version, bool quiet) {
std::shared_lock rdlock(_meta_lock);
return capture_consistent_versions(version, nullptr, quiet);
return capture_consistent_versions(version, nullptr, false, quiet);
}
bool Tablet::exceed_version_limit(int32_t limit) const {
@ -919,7 +924,7 @@ void Tablet::acquire_version_and_rowsets(
Status Tablet::capture_consistent_rowsets(const Version& spec_version,
std::vector<RowsetSharedPtr>* rowsets) const {
std::vector<Version> version_path;
RETURN_IF_ERROR(capture_consistent_versions(spec_version, &version_path));
RETURN_IF_ERROR(capture_consistent_versions(spec_version, &version_path, false, false));
RETURN_IF_ERROR(_capture_consistent_rowsets_unlocked(version_path, rowsets));
return Status::OK();
}
@ -955,10 +960,11 @@ Status Tablet::_capture_consistent_rowsets_unlocked(const std::vector<Version>&
return Status::OK();
}
Status Tablet::capture_rs_readers(const Version& spec_version,
std::vector<RowSetSplits>* rs_splits) const {
Status Tablet::capture_rs_readers(const Version& spec_version, std::vector<RowSetSplits>* rs_splits,
bool skip_missing_version) const {
std::vector<Version> version_path;
RETURN_IF_ERROR(capture_consistent_versions(spec_version, &version_path));
RETURN_IF_ERROR(
capture_consistent_versions(spec_version, &version_path, skip_missing_version, false));
RETURN_IF_ERROR(capture_rs_readers(version_path, rs_splits));
return Status::OK();
}
@ -3576,7 +3582,8 @@ Status Tablet::check_rowid_conversion(
Status Tablet::all_rs_id(int64_t max_version, RowsetIdUnorderedSet* rowset_ids) const {
// Ensure that the obtained versions of rowsets are continuous
std::vector<Version> version_path;
RETURN_IF_ERROR(capture_consistent_versions(Version(0, max_version), &version_path));
RETURN_IF_ERROR(
capture_consistent_versions(Version(0, max_version), &version_path, false, false));
for (auto& ver : version_path) {
if (ver.second == 1) {
// [0-1] rowset is empty for each tablet, skip it

8
be/src/olap/tablet.h
View File

@ -173,9 +173,10 @@ public:
// Given spec_version, find a continuous version path and store it in version_path.
// If quiet is true, then only "does this path exist" is returned.
// If skip_missing_version is true, return ok even there are missing versions.
Status capture_consistent_versions(const Version& spec_version,
std::vector<Version>* version_path,
bool quiet = false) const;
bool skip_missing_version, bool quiet) const;
// if quiet is true, no error log will be printed if there are missing versions
Status check_version_integrity(const Version& version, bool quiet = false);
bool check_version_exist(const Version& version) const;
@ -184,8 +185,9 @@ public:
Status capture_consistent_rowsets(const Version& spec_version,
std::vector<RowsetSharedPtr>* rowsets) const;
Status capture_rs_readers(const Version& spec_version,
std::vector<RowSetSplits>* rs_splits) const override;
// If skip_missing_version is true, skip versions if they are missing.
Status capture_rs_readers(const Version& spec_version, std::vector<RowSetSplits>* rs_splits,
bool skip_missing_version) const override;
Status capture_rs_readers(const std::vector<Version>& version_path,
std::vector<RowSetSplits>* rs_splits) const;

4
be/src/runtime/runtime_state.h
View File

@ -367,6 +367,10 @@ public:
return _query_options.__isset.skip_delete_bitmap && _query_options.skip_delete_bitmap;
}
bool skip_missing_version() const {
return _query_options.__isset.skip_missing_version && _query_options.skip_missing_version;
}
bool enable_page_cache() const;
int partitioned_hash_join_rows_threshold() const {

2
be/src/vec/exec/scan/new_olap_scan_node.cpp
View File

@ -483,7 +483,7 @@ Status NewOlapScanNode::_init_scanners(std::list<VScannerSPtr>* scanners) {
auto& read_source = tablets_read_source.emplace_back();
{
std::shared_lock rdlock(tablet->get_header_lock());
auto st = tablet->capture_rs_readers({0, version}, &read_source.rs_splits);
auto st = tablet->capture_rs_readers({0, version}, &read_source.rs_splits, false);
if (!st.ok()) {
LOG(WARNING) << "fail to init reader.res=" << st;
return Status::InternalError(

3
be/src/vec/exec/scan/new_olap_scanner.cpp
View File

@ -182,7 +182,8 @@ Status NewOlapScanner::init() {
{
std::shared_lock rdlock(tablet->get_header_lock());
auto st = tablet->capture_rs_readers(_tablet_reader_params.version,
&read_source.rs_splits);
&read_source.rs_splits,
_state->skip_missing_version());
if (!st.ok()) {
LOG(WARNING) << "fail to init reader.res=" << st;
return Status::InternalError(

4
be/test/olap/tablet_test.cpp
View File

@ -282,12 +282,12 @@ TEST_F(TestTablet, pad_rowset) {
Version version(5, 5);
std::vector<RowSetSplits> splits;
ASSERT_FALSE(_tablet->capture_rs_readers(version, &splits).ok());
ASSERT_FALSE(_tablet->capture_rs_readers(version, &splits, false).ok());
splits.clear();
PadRowsetAction action(nullptr, TPrivilegeHier::GLOBAL, TPrivilegeType::ADMIN);
static_cast<void>(action._pad_rowset(_tablet, version));
ASSERT_TRUE(_tablet->capture_rs_readers(version, &splits).ok());
ASSERT_TRUE(_tablet->capture_rs_readers(version, &splits, false).ok());
}
TEST_F(TestTablet, cooldown_policy) {

14
docs/en/docs/admin-manual/config/fe-config.md
View File

@ -1814,20 +1814,6 @@ In some very special circumstances, such as code bugs, or human misoperation, et
Set to true so that Doris will automatically use blank replicas to fill tablets which all replicas have been damaged or missing
#### `recover_with_skip_missing_version`
Default：disable
IsMutable：true
MasterOnly：true
In some scenarios, there is an unrecoverable metadata problem in the cluster, and the visibleVersion of the data does not match be. In this case, it is still necessary to restore the remaining data (which may cause problems with the correctness of the data). This configuration is the same as` recover_with_empty_tablet` should only be used in emergency situations
This configuration has three values:
* disable : If an exception occurs, an error will be reported normally.
* ignore_version: ignore the visibleVersion information recorded in fe partition, use replica version
* ignore_all: In addition to ignore_version, when encountering no queryable replica, skip it directly instead of throwing an exception
#### `min_clone_task_timeout_sec` `And max_clone_task_timeout_sec`
Default：Minimum 3 minutes, maximum two hours

4
docs/en/docs/advanced/variables.md
View File

@ -575,6 +575,10 @@ Note that the comment must start with /*+ and can only follow the SELECT.
For debugging purpose. In Unique Key MoW table, in case of problems of reading data, setting value to `true` will also read deleted data.
* `skip_missing_version`
In some scenarios, all replicas of tablet are having missing versions, and the tablet is unable to recover. This config can control the behavior of query. When it is opened, the query will ignore the visible version recorded in FE partition, use the replica version. If the replica on be has missing versions, the query will directly skip this missing version, and only return the data of the existing version, In addition, the query will always try to select the one with the highest lastSuccessVersion among all surviving BE replicas, so as to recover as much data as possible. You should only open it in the emergency scenarios mentioned above, only used for temporary recovery queries. Note that, this variable conflicts with the a variable, when the a variable is not -1, this variable will not work.
* `default_password_lifetime`
Default password expiration time. The default value is 0, which means no expiration. The unit is days. This parameter is only enabled if the user's password expiration property has a value of DEFAULT. like:

18
docs/zh-CN/docs/admin-manual/config/fe-config.md
View File

@ -1815,24 +1815,6 @@ show data （其他用法：HELP SHOW DATA）
在这种情况下，您可以将此配置设置为 true。 系统会将损坏的 tablet 替换为空 tablet，以确保查询可以执行。 （但此时数据已经丢失，所以查询结果可能不准确）
#### `recover_with_skip_missing_version`
默认值：disable
是否可以动态配置：true
是否为 Master FE 节点独有的配置项：true
有些场景下集群出现了不可恢复的元数据问题，数据已的visibleversion 已经和be 不匹配，
这种情况下仍然需要恢复剩余的数据（可能能会导致数据的正确性有问题），这个配置同`recover_with_empty_tablet` 一样只能在紧急情况下使用
这个配置有三个值：
* disable ：出现异常会正常报错。
* ignore_version: 忽略 fe partition 中记录的visibleVersion 信息， 使用replica version
* ignore_all: 除了ignore_version， 在遇到找不到可查询的replica 时，直接跳过而不是抛出异常
#### `min_clone_task_timeout_sec` 和 `max_clone_task_timeout_sec`
默认值：最小3分钟，最大两小时

4
docs/zh-CN/docs/advanced/variables.md
View File

@ -563,6 +563,10 @@ try (Connection conn = DriverManager.getConnection("jdbc:mysql://127.0.0.1:9030/
用于调试目的。在Unique Key MoW表中，当发现读取表的数据结果有误的时候，把此变量的值设置为`true`，将会把被delete bitmap标记删除的数据当成正常数据读取。
* `skip_missing_version`
有些极端场景下，表的 Tablet 下的所有的所有副本都有版本缺失，使得这些 Tablet 没有办法被恢复，导致整张表都不能查询。这个变量可以用来控制查询的行为，打设置为`true`时，查询会忽略 FE partition 中记录的 visibleVersion，使用 replica version。如果 Be 上的 Replica 有缺失的版本，则查询会直接跳过这些缺失的版本，只返回仍存在版本的数据。此外，查询将会总是选择所有存活的 BE 中所有 Replica 里 lastSuccessVersion 最大的那一个，这样可以尽可能的恢复更多的数据。这个变量应该只在上述紧急情况下才被设置为`true`，仅用于临时让表恢复查询。注意，此变量与 use_fix_replica 变量冲突，当 use_fix_replica 变量不等于 -1 时，此变量会不起作用
* `default_password_lifetime`
默认的密码过期时间。默认值为 0，即表示不过期。单位为天。该参数只有当用户的密码过期属性为 DEFAULT 值时，才启用。如：

14
fe/fe-common/src/main/java/org/apache/doris/common/Config.java
View File

@ -1253,20 +1253,6 @@ public class Config extends ConfigBase {
@ConfField(mutable = true, masterOnly = true)
public static boolean recover_with_empty_tablet = false;
/**
* In some scenarios, there is an unrecoverable metadata problem in the cluster,
* and the visibleVersion of the data does not match be. In this case, it is still
* necessary to restore the remaining data (which may cause problems with the correctness of the data).
* This configuration is the same as` recover_with_empty_tablet` should only be used in emergency situations
* This configuration has three values:
* disable : If an exception occurs, an error will be reported normally.
* ignore_version: ignore the visibleVersion information recorded in fe partition, use replica version
* ignore_all: In addition to ignore_version, when encountering no queryable replica,
* skip it directly instead of throwing an exception
*/
@ConfField(mutable = true, masterOnly = true)
public static String recover_with_skip_missing_version = "disable";
/**
* Whether to add a delete sign column when create unique table
*/

18
fe/fe-core/src/main/java/org/apache/doris/catalog/Replica.java
View File

@ -38,6 +38,8 @@ import java.util.Comparator;
public class Replica implements Writable {
private static final Logger LOG = LogManager.getLogger(Replica.class);
public static final VersionComparator<Replica> VERSION_DESC_COMPARATOR = new VersionComparator<Replica>();
public static final LastSuccessVersionComparator<Replica> LAST_SUCCESS_VERSION_COMPARATOR =
new LastSuccessVersionComparator<Replica>();
public static final IdComparator<Replica> ID_COMPARATOR = new IdComparator<Replica>();
public enum ReplicaState {
@ -682,6 +684,22 @@ public class Replica implements Writable {
}
}
private static class LastSuccessVersionComparator<T extends Replica> implements Comparator<T> {
public LastSuccessVersionComparator() {
}
@Override
public int compare(T replica1, T replica2) {
if (replica1.getLastSuccessVersion() < replica2.getLastSuccessVersion()) {
return 1;
} else if (replica1.getLastSuccessVersion() == replica2.getLastSuccessVersion()) {
return 0;
} else {
return -1;
}
}
}
private static class IdComparator<T extends Replica> implements Comparator<T> {
public IdComparator() {
}

4
fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java
View File

@ -238,7 +238,7 @@ public class Tablet extends MetaObject implements Writable {
}
// for query
public List<Replica> getQueryableReplicas(long visibleVersion) {
public List<Replica> getQueryableReplicas(long visibleVersion, boolean allowFailedVersion) {
List<Replica> allQueryableReplica = Lists.newArrayListWithCapacity(replicas.size());
List<Replica> auxiliaryReplica = Lists.newArrayListWithCapacity(replicas.size());
for (Replica replica : replicas) {
@ -247,7 +247,7 @@ public class Tablet extends MetaObject implements Writable {
}
// Skip the missing version replica
if (replica.getLastFailedVersion() > 0) {
if (replica.getLastFailedVersion() > 0 && !allowFailedVersion) {
continue;
}

34
fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
View File

@ -713,14 +713,19 @@ public class OlapScanNode extends ScanNode {
String visibleVersionStr = String.valueOf(visibleVersion);
Set<Tag> allowedTags = Sets.newHashSet();
int useFixReplica = -1;
boolean needCheckTags = false;
boolean skipMissingVersion = false;
if (ConnectContext.get() != null) {
allowedTags = ConnectContext.get().getResourceTags();
needCheckTags = ConnectContext.get().isResourceTagsSet();
useFixReplica = ConnectContext.get().getSessionVariable().useFixReplica;
// if use_fix_replica is set to true, set skip_missing_version to false
skipMissingVersion = useFixReplica == -1 && ConnectContext.get().getSessionVariable().skipMissingVersion;
}
for (Tablet tablet : tablets) {
long tabletId = tablet.getId();
if (!Config.recover_with_skip_missing_version.equalsIgnoreCase("disable")) {
if (skipMissingVersion) {
long tabletVersion = -1L;
for (Replica replica : tablet.getReplicas()) {
if (replica.getVersion() > tabletVersion) {
@ -743,7 +748,7 @@ public class OlapScanNode extends ScanNode {
paloRange.setTabletId(tabletId);
// random shuffle List && only collect one copy
List<Replica> replicas = tablet.getQueryableReplicas(visibleVersion);
List<Replica> replicas = tablet.getQueryableReplicas(visibleVersion, skipMissingVersion);
if (replicas.isEmpty()) {
LOG.warn("no queryable replica found in tablet {}. visible version {}", tabletId, visibleVersion);
StringBuilder sb = new StringBuilder(
@ -757,12 +762,13 @@ public class OlapScanNode extends ScanNode {
throw new UserException(sb.toString());
}
int useFixReplica = -1;
if (ConnectContext.get() != null) {
useFixReplica = ConnectContext.get().getSessionVariable().useFixReplica;
}
if (useFixReplica == -1) {
Collections.shuffle(replicas);
if (skipMissingVersion) {
// sort by replica's last success version, higher success version in the front.
replicas.sort(Replica.LAST_SUCCESS_VERSION_COMPARATOR);
} else {
Collections.shuffle(replicas);
}
} else {
LOG.debug("use fix replica, value: {}, replica num: {}", useFixReplica, replicas.size());
// sort by replica id
@ -771,6 +777,7 @@ public class OlapScanNode extends ScanNode {
replicas.clear();
replicas.add(replica);
}
final long coolDownReplicaId = tablet.getCooldownReplicaId();
// we prefer to query using cooldown replica to make sure the cache is fully utilized
// for example: consider there are 3BEs(A,B,C) and each has one replica for tablet X. and X
@ -832,14 +839,15 @@ public class OlapScanNode extends ScanNode {
collectedStat = true;
}
scanBackendIds.add(backend.getId());
// For skipping missing version of tablet, we only select the backend with the highest last
// success version replica to save as much data as possible.
if (skipMissingVersion) {
break;
}
}
if (tabletIsNull) {
if (Config.recover_with_skip_missing_version.equalsIgnoreCase("ignore_all")) {
continue;
} else {
throw new UserException(tabletId + " have no queryable replicas. err: "
+ Joiner.on(", ").join(errs));
}
throw new UserException(tabletId + " have no queryable replicas. err: "
+ Joiner.on(", ").join(errs));
}
TScanRange scanRange = new TScanRange();
scanRange.setPaloScanRange(paloRange);

17
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
View File

@ -293,6 +293,8 @@ public class SessionVariable implements Serializable, Writable {
public static final String SKIP_DELETE_BITMAP = "skip_delete_bitmap";
public static final String SKIP_MISSING_VERSION = "skip_missing_version";
public static final String ENABLE_PUSH_DOWN_NO_GROUP_AGG = "enable_push_down_no_group_agg";
public static final String ENABLE_CBO_STATISTICS = "enable_cbo_statistics";
@ -976,6 +978,19 @@ public class SessionVariable implements Serializable, Writable {
@VariableMgr.VarAttr(name = SKIP_DELETE_BITMAP)
public boolean skipDeleteBitmap = false;
// This variable replace the original FE config `recover_with_skip_missing_version`.
// In some scenarios, all replicas of tablet are having missing versions, and the tablet is unable to recover.
// This config can control the behavior of query. When it is set to `true`, the query will ignore the
// visible version recorded in FE partition, use the replica version. If the replica on BE has missing versions,
// the query will directly skip this missing version, and only return the data of the existing versions.
// Besides, the query will always try to select the one with the highest lastSuccessVersion among all surviving
// BE replicas, so as to recover as much data as possible.
// You should only open it in the emergency scenarios mentioned above, only used for temporary recovery queries.
// This variable conflicts with the use_fix_replica variable, when the use_fix_replica variable is not -1,
// this variable will not work.
@VariableMgr.VarAttr(name = SKIP_MISSING_VERSION)
public boolean skipMissingVersion = false;
// This variable is used to avoid FE fallback to the original parser. When we execute SQL in regression tests
// for nereids, fallback will cause the Doris return the correct result although the syntax is unsupported
// in nereids for some mistaken modification. You should set it on the
@ -2436,6 +2451,8 @@ public class SessionVariable implements Serializable, Writable {
tResult.setEnableDecimal256(enableNereidsPlanner && enableDecimal256);
tResult.setSkipMissingVersion(skipMissingVersion);
return tResult;
}

2
gensrc/thrift/PaloInternalService.thrift
View File

@ -253,6 +253,8 @@ struct TQueryOptions {
88: optional bool enable_decimal256 = false;
89: optional bool enable_local_shuffle = false;
// For emergency use, skip missing version when reading rowsets
90: optional bool skip_missing_version = false;
}

5
regression-test/data/query_p0/session_variable/test_skip_missing_version.out Normal file
View File

@ -0,0 +1,5 @@
-- This file is automatically generated. You should know what you did if you want to edit this
-- !select_all --
1000 a 10
2000 b 10

42
regression-test/suites/query_p0/session_variable/test_skip_missing_version.groovy Normal file
View File

@ -0,0 +1,42 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
suite("test_skip_missing_version") {
def test_tbl = "test_skip_missing_version_tbl"
sql """ DROP TABLE IF EXISTS ${test_tbl}"""
sql """
CREATE TABLE ${test_tbl} (
`k1` int(11) NULL,
`k2` char(5) NULL,
`k3` tinyint(4) NULL
) ENGINE=OLAP
DUPLICATE KEY(`k1`, `k2`, `k3`)
DISTRIBUTED BY HASH(`k1`) BUCKETS 5
PROPERTIES (
"replication_num"="1"
);
"""
sql """ INSERT INTO ${test_tbl} VALUES(1000, 'a', 10); """
sql """ INSERT INTO ${test_tbl} VALUES(2000, 'b', 10); """
// This case cannot verify the results, but it can verify abnormalities after
// SET skip_missing_version=true
sql """ SET skip_missing_version=true """
qt_select_all """ select * from ${test_tbl} order by k1 """
}