be/src/olap/version_graph.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 "olap/version_graph.h"

#include <cctz/time_zone.h>
#include <stddef.h>

#include <algorithm>
// IWYU pragma: no_include <bits/chrono.h>
#include <chrono> // IWYU pragma: keep
#include <list>
#include <memory>
#include <optional>
#include <ostream>
#include <ranges>
#include <utility>

#include "common/logging.h"

namespace doris {
#include "common/compile_check_begin.h"

using namespace ErrorCode;

void TimestampedVersionTracker::construct_versioned_tracker(
        const RowsetMetaMapContainer& rs_metas) {
    construct_versioned_tracker(std::views::values(rs_metas));
}

void TimestampedVersionTracker::construct_versioned_tracker(
        const RowsetMetaMapContainer& rs_metas, const RowsetMetaMapContainer& stale_metas) {
    construct_versioned_tracker(std::views::values(rs_metas), std::views::values(stale_metas));
}

bool TimestampedVersionTracker::_find_path_from_stale_map(
        const std::unordered_map<int64_t, std::unordered_map<int64_t, RowsetMetaSharedPtr>>&
                stale_map,
        int64_t first_version, int64_t second_version,
        std::vector<RowsetMetaSharedPtr>* stale_path) {
    auto first_iter = stale_map.find(first_version);
    // If `first_version` not in `stale_map`, there is no path.
    if (first_iter == stale_map.end()) {
        return false;
    }
    auto& second_version_map = first_iter->second;
    auto second_iter = second_version_map.find(second_version);
    // If second_version in `stale_map`, find a path.
    if (second_iter != second_version_map.end()) {
        auto row_meta = second_iter->second;
        // Add rowset to path.
        stale_path->push_back(row_meta);
        return true;
    }

    // Traverse the first version map to backtracking  `_find_path_from_stale_map`.
    auto map_iter = second_version_map.begin();
    while (map_iter != second_version_map.end()) {
        // The version greater than `second_version`, we can't find path in `stale_map`.
        if (map_iter->first > second_version) {
            map_iter++;
            continue;
        }
        // Backtracking `_find_path_from_stale_map` find from `map_iter->first + 1` to `second_version`.
        stale_path->push_back(map_iter->second);
        bool r = _find_path_from_stale_map(stale_map, map_iter->first + 1, second_version,
                                           stale_path);
        if (r) {
            return true;
        }
        // There is no path in current version, pop and continue.
        stale_path->pop_back();
        map_iter++;
    }

    return false;
}

void TimestampedVersionTracker::get_stale_version_path_json_doc(rapidjson::Document& path_arr) {
    auto path_arr_iter = _stale_version_path_map.begin();

    // Do loop version path.
    while (path_arr_iter != _stale_version_path_map.end()) {
        auto path_id = path_arr_iter->first;
        auto path_version_path = path_arr_iter->second;

        rapidjson::Document item;
        item.SetObject();
        // Add `path_id` to item.
        auto path_id_str = std::to_string(path_id);
        rapidjson::Value path_id_value;
        path_id_value.SetString(path_id_str.c_str(),
                                static_cast<rapidjson::SizeType>(path_id_str.length()),
                                path_arr.GetAllocator());
        item.AddMember("path id", path_id_value, path_arr.GetAllocator());

        // Add max create time to item.
        auto time_zone = cctz::local_time_zone();

        auto tp = std::chrono::system_clock::from_time_t(path_version_path->max_create_time());
        auto create_time_str = cctz::format("%Y-%m-%d %H:%M:%S %z", tp, time_zone);

        rapidjson::Value create_time_value;
        create_time_value.SetString(create_time_str.c_str(),
                                    static_cast<rapidjson::SizeType>(create_time_str.length()),
                                    path_arr.GetAllocator());
        item.AddMember("last create time", create_time_value, path_arr.GetAllocator());

        // Add path list to item.
        std::stringstream path_list_stream;
        path_list_stream << path_id_str;
        auto path_list_ptr = path_version_path->timestamped_versions();
        auto path_list_iter = path_list_ptr.begin();
        while (path_list_iter != path_list_ptr.end()) {
            path_list_stream << " -> ";
            path_list_stream << "[";
            path_list_stream << (*path_list_iter)->version().first;
            path_list_stream << "-";
            path_list_stream << (*path_list_iter)->version().second;
            path_list_stream << "]";
            path_list_iter++;
        }
        std::string path_list = path_list_stream.str();
        rapidjson::Value path_list_value;
        path_list_value.SetString(path_list.c_str(),
                                  static_cast<rapidjson::SizeType>(path_list.length()),
                                  path_arr.GetAllocator());
        item.AddMember("path list", path_list_value, path_arr.GetAllocator());

        // Add item to `path_arr`.
        path_arr.PushBack(item, path_arr.GetAllocator());

        path_arr_iter++;
    }
}

void TimestampedVersionTracker::recover_versioned_tracker(
        const std::map<int64_t, PathVersionListSharedPtr>& stale_version_path_map) {
    auto _path_map_iter = stale_version_path_map.begin();
    // Recover `stale_version_path_map`.
    while (_path_map_iter != stale_version_path_map.end()) {
        // Add `PathVersionListSharedPtr` to map.
        _stale_version_path_map[_path_map_iter->first] = _path_map_iter->second;

        std::vector<TimestampedVersionSharedPtr>& timestamped_versions =
                _path_map_iter->second->timestamped_versions();
        std::vector<TimestampedVersionSharedPtr>::iterator version_path_iter =
                timestamped_versions.begin();
        while (version_path_iter != timestamped_versions.end()) {
            // Add version to `_version_graph`.
            _version_graph.add_version_to_graph((*version_path_iter)->version());
            ++version_path_iter;
        }
        ++_path_map_iter;
    }
    LOG(INFO) << "recover_versioned_tracker current map info " << get_current_path_map_str();
}

void TimestampedVersionTracker::add_version(const Version& version) {
    _version_graph.add_version_to_graph(version);
}

void TimestampedVersionTracker::add_stale_path_version(
        const std::vector<RowsetMetaSharedPtr>& stale_rs_metas) {
    if (stale_rs_metas.empty()) {
        VLOG_NOTICE << "there is no version in the stale_rs_metas.";
        return;
    }

    PathVersionListSharedPtr ptr(new TimestampedVersionPathContainer());
    for (auto rs : stale_rs_metas) {
        TimestampedVersionSharedPtr vt_ptr(new TimestampedVersion(rs->version(), rs->stale_at()));
        ptr->add_timestamped_version(vt_ptr);
    }

    std::vector<TimestampedVersionSharedPtr>& timestamped_versions = ptr->timestamped_versions();

    struct TimestampedVersionPtrCompare {
        bool operator()(const TimestampedVersionSharedPtr ptr1,
                        const TimestampedVersionSharedPtr ptr2) {
            return ptr1->version().first < ptr2->version().first;
        }
    };
    sort(timestamped_versions.begin(), timestamped_versions.end(), TimestampedVersionPtrCompare());
    _stale_version_path_map[_next_path_id] = ptr;
    _next_path_id++;
}

// Capture consistent versions from graph.
Status TimestampedVersionTracker::capture_consistent_versions(
        const Version& spec_version, std::vector<Version>* version_path) const {
    return _version_graph.capture_consistent_versions(spec_version, version_path);
}

Status TimestampedVersionTracker::capture_consistent_versions_with_validator(
        const Version& spec_version, std::vector<Version>& version_path,
        const std::function<bool(int64_t, int64_t)>& validator) const {
    return _version_graph.capture_consistent_versions_with_validator(spec_version, version_path,
                                                                     validator);
}

Status TimestampedVersionTracker::capture_consistent_versions_prefer_cache(
        const Version& spec_version, std::vector<Version>& version_path,
        const std::function<bool(int64_t, int64_t)>& validator) const {
    return _version_graph.capture_consistent_versions_prefer_cache(spec_version, version_path,
                                                                   validator);
}

Status TimestampedVersionTracker::capture_consistent_versions_with_validator_mow(
        const Version& spec_version, std::vector<Version>& version_path,
        const std::function<bool(int64_t, int64_t)>& validator) const {
    return _version_graph.capture_consistent_versions_with_validator_mow(spec_version, version_path,
                                                                         validator);
}

void TimestampedVersionTracker::capture_expired_paths(
        int64_t stale_sweep_endtime, std::vector<int64_t>* path_version_vec) const {
    std::map<int64_t, PathVersionListSharedPtr>::const_iterator iter =
            _stale_version_path_map.begin();

    while (iter != _stale_version_path_map.end()) {
        int64_t max_create_time = iter->second->max_create_time();
        if (max_create_time <= stale_sweep_endtime) {
            int64_t path_version = iter->first;
            path_version_vec->push_back(path_version);
        }
        ++iter;
    }
}

PathVersionListSharedPtr TimestampedVersionTracker::fetch_path_version_by_id(int64_t path_id) {
    if (_stale_version_path_map.count(path_id) == 0) {
        VLOG_NOTICE << "path version " << path_id << " does not exist!";
        return nullptr;
    }

    return _stale_version_path_map[path_id];
}

PathVersionListSharedPtr TimestampedVersionTracker::fetch_and_delete_path_by_id(int64_t path_id) {
    if (_stale_version_path_map.count(path_id) == 0) {
        VLOG_NOTICE << "path version " << path_id << " does not exist!";
        return nullptr;
    }

    VLOG_NOTICE << get_current_path_map_str();
    PathVersionListSharedPtr ptr = fetch_path_version_by_id(path_id);

    _stale_version_path_map.erase(path_id);

    for (auto& version : ptr->timestamped_versions()) {
        static_cast<void>(_version_graph.delete_version_from_graph(version->version()));
    }
    return ptr;
}

std::string TimestampedVersionTracker::get_current_path_map_str() {
    std::stringstream tracker_info;
    tracker_info << "current expired next_path_id " << _next_path_id << std::endl;

    std::map<int64_t, PathVersionListSharedPtr>::const_iterator iter =
            _stale_version_path_map.begin();
    while (iter != _stale_version_path_map.end()) {
        tracker_info << "current expired path_version " << iter->first;
        std::vector<TimestampedVersionSharedPtr>& timestamped_versions =
                iter->second->timestamped_versions();
        std::vector<TimestampedVersionSharedPtr>::iterator version_path_iter =
                timestamped_versions.begin();
        int64_t max_create_time = -1;
        while (version_path_iter != timestamped_versions.end()) {
            if (max_create_time < (*version_path_iter)->get_create_time()) {
                max_create_time = (*version_path_iter)->get_create_time();
            }
            tracker_info << " -> [";
            tracker_info << (*version_path_iter)->version().first;
            tracker_info << ",";
            tracker_info << (*version_path_iter)->version().second;
            tracker_info << "]";

            ++version_path_iter;
        }

        tracker_info << std::endl;
        ++iter;
    }
    return tracker_info.str();
}

double TimestampedVersionTracker::get_orphan_vertex_ratio() {
    return _version_graph.get_orphan_vertex_ratio();
}

std::string TimestampedVersionTracker::debug_string() const {
    return _version_graph.debug_string();
}

void TimestampedVersionPathContainer::add_timestamped_version(TimestampedVersionSharedPtr version) {
    // Compare and refresh `_max_create_time`.
    if (version->get_create_time() > _max_create_time) {
        _max_create_time = version->get_create_time();
    }
    _timestamped_versions_container.push_back(version);
}

std::vector<TimestampedVersionSharedPtr>& TimestampedVersionPathContainer::timestamped_versions() {
    return _timestamped_versions_container;
}

void VersionGraph::construct_version_graph(const RowsetMetaMapContainer& rs_metas,
                                           int64_t* max_version) {
    construct_version_graph(std::views::values(rs_metas), max_version);
}

void VersionGraph::reconstruct_version_graph(const RowsetMetaMapContainer& rs_metas,
                                             int64_t* max_version) {
    reconstruct_version_graph(std::views::values(rs_metas), max_version);
}

void VersionGraph::add_version_to_graph(const Version& version) {
    // Add version.first as new vertex of version graph if not exist.
    int64_t start_vertex_value = version.first;
    int64_t end_vertex_value = version.second + 1;

    // Add vertex to graph.
    _add_vertex_to_graph(start_vertex_value);
    _add_vertex_to_graph(end_vertex_value);

    int64_t start_vertex_index = _vertex_index_map[start_vertex_value];
    int64_t end_vertex_index = _vertex_index_map[end_vertex_value];

    // We assume this version is new version, so we just add two edges
    // into version graph. add one edge from `start_version` to `end_version`
    // Make sure the vertex's edges are sorted by version in descending order when inserting.
    auto end_vertex_it = _version_graph[start_vertex_index].edges.begin();
    while (end_vertex_it != _version_graph[start_vertex_index].edges.end()) {
        if (_version_graph[*end_vertex_it].value < _version_graph[end_vertex_index].value) {
            break;
        }
        end_vertex_it++;
    }
    _version_graph[start_vertex_index].edges.insert(end_vertex_it, end_vertex_index);

    // We add reverse edge(from end_version to start_version) to graph
    // Make sure the vertex's edges are sorted by version in descending order when inserting.
    auto start_vertex_it = _version_graph[end_vertex_index].edges.begin();
    while (start_vertex_it != _version_graph[end_vertex_index].edges.end()) {
        if (_version_graph[*start_vertex_it].value < _version_graph[start_vertex_index].value) {
            break;
        }
        start_vertex_it++;
    }
    _version_graph[end_vertex_index].edges.insert(start_vertex_it, start_vertex_index);
}

Status VersionGraph::delete_version_from_graph(const Version& version) {
    int64_t start_vertex_value = version.first;
    int64_t end_vertex_value = version.second + 1;

    if (_vertex_index_map.find(start_vertex_value) == _vertex_index_map.end() ||
        _vertex_index_map.find(end_vertex_value) == _vertex_index_map.end()) {
        return Status::Error<HEADER_DELETE_VERSION>(
                "vertex for version does not exists. version={}-{}", version.first, version.second);
    }

    int64_t start_vertex_index = _vertex_index_map[start_vertex_value];
    int64_t end_vertex_index = _vertex_index_map[end_vertex_value];
    // Remove edge and its reverse edge.
    // When there are same versions in edges, just remove the first version.
    auto start_edges_iter = _version_graph[start_vertex_index].edges.begin();
    while (start_edges_iter != _version_graph[start_vertex_index].edges.end()) {
        if (*start_edges_iter == end_vertex_index) {
            _version_graph[start_vertex_index].edges.erase(start_edges_iter);
            break;
        }
        start_edges_iter++;
    }

    auto end_edges_iter = _version_graph[end_vertex_index].edges.begin();
    while (end_edges_iter != _version_graph[end_vertex_index].edges.end()) {
        if (*end_edges_iter == start_vertex_index) {
            _version_graph[end_vertex_index].edges.erase(end_edges_iter);
            break;
        }
        end_edges_iter++;
    }

    // Here we do not delete vertex in `_version_graph` even if its edges are empty.
    // the `_version_graph` will be rebuilt when doing trash sweep.
    return Status::OK();
}

void VersionGraph::_add_vertex_to_graph(int64_t vertex_value) {
    // Vertex with vertex_value already exists.
    if (_vertex_index_map.find(vertex_value) != _vertex_index_map.end()) {
        VLOG_NOTICE << "vertex with vertex value already exists. value=" << vertex_value;
        return;
    }

    _version_graph.emplace_back(Vertex(vertex_value));
    _vertex_index_map[vertex_value] = _version_graph.size() - 1;
}

Status VersionGraph::capture_consistent_versions(const Version& spec_version,
                                                 std::vector<Version>* version_path) const {
    if (spec_version.first > spec_version.second) {
        return Status::Error<INVALID_ARGUMENT, false>(
                "invalid specified version. spec_version={}-{}", spec_version.first,
                spec_version.second);
    }

    int64_t cur_idx = -1;
    for (size_t i = 0; i < _version_graph.size(); i++) {
        if (_version_graph[i].value == spec_version.first) {
            cur_idx = i;
            break;
        }
    }

    if (cur_idx < 0) {
        return Status::InternalError<false>(
                "failed to find path in version_graph. spec_version: {}-{}", spec_version.first,
                spec_version.second);
    }

    int64_t end_value = spec_version.second + 1;
    while (_version_graph[cur_idx].value < end_value) {
        int64_t next_idx = -1;
        for (const auto& it : _version_graph[cur_idx].edges) {
            // Only consider incremental versions.
            if (_version_graph[it].value < _version_graph[cur_idx].value) {
                break;
            }

            if (_version_graph[it].value > end_value) {
                continue;
            }

            // Considering edges had been sorted by version in descending order,
            // This version is the largest version that smaller than `end_version`.
            next_idx = it;
            break;
        }

        if (next_idx > -1) {
            if (version_path != nullptr) {
                version_path->emplace_back(_version_graph[cur_idx].value,
                                           _version_graph[next_idx].value - 1);
            }
            cur_idx = next_idx;
        } else {
            return Status::InternalError<false>(
                    "fail to find path in version_graph. spec_version: {}-{}", spec_version.first,
                    spec_version.second);
        }
    }

    if (VLOG_TRACE_IS_ON && version_path != nullptr) {
        std::stringstream shortest_path_for_debug;
        for (const auto& version : *version_path) {
            shortest_path_for_debug << version << ' ';
        }
        VLOG_TRACE << "success to find path for spec_version. spec_version=" << spec_version
                   << ", path=" << shortest_path_for_debug.str();
    }

    return Status::OK();
}

Status VersionGraph::capture_consistent_versions_prefer_cache(
        const Version& spec_version, std::vector<Version>& version_path,
        const std::function<bool(int64_t, int64_t)>& validator) const {
    if (spec_version.first > spec_version.second) {
        return Status::Error<INVALID_ARGUMENT, false>(
                "invalid specified version. spec_version={}-{}", spec_version.first,
                spec_version.second);
    }

    int64_t cur_idx = -1;
    for (size_t i = 0; i < _version_graph.size(); i++) {
        if (_version_graph[i].value == spec_version.first) {
            cur_idx = i;
            break;
        }
    }

    if (cur_idx < 0) {
        return Status::InternalError<false>("failed to find path in version_graph. spec_version={}",
                                            spec_version.to_string());
    }

    int64_t end_value = spec_version.second + 1;
    while (_version_graph[cur_idx].value < end_value) {
        int64_t next_idx = -1;
        int64_t first_idx = -1;
        for (const auto& it : _version_graph[cur_idx].edges) {
            // Only consider incremental versions.
            if (_version_graph[it].value < _version_graph[cur_idx].value) {
                break;
            }
            if (first_idx == -1) {
                first_idx = it;
            }

            if (!validator(_version_graph[cur_idx].value, _version_graph[it].value - 1)) {
                continue;
            }

            next_idx = it;
            break;
        }

        if (next_idx > -1) {
            version_path.emplace_back(_version_graph[cur_idx].value,
                                      _version_graph[next_idx].value - 1);

            cur_idx = next_idx;
        } else if (first_idx != -1) {
            // if all edges are not in cache, use the first edge if possible
            version_path.emplace_back(_version_graph[cur_idx].value,
                                      _version_graph[first_idx].value - 1);
            cur_idx = first_idx;
        } else {
            return Status::OK();
        }
    }
    return Status::OK();
}

Status VersionGraph::capture_consistent_versions_with_validator(
        const Version& spec_version, std::vector<Version>& version_path,
        const std::function<bool(int64_t, int64_t)>& validator) const {
    if (spec_version.first > spec_version.second) {
        return Status::Error<INVALID_ARGUMENT, false>(
                "invalid specified version. spec_version={}-{}", spec_version.first,
                spec_version.second);
    }

    int64_t cur_idx = -1;
    for (size_t i = 0; i < _version_graph.size(); i++) {
        if (_version_graph[i].value == spec_version.first) {
            cur_idx = i;
            break;
        }
    }

    if (cur_idx < 0) {
        return Status::InternalError<false>("failed to find path in version_graph. spec_version={}",
                                            spec_version.to_string());
    }

    int64_t end_value = spec_version.second + 1;
    while (_version_graph[cur_idx].value < end_value) {
        int64_t next_idx = -1;
        for (const auto& it : _version_graph[cur_idx].edges) {
            // Only consider incremental versions.
            if (_version_graph[it].value < _version_graph[cur_idx].value) {
                break;
            }

            if (!validator(_version_graph[cur_idx].value, _version_graph[it].value - 1)) {
                continue;
            }

            next_idx = it;
            break;
        }

        if (next_idx > -1) {
            version_path.emplace_back(_version_graph[cur_idx].value,
                                      _version_graph[next_idx].value - 1);

            cur_idx = next_idx;
        } else {
            return Status::OK();
        }
    }
    return Status::OK();
}

Status VersionGraph::capture_consistent_versions_with_validator_mow(
        const Version& spec_version, std::vector<Version>& version_path,
        const std::function<bool(int64_t, int64_t)>& validator) const {
    if (spec_version.first > spec_version.second) {
        return Status::Error<INVALID_ARGUMENT, false>(
                "invalid specified version. spec_version={}-{}", spec_version.first,
                spec_version.second);
    }

    int64_t cur_idx = -1;
    for (size_t i = 0; i < _version_graph.size(); i++) {
        if (_version_graph[i].value == spec_version.first) {
            cur_idx = i;
            break;
        }
    }

    if (cur_idx < 0) {
        return Status::InternalError<false>("failed to find path in version_graph. spec_version={}",
                                            spec_version.to_string());
    }

    int64_t end_value = spec_version.second + 1;
    while (_version_graph[cur_idx].value < end_value) {
        int64_t next_idx = -1;
        for (const auto& it : _version_graph[cur_idx].edges) {
            // Only consider incremental versions.
            if (_version_graph[it].value < _version_graph[cur_idx].value) {
                break;
            }

            if (!validator(_version_graph[cur_idx].value, _version_graph[it].value - 1)) {
                if (_version_graph[cur_idx].value + 1 == _version_graph[it].value) {
                    break;
                }
                end_value = std::min(_version_graph[it].value, end_value);
                continue;
            }

            next_idx = it;
            break;
        }

        if (next_idx > -1) {
            version_path.emplace_back(_version_graph[cur_idx].value,
                                      _version_graph[next_idx].value - 1);

            cur_idx = next_idx;
        } else {
            return Status::OK();
        }
    }
    return Status::OK();
}

double VersionGraph::get_orphan_vertex_ratio() {
    int64_t vertex_num = _version_graph.size();
    int64_t orphan_vertex_num = 0;
    for (auto& iter : _version_graph) {
        if (iter.edges.empty()) {
            ++orphan_vertex_num;
        }
    }
    return static_cast<double>(orphan_vertex_num) / static_cast<double>(vertex_num);
}

std::string VersionGraph::debug_string() const {
    std::stringstream ss;
    ss << "VersionGraph: [";
    for (size_t i = 0; i < _version_graph.size(); ++i) {
        ss << "{value: " << _version_graph[i].value << ", edges: [";
        for (const auto& edge : _version_graph[i].edges) {
            if (_version_graph[edge].value > _version_graph[i].value) {
                ss << _version_graph[edge].value << ", ";
            }
        }
        ss << "]}, ";
    }
    ss << "]";
    return ss.str();
}

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