be/src/exec/sink/vtablet_block_convertor.cpp

Source
// 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 "exec/sink/vtablet_block_convertor.h"

#include <fmt/format.h>
#include <gen_cpp/FrontendService.h>
#include <glog/logging.h>
#include <google/protobuf/stubs/common.h>

#include <algorithm>
#include <memory>
#include <string>
#include <unordered_map>
#include <utility>

#include "common/compiler_util.h" // IWYU pragma: keep
#include "common/consts.h"
#include "common/status.h"
#include "core/assert_cast.h"
#include "core/binary_cast.hpp"
#include "core/block/block.h"
#include "core/column/column.h"
#include "core/column/column_array.h"
#include "core/column/column_const.h"
#include "core/column/column_decimal.h"
#include "core/column/column_map.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_struct.h"
#include "core/data_type/data_type.h"
#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_decimal.h"
#include "core/data_type/data_type_factory.hpp"
#include "core/data_type/data_type_map.h"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_struct.h"
#include "core/data_type/define_primitive_type.h"
#include "core/data_type/primitive_type.h"
#include "core/types.h"
#include "core/wide_integer_to_string.h"
#include "exprs/function/function_helpers.h"
#include "exprs/function/simple_function_factory.h"
#include "exprs/vexpr.h"
#include "exprs/vexpr_context.h"
#include "runtime/descriptors.h"
#include "runtime/runtime_state.h"
#include "service/brpc.h"
#include "storage/olap_common.h"
#include "util/brpc_client_cache.h"
#include "util/thread.h"

namespace doris {

// !FIXME: Here we should consider using MutableBlock, due to potential data reorganization
Status OlapTableBlockConvertor::validate_and_convert_block(RuntimeState* state, Block* input_block,
                                                           std::shared_ptr<Block>& block,
                                                           VExprContextSPtrs output_vexpr_ctxs,
                                                           size_t rows, bool& has_filtered_rows) {
    DCHECK(input_block->rows() > 0);

    block = Block::create_shared(input_block->get_columns_with_type_and_name());
    if (!output_vexpr_ctxs.empty()) {
        // Do vectorized expr here to speed up load
        RETURN_IF_ERROR(VExprContext::get_output_block_after_execute_exprs(
                output_vexpr_ctxs, *input_block, block.get()));
    }

    if (_is_partial_update_and_auto_inc) {
        // If this load is partial update and this table has a auto inc column,
        // e.g. table schema: k1, v1, v2(auto inc)
        // 1. insert columns include auto inc column
        // e.g. insert into table (k1, v2) value(a, 1);
        // we do nothing.
        // 2. insert columns do not include auto inc column
        // e.g. insert into table (k1, v1) value(a, a);
        // we need to fill auto_inc_cols by creating a new column.
        if (!_auto_inc_col_idx.has_value()) {
            RETURN_IF_ERROR(_partial_update_fill_auto_inc_cols(block.get(), rows));
        }
    } else if (_auto_inc_col_idx.has_value()) {
        // fill the valus for auto-increment columns
        DCHECK_EQ(_is_partial_update_and_auto_inc, false);
        RETURN_IF_ERROR(_fill_auto_inc_cols(block.get(), rows));
    }

    int filtered_rows = 0;
    {
        SCOPED_RAW_TIMER(&_validate_data_ns);
        _filter_map.clear();
        _filter_map.resize(rows, 0);
        auto st = _validate_data(state, block.get(), rows, filtered_rows);
        _num_filtered_rows += filtered_rows;
        has_filtered_rows = filtered_rows > 0;
        if (!st.ok()) {
            return st;
        }
        _convert_to_dest_desc_block(block.get());
    }

    return Status::OK();
}

void OlapTableBlockConvertor::init_autoinc_info(int64_t db_id, int64_t table_id, int batch_size,
                                                bool is_partial_update_and_auto_inc,
                                                int32_t auto_increment_column_unique_id) {
    _batch_size = batch_size;
    if (is_partial_update_and_auto_inc) {
        _is_partial_update_and_auto_inc = is_partial_update_and_auto_inc;
        _auto_inc_id_buffer = GlobalAutoIncBuffers::GetInstance()->get_auto_inc_buffer(
                db_id, table_id, auto_increment_column_unique_id);
        return;
    }
    for (size_t idx = 0; idx < _output_tuple_desc->slots().size(); idx++) {
        if (_output_tuple_desc->slots()[idx]->is_auto_increment()) {
            _auto_inc_col_idx = idx;
            _auto_inc_id_buffer = GlobalAutoIncBuffers::GetInstance()->get_auto_inc_buffer(
                    db_id, table_id, _output_tuple_desc->slots()[idx]->col_unique_id());
            _auto_inc_id_buffer->set_batch_size_at_least(_batch_size);
            break;
        }
    }
}

template <bool is_min>
DecimalV2Value OlapTableBlockConvertor::_get_decimalv2_min_or_max(const DataTypePtr& type) {
    std::map<std::pair<int, int>, DecimalV2Value>* pmap;
    if constexpr (is_min) {
        pmap = &_min_decimalv2_val;
    } else {
        pmap = &_max_decimalv2_val;
    }

    // found
    auto iter = pmap->find(
            {remove_nullable(type)->get_precision(), remove_nullable(type)->get_scale()});
    if (iter != pmap->end()) {
        return iter->second;
    }

    // save min or max DecimalV2Value for next time
    DecimalV2Value value;
    if constexpr (is_min) {
        value.to_min_decimal(type->get_precision(), type->get_scale());
    } else {
        value.to_max_decimal(type->get_precision(), type->get_scale());
    }
    pmap->emplace(std::pair<int, int> {type->get_precision(), type->get_scale()}, value);
    return value;
}

template <typename DecimalType, bool IsMin>
DecimalType OlapTableBlockConvertor::_get_decimalv3_min_or_max(const DataTypePtr& type) {
    std::map<int, typename DecimalType::NativeType>* pmap;
    if constexpr (std::is_same_v<DecimalType, Decimal32>) {
        pmap = IsMin ? &_min_decimal32_val : &_max_decimal32_val;
    } else if constexpr (std::is_same_v<DecimalType, Decimal64>) {
        pmap = IsMin ? &_min_decimal64_val : &_max_decimal64_val;
    } else if constexpr (std::is_same_v<DecimalType, Decimal128V3>) {
        pmap = IsMin ? &_min_decimal128_val : &_max_decimal128_val;
    } else {
        pmap = IsMin ? &_min_decimal256_val : &_max_decimal256_val;
    }

    // found
    auto iter = pmap->find(type->get_precision());
    if (iter != pmap->end()) {
        return DecimalType(iter->second);
    }

    DecimalType value;
    if constexpr (IsMin) {
        value = min_decimal_value<DecimalType::PType>(type->get_precision());
    } else {
        value = max_decimal_value<DecimalType::PType>(type->get_precision());
    }
    pmap->emplace(type->get_precision(), value.value);
    return value;
}

Status OlapTableBlockConvertor::_internal_validate_column(RuntimeState* state, Block* block,
                                                          const DataTypePtr& type, ColumnPtr column,
                                                          size_t slot_index,
                                                          fmt::memory_buffer& error_prefix,
                                                          const size_t row_count,
                                                          IColumn::Permutation* rows) {
    DCHECK((rows == nullptr) || (rows->size() == row_count));
    fmt::memory_buffer error_msg;
    auto set_invalid_and_append_error_msg = [&](size_t row) {
        _filter_map[row] = true;
        auto ret = state->append_error_msg_to_file([]() -> std::string { return ""; },
                                                   [&error_prefix, &error_msg]() -> std::string {
                                                       return fmt::to_string(error_prefix) +
                                                              fmt::to_string(error_msg);
                                                   });
        error_msg.clear();
        return ret;
    };

    const auto* column_ptr = check_and_get_column<ColumnNullable>(*column);
    const auto& real_column_ptr =
            column_ptr == nullptr ? column : (column_ptr->get_nested_column_ptr());
    const auto* null_map = column_ptr == nullptr ? nullptr : column_ptr->get_null_map_data().data();
    auto need_to_validate = [](size_t j, size_t row, const std::vector<char>& filter_map,
                               const unsigned char* null_map) {
        return !filter_map[row] && (null_map == nullptr || null_map[j] == 0);
    };

    // may change orig_column if substring function is performed
    auto string_column_checker = [&state, &error_msg, need_to_validate,
                                  set_invalid_and_append_error_msg](
                                         ColumnPtr& orig_column, const DataTypePtr& orig_type,
                                         IColumn::Permutation* rows,
                                         const std::vector<char>& filter_map) {
        int limit = config::string_type_length_soft_limit_bytes;
        int len = -1;
        // when type.len is negative, std::min will return overflow value, so we need to check it
        const auto* type_str =
                check_and_get_data_type<DataTypeString>(remove_nullable(orig_type).get());
        if (type_str) {
            if (type_str->len() >= 0) {
                len = type_str->len();
                limit = std::min(limit, type_str->len());
            }
        }

        const auto* tmp_column_ptr = check_and_get_column<ColumnNullable>(*orig_column);
        const auto& tmp_real_column_ptr =
                tmp_column_ptr == nullptr ? orig_column : (tmp_column_ptr->get_nested_column_ptr());
        const auto* column_string = assert_cast<const ColumnString*>(tmp_real_column_ptr.get());
        const auto* null_map =
                tmp_column_ptr == nullptr ? nullptr : tmp_column_ptr->get_null_map_data().data();

        const auto* __restrict offsets = column_string->get_offsets().data();
        int invalid_count = 0;
        size_t row_count = orig_column->size();
        for (int64_t j = 0; j < row_count; ++j) {
            invalid_count += (offsets[j] - offsets[j - 1]) > limit;
        }

        if (invalid_count) {
            // For string column, if in non-strict load mode(for both insert stmt and stream load),
            // truncate the string to schema len.
            // After truncation, still need to check if byte len of each row exceed the schema len,
            // because currently the schema len is defined in bytes, and substring works by unit of chars.
            // This is a workaround for now, need to improve it after better support of multi-byte chars.
            if (type_str && !state->enable_insert_strict()) {
                ColumnsWithTypeAndName argument_template;
                auto input_type = remove_nullable(orig_type);
                auto pos_type = DataTypeFactory::instance().create_data_type(
                        FieldType::OLAP_FIELD_TYPE_INT, 0, 0);
                auto len_type = DataTypeFactory::instance().create_data_type(
                        FieldType::OLAP_FIELD_TYPE_INT, 0, 0);
                argument_template.emplace_back(nullptr, input_type, "string column");
                argument_template.emplace_back(nullptr, pos_type, "pos column");
                argument_template.emplace_back(nullptr, len_type, "len column");
                auto func = SimpleFunctionFactory::instance().get_function(
                        "substring", argument_template, input_type, {}, state->be_exec_version());
                if (!func) {
                    return Status::InternalError("get function substring failed");
                }
                auto pos_column = pos_type->create_column_const(row_count, to_field<TYPE_INT>(1));
                auto len_column =
                        len_type->create_column_const(row_count, to_field<TYPE_INT>(limit));
                Block tmp_block({{remove_nullable(orig_column), input_type, "string column"},
                                 {pos_column, pos_type, "pos"},
                                 {len_column, len_type, "len"},
                                 {nullptr, input_type, "result"}});
                RETURN_IF_ERROR(func->execute(nullptr, tmp_block, {0, 1, 2}, 3, row_count));
                column_string =
                        assert_cast<const ColumnString*>(tmp_block.get_by_position(3).column.get());
                orig_column =
                        orig_column->is_nullable()
                                ? ColumnNullable::create(tmp_block.get_by_position(3).column,
                                                         tmp_column_ptr->get_null_map_column_ptr())
                                : std::move(tmp_block.get_by_position(3).column);
            }
            for (size_t j = 0; j < row_count; ++j) {
                auto row = rows ? (*rows)[j] : j;
                if (need_to_validate(j, row, filter_map, null_map)) {
                    auto str_val = column_string->get_data_at(j);
                    bool invalid = str_val.size > limit;
                    if (invalid) {
                        if (str_val.size > len) {
                            fmt::format_to(error_msg, "{}",
                                           "the length of input is too long than schema. ");
                            fmt::format_to(error_msg, "first 32 bytes of input str: [{}] ",
                                           str_val.to_prefix(32));
                            fmt::format_to(error_msg, "schema length: {}; ", len);
                            fmt::format_to(error_msg, "actual length: {}; ", str_val.size);
                        } else if (str_val.size > limit) {
                            fmt::format_to(
                                    error_msg, "{}",
                                    "the length of input string is too long than vec schema. ");
                            fmt::format_to(error_msg, "first 32 bytes of input str: [{}] ",
                                           str_val.to_prefix(32));
                            fmt::format_to(error_msg, "schema length: {}; ", len);
                            fmt::format_to(error_msg, "limit length: {}; ", limit);
                            fmt::format_to(error_msg, "actual length: {}; ", str_val.size);
                        }
                        RETURN_IF_ERROR(set_invalid_and_append_error_msg(row));
                    }
                }
            }
        }
        return Status::OK();
    };

    switch (type->get_primitive_type()) {
    case TYPE_CHAR:
    case TYPE_VARCHAR:
    case TYPE_STRING: {
        RETURN_IF_ERROR(string_column_checker(column, type, rows, _filter_map));
        block->get_by_position(slot_index).column = std::move(column);
        break;
    }
    case TYPE_JSONB: {
        const auto* column_string = assert_cast<const ColumnString*>(real_column_ptr.get());
        for (size_t j = 0; j < row_count; ++j) {
            if (!_filter_map[j]) {
                if (type->is_nullable() && column_ptr && column_ptr->is_null_at(j)) {
                    continue;
                }
                auto str_val = column_string->get_data_at(j);
                bool invalid = str_val.size == 0;
                if (invalid) {
                    error_msg.clear();
                    fmt::format_to(error_msg, "{}", "jsonb with size 0 is invalid");
                    RETURN_IF_ERROR(set_invalid_and_append_error_msg(j));
                }
            }
        }
        break;
    }
    case TYPE_DECIMALV2: {
        // column_decimal utilizes the ColumnPtr from the block* block in _validate_data and can be modified.
        auto* column_decimal = const_cast<ColumnDecimal128V2*>(
                assert_cast<const ColumnDecimal128V2*>(real_column_ptr.get()));
        const auto& max_decimalv2 = _get_decimalv2_min_or_max<false>(type);
        const auto& min_decimalv2 = _get_decimalv2_min_or_max<true>(type);
        for (size_t j = 0; j < row_count; ++j) {
            auto row = rows ? (*rows)[j] : j;
            if (need_to_validate(j, row, _filter_map, null_map)) {
                auto dec_val = binary_cast<Int128, DecimalV2Value>(column_decimal->get_data()[j]);
                bool invalid = false;

                if (dec_val.greater_than_scale(type->get_scale())) {
                    auto code =
                            dec_val.round(&dec_val, remove_nullable(type)->get_scale(), HALF_UP);
                    column_decimal->get_data()[j] = dec_val;

                    if (code != E_DEC_OK) {
                        fmt::format_to(error_msg, "round one decimal failed.value={}; ",
                                       dec_val.to_string());
                        invalid = true;
                    }
                }
                if (dec_val > max_decimalv2 || dec_val < min_decimalv2) {
                    fmt::format_to(error_msg, "{}", "decimal value is not valid for definition");
                    fmt::format_to(error_msg, ", value={}", dec_val.to_string());
                    fmt::format_to(error_msg, ", precision={}, scale={}", type->get_precision(),
                                   type->get_scale());
                    fmt::format_to(error_msg, ", min={}, max={}; ", min_decimalv2.to_string(),
                                   max_decimalv2.to_string());
                    invalid = true;
                }

                if (invalid) {
                    RETURN_IF_ERROR(set_invalid_and_append_error_msg(row));
                }
            }
        }
        break;
    }
    case TYPE_DECIMAL32: {
#define CHECK_VALIDATION_FOR_DECIMALV3(DecimalType)                                               \
    auto column_decimal =                                                                         \
            assert_cast<const ColumnDecimal<DecimalType::PType>*>(real_column_ptr.get());         \
    const auto& max_decimal = _get_decimalv3_min_or_max<DecimalType, false>(type);                \
    const auto& min_decimal = _get_decimalv3_min_or_max<DecimalType, true>(type);                 \
    const auto* __restrict datas = column_decimal->get_data().data();                             \
    int invalid_count = 0;                                                                        \
    for (int j = 0; j < row_count; ++j) {                                                         \
        const auto dec_val = datas[j];                                                            \
        invalid_count += dec_val > max_decimal || dec_val < min_decimal;                          \
    }                                                                                             \
    if (invalid_count) {                                                                          \
        for (size_t j = 0; j < row_count; ++j) {                                                  \
            auto row = rows ? (*rows)[j] : j;                                                     \
            if (need_to_validate(j, row, _filter_map, null_map)) {                                \
                auto dec_val = column_decimal->get_data()[j];                                     \
                bool invalid = false;                                                             \
                if (dec_val > max_decimal || dec_val < min_decimal) {                             \
                    fmt::format_to(error_msg, "{}", "decimal value is not valid for definition"); \
                    fmt::format_to(error_msg, ", value={}", dec_val.value);                       \
                    fmt::format_to(error_msg, ", precision={}, scale={}", type->get_precision(),  \
                                   type->get_scale());                                            \
                    fmt::format_to(error_msg, ", min={}, max={}; ", min_decimal.value,            \
                                   max_decimal.value);                                            \
                    invalid = true;                                                               \
                }                                                                                 \
                if (invalid) {                                                                    \
                    RETURN_IF_ERROR(set_invalid_and_append_error_msg(row));                       \
                }                                                                                 \
            }                                                                                     \
        }                                                                                         \
    }
        CHECK_VALIDATION_FOR_DECIMALV3(Decimal32);
        break;
    }
    case TYPE_DECIMAL64: {
        CHECK_VALIDATION_FOR_DECIMALV3(Decimal64);
        break;
    }
    case TYPE_DECIMAL128I: {
        CHECK_VALIDATION_FOR_DECIMALV3(Decimal128V3);
        break;
    }
    case TYPE_DECIMAL256: {
        CHECK_VALIDATION_FOR_DECIMALV3(Decimal256);
        break;
    }
#undef CHECK_VALIDATION_FOR_DECIMALV3
    case TYPE_ARRAY: {
        const auto* column_array = assert_cast<const ColumnArray*>(real_column_ptr.get());
        const auto* type_array = assert_cast<const DataTypeArray*>(remove_nullable(type).get());
        auto nested_type = type_array->get_nested_type();
        const auto& offsets = column_array->get_offsets();
        IColumn::Permutation permutation(offsets.back());
        for (size_t r = 0; r < row_count; ++r) {
            for (size_t c = offsets[r - 1]; c < offsets[r]; ++c) {
                permutation[c] = rows ? (*rows)[r] : r;
            }
        }
        fmt::format_to(error_prefix, "ARRAY type failed: ");
        auto data_column_ptr = column_array->get_data_ptr();
        switch (nested_type->get_primitive_type()) {
        case TYPE_CHAR:
        case TYPE_VARCHAR:
        case TYPE_STRING: {
            RETURN_IF_ERROR(
                    string_column_checker(data_column_ptr, nested_type, &permutation, _filter_map));
            const_cast<ColumnArray*>(column_array)->get_data_ptr() = std::move(data_column_ptr);
            break;
        }
        default:
            RETURN_IF_ERROR(_validate_column(state, block, nested_type, data_column_ptr, slot_index,
                                             error_prefix, permutation.size(), &permutation));
            break;
        }
        break;
    }
    case TYPE_MAP: {
        const auto* column_map = assert_cast<const ColumnMap*>(real_column_ptr.get());
        // column_map utilizes the ColumnPtr from the block* block in _validate_data and can be modified.
        RETURN_IF_ERROR((const_cast<ColumnMap*>(column_map))->deduplicate_keys(true));

        const auto* type_map = assert_cast<const DataTypeMap*>(remove_nullable(type).get());
        auto key_type = type_map->get_key_type();
        auto val_type = type_map->get_value_type();
        const auto& offsets = column_map->get_offsets();
        IColumn::Permutation permutation(offsets.back());
        for (size_t r = 0; r < row_count; ++r) {
            for (size_t c = offsets[r - 1]; c < offsets[r]; ++c) {
                permutation[c] = rows ? (*rows)[r] : r;
            }
        }

        fmt::format_to(error_prefix, "MAP type failed: ");
        switch (key_type->get_primitive_type()) {
        case TYPE_CHAR:
        case TYPE_VARCHAR:
        case TYPE_STRING: {
            auto key_column_ptr = column_map->get_keys_ptr();
            RETURN_IF_ERROR(
                    string_column_checker(key_column_ptr, key_type, &permutation, _filter_map));
            const_cast<ColumnMap*>(column_map)->get_keys_ptr() = std::move(key_column_ptr);
            break;
        }
        default:
            RETURN_IF_ERROR(_validate_column(state, block, key_type, column_map->get_keys_ptr(),
                                             slot_index, error_prefix, permutation.size(),
                                             &permutation));
            break;
        }

        switch (val_type->get_primitive_type()) {
        case TYPE_CHAR:
        case TYPE_VARCHAR:
        case TYPE_STRING: {
            auto value_column_ptr = column_map->get_values_ptr();
            RETURN_IF_ERROR(
                    string_column_checker(value_column_ptr, val_type, &permutation, _filter_map));
            const_cast<ColumnMap*>(column_map)->get_values_ptr() = std::move(value_column_ptr);
            break;
        }
        default:
            RETURN_IF_ERROR(_validate_column(state, block, val_type, column_map->get_values_ptr(),
                                             slot_index, error_prefix, permutation.size(),
                                             &permutation));
            break;
        }
        break;
    }
    case TYPE_STRUCT: {
        const auto column_struct = assert_cast<const ColumnStruct*>(real_column_ptr.get());
        const auto* type_struct = assert_cast<const DataTypeStruct*>(remove_nullable(type).get());
        DCHECK(type_struct->get_elements().size() == column_struct->tuple_size());
        fmt::format_to(error_prefix, "STRUCT type failed: ");
        for (size_t sc = 0; sc < column_struct->tuple_size(); ++sc) {
            auto element_type = type_struct->get_element(sc);
            switch (element_type->get_primitive_type()) {
            case TYPE_CHAR:
            case TYPE_VARCHAR:
            case TYPE_STRING: {
                auto element_column_ptr = column_struct->get_column_ptr(sc);
                RETURN_IF_ERROR(string_column_checker(element_column_ptr, element_type, nullptr,
                                                      _filter_map));
                const_cast<ColumnStruct*>(column_struct)->get_column_ptr(sc) =
                        std::move(element_column_ptr);
                break;
            }
            default:
                RETURN_IF_ERROR(_validate_column(state, block, type_struct->get_element(sc),
                                                 column_struct->get_column_ptr(sc), slot_index,
                                                 error_prefix,
                                                 column_struct->get_column_ptr(sc)->size()));
                break;
            }
        }
        break;
    }
    case TYPE_AGG_STATE: {
        auto* column_string = check_and_get_column<ColumnString>(*real_column_ptr);
        if (column_string) {
            RETURN_IF_ERROR(string_column_checker(column, type, rows, _filter_map));
        }
        break;
    }
    default:
        break;
    }

    // Dispose the column should do not contain the NULL value
    // Only two case:
    // 1. column is nullable but the desc is not nullable
    // 2. desc->type is BITMAP
    if ((!type->is_nullable() || type->get_primitive_type() == TYPE_BITMAP) && column_ptr) {
        for (int j = 0; j < row_count; ++j) {
            auto row = rows ? (*rows)[j] : j;
            if (null_map[j] && !_filter_map[row]) {
                fmt::format_to(error_msg, "null value for not null column, type={}",
                               type->get_name());
                RETURN_IF_ERROR(set_invalid_and_append_error_msg(row));
            }
        }
    }

    return Status::OK();
}

Status OlapTableBlockConvertor::_validate_data(RuntimeState* state, Block* block, const size_t rows,
                                               int& filtered_rows) {
    filtered_rows = 0;
    Defer defer {[&] {
        for (int i = 0; i < rows; ++i) {
            filtered_rows += _filter_map[i];
        }
    }};
    for (int i = 0; i < _output_tuple_desc->slots().size(); ++i) {
        SlotDescriptor* desc = _output_tuple_desc->slots()[i];
        block->get_by_position(i).column =
                block->get_by_position(i).column->convert_to_full_column_if_const();
        const auto& column = block->get_by_position(i).column;

        fmt::memory_buffer error_prefix;
        fmt::format_to(error_prefix, "column_name[{}], ", desc->col_name());
        RETURN_IF_ERROR(
                _validate_column(state, block, desc->type(), column, i, error_prefix, rows));
    }
    return Status::OK();
}

void OlapTableBlockConvertor::_convert_to_dest_desc_block(doris::Block* block) {
    for (int i = 0; i < _output_tuple_desc->slots().size() && i < block->columns(); ++i) {
        SlotDescriptor* desc = _output_tuple_desc->slots()[i];
        if (desc->is_nullable() != block->get_by_position(i).type->is_nullable()) {
            if (desc->is_nullable()) {
                block->get_by_position(i).type = make_nullable(block->get_by_position(i).type);
                block->get_by_position(i).column = make_nullable(block->get_by_position(i).column);
            } else {
                block->get_by_position(i).type =
                        assert_cast<const DataTypeNullable&>(*block->get_by_position(i).type)
                                .get_nested_type();
                block->get_by_position(i).column =
                        assert_cast<const ColumnNullable&>(*block->get_by_position(i).column)
                                .get_nested_column_ptr();
            }
        }
    }
}

Status OlapTableBlockConvertor::_fill_auto_inc_cols(Block* block, size_t rows) {
    size_t idx = _auto_inc_col_idx.value();
    SlotDescriptor* slot = _output_tuple_desc->slots()[idx];
    DCHECK(slot->type()->get_primitive_type() == PrimitiveType::TYPE_BIGINT);
    DCHECK(!slot->is_nullable());

    size_t null_value_count = 0;
    auto dst_column = ColumnInt64::create();
    ColumnInt64::Container& dst_values = dst_column->get_data();

    ColumnPtr src_column_ptr = block->get_by_position(idx).column;
    if (const auto* const_column = check_and_get_column<ColumnConst>(src_column_ptr.get())) {
        // for insert stmt like "insert into tbl1 select null,col1,col2,... from tbl2" or
        // "insert into tbl1 select 1,col1,col2,... from tbl2", the type of literal's column
        // will be `ColumnConst`
        if (const_column->is_null_at(0)) {
            // the input of autoinc column are all null literals
            // fill the column with generated ids
            null_value_count = rows;
            std::vector<std::pair<int64_t, size_t>> res;
            RETURN_IF_ERROR(_auto_inc_id_buffer->sync_request_ids(null_value_count, &res));
            for (auto [start, length] : res) {
                _auto_inc_id_allocator.insert_ids(start, length);
            }

            for (size_t i = 0; i < rows; i++) {
                dst_values.emplace_back(_auto_inc_id_allocator.next_id());
            }
        } else {
            // the input of autoinc column are all int64 literals
            // fill the column with that literal
            int64_t value = const_column->get_int(0);
            dst_values.resize_fill(rows, value);
        }
    } else if (const auto* src_nullable_column =
                       check_and_get_column<ColumnNullable>(src_column_ptr.get())) {
        auto src_nested_column_ptr = src_nullable_column->get_nested_column_ptr();
        const auto& null_map_data = src_nullable_column->get_null_map_data();
        dst_values.reserve(rows);
        for (size_t i = 0; i < rows; i++) {
            null_value_count += null_map_data[i];
        }
        std::vector<std::pair<int64_t, size_t>> res;
        RETURN_IF_ERROR(_auto_inc_id_buffer->sync_request_ids(null_value_count, &res));
        for (auto [start, length] : res) {
            _auto_inc_id_allocator.insert_ids(start, length);
        }

        for (size_t i = 0; i < rows; i++) {
            dst_values.emplace_back((null_map_data[i] != 0) ? _auto_inc_id_allocator.next_id()
                                                            : src_nested_column_ptr->get_int(i));
        }
    } else {
        return Status::OK();
    }
    block->get_by_position(idx).column = std::move(dst_column);
    block->get_by_position(idx).type = remove_nullable(slot->type());
    return Status::OK();
}

Status OlapTableBlockConvertor::_partial_update_fill_auto_inc_cols(Block* block, size_t rows) {
    auto dst_column = ColumnInt64::create();
    ColumnInt64::Container& dst_values = dst_column->get_data();
    size_t null_value_count = rows;
    std::vector<std::pair<int64_t, size_t>> res;
    RETURN_IF_ERROR(_auto_inc_id_buffer->sync_request_ids(null_value_count, &res));
    for (auto [start, length] : res) {
        _auto_inc_id_allocator.insert_ids(start, length);
    }

    for (size_t i = 0; i < rows; i++) {
        dst_values.emplace_back(_auto_inc_id_allocator.next_id());
    }
    block->insert(ColumnWithTypeAndName(std::move(dst_column), std::make_shared<DataTypeInt64>(),
                                        BeConsts::PARTIAL_UPDATE_AUTO_INC_COL));
    return Status::OK();
}

} // namespace doris

Coverage Report

Created: 2026-04-14 20:14