Skip to content

[branch-2.0](delete) Fix delete job timeout when executing delete from ... #40028

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

Merged
dataroaring merged 2 commits into from
Aug 28, 2024
Merged

[branch-2.0](delete) Fix delete job timeout when executing delete from ... #40028

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ac2172a
pick
bobhan1 Aug 28, 2024
1dff0f9
modify case
bobhan1 Aug 28, 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
29 changes: 17 additions & 12 deletions be/src/olap/delete_handler.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
#include "olap/olap_common.h"
#include "olap/predicate_creator.h"
#include "olap/utils.h"
#include "util/debug_points.h"

using apache::thrift::ThriftDebugString;
using std::vector;
Expand All @@ -48,17 +49,21 @@ using namespace ErrorCode;
Status DeleteHandler::generate_delete_predicate(const TabletSchema& schema,
const std::vector<TCondition>& conditions,
DeletePredicatePB* del_pred) {
DBUG_EXECUTE_IF("DeleteHandler::generate_delete_predicate.inject_failure", {
return Status::Error<false>(dp->param<int>("error_code"),
dp->param<std::string>("error_msg"));
})
if (conditions.empty()) {
return Status::Error<DELETE_INVALID_PARAMETERS>(
return Status::Error<ErrorCode::INVALID_ARGUMENT>(
"invalid parameters for store_cond. condition_size={}", conditions.size());
}

// Check whether the delete condition meets the requirements
for (const TCondition& condition : conditions) {
if (check_condition_valid(schema, condition) != Status::OK()) {
// Error will print log, no need to do it manually.
return Status::Error<DELETE_INVALID_CONDITION>("invalid condition. condition={}",
ThriftDebugString(condition));
return Status::Error<ErrorCode::INVALID_ARGUMENT>("invalid condition. condition={}",
ThriftDebugString(condition));
}
}

Expand All @@ -80,7 +85,7 @@ Status DeleteHandler::generate_delete_predicate(const TabletSchema& schema,
if (TCondition tmp; !DeleteHandler::parse_delete_condition(condition_str, &tmp)) {
LOG(WARNING) << "failed to parse condition_str, condtion="
<< ThriftDebugString(condition);
return Status::Error<DELETE_INVALID_CONDITION>(
return Status::Error<ErrorCode::INVALID_ARGUMENT>(
"failed to parse condition_str, condtion={}", ThriftDebugString(condition));
}
VLOG_NOTICE << __PRETTY_FUNCTION__ << " condition_str: " << condition_str;
Expand Down Expand Up @@ -172,8 +177,8 @@ Status DeleteHandler::check_condition_valid(const TabletSchema& schema, const TC
// Check whether the column exists
int32_t field_index = schema.field_index(cond.column_name);
if (field_index < 0) {
return Status::Error<DELETE_INVALID_CONDITION>("field is not existent. [field_index={}]",
field_index);
return Status::Error<ErrorCode::INVALID_ARGUMENT>("field is not existent. [field_index={}]",
field_index);
}

// Delete condition should only applied on key columns or duplicate key table, and
Expand All @@ -182,21 +187,21 @@ Status DeleteHandler::check_condition_valid(const TabletSchema& schema, const TC

if (column.type() == FieldType::OLAP_FIELD_TYPE_DOUBLE ||
column.type() == FieldType::OLAP_FIELD_TYPE_FLOAT) {
return Status::Error<DELETE_INVALID_CONDITION>("data type is float or double.");
return Status::Error<ErrorCode::INVALID_ARGUMENT>("data type is float or double.");
}

// Check operator and operands size are matched.
if ("*=" != cond.condition_op && "!*=" != cond.condition_op &&
cond.condition_values.size() != 1) {
return Status::Error<DELETE_INVALID_CONDITION>("invalid condition value size. [size={}]",
cond.condition_values.size());
return Status::Error<ErrorCode::INVALID_ARGUMENT>("invalid condition value size. [size={}]",
cond.condition_values.size());
}

// Check each operand is valid
for (const auto& condition_value : cond.condition_values) {
if (!is_condition_value_valid(column, cond.condition_op, condition_value)) {
return Status::Error<DELETE_INVALID_CONDITION>("invalid condition value. [value={}]",
condition_value);
return Status::Error<ErrorCode::INVALID_ARGUMENT>("invalid condition value. [value={}]",
condition_value);
}
}

Expand Down Expand Up @@ -267,7 +272,7 @@ Status DeleteHandler::init(TabletSchemaSPtr tablet_schema,
for (const auto& sub_predicate : delete_condition.sub_predicates()) {
TCondition condition;
if (!parse_delete_condition(sub_predicate, &condition)) {
return Status::Error<DELETE_INVALID_PARAMETERS>(
return Status::Error<ErrorCode::INVALID_ARGUMENT>(
"fail to parse condition. condition={}", sub_predicate);
}
int32_t col_unique_id =
Expand Down
4 changes: 4 additions & 0 deletions be/src/olap/push_handler.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,10 @@ Status PushHandler::_do_streaming_ingestion(TabletSharedPtr tablet, const TPushR
}

std::shared_lock base_migration_rlock(tablet->get_migration_lock(), std::try_to_lock);
DBUG_EXECUTE_IF("PushHandler::_do_streaming_ingestion.try_lock_fail", {
return Status::Error<TRY_LOCK_FAILED>(
"PushHandler::_do_streaming_ingestion get lock failed");
})
if (!base_migration_rlock.owns_lock()) {
return Status::Error<TRY_LOCK_FAILED>(
"PushHandler::_do_streaming_ingestion get lock failed");
Expand Down
52 changes: 26 additions & 26 deletions be/test/olap/delete_handler_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -408,7 +408,7 @@ TEST_F(TestDeleteConditionHandler, StoreCondInvalidParameters) {
DeletePredicatePB del_pred;
Status failed_res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(),
conditions, &del_pred);
EXPECT_EQ(Status::Error<DELETE_INVALID_PARAMETERS>(""), failed_res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), failed_res);
}

// 检测过滤条件中指定的列不存在,或者列不符合要求
Expand All @@ -424,7 +424,7 @@ TEST_F(TestDeleteConditionHandler, StoreCondNonexistentColumn) {
DeletePredicatePB del_pred;
Status failed_res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(),
conditions, &del_pred);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), failed_res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), failed_res);

// 'v'是value列
conditions.clear();
Expand Down Expand Up @@ -630,15 +630,15 @@ TEST_F(TestDeleteConditionHandler2, InvalidConditionValue) {
DeletePredicatePB del_pred_1;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_1);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k1的值越下界，k1类型为int8
conditions[0].condition_values.clear();
conditions[0].condition_values.push_back("-1000");
DeletePredicatePB del_pred_2;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_2);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k2的值越上界，k2类型为int16
conditions[0].condition_values.clear();
Expand All @@ -647,15 +647,15 @@ TEST_F(TestDeleteConditionHandler2, InvalidConditionValue) {
DeletePredicatePB del_pred_3;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_3);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k2的值越下界，k2类型为int16
conditions[0].condition_values.clear();
conditions[0].condition_values.push_back("-32769");
DeletePredicatePB del_pred_4;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_4);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k3的值越上界，k3类型为int32
conditions[0].condition_values.clear();
Expand All @@ -664,15 +664,15 @@ TEST_F(TestDeleteConditionHandler2, InvalidConditionValue) {
DeletePredicatePB del_pred_5;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_5);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k3的值越下界，k3类型为int32
conditions[0].condition_values.clear();
conditions[0].condition_values.push_back("-2147483649");
DeletePredicatePB del_pred_6;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_6);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k4的值越上界，k2类型为int64
conditions[0].condition_values.clear();
Expand All @@ -681,15 +681,15 @@ TEST_F(TestDeleteConditionHandler2, InvalidConditionValue) {
DeletePredicatePB del_pred_7;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_7);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k4的值越下界，k1类型为int64
conditions[0].condition_values.clear();
conditions[0].condition_values.push_back("-9223372036854775809");
DeletePredicatePB del_pred_8;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_8);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k5的值越上界，k5类型为int128
conditions[0].condition_values.clear();
Expand All @@ -698,15 +698,15 @@ TEST_F(TestDeleteConditionHandler2, InvalidConditionValue) {
DeletePredicatePB del_pred_9;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_9);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k5的值越下界，k5类型为int128
conditions[0].condition_values.clear();
conditions[0].condition_values.push_back("-170141183460469231731687303715884105729");
DeletePredicatePB del_pred_10;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_10);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k9整数部分长度过长，k9类型为decimal, precision=6, frac=3
conditions[0].condition_values.clear();
Expand All @@ -715,23 +715,23 @@ TEST_F(TestDeleteConditionHandler2, InvalidConditionValue) {
DeletePredicatePB del_pred_11;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_11);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k9小数部分长度过长，k9类型为decimal, precision=6, frac=3
conditions[0].condition_values.clear();
conditions[0].condition_values.push_back("1.2345678");
DeletePredicatePB del_pred_12;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_12);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k9没有小数部分，但包含小数点
conditions[0].condition_values.clear();
conditions[0].condition_values.push_back("1.");
DeletePredicatePB del_pred_13;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_13);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k10类型的过滤值不符合对应格式，k10为date
conditions[0].condition_values.clear();
Expand All @@ -740,21 +740,21 @@ TEST_F(TestDeleteConditionHandler2, InvalidConditionValue) {
DeletePredicatePB del_pred_14;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_14);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

conditions[0].condition_values.clear();
conditions[0].condition_values.push_back("2013-64-01");
DeletePredicatePB del_pred_15;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_15);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

conditions[0].condition_values.clear();
conditions[0].condition_values.push_back("2013-01-40");
DeletePredicatePB del_pred_16;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_16);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k11类型的过滤值不符合对应格式，k11为datetime
conditions[0].condition_values.clear();
Expand All @@ -763,42 +763,42 @@ TEST_F(TestDeleteConditionHandler2, InvalidConditionValue) {
DeletePredicatePB del_pred_17;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_17);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

conditions[0].condition_values.clear();
conditions[0].condition_values.push_back("2013-64-01 00:00:00");
DeletePredicatePB del_pred_18;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_18);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

conditions[0].condition_values.clear();
conditions[0].condition_values.push_back("2013-01-40 00:00:00");
DeletePredicatePB del_pred_19;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_19);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

conditions[0].condition_values.clear();
conditions[0].condition_values.push_back("2013-01-01 24:00:00");
DeletePredicatePB del_pred_20;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_20);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

conditions[0].condition_values.clear();
conditions[0].condition_values.push_back("2013-01-01 00:60:00");
DeletePredicatePB del_pred_21;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_21);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

conditions[0].condition_values.clear();
conditions[0].condition_values.push_back("2013-01-01 00:00:60");
DeletePredicatePB del_pred_22;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_22);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

// 测试k12和k13类型的过滤值过长，k12,k13类型分别为string(64), varchar(64)
conditions[0].condition_values.clear();
Expand All @@ -810,7 +810,7 @@ TEST_F(TestDeleteConditionHandler2, InvalidConditionValue) {
DeletePredicatePB del_pred_23;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_23);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);

conditions[0].condition_values.clear();
conditions[0].column_name = "k13";
Expand All @@ -821,7 +821,7 @@ TEST_F(TestDeleteConditionHandler2, InvalidConditionValue) {
DeletePredicatePB del_pred_24;
res = DeleteHandler::generate_delete_predicate(*tablet->tablet_schema(), conditions,
&del_pred_24);
EXPECT_EQ(Status::Error<DELETE_INVALID_CONDITION>(""), res);
EXPECT_EQ(Status::Error<INVALID_ARGUMENT>(""), res);
}

class TestDeleteHandler : public testing::Test {
Expand Down
6 changes: 6 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/load/DeleteHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -406,6 +406,12 @@ public void process(DeleteStmt stmt) throws DdlException, QueryStateException {
break;
}
} else {
if (!countDownLatch.getStatus().ok()) {
// encounter some errors that don't need to retry, abort directly
LOG.warn("delete job failed, errmsg={}", countDownLatch.getStatus().getErrorMsg());
throw new DdlException(String.format("delete job failed, errmsg:%s",
countDownLatch.getStatus().getErrorMsg()));
}
commitJob(deleteJob, db, olapTable, timeoutMs);
}
} finally {
Expand Down
Loading