Method tryProcessTable

programs/copier/ClusterCopier.cpp:909–1138

}

bool ClusterCopier::tryProcessTable(const ConnectionTimeouts & timeouts, TaskTable & task_table)
{
    /// Create destination table
    TaskStatus task_status = TaskStatus::Error;

    task_status = tryCreateDestinationTable(timeouts, task_table);
    /// Exit if success
    if (task_status != TaskStatus::Finished)
    {
        LOG_WARNING(log, "Create destination Tale Failed ");
        return false;
    }

    /// Set all_partitions_count for table in Zookeeper
    auto zookeeper = getContext()->getZooKeeper();
    while (true)
    {
        Coordination::Stat stat;
        auto status_json = zookeeper->get(task_zookeeper_path + "/status", &stat);
        auto statuses = StatusAccumulator::fromJSON(status_json);

        /// Exit if someone already set the initial value for this table.
        if (statuses->find(task_table.name_in_config) != statuses->end())
            break;
        (*statuses)[task_table.name_in_config] = StatusAccumulator::TableStatus
        {
            /*all_partitions_count=*/task_table.ordered_partition_names.size(),
            /*processed_partition_count=*/0
        };

        auto statuses_to_commit = StatusAccumulator::serializeToJSON(statuses);
        auto error = zookeeper->trySet(task_zookeeper_path + "/status", statuses_to_commit, stat.version);
        if (error == Coordination::Error::ZOK)
            break;
    }


    /// An heuristic: if previous shard is already done, then check next one without sleeps due to max_workers constraint
    bool previous_shard_is_instantly_finished = false;

    /// Process each partition that is present in cluster
    for (const String & partition_name : task_table.ordered_partition_names)
    {
        if (!task_table.cluster_partitions.count(partition_name))
            throw Exception("There are no expected partition " + partition_name + ". It is a bug", ErrorCodes::LOGICAL_ERROR);

        ClusterPartition & cluster_partition = task_table.cluster_partitions[partition_name];

        Stopwatch watch;
        /// We will check all the shards of the table and check if they contain current partition.
        TasksShard expected_shards;
        UInt64 num_failed_shards = 0;

        ++cluster_partition.total_tries;

        LOG_INFO(log, "Processing partition {} for the whole cluster", partition_name);

        /// Process each source shard having current partition and copy current partition