This query does not show realtime metrics:

```
INSERT INTO wikistat 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,
    CAST(values[3], 'UInt64') AS hits
FROM s3('https://....s3.eu-central-1.amazonaws.com/wikistat/original/*.gz', '...', '...', LineAsString, auto)
WHERE length(values) >= 3
```

Although this query does:

```
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,
    CAST(values[3], 'UInt64') AS hits
FROM s3('https://....s3.eu-central-1.amazonaws.com/wikistat/original/*.gz', '...', '...', LineAsString, auto)
WHERE length(values) >= 3
```

I also was not able to cancel the first query.
auto file_name = current_path.substr(last_slash_pos + 1);
            auto column = DataTypeLowCardinality{std::make_shared<DataTypeString>()}.createColumnConst(num_rows, std::move(file_name));
            columns.push_back(column->convertToFullColumnIfConst());
        }
        return Chunk(std::move(columns), num_rows);
    }
    {
        std::lock_guard lock(reader_mutex);
        reader.reset();
        pipeline.reset();
        read_buf.reset();
        if (!initialize())
            return {};
    }
    return generate();
}
class HDFSSink : public SinkToStorage
{
public:
    HDFSSink(const String & uri,
        const String & format,
        const Block & sample_block,
        ContextPtr context,
        const CompressionMethod compression_method)
        : SinkToStorage(sample_block)
    {
        write_buf = wrapWriteBufferWithCompressionMethod(std::make_unique<WriteBufferFromHDFS>(uri, context->getGlobalContext()->getConfigRef(), context->getSettingsRef().hdfs_replication), compression_method, 3);
        writer = FormatFactory::instance().getOutputFormatParallelIfPossible(format, *write_buf, sample_block, context);
    }
    String getName() const override { return "HDFSSink"; }