Method loadAndDumpPart

programs/dumper/Dumper.cpp:478–534

}

void ClickHouseDumper::loadAndDumpPart(StorageCloudMergeTree & cloud, MergeTreeCNCHDataDumper & dumper, const String & part_name)
{
    DiskPtr local_disk = global_context->getStoragePolicy("default")->getAnyDisk();
    DiskPtr remote_disk = global_context->getStoragePolicy("cnch_default_hdfs")->getAnyDisk();
    if (remote_disk->getType() == DiskType::Type::ByteS3)
        throw Exception("Currently dump to  " + DiskType::toString(remote_disk->getType()) + " doesn't supported.", ErrorCodes::UNKNOWN_TABLE);

    auto volume = std::make_shared<SingleDiskVolume>("volume_single", local_disk, 0);
    const String to_path = '/' + cloud.getDatabaseName() + '/' + cloud.getTableName() + '/';
    const String relative_path = config().getString("path") + "/data/" +  cloud.getDatabaseName() + "/" + cloud.getTableName() + "/" + part_name;
    const String from_path =  cloud.getRelativeDataPath(IStorage::StorageLocation::MAIN) + '/' + part_name;
    bool overwrite = config().has("overwrite");

    LOG_TRACE(log, "loadAndDumpPart local disk path = {}, remote disk path = {}, relative_path = {}", local_disk->getPath(), remote_disk->getPath(), relative_path);
    const String full_path = remote_disk->getPath() + to_path + part_name;
    if(!overwrite && remote_disk->exists(full_path))
    {
        LOG_WARNING(log, "Part " + part_name + " already exists. Ignore it.");
        return;
    }

    auto part_info = MergeTreePartInfo::fromPartName(part_name, cloud.format_version);
    auto local_part = std::make_shared<MergeTreeDataPartWide>(
        cloud,
        part_name,
        part_info,
        volume,
        relative_path,
        nullptr,
        IStorage::StorageLocation::AUXILITY);

    bool skip_corrupt_parts = config().has("skip_corrupt_parts");
    try
    {
        LOG_TRACE(log, "Loading {}", part_name);

        local_part->loadColumnsChecksumsIndexes(true, true);

        LOG_TRACE(log, "Dumping {} remote_disk path {}", local_part->name, remote_disk->getPath());

        auto dumped_part = dumper.dumpTempPart(local_part, remote_disk, true);
        dumped_part->is_temp = false;
        dumped_part->renameTo(local_part->name, true);

        if (remote_disk->exists(full_path))
            remote_disk->removeRecursive(full_path);

        /// remote disk from uuid path move to {database}/{table}
        remote_disk->moveDirectory(from_path, remote_disk->getPath() + to_path);
    }
    catch (const DB::Exception & e)
    {
        if (!skip_corrupt_parts)
            throw;
        LOG_ERROR(log, "Failed to load or dump the part: {}, exception: {}", part_name, e.what());
    }
}