Method buildInputStreamFromSource

src/Interpreters/InterpreterInsertQuery.cpp:757–828

}

BlockInputStreamPtr InterpreterInsertQuery::buildInputStreamFromSource(
    const ContextPtr context_ptr,
    const ColumnsDescription & columns,
    const Block & sample,
    const Settings & settings,
    const String & source_uri,
    const String & format,
    bool is_enable_squash,
    const String & compression_method)
{
    Poco::URI uri(source_uri);
    const String & scheme = uri.getScheme();

    BlockInputStreams inputs;
    {
        std::vector<String> file_path_list;
        parseFuzzyName(context_ptr, file_path_list, source_uri, scheme);

        for (auto & file_path : file_path_list)
        {
            std::unique_ptr<ReadBuffer> read_buf = nullptr;

            if (scheme.empty() || scheme == "file")
            {
                read_buf = std::make_unique<ReadBufferFromFile>(Poco::URI(file_path).getPath());
            }
#if USE_HDFS
            else if (DB::isHdfsOrCfsScheme(scheme))
            {
                ReadSettings read_settings;
                read_settings.remote_throttler = context_ptr->getProcessList().getHDFSDownloadThrottler();
                read_buf = std::make_unique<ReadBufferFromByteHDFS>(file_path, context_ptr->getHdfsConnectionParams(), read_settings);
            }
#endif
#if USE_AWS_S3
            else if (isS3URIScheme(scheme))
            {
                S3::URI s3_uri(file_path);
                String endpoint = s3_uri.endpoint.empty() ? context_ptr->getSettingsRef().s3_endpoint.toString() : s3_uri.endpoint;
                String bucket = s3_uri.bucket;
                String key = s3_uri.key;
                S3::S3Config s3_cfg(
                    endpoint,
                    context_ptr->getSettingsRef().s3_region.toString(),
                    bucket,
                    context_ptr->getSettingsRef().s3_ak_id.toString(),
                    context_ptr->getSettingsRef().s3_ak_secret.toString(),
                    "",
                    "",
                    context_ptr->getSettingsRef().s3_use_virtual_hosted_style);
                const std::shared_ptr<Aws::S3::S3Client> client = s3_cfg.create();
                read_buf = std::make_unique<ReadBufferFromS3>(client, bucket, key, context_ptr->getReadSettings());
            }
#endif
            else
            {
                throw Exception("URI scheme " + scheme + " is not supported with insert statement yet", ErrorCodes::NOT_IMPLEMENTED);
            }