Method pullLogsToQueue

src/Storages/MergeTree/ReplicatedMergeTreeQueue.cpp:819–984

}

std::pair<int32_t, int32_t> ReplicatedMergeTreeQueue::pullLogsToQueue(zkutil::ZooKeeperPtr zookeeper, Coordination::WatchCallbackPtr watch_callback, PullLogsReason reason)
{
    std::lock_guard lock(pull_logs_to_queue_mutex);

    if (reason != LOAD && reason != FIX_METADATA_VERSION)
    {
        /// It's totally ok to load queue on readonly replica (that's what RestartingThread does on initialization).
        /// It's ok if replica became readonly due to connection loss after we got current zookeeper (in this case zookeeper must be expired).
        /// And it's ok if replica became readonly after shutdown.
        /// In other cases it's likely that someone called pullLogsToQueue(...) when queue is not initialized yet by RestartingThread.
        bool not_completely_initialized = storage.is_readonly && !zookeeper->expired() && !storage.shutdown_prepared_called;
        if (not_completely_initialized)
            throw Exception(ErrorCodes::LOGICAL_ERROR, "Tried to pull logs to queue (reason: {}) on readonly replica {}, it's a bug",
                            reason, storage.getStorageID().getNameForLogs());
    }

    if (pull_log_blocker.isCancelled())
        throw Exception(ErrorCodes::ABORTED, "Log pulling is cancelled");

    String index_str = zookeeper->get(fs::path(replica_path) / "log_pointer");
    UInt64 index = 0;

    /// The version of "/log" is modified when new entries to merge/mutate/drop appear.
    Coordination::Stat stat;
    zookeeper->get(fs::path(zookeeper_path) / "log", &stat);

    Strings log_entries = zookeeper->getChildrenWatch(
        fs::path(zookeeper_path) / "log",
        nullptr,
        Coordination::WatchCallbackPtrOrEventPtr{watch_callback, ProfileEvents::ZooKeeperWatchTriggeredReplicatedMergeTreeLog});

    /// We update mutations after we have loaded the list of log entries, but before we insert them
    /// in the queue.
    /// With this we ensure that if you read the log state L1 and then the state of mutations M1,
    /// then L1 "happened-before" M1.
    int32_t mutations_version = updateMutations(zookeeper);

    if (index_str.empty())
    {
        /// If we do not already have a pointer to the log, put a pointer to the first entry in it.
        index = log_entries.empty() ? 0 : parse<UInt64>(std::min_element(log_entries.begin(), log_entries.end())->substr(strlen("log-")));

        zookeeper->set(fs::path(replica_path) / "log_pointer", toString(index));
    }
    else
    {
        index = parse<UInt64>(index_str);
    }

    String min_log_entry = "log-" + padIndex(index);

    /// Multiple log entries that must be copied to the queue.

    std::erase_if(log_entries, [&min_log_entry](const String & entry) { return entry < min_log_entry; });

    if (!log_entries.empty())
    {
        ::sort(log_entries.begin(), log_entries.end());