| // 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. |
| |
| #include "olap/delete_handler.h" |
| |
| #include <gen_cpp/PaloInternalService_types.h> |
| #include <gen_cpp/olap_file.pb.h> |
| #include <thrift/protocol/TDebugProtocol.h> |
| |
| #include <string> |
| #include <vector> |
| |
| #include "common/config.h" |
| #include "common/logging.h" |
| #include "common/status.h" |
| #include "olap/block_column_predicate.h" |
| #include "olap/olap_common.h" |
| #include "olap/predicate_creator.h" |
| #include "olap/tablet_schema.h" |
| #include "olap/utils.h" |
| #include "util/debug_points.h" |
| #include "vec/functions/cast/cast_parameters.h" |
| #include "vec/functions/cast/cast_to_boolean.h" |
| #include "vec/functions/cast/cast_to_date_or_datetime_impl.hpp" |
| #include "vec/functions/cast/cast_to_datetimev2_impl.hpp" |
| #include "vec/functions/cast/cast_to_datev2_impl.hpp" |
| #include "vec/functions/cast/cast_to_decimal.h" |
| #include "vec/functions/cast/cast_to_float.h" |
| #include "vec/functions/cast/cast_to_int.h" |
| #include "vec/functions/cast/cast_to_ip.h" |
| |
| using apache::thrift::ThriftDebugString; |
| using std::vector; |
| using std::string; |
| |
| using ::google::protobuf::RepeatedPtrField; |
| |
| namespace doris { |
| |
| template <PrimitiveType PType> |
| Status convert(const vectorized::DataTypePtr& data_type, const std::string& str, |
| vectorized::Arena& arena, typename PrimitiveTypeTraits<PType>::CppType& res) { |
| if constexpr (PType == TYPE_TINYINT || PType == TYPE_SMALLINT || PType == TYPE_INT || |
| PType == TYPE_BIGINT || PType == TYPE_LARGEINT) { |
| vectorized::CastParameters parameters; |
| if (!vectorized::CastToInt::from_string({str.data(), str.size()}, res, parameters)) { |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "invalid {} string. str={}", type_to_string(data_type->get_primitive_type()), |
| str); |
| } |
| return Status::OK(); |
| } |
| if constexpr (PType == TYPE_FLOAT || PType == TYPE_DOUBLE) { |
| vectorized::CastParameters parameters; |
| if (!vectorized::CastToFloat::from_string({str.data(), str.size()}, res, parameters)) { |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "invalid {} string. str={}", type_to_string(data_type->get_primitive_type()), |
| str); |
| } |
| return Status::OK(); |
| } |
| if constexpr (PType == TYPE_DATE) { |
| vectorized::CastParameters parameters; |
| if (!vectorized::CastToDateOrDatetime::from_string<false>({str.data(), str.size()}, res, |
| nullptr, parameters)) { |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "invalid {} string. str={}", type_to_string(data_type->get_primitive_type()), |
| str); |
| } |
| return Status::OK(); |
| } |
| if constexpr (PType == TYPE_DATETIME) { |
| vectorized::CastParameters parameters; |
| if (!vectorized::CastToDateOrDatetime::from_string<true>({str.data(), str.size()}, res, |
| nullptr, parameters)) { |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "invalid {} string. str={}", type_to_string(data_type->get_primitive_type()), |
| str); |
| } |
| return Status::OK(); |
| } |
| if constexpr (PType == TYPE_DATEV2) { |
| vectorized::CastParameters parameters; |
| if (!vectorized::CastToDateV2::from_string({str.data(), str.size()}, res, nullptr, |
| parameters)) { |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "invalid {} string. str={}", type_to_string(data_type->get_primitive_type()), |
| str); |
| } |
| return Status::OK(); |
| } |
| if constexpr (PType == TYPE_DATETIMEV2) { |
| vectorized::CastParameters parameters; |
| if (!vectorized::CastToDatetimeV2::from_string({str.data(), str.size()}, res, nullptr, |
| data_type->get_scale(), parameters)) { |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "invalid {} string. str={}", type_to_string(data_type->get_primitive_type()), |
| str); |
| } |
| return Status::OK(); |
| } |
| if constexpr (PType == TYPE_TIMESTAMPTZ) { |
| vectorized::CastParameters parameters; |
| if (!vectorized::CastToTimstampTz::from_string({str.data(), str.size()}, res, parameters, |
| nullptr, data_type->get_scale())) { |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "invalid {} string. str={}", type_to_string(data_type->get_primitive_type()), |
| str); |
| } |
| return Status::OK(); |
| } |
| if constexpr (PType == TYPE_CHAR) { |
| size_t target = assert_cast<const vectorized::DataTypeString*>( |
| vectorized::remove_nullable(data_type).get()) |
| ->len(); |
| res = {str.data(), str.size()}; |
| if (target > str.size()) { |
| char* buffer = arena.alloc(target); |
| memset(buffer, 0, target); |
| memcpy(buffer, str.data(), str.size()); |
| res = {buffer, target}; |
| } |
| return Status::OK(); |
| } |
| if constexpr (PType == TYPE_STRING || PType == TYPE_VARCHAR) { |
| char* buffer = arena.alloc(str.size()); |
| memcpy(buffer, str.data(), str.size()); |
| res = {buffer, str.size()}; |
| return Status::OK(); |
| } |
| if constexpr (PType == TYPE_BOOLEAN) { |
| vectorized::CastParameters parameters; |
| vectorized::UInt8 tmp; |
| if (!vectorized::CastToBool::from_string({str.data(), str.size()}, tmp, parameters)) { |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "invalid {} string. str={}", type_to_string(data_type->get_primitive_type()), |
| str); |
| } |
| res = tmp != 0; |
| return Status::OK(); |
| } |
| if constexpr (PType == TYPE_IPV4) { |
| vectorized::CastParameters parameters; |
| if (!vectorized::CastToIPv4::from_string({str.data(), str.size()}, res, parameters)) { |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "invalid {} string. str={}", type_to_string(data_type->get_primitive_type()), |
| str); |
| } |
| return Status::OK(); |
| } |
| if constexpr (PType == TYPE_IPV6) { |
| vectorized::CastParameters parameters; |
| if (!vectorized::CastToIPv6::from_string({str.data(), str.size()}, res, parameters)) { |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "invalid {} string. str={}", type_to_string(data_type->get_primitive_type()), |
| str); |
| } |
| return Status::OK(); |
| } |
| if constexpr (PType == TYPE_DECIMALV2) { |
| vectorized::CastParameters parameters; |
| vectorized::Decimal128V2 tmp; |
| if (!vectorized::CastToDecimal::from_string({str.data(), str.size()}, tmp, |
| data_type->get_precision(), |
| data_type->get_scale(), parameters)) { |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "invalid {} string. str={}", type_to_string(data_type->get_primitive_type()), |
| str); |
| } |
| res = DecimalV2Value(tmp.value); |
| return Status::OK(); |
| } else if constexpr (is_decimal(PType)) { |
| vectorized::CastParameters parameters; |
| if (!vectorized::CastToDecimal::from_string({str.data(), str.size()}, res, |
| data_type->get_precision(), |
| data_type->get_scale(), parameters)) { |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "invalid {} string. str={}", type_to_string(data_type->get_primitive_type()), |
| str); |
| } |
| return Status::OK(); |
| } |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "unsupported data type in delete handler. type={}", |
| type_to_string(data_type->get_primitive_type())); |
| } |
| |
| #define CONVERT_CASE(PType) \ |
| case PType: { \ |
| set = build_set<PType>(); \ |
| for (const auto& s : str) { \ |
| typename PrimitiveTypeTraits<PType>::CppType tmp; \ |
| RETURN_IF_ERROR(convert<PType>(data_type, s, arena, tmp)); \ |
| set->insert(reinterpret_cast<const void*>(&tmp)); \ |
| } \ |
| return Status::OK(); \ |
| } |
| Status convert(const vectorized::DataTypePtr& data_type, const std::list<std::string>& str, |
| vectorized::Arena& arena, std::shared_ptr<HybridSetBase>& set) { |
| switch (data_type->get_primitive_type()) { |
| CONVERT_CASE(TYPE_TINYINT); |
| CONVERT_CASE(TYPE_SMALLINT); |
| CONVERT_CASE(TYPE_INT); |
| CONVERT_CASE(TYPE_BIGINT); |
| CONVERT_CASE(TYPE_LARGEINT); |
| CONVERT_CASE(TYPE_FLOAT); |
| CONVERT_CASE(TYPE_DOUBLE); |
| CONVERT_CASE(TYPE_DATE); |
| CONVERT_CASE(TYPE_DATETIME); |
| CONVERT_CASE(TYPE_DATEV2); |
| CONVERT_CASE(TYPE_DATETIMEV2); |
| CONVERT_CASE(TYPE_TIMESTAMPTZ); |
| CONVERT_CASE(TYPE_BOOLEAN); |
| CONVERT_CASE(TYPE_IPV4); |
| CONVERT_CASE(TYPE_IPV6); |
| CONVERT_CASE(TYPE_DECIMALV2); |
| CONVERT_CASE(TYPE_DECIMAL32); |
| CONVERT_CASE(TYPE_DECIMAL64); |
| CONVERT_CASE(TYPE_DECIMAL128I); |
| CONVERT_CASE(TYPE_DECIMAL256); |
| CONVERT_CASE(TYPE_CHAR); |
| CONVERT_CASE(TYPE_VARCHAR); |
| CONVERT_CASE(TYPE_STRING); |
| default: |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "unsupported data type in delete handler. type={}", |
| type_to_string(data_type->get_primitive_type())); |
| } |
| return Status::OK(); |
| } |
| #undef CONVERT_CASE |
| |
| #define CONVERT_CASE(PType) \ |
| case PType: { \ |
| typename PrimitiveTypeTraits<PType>::CppType tmp; \ |
| RETURN_IF_ERROR(convert<PType>(type, res.value_str.front(), arena, tmp)); \ |
| v.data = reinterpret_cast<const char*>(&tmp); \ |
| v.size = sizeof(tmp); \ |
| switch (res.condition_op) { \ |
| case PredicateType::EQ: \ |
| predicate = \ |
| create_comparison_predicate0<PredicateType::EQ>(index, type, v, true, arena); \ |
| return Status::OK(); \ |
| case PredicateType::NE: \ |
| predicate = \ |
| create_comparison_predicate0<PredicateType::NE>(index, type, v, true, arena); \ |
| return Status::OK(); \ |
| case PredicateType::GT: \ |
| predicate = \ |
| create_comparison_predicate0<PredicateType::GT>(index, type, v, true, arena); \ |
| return Status::OK(); \ |
| case PredicateType::GE: \ |
| predicate = \ |
| create_comparison_predicate0<PredicateType::GE>(index, type, v, true, arena); \ |
| return Status::OK(); \ |
| case PredicateType::LT: \ |
| predicate = \ |
| create_comparison_predicate0<PredicateType::LT>(index, type, v, true, arena); \ |
| return Status::OK(); \ |
| case PredicateType::LE: \ |
| predicate = \ |
| create_comparison_predicate0<PredicateType::LE>(index, type, v, true, arena); \ |
| return Status::OK(); \ |
| default: \ |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( \ |
| "invalid condition operator. operator={}", type_to_op_str(res.condition_op)); \ |
| } \ |
| } |
| Status parse_to_predicate(const uint32_t index, const vectorized::DataTypePtr& type, |
| DeleteHandler::ConditionParseResult& res, vectorized::Arena& arena, |
| std::shared_ptr<ColumnPredicate>& predicate) { |
| DCHECK_EQ(res.value_str.size(), 1); |
| if (res.condition_op == PredicateType::IS_NULL || |
| res.condition_op == PredicateType::IS_NOT_NULL) { |
| predicate = NullPredicate::create_shared( |
| index, res.condition_op == PredicateType::IS_NOT_NULL, type->get_primitive_type()); |
| return Status::OK(); |
| } |
| StringRef v; |
| switch (type->get_primitive_type()) { |
| CONVERT_CASE(TYPE_TINYINT); |
| CONVERT_CASE(TYPE_SMALLINT); |
| CONVERT_CASE(TYPE_INT); |
| CONVERT_CASE(TYPE_BIGINT); |
| CONVERT_CASE(TYPE_LARGEINT); |
| CONVERT_CASE(TYPE_FLOAT); |
| CONVERT_CASE(TYPE_DOUBLE); |
| CONVERT_CASE(TYPE_DATE); |
| CONVERT_CASE(TYPE_DATETIME); |
| CONVERT_CASE(TYPE_DATEV2); |
| CONVERT_CASE(TYPE_DATETIMEV2); |
| CONVERT_CASE(TYPE_TIMESTAMPTZ); |
| CONVERT_CASE(TYPE_BOOLEAN); |
| CONVERT_CASE(TYPE_IPV4); |
| CONVERT_CASE(TYPE_IPV6); |
| CONVERT_CASE(TYPE_DECIMALV2); |
| CONVERT_CASE(TYPE_DECIMAL32); |
| CONVERT_CASE(TYPE_DECIMAL64); |
| CONVERT_CASE(TYPE_DECIMAL128I); |
| CONVERT_CASE(TYPE_DECIMAL256); |
| case TYPE_CHAR: |
| case TYPE_VARCHAR: |
| case TYPE_STRING: { |
| RETURN_IF_ERROR(convert<TYPE_STRING>(type, res.value_str.front(), arena, v)); |
| switch (res.condition_op) { |
| case PredicateType::EQ: |
| predicate = |
| create_comparison_predicate0<PredicateType::EQ>(index, type, v, true, arena); |
| return Status::OK(); |
| case PredicateType::NE: |
| predicate = |
| create_comparison_predicate0<PredicateType::NE>(index, type, v, true, arena); |
| return Status::OK(); |
| case PredicateType::GT: |
| predicate = |
| create_comparison_predicate0<PredicateType::GT>(index, type, v, true, arena); |
| return Status::OK(); |
| case PredicateType::GE: |
| predicate = |
| create_comparison_predicate0<PredicateType::GE>(index, type, v, true, arena); |
| return Status::OK(); |
| case PredicateType::LT: |
| predicate = |
| create_comparison_predicate0<PredicateType::LT>(index, type, v, true, arena); |
| return Status::OK(); |
| case PredicateType::LE: |
| predicate = |
| create_comparison_predicate0<PredicateType::LE>(index, type, v, true, arena); |
| return Status::OK(); |
| default: |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "invalid condition operator. operator={}", type_to_op_str(res.condition_op)); |
| } |
| break; |
| } |
| default: |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "unsupported data type in delete handler. type={}", |
| type_to_string(type->get_primitive_type())); |
| } |
| return Status::OK(); |
| #undef CONVERT_CASE |
| } |
| |
| Status parse_to_in_predicate(const uint32_t index, const vectorized::DataTypePtr& type, |
| DeleteHandler::ConditionParseResult& res, vectorized::Arena& arena, |
| std::shared_ptr<ColumnPredicate>& predicate) { |
| DCHECK_GT(res.value_str.size(), 1); |
| switch (res.condition_op) { |
| case PredicateType::IN_LIST: { |
| std::shared_ptr<HybridSetBase> set; |
| RETURN_IF_ERROR(convert(type, res.value_str, arena, set)); |
| predicate = create_in_list_predicate<PredicateType::IN_LIST>(index, type, set, true); |
| break; |
| } |
| case PredicateType::NOT_IN_LIST: { |
| std::shared_ptr<HybridSetBase> set; |
| RETURN_IF_ERROR(convert(type, res.value_str, arena, set)); |
| predicate = create_in_list_predicate<PredicateType::NOT_IN_LIST>(index, type, set, true); |
| break; |
| } |
| default: |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>("invalid condition operator. operator={}", |
| type_to_op_str(res.condition_op)); |
| } |
| return Status::OK(); |
| } |
| |
| // construct sub condition from TCondition |
| std::string construct_sub_predicate(const TCondition& condition) { |
| string op = condition.condition_op; |
| if (op == "<") { |
| op += "<"; |
| } else if (op == ">") { |
| op += ">"; |
| } |
| string condition_str; |
| if ("IS" == op) { |
| // ATTN: tricky! Surround IS with spaces to make it "special" |
| condition_str = condition.column_name + " IS " + condition.condition_values[0]; |
| } else { // multi-elements IN expr has been processed with InPredicatePB |
| if (op == "*=") { |
| op = "="; |
| } else if (op == "!*=") { |
| op = "!="; |
| } |
| condition_str = condition.column_name + op + "'" + condition.condition_values[0] + "'"; |
| } |
| return condition_str; |
| } |
| |
| // make operators from FE adaptive to BE |
| std::string trans_op(const std::string& opt) { |
| std::string op = string(opt); |
| if (op == "<") { |
| op += "<"; |
| } else if (op == ">") { |
| op += ">"; |
| } |
| if ("IS" != op) { |
| if (op == "*=") { |
| op = "="; |
| } else if (op == "!*=") { |
| op = "!="; |
| } |
| } |
| return op; |
| } |
| |
| 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<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) { |
| RETURN_IF_ERROR(check_condition_valid(schema, condition)); |
| } |
| |
| // Store delete condition |
| for (const TCondition& condition : conditions) { |
| if (condition.condition_values.size() > 1) { |
| InPredicatePB* in_pred = del_pred->add_in_predicates(); |
| if (condition.__isset.column_unique_id) { |
| in_pred->set_column_unique_id(condition.column_unique_id); |
| } |
| in_pred->set_column_name(condition.column_name); |
| bool is_not_in = condition.condition_op == "!*="; |
| in_pred->set_is_not_in(is_not_in); |
| for (const auto& condition_value : condition.condition_values) { |
| in_pred->add_values(condition_value); |
| } |
| |
| LOG(INFO) << "store one sub-delete condition. condition name=" << in_pred->column_name() |
| << "condition size=" << in_pred->values().size(); |
| } else { |
| // write sub predicate v1 for compactbility |
| std::string condition_str = construct_sub_predicate(condition); |
| VLOG_NOTICE << __PRETTY_FUNCTION__ << " condition_str: " << condition_str; |
| del_pred->add_sub_predicates(condition_str); |
| DeleteSubPredicatePB* sub_predicate = del_pred->add_sub_predicates_v2(); |
| if (condition.__isset.column_unique_id) { |
| // only light schema change capable table set this field |
| sub_predicate->set_column_unique_id(condition.column_unique_id); |
| } else { |
| try { |
| [[maybe_unused]] auto parsed_cond = parse_condition(condition_str); |
| } catch (const Exception& e) { |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "failed to parse condition_str, condition={}, error={}", |
| ThriftDebugString(condition), e.to_string()); |
| } |
| } |
| |
| sub_predicate->set_column_name(condition.column_name); |
| sub_predicate->set_op(trans_op(condition.condition_op)); |
| sub_predicate->set_cond_value(condition.condition_values[0]); |
| LOG(INFO) << "store one sub-delete condition. condition=" |
| << fmt::format(" {} {} {}", condition.column_name, condition.condition_op, |
| condition.condition_values[0]); |
| } |
| } |
| del_pred->set_version(-1); |
| |
| return Status::OK(); |
| } |
| |
| Status DeleteHandler::convert_to_sub_pred_v2(DeletePredicatePB* delete_pred, |
| TabletSchemaSPtr schema) { |
| if (!delete_pred->sub_predicates().empty() && delete_pred->sub_predicates_v2().empty()) { |
| for (const auto& condition_str : delete_pred->sub_predicates()) { |
| auto* sub_pred = delete_pred->add_sub_predicates_v2(); |
| auto condition = parse_condition(condition_str); |
| const auto& column = *DORIS_TRY(schema->column(condition.column_name)); |
| sub_pred->set_column_unique_id(column.unique_id()); |
| sub_pred->set_column_name(condition.column_name); |
| sub_pred->set_op(type_to_op_str(condition.condition_op)); |
| sub_pred->set_cond_value(condition.value_str.front()); |
| } |
| } |
| |
| auto* in_pred_list = delete_pred->mutable_in_predicates(); |
| for (auto& in_pred : *in_pred_list) { |
| const auto& column = *DORIS_TRY(schema->column(in_pred.column_name())); |
| in_pred.set_column_unique_id(column.unique_id()); |
| } |
| return Status::OK(); |
| } |
| |
| bool DeleteHandler::is_condition_value_valid(const TabletColumn& column, |
| const std::string& condition_op, |
| const string& value_str) { |
| if ("IS" == condition_op && ("NULL" == value_str || "NOT NULL" == value_str)) { |
| return true; |
| } |
| |
| FieldType field_type = column.type(); |
| switch (field_type) { |
| case FieldType::OLAP_FIELD_TYPE_TINYINT: |
| return valid_signed_number<int8_t>(value_str); |
| case FieldType::OLAP_FIELD_TYPE_SMALLINT: |
| return valid_signed_number<int16_t>(value_str); |
| case FieldType::OLAP_FIELD_TYPE_INT: |
| return valid_signed_number<int32_t>(value_str); |
| case FieldType::OLAP_FIELD_TYPE_BIGINT: |
| return valid_signed_number<int64_t>(value_str); |
| case FieldType::OLAP_FIELD_TYPE_LARGEINT: |
| return valid_signed_number<int128_t>(value_str); |
| case FieldType::OLAP_FIELD_TYPE_UNSIGNED_TINYINT: |
| return valid_unsigned_number<uint8_t>(value_str); |
| case FieldType::OLAP_FIELD_TYPE_UNSIGNED_SMALLINT: |
| return valid_unsigned_number<uint16_t>(value_str); |
| case FieldType::OLAP_FIELD_TYPE_UNSIGNED_INT: |
| return valid_unsigned_number<uint32_t>(value_str); |
| case FieldType::OLAP_FIELD_TYPE_UNSIGNED_BIGINT: |
| return valid_unsigned_number<uint64_t>(value_str); |
| case FieldType::OLAP_FIELD_TYPE_DECIMAL: |
| return valid_decimal(value_str, column.precision(), column.frac()); |
| case FieldType::OLAP_FIELD_TYPE_DECIMAL32: |
| return valid_decimal(value_str, column.precision(), column.frac()); |
| case FieldType::OLAP_FIELD_TYPE_DECIMAL64: |
| return valid_decimal(value_str, column.precision(), column.frac()); |
| case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: |
| return valid_decimal(value_str, column.precision(), column.frac()); |
| case FieldType::OLAP_FIELD_TYPE_DECIMAL256: |
| return valid_decimal(value_str, column.precision(), column.frac()); |
| case FieldType::OLAP_FIELD_TYPE_CHAR: |
| case FieldType::OLAP_FIELD_TYPE_VARCHAR: |
| return value_str.size() <= column.length(); |
| case FieldType::OLAP_FIELD_TYPE_STRING: |
| return value_str.size() <= config::string_type_length_soft_limit_bytes; |
| case FieldType::OLAP_FIELD_TYPE_DATE: |
| case FieldType::OLAP_FIELD_TYPE_DATETIME: |
| case FieldType::OLAP_FIELD_TYPE_DATEV2: |
| case FieldType::OLAP_FIELD_TYPE_DATETIMEV2: |
| case FieldType::OLAP_FIELD_TYPE_TIMESTAMPTZ: |
| return valid_datetime(value_str, column.frac()); |
| case FieldType::OLAP_FIELD_TYPE_BOOL: |
| return valid_bool(value_str); |
| case FieldType::OLAP_FIELD_TYPE_IPV4: |
| return valid_ipv4(value_str); |
| case FieldType::OLAP_FIELD_TYPE_IPV6: |
| return valid_ipv6(value_str); |
| default: |
| LOG(WARNING) << "unknown field type. [type=" << int(field_type) << "]"; |
| } |
| return false; |
| } |
| |
| Status DeleteHandler::check_condition_valid(const TabletSchema& schema, const TCondition& cond) { |
| // Check whether the column exists |
| int32_t field_index = schema.field_index(cond.column_name); |
| if (field_index < 0) { |
| 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 |
| // the condition column type should not be float or double. |
| const TabletColumn& column = schema.column(field_index); |
| |
| if (column.type() == FieldType::OLAP_FIELD_TYPE_DOUBLE || |
| column.type() == FieldType::OLAP_FIELD_TYPE_FLOAT) { |
| 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<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<ErrorCode::INVALID_ARGUMENT>("invalid condition value. [value={}]", |
| condition_value); |
| } |
| } |
| |
| if (!cond.__isset.column_unique_id) { |
| LOG(WARNING) << "column=" << cond.column_name |
| << " in predicate does not have uid, table id=" << schema.table_id(); |
| // TODO(tsy): make it fail here after FE forbidding hard-link-schema-change |
| return Status::OK(); |
| } |
| if (schema.field_index(cond.column_unique_id) == -1) { |
| const auto& err_msg = |
| fmt::format("column id does not exists in table={}, schema version={},", |
| schema.table_id(), schema.schema_version()); |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>(err_msg); |
| } |
| if (!iequal(schema.column_by_uid(cond.column_unique_id).name(), cond.column_name)) { |
| const auto& err_msg = fmt::format( |
| "colum name={} does not belongs to column uid={}, which " |
| "column name={}, " |
| "delete_cond.column_name ={}", |
| cond.column_name, cond.column_unique_id, |
| schema.column_by_uid(cond.column_unique_id).name(), cond.column_name); |
| return Status::Error<ErrorCode::INVALID_ARGUMENT>(err_msg); |
| } |
| |
| return Status::OK(); |
| } |
| |
| PredicateType DeleteHandler::parse_condition_op(const std::string& op_str, |
| const std::list<std::string>& cond_values) { |
| if (trim(to_lower(op_str)) == "=") { |
| return PredicateType::EQ; |
| } else if (trim(to_lower(op_str)) == "!=") { |
| return PredicateType::NE; |
| } else if (trim(to_lower(op_str)) == ">>") { |
| return PredicateType::GT; |
| } else if (trim(to_lower(op_str)) == "<<") { |
| return PredicateType::LT; |
| } else if (trim(to_lower(op_str)) == ">=") { |
| return PredicateType::GE; |
| } else if (trim(to_lower(op_str)) == "<=") { |
| return PredicateType::LE; |
| } else if (trim(to_lower(op_str)) == "*=") { |
| return cond_values.size() > 1 ? PredicateType::IN_LIST : PredicateType::EQ; |
| } else if (trim(to_lower(op_str)) == "!*=") { |
| return cond_values.size() > 1 ? PredicateType::NOT_IN_LIST : PredicateType::NE; |
| } else if (trim(to_lower(op_str)) == "is") { |
| return to_lower(cond_values.front()) == "null" ? PredicateType::IS_NULL |
| : PredicateType::IS_NOT_NULL; |
| } else { |
| throw Exception(Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "invalid condition operator. operator={}", op_str)); |
| } |
| return PredicateType::UNKNOWN; |
| } |
| |
| DeleteHandler::ConditionParseResult DeleteHandler::parse_condition( |
| const DeleteSubPredicatePB& sub_cond) { |
| ConditionParseResult res; |
| if (!sub_cond.has_column_name() || !sub_cond.has_op() || !sub_cond.has_cond_value()) { |
| throw Exception(Status::Error<ErrorCode::INVALID_ARGUMENT>( |
| "fail to parse condition. condition={} {} {}", sub_cond.column_name(), |
| sub_cond.op(), sub_cond.cond_value())); |
| } |
| if (sub_cond.has_column_unique_id()) { |
| res.col_unique_id = sub_cond.column_unique_id(); |
| } |
| res.column_name = sub_cond.column_name(); |
| res.value_str.push_back(sub_cond.cond_value()); |
| res.condition_op = parse_condition_op(sub_cond.op(), res.value_str); |
| return res; |
| } |
| |
| // clang-format off |
| // Condition string format, the format is (column_name)(op)(value) |
| // eg: condition_str="c1 = 1597751948193618247 and length(source)<1;\n;\n" |
| // column_name: matches "c1", must include FeNameFormat.java COLUMN_NAME_REGEX |
| // and compactible with any the lagacy |
| // operator: matches "=" |
| // value: matches "1597751948193618247 and length(source)<1;\n;\n" |
| // |
| // For more info, see DeleteHandler::construct_sub_predicates |
| // FIXME(gavin): This is a tricky implementation, it should not be the final resolution, refactor it. |
| const char* const CONDITION_STR_PATTERN = |
| // .----------------- column-name --------------------------. .----------------------- operator ------------------------. .------------ value ----------. |
| R"(([_a-zA-Z@0-9\s/\p{L}][.a-zA-Z0-9_+-/?@#$%^&*"\s,:\p{L}]*)\s*((?:=)|(?:!=)|(?:>>)|(?:<<)|(?:>=)|(?:<=)|(?:\*=)|(?: IS ))\s*('((?:[\s\S]+)?)'|(?:[\s\S]+)?))"; |
| // '----------------- group 1 ------------------------------' '--------------------- group 2 ---------------------------' | '-- group 4--' | |
| // match any of: = != >> << >= <= *= " IS " '----------- group 3 ---------' |
| // match **ANY THING** without(4) |
| // or with(3) single quote |
| // clang-format on |
| RE2 DELETE_HANDLER_REGEX(CONDITION_STR_PATTERN); |
| |
| DeleteHandler::ConditionParseResult DeleteHandler::parse_condition( |
| const std::string& condition_str) { |
| ConditionParseResult res; |
| std::string col_name, op, value, g4; |
| |
| bool matched = RE2::FullMatch(condition_str, DELETE_HANDLER_REGEX, &col_name, &op, &value, |
| &g4); // exact match |
| |
| if (!matched) { |
| throw Exception( |
| Status::InvalidArgument("fail to sub condition. condition={}", condition_str)); |
| } |
| |
| res.column_name = col_name; |
| |
| // match string with single quotes, a = b or a = 'b' |
| if (!g4.empty()) { |
| res.value_str.push_back(g4); |
| } else { |
| res.value_str.push_back(value); |
| } |
| res.condition_op = DeleteHandler::parse_condition_op(op, res.value_str); |
| VLOG_NOTICE << "parsed condition_str: col_name={" << col_name << "} op={" << op << "} val={" |
| << res.value_str.back() << "}"; |
| return res; |
| } |
| |
| template <typename SubPredType> |
| requires(std::is_same_v<SubPredType, DeleteSubPredicatePB> or |
| std::is_same_v<SubPredType, std::string>) |
| Status DeleteHandler::_parse_column_pred(TabletSchemaSPtr complete_schema, |
| TabletSchemaSPtr delete_pred_related_schema, |
| const RepeatedPtrField<SubPredType>& sub_pred_list, |
| DeleteConditions* delete_conditions) { |
| for (const auto& sub_predicate : sub_pred_list) { |
| auto condition = parse_condition(sub_predicate); |
| int32_t col_unique_id = -1; |
| if constexpr (std::is_same_v<SubPredType, DeleteSubPredicatePB>) { |
| if (sub_predicate.has_column_unique_id()) [[likely]] { |
| col_unique_id = sub_predicate.column_unique_id(); |
| } |
| } |
| if (col_unique_id < 0) { |
| const auto& column = |
| *DORIS_TRY(delete_pred_related_schema->column(condition.column_name)); |
| col_unique_id = column.unique_id(); |
| } |
| condition.col_unique_id = col_unique_id; |
| const auto& column = complete_schema->column_by_uid(col_unique_id); |
| uint32_t index = complete_schema->field_index(col_unique_id); |
| std::shared_ptr<ColumnPredicate> predicate; |
| RETURN_IF_ERROR(parse_to_predicate(index, column.get_vec_type(), condition, |
| _predicate_arena, predicate)); |
| if (predicate != nullptr) { |
| delete_conditions->column_predicate_vec.push_back(predicate); |
| } |
| } |
| return Status::OK(); |
| } |
| |
| Status DeleteHandler::init(TabletSchemaSPtr tablet_schema, |
| const std::vector<RowsetMetaSharedPtr>& delete_preds, int64_t version) { |
| DCHECK(!_is_inited) << "reinitialize delete handler."; |
| DCHECK(version >= 0) << "invalid parameters. version=" << version; |
| |
| for (const auto& delete_pred : delete_preds) { |
| // Skip the delete condition with large version |
| if (delete_pred->version().first > version) { |
| continue; |
| } |
| // Need the tablet schema at the delete condition to parse the accurate column |
| const auto& delete_pred_related_schema = delete_pred->tablet_schema(); |
| const auto& delete_condition = delete_pred->delete_predicate(); |
| DeleteConditions temp; |
| temp.filter_version = delete_pred->version().first; |
| if (!delete_condition.sub_predicates_v2().empty()) { |
| RETURN_IF_ERROR(_parse_column_pred(tablet_schema, delete_pred_related_schema, |
| delete_condition.sub_predicates_v2(), &temp)); |
| } else { |
| // make it compatible with the former versions |
| RETURN_IF_ERROR(_parse_column_pred(tablet_schema, delete_pred_related_schema, |
| delete_condition.sub_predicates(), &temp)); |
| } |
| for (const auto& in_predicate : delete_condition.in_predicates()) { |
| ConditionParseResult condition; |
| condition.column_name = in_predicate.column_name(); |
| |
| int32_t col_unique_id = -1; |
| if (in_predicate.has_column_unique_id()) { |
| col_unique_id = in_predicate.column_unique_id(); |
| } else { |
| // if upgrade from version 2.0.x, column_unique_id maybe not set |
| const auto& pre_column = |
| *DORIS_TRY(delete_pred_related_schema->column(condition.column_name)); |
| col_unique_id = pre_column.unique_id(); |
| } |
| if (col_unique_id == -1) { |
| return Status::Error<ErrorCode::DELETE_INVALID_CONDITION>( |
| "cannot get column_unique_id for column {}", condition.column_name); |
| } |
| condition.col_unique_id = col_unique_id; |
| |
| condition.condition_op = |
| in_predicate.is_not_in() ? PredicateType::NOT_IN_LIST : PredicateType::IN_LIST; |
| for (const auto& value : in_predicate.values()) { |
| condition.value_str.push_back(value); |
| } |
| const auto& column = tablet_schema->column_by_uid(col_unique_id); |
| uint32_t index = tablet_schema->field_index(col_unique_id); |
| std::shared_ptr<ColumnPredicate> predicate; |
| RETURN_IF_ERROR(parse_to_in_predicate(index, column.get_vec_type(), condition, |
| _predicate_arena, predicate)); |
| temp.column_predicate_vec.push_back(predicate); |
| } |
| |
| _del_conds.emplace_back(std::move(temp)); |
| } |
| |
| _is_inited = true; |
| |
| return Status::OK(); |
| } |
| |
| DeleteHandler::~DeleteHandler() { |
| if (!_is_inited) { |
| return; |
| } |
| |
| _del_conds.clear(); |
| _is_inited = false; |
| } |
| |
| void DeleteHandler::get_delete_conditions_after_version( |
| int64_t version, AndBlockColumnPredicate* and_block_column_predicate_ptr, |
| std::unordered_map<int32_t, std::vector<std::shared_ptr<const ColumnPredicate>>>* |
| del_predicates_for_zone_map) const { |
| for (const auto& del_cond : _del_conds) { |
| if (del_cond.filter_version > version) { |
| // now, only query support delete column predicate operator |
| if (!del_cond.column_predicate_vec.empty()) { |
| if (del_cond.column_predicate_vec.size() == 1) { |
| auto single_column_block_predicate = SingleColumnBlockPredicate::create_unique( |
| del_cond.column_predicate_vec[0]); |
| and_block_column_predicate_ptr->add_column_predicate( |
| std::move(single_column_block_predicate)); |
| if (del_predicates_for_zone_map->count( |
| del_cond.column_predicate_vec[0]->column_id()) < 1) { |
| del_predicates_for_zone_map->insert( |
| {del_cond.column_predicate_vec[0]->column_id(), |
| std::vector<std::shared_ptr<const ColumnPredicate>> {}}); |
| } |
| (*del_predicates_for_zone_map)[del_cond.column_predicate_vec[0]->column_id()] |
| .push_back(del_cond.column_predicate_vec[0]); |
| } else { |
| auto or_column_predicate = OrBlockColumnPredicate::create_unique(); |
| |
| // build or_column_predicate |
| // when delete from where a = 1 and b = 2, we can not use del_predicates_for_zone_map to filter zone page, |
| // so here do not put predicate to del_predicates_for_zone_map, |
| // refer #17145 for more details. |
| // // TODO: need refactor design and code to use more version delete and more column delete to filter zone page. |
| std::for_each(del_cond.column_predicate_vec.cbegin(), |
| del_cond.column_predicate_vec.cend(), |
| [&or_column_predicate]( |
| const std::shared_ptr<const ColumnPredicate> predicate) { |
| or_column_predicate->add_column_predicate( |
| SingleColumnBlockPredicate::create_unique(predicate)); |
| }); |
| and_block_column_predicate_ptr->add_column_predicate( |
| std::move(or_column_predicate)); |
| } |
| } |
| } |
| } |
| } |
| |
| } // namespace doris |
