```
ubuntu@ip-172-31-79-147:/opt/downloads$ clickhouse-local --max_threads 8 --query "WITH parseDateTimeBestEffort(extract(_path, 'pageviews-([\\d\\-]+)\\.gz$')) AS time, splitByChar(' ', line) AS values, splitByChar('.', values[1]) AS projects SELECT time, projects[1] AS project, projects[2] AS subproject, values[2] AS path, (values[3])::UInt64 AS hits FROM file('pageviews*.gz', LineAsString) WHERE length(values) >= 3" | pv > /dev/null
^C01GiB 0:01:08 [ 129MiB/s] [                                                                                                                                                                   <=>                                                                                    ]

ubuntu@ip-172-31-79-147:/opt/downloads$ clickhouse-local --max_threads 8 --query "WITH parseDateTimeBestEffort(extract(toLowCardinality(_path), 'pageviews-([\\d\\-]+)\\.gz$')) AS time, splitByChar(' ', line) AS values, splitByChar('.', values[1]) AS projects SELECT time, projects[1] AS project, projects[2] AS subproject, values[2] AS path, (values[3])::UInt64 AS hits FROM file('pageviews*.gz', LineAsString) WHERE length(values) >= 3" | pv > /dev/null
^C46GiB 0:00:48 [ 175MiB/s] [                                                                                                                                                                                                                                        <=>               ]
```

CHECK_KAFKA_STORAGE_ARGUMENT(6, kafka_schema, 2)
            CHECK_KAFKA_STORAGE_ARGUMENT(7, kafka_num_consumers, 0)
            CHECK_KAFKA_STORAGE_ARGUMENT(8, kafka_max_block_size, 0)
            CHECK_KAFKA_STORAGE_ARGUMENT(9, kafka_skip_broken_messages, 0)
            CHECK_KAFKA_STORAGE_ARGUMENT(10, kafka_commit_every_batch, 0)
        }
        #undef CHECK_KAFKA_STORAGE_ARGUMENT
        auto num_consumers = kafka_settings->kafka_num_consumers.value;
        auto physical_cpu_cores = getNumberOfPhysicalCPUCores();
        if (num_consumers > physical_cpu_cores)
        {
            throw Exception(ErrorCodes::BAD_ARGUMENTS, "Number of consumers can not be bigger than {}", physical_cpu_cores);
        }
        else if (num_consumers < 1)
        {
            throw Exception("Number of consumers can not be lower than 1", ErrorCodes::BAD_ARGUMENTS);
        }
        if (kafka_settings->kafka_max_block_size.changed && kafka_settings->kafka_max_block_size.value < 1)
        {
            throw Exception("kafka_max_block_size can not be lower than 1", ErrorCodes::BAD_ARGUMENTS);
        }
        if (kafka_settings->kafka_poll_max_batch_size.changed && kafka_settings->kafka_poll_max_batch_size.value < 1)
        {
            throw Exception("kafka_poll_max_batch_size can not be lower than 1", ErrorCodes::BAD_ARGUMENTS);
        }
        return StorageKafka::create(args.table_id, args.getContext(), args.columns, std::move(kafka_settings), collection_name);
    };
    factory.registerStorage("Kafka", creator_fn, StorageFactory::StorageFeatures{ .supports_settings = true, });
}
NamesAndTypesList StorageKafka::getVirtuals() const