```
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] [                                                                                                                                                                                                                                        <=>               ]
```

else
        storage_metadata.setColumns(columns_);
    storage_metadata.setConstraints(constraints_);
    storage_metadata.setComment(comment);
    setInMemoryMetadata(storage_metadata);
    if (sharding_key_)
    {
        sharding_key_expr = buildShardingKeyExpression(sharding_key_, getContext(), storage_metadata.getColumns().getAllPhysical(), false);
        sharding_key_column_name = sharding_key_->getColumnName();
        sharding_key_is_deterministic = isExpressionActionsDeterministic(sharding_key_expr);
    }
    if (!relative_data_path.empty())
    {
        storage_policy = getContext()->getStoragePolicy(storage_policy_name_);
        data_volume = storage_policy->getVolume(0);
        if (storage_policy->getVolumes().size() > 1)
            LOG_WARNING(log, "Storage policy for Distributed table has multiple volumes. "
                             "Only {} volume will be used to store data. Other will be ignored.", data_volume->getName());
    }
    
    if (!attach_)
    {
        if (remote_database.empty() && !remote_table_function_ptr && !getCluster()->maybeCrossReplication())
            LOG_WARNING(log, "Name of remote database is empty. Default database will be used implicitly.");
        size_t num_local_shards = getCluster()->getLocalShardCount();
        if (num_local_shards && (remote_database.empty() || remote_database == id_.database_name) && remote_table == id_.table_name)
            throw Exception("Distributed table " + id_.table_name + " looks at itself", ErrorCodes::INFINITE_LOOP);
    }
}
StorageDistributed::StorageDistributed(