wal_reader.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 "wal_reader.h"

#include <absl/strings/str_split.h>

#include "agent/be_exec_version_manager.h"
#include "common/logging.h"
#include "core/block/block.h"
#include "cpp/sync_point.h"
#include "load/group_commit/wal/wal_manager.h"
#include "runtime/runtime_state.h"

namespace doris {
WalReader::WalReader(RuntimeState* state) : _state(state) {
    _wal_id = state->wal_id();
}

Status WalReader::init_reader(const TupleDescriptor* tuple_descriptor) {
    _tuple_descriptor = tuple_descriptor;
    RETURN_IF_ERROR(_state->exec_env()->wal_mgr()->get_wal_path(_wal_id, _wal_path));
    _wal_reader = std::make_shared<doris::WalFileReader>(_wal_path);
    RETURN_IF_ERROR(_wal_reader->init());
    return Status::OK();
}

// ---- Unified init_reader(ReaderInitContext*) overrides ----

Status WalReader::_open_file_reader(ReaderInitContext* ctx) {
    auto* wal_ctx = checked_context_cast<WalInitContext>(ctx);
    _tuple_descriptor = wal_ctx->output_tuple_descriptor;
    RETURN_IF_ERROR(_state->exec_env()->wal_mgr()->get_wal_path(_wal_id, _wal_path));
    _wal_reader = std::make_shared<doris::WalFileReader>(_wal_path);
    RETURN_IF_ERROR(_wal_reader->init());
    return Status::OK();
}

Status WalReader::_do_init_reader(ReaderInitContext* /*base_ctx*/) {
    return Status::OK();
}

Status WalReader::_do_get_next_block(Block* block, size_t* read_rows, bool* eof) {
    //read src block
    PBlock pblock;
    auto st = _wal_reader->read_block(pblock);
    if (st.is<ErrorCode::END_OF_FILE>()) {
        LOG(INFO) << "read eof on wal:" << _wal_path;
        *read_rows = 0;
        *eof = true;
        return Status::OK();
    }
    if (!st.ok()) {
        LOG(WARNING) << "Failed to read wal on path = " << _wal_path;
        return st;
    }
    int be_exec_version = pblock.has_be_exec_version() ? pblock.be_exec_version() : 0;
    if (!BeExecVersionManager::check_be_exec_version(be_exec_version)) {
        return Status::DataQualityError("check be exec version fail when reading wal file {}",
                                        _wal_path);
    }
    Block src_block;
    size_t uncompressed_size = 0;
    int64_t uncompressed_time = 0;
    RETURN_IF_ERROR(src_block.deserialize(pblock, &uncompressed_size, &uncompressed_time));
    //convert to dst block
    Block dst_block;
    int index = 0;
    auto output_block_columns = block->get_columns_with_type_and_name();
    size_t output_block_column_size = output_block_columns.size();
    TEST_SYNC_POINT_CALLBACK("WalReader::set_column_id_count", &_column_id_count);
    TEST_SYNC_POINT_CALLBACK("WalReader::set_out_block_column_size", &output_block_column_size);
    if (_column_id_count != src_block.columns() ||
        output_block_column_size != _tuple_descriptor->slots().size()) {
        return Status::InternalError(
                "not equal wal _column_id_count={} vs wal block columns size={}, "
                "output block columns size={} vs tuple_descriptor size={}",
                std::to_string(_column_id_count), std::to_string(src_block.columns()),
                std::to_string(output_block_column_size),
                std::to_string(_tuple_descriptor->slots().size()));
    }
    for (auto* slot_desc : _tuple_descriptor->slots()) {
        auto pos = _column_pos_map[slot_desc->col_unique_id()];
        if (pos >= src_block.columns()) {
            return Status::InternalError("read wal {} fail, pos {}, columns size {}", _wal_path,
                                         pos, src_block.columns());
        }
        ColumnPtr column_ptr = src_block.get_by_position(pos).column;
        if (!column_ptr && slot_desc->is_nullable()) {
            column_ptr = make_nullable(column_ptr);
        }
        dst_block.insert(index, ColumnWithTypeAndName(std::move(column_ptr),
                                                      output_block_columns[index].type,
                                                      output_block_columns[index].name));
        index++;
    }
    block->swap(dst_block);
    *read_rows = block->rows();
    VLOG_DEBUG << "read block rows:" << *read_rows;

    return Status::OK();
}

Status WalReader::_get_columns_impl(std::unordered_map<std::string, DataTypePtr>* name_to_type) {
    std::string col_ids;
    RETURN_IF_ERROR(_wal_reader->read_header(_version, col_ids));
    std::vector<std::string> column_id_vector =
            absl::StrSplit(col_ids, ",", absl::SkipWhitespace());
    _column_id_count = column_id_vector.size();
    try {
        int64_t pos = 0;
        for (auto col_id_str : column_id_vector) {
            auto col_id = std::strtoll(col_id_str.c_str(), nullptr, 10);
            _column_pos_map.emplace(col_id, pos);
            pos++;
        }
    } catch (const std::invalid_argument& e) {
        return Status::InvalidArgument("Invalid format, {}", e.what());
    }
    // Report WAL columns so on_before_init_reader does not mark them as missing.
    if (_tuple_descriptor) {
        for (auto* slot_desc : _tuple_descriptor->slots()) {
            if (_column_pos_map.contains(slot_desc->col_unique_id())) {
                name_to_type->emplace(slot_desc->col_name(), slot_desc->get_data_type_ptr());
            }
        }
    }
    return Status::OK();
}

} // namespace doris