Method tryCreateDestinationTable

programs/copier/ClusterCopier.cpp:1140–1172

}

TaskStatus ClusterCopier::tryCreateDestinationTable(const ConnectionTimeouts & timeouts, TaskTable & task_table)
{
    /// Try create original table (if not exists) on each shard

    //TaskTable & task_table = task_shard.task_table;
    const TaskShardPtr task_shard = task_table.all_shards.at(0);
    /// We need to update table definitions for each part, it could be changed after ALTER
    task_shard->current_pull_table_create_query = getCreateTableForPullShard(timeouts, *task_shard);
    try
    {
        auto create_query_push_ast
            = rewriteCreateQueryStorage(task_shard->current_pull_table_create_query, task_table.table_push, task_table.engine_push_ast);
        auto & create = create_query_push_ast->as<ASTCreateQuery &>();
        create.if_not_exists = true;
        InterpreterCreateQuery::prepareOnClusterQuery(create, getContext(), task_table.cluster_push_name);
        String query = queryToString(create_query_push_ast);

        LOG_INFO(log, "Create destination tables. Query: \n {}", query);
        UInt64 shards = executeQueryOnCluster(task_table.cluster_push, query, task_cluster->settings_push, ClusterExecutionMode::ON_EACH_NODE);
        LOG_INFO(
            log,
            "Destination tables {} have been created on {} shards of {}",
            getQuotedTable(task_table.table_push),
            shards,
            task_table.cluster_push->getShardCount());
    }
    catch (...)
    {
        tryLogCurrentException(log, "Error while creating original table. Maybe we are not first.");
    }

    return TaskStatus::Finished;
}

/// Job for copying partition from particular shard.
TaskStatus ClusterCopier::tryProcessPartitionTask(const ConnectionTimeouts & timeouts, ShardPartition & task_partition, bool is_unprioritized_task)