ClickHouse / ClickHouse/ClickHouse
Query streaming proposal
- Dominant language
- C++
- Stars
- 49.9k
- Forks
- 9k
- Avg merge
- 21h 32m
- Merged PRs (30d)
- 515
Description
### Company or project name
_No response_
### Use case
I'm dreaming about streaming queries in ClickHouse. I've seen a few issues and pull requests (such as #42990 and #63312), but they focus more on persistent cursors.
Making the cursor the first part of the MergeTree ordering key kind of kills the index, as all the other parts of the ordering key are no longer searchable.
Instead, I would be happy with session-bound streaming, so that I can make a query and then keep the results up to date.
### Describe the solution you'd like
Implementation of session-bound streaming should be significantly simpler too, and I guess it can be implemented by heavily reusing existing components with moderate changes. The way I see it working under the hood is:
1. Create a temporary `Null` or `Memory` table with the schema of the query result.
2. Create a temporary materialized view that inserts data into that temporary table.
3. Populate the materialized view from the whole table and seamlessly process new blocks.
4. On writes to that temporary table, data is streamed to the client.
The simplest example is getting some last events and subscribing to new ones.
```sql
SELECT * FROM events STREAM WHERE timestamp >= now() - toIntervalSeconds(60)
```
Updates for `GROUP BY` or window functions won't be complete, but that can be worked out on the client. This is still more convenient and requires less computation than re-trying the full query.
```sql
SELECT day, domain, sum(hits) AS total FROM visits STREAM GROUP BY day, domain
```
Updates can be pretty trivially processed on the client.
Even queries that can not be easily maintained on the client can benefit from this approach. The client can still use streaming as a notification channel and rerun the query when the source data has changed.
```sql
SELECT domain, sum(hits) AS total FROM visits STREAM GROUP BY ALL ORDER BY total DESC LIMIT 11
```
Without maintaining counters for all domains, we can't be sure that a small domain won't slowly get ahead of the top 10. However, we can calculate when we should rerun the query.
If we receive updates for the top 10, we update them. All other updates we add to the 11th row and rerun the query if it reaches the 10th row.
Not sure how to get the last N events and all the updates. We need to apply a limit to the initial query but not to the updates, which is less of a challenge. Maybe we can make it similar to recursive CTEs; if the query is a `UNION ALL`, the first part is applied to the initial data and the second to updates.
And if these "under the hood" operations can be manually done one by one, that would be a complete banger. We create a temporary queue table, populate it with what we want, create one or more temporary materialized views (including cascading and possibly `-State` columns), and then block on reading the queue table indefinitely.
As I understand, parts have a version number, and if that can be used to seamlessly handle both the initial query and future updates, that would be perfect.
The same logic can be run on each replica in a cluster, and the results can be merged in the same way as for usual queries and updates in the stream.
How does it sound?
### Describe alternatives you've considered
I think proton is losing users not keeping protocol backward compatible as they have some interesting features and the streaming is the main of them.
### Additional context
If this proposal aligns with the project's plans and priorities, I would be happy to try to implement it all or some parts of it.
Contributor guide
Assessment
This issue has not been assessed yet.