be/src/storage/row_cursor.cpp

// 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 "storage/row_cursor.h"

#include <glog/logging.h>
#include <stdlib.h>

#include <algorithm>
#include <new>
#include <numeric>
#include <ostream>

#include "common/cast_set.h"
#include "storage/field.h"
#include "storage/olap_common.h"
#include "storage/olap_define.h"
#include "storage/tablet/tablet_schema.h"
#include "util/slice.h"

using std::nothrow;
using std::string;
using std::vector;

namespace doris {
#include "common/compile_check_begin.h"
using namespace ErrorCode;
RowCursor::RowCursor()
        : _fixed_len(0), _variable_len(0), _string_field_count(0), _long_text_buf(nullptr) {}

RowCursor::~RowCursor() {
    delete[] _owned_fixed_buf;
    delete[] _variable_buf;
    if (_string_field_count > 0 && _long_text_buf != nullptr) {
        for (int i = 0; i < _string_field_count; ++i) {
            free(_long_text_buf[i]);
        }
        free(_long_text_buf);
    }
}

Status RowCursor::_init(const std::vector<uint32_t>& columns) {
    _variable_len = 0;
    for (auto cid : columns) {
        if (_schema->column(cid) == nullptr) {
            return Status::Error<INIT_FAILED>("Fail to malloc _fixed_buf.");
        }
        _variable_len += column_schema(cid)->get_variable_len();
        if (_schema->column(cid)->type() == FieldType::OLAP_FIELD_TYPE_STRING) {
            ++_string_field_count;
        }
    }

    _fixed_len = _schema->schema_size();
    _fixed_buf = new (nothrow) char[_fixed_len]();
    if (_fixed_buf == nullptr) {
        return Status::Error<MEM_ALLOC_FAILED>("Fail to malloc _fixed_buf.");
    }
    _owned_fixed_buf = _fixed_buf;

    return Status::OK();
}

Status RowCursor::_init(const std::shared_ptr<Schema>& shared_schema,
                        const std::vector<uint32_t>& columns) {
    _schema.reset(new Schema(*shared_schema));
    return _init(columns);
}

Status RowCursor::_init(const std::vector<TabletColumnPtr>& schema,
                        const std::vector<uint32_t>& columns) {
    _schema.reset(new Schema(schema, columns));
    return _init(columns);
}

Status RowCursor::_init_scan_key(TabletSchemaSPtr schema,
                                 const std::vector<std::string>& scan_keys) {
    // NOTE: cid equal with column index
    // Hyperloglog cannot be key, no need to handle it
    _variable_len = 0;
    for (auto cid : _schema->column_ids()) {
        const TabletColumn& column = schema->column(cid);
        FieldType type = column.type();
        if (type == FieldType::OLAP_FIELD_TYPE_VARCHAR) {
            _variable_len += scan_keys[cid].length();
        } else if (type == FieldType::OLAP_FIELD_TYPE_CHAR ||
                   type == FieldType::OLAP_FIELD_TYPE_ARRAY) {
            _variable_len +=
                    std::max(scan_keys[cid].length(), static_cast<size_t>(column.length()));
        } else if (type == FieldType::OLAP_FIELD_TYPE_STRING) {
            ++_string_field_count;
        }
    }

    // variable_len for null bytes
    RETURN_IF_ERROR(_alloc_buf());
    char* fixed_ptr = _fixed_buf;
    char* variable_ptr = _variable_buf;
    char** long_text_ptr = _long_text_buf;
    for (auto cid : _schema->column_ids()) {
        const TabletColumn& column = schema->column(cid);
        fixed_ptr = _fixed_buf + _schema->column_offset(cid);
        FieldType type = column.type();
        if (type == FieldType::OLAP_FIELD_TYPE_VARCHAR) {
            // Use memcpy to avoid misaligned store on fixed_ptr + 1
            Slice slice(variable_ptr, scan_keys[cid].length());
            memcpy(fixed_ptr + 1, &slice, sizeof(Slice));
            variable_ptr += scan_keys[cid].length();
        } else if (type == FieldType::OLAP_FIELD_TYPE_CHAR) {
            // Use memcpy to avoid misaligned store on fixed_ptr + 1
            size_t len = std::max(scan_keys[cid].length(), static_cast<size_t>(column.length()));
            Slice slice(variable_ptr, len);
            memcpy(fixed_ptr + 1, &slice, sizeof(Slice));
            variable_ptr += len;
        } else if (type == FieldType::OLAP_FIELD_TYPE_STRING) {
            // Use memcpy to avoid misaligned store on fixed_ptr + 1
            _schema->mutable_column(cid)->set_long_text_buf(long_text_ptr);
            Slice slice(*(long_text_ptr), DEFAULT_TEXT_LENGTH);
            memcpy(fixed_ptr + 1, &slice, sizeof(Slice));
            ++long_text_ptr;
        }
    }

    return Status::OK();
}

Status RowCursor::_init(TabletSchemaSPtr schema, uint32_t column_count) {
    if (column_count > schema->num_columns()) {
        return Status::Error<INVALID_ARGUMENT>(
                "Input param are invalid. Column count is bigger than num_columns of schema. "
                "column_count={}, schema.num_columns={}",
                column_count, schema->num_columns());
    }
    std::vector<uint32_t> columns;
    for (auto i = 0; i < column_count; ++i) {
        columns.push_back(i);
    }
    RETURN_IF_ERROR(_init(schema->columns(), columns));
    return Status::OK();
}

Status RowCursor::init_scan_key(TabletSchemaSPtr schema,
                                const std::vector<std::string>& scan_keys) {
    size_t scan_key_size = scan_keys.size();
    if (scan_key_size > schema->num_columns()) {
        return Status::Error<INVALID_ARGUMENT>(
                "Input param are invalid. Column count is bigger than num_columns of schema. "
                "column_count={}, schema.num_columns={}",
                scan_key_size, schema->num_columns());
    }

    std::vector<uint32_t> columns(scan_key_size);
    std::iota(columns.begin(), columns.end(), 0);

    RETURN_IF_ERROR(_init(schema->columns(), columns));

    return _init_scan_key(schema, scan_keys);
}

Status RowCursor::init_scan_key(TabletSchemaSPtr schema, const std::vector<std::string>& scan_keys,
                                const std::shared_ptr<Schema>& shared_schema) {
    size_t scan_key_size = scan_keys.size();

    std::vector<uint32_t> columns;
    for (uint32_t i = 0; i < scan_key_size; ++i) {
        columns.push_back(i);
    }

    RETURN_IF_ERROR(_init(shared_schema, columns));

    return _init_scan_key(schema, scan_keys);
}

Status RowCursor::from_tuple(const OlapTuple& tuple) {
    if (tuple.size() != _schema->num_column_ids()) {
        return Status::Error<INVALID_ARGUMENT>(
                "column count does not match. tuple_size={}, field_count={}", tuple.size(),
                _schema->num_column_ids());
    }
    _row_string.resize(tuple.size());

    for (size_t i = 0; i < tuple.size(); ++i) {
        auto cid = _schema->column_ids()[i];
        const StorageField* field = column_schema(cid);
        if (tuple.is_null(i)) {
            _set_null(cid);
            continue;
        }
        _set_not_null(cid);
        _row_string[i] = tuple.get_value(i);
        char* buf = _cell_ptr(cid);
        Status res = field->from_string(buf, tuple.get_value(i), field->get_precision(),
                                        field->get_scale());
        if (!res.ok()) {
            LOG(WARNING) << "fail to convert field from string. string=" << tuple.get_value(i)
                         << ", res=" << res;
            return res;
        }
    }

    return Status::OK();
}

std::string RowCursor::to_string() const {
    std::string result;
    size_t i = 0;
    for (auto cid : _schema->column_ids()) {
        if (i > 0) {
            result.append("|");
        }

        result.append(std::to_string(_is_null(cid)));
        result.append("&");
        if (_is_null(cid)) {
            result.append("NULL");
        } else {
            result.append(_row_string[i]);
        }
        ++i;
    }

    return result;
}
Status RowCursor::_alloc_buf() {
    // variable_len for null bytes
    _variable_buf = new (nothrow) char[_variable_len]();
    if (_variable_buf == nullptr) {
        return Status::Error<MEM_ALLOC_FAILED>("Fail to malloc _variable_buf.");
    }
    if (_string_field_count > 0) {
        _long_text_buf = (char**)malloc(_string_field_count * sizeof(char*));
        if (_long_text_buf == nullptr) {
            return Status::Error<MEM_ALLOC_FAILED>("Fail to malloc _long_text_buf.");
        }
        for (int i = 0; i < _string_field_count; ++i) {
            _long_text_buf[i] = (char*)malloc(DEFAULT_TEXT_LENGTH * sizeof(char));
            if (_long_text_buf[i] == nullptr) {
                return Status::Error<MEM_ALLOC_FAILED>("Fail to malloc _long_text_buf.");
            }
        }
    }
    return Status::OK();
}

#include "common/compile_check_end.h"
} // namespace doris