FOSS data stack to perform complex custom logic on billions of ordered rows

Question

Is there a purely analytics-focused (i.e. a handful of technial users, no business users or clients) FOSS data stack that would enable me to

1. ingest several hundred million up to a billion of rows per day (with all primitive column types),
2. keep the overall size reasonable (similar to e.g. parquet files or clickhouse),
3. perform typical SQL filters and aggregations, and
4. (this is the critical point) run ad-hoc "queries" to read the data in order and perform complex custom logic on at least one day of data at once; preferably parallelized on some partition column.

A typical use case for point 4: There are prices for several symbols from several sources. Most rows have a reference key, meaning that they are an update for (and hence replace) the price of a previous row. At a given time a symbol has a price from many sources, often including multiple prices from a single source.

┌──time─┬─key─┬─referenceKey─┬─symbol─┬─price─┬─source─┐
│     2 │   1 │              │      A |   3.5 |      X |
│     4 │   2 │              │      B |   1.3 |      X |
│     5 │   3 │              │      A |   3.6 |      Y |
│     7 │   4 │            1 │      A |   3.7 |      X |
│     9 │   6 │            3 │      A |   3.8 |      Y |
│    11 │   5 │            2 │      B |   1.4 |      X |
└───────┴─────┴──────────────┴────────┴───────┴────────┘

The challenge is to capture the best (lowest) price per symbol whenever it changes, i.e.

┌──time─┬─symbol─┬─lowestPrice─┬─source─┐
│     2 │      A │         3.5 │      X |
│     4 │      B │         1.3 │      X |
│     7 │      A │         3.6 │      Y |
│     9 │      A │         3.7 │      X |
│    11 │      B │         1.4 │      X |
└───────┴────────┴─────────────┴────────┘

In this example, each symbol could be treated as it's own separate partition, since the aggregation looks at each symbol individually - which should allow for concurrent computation.

What I have tried so far:

1. Spark (with parquet data). While Spark UDFs are not intended (and from my experience not suitable) to be used in this way, there are often combinations of Spark functions and array trickery that one can do, for example in the case above. The problem is that a) these queries are long, complex and hardly maintainable, b) they take a lot of time and resources to run and c) they often don't exist for more challenging use cases.
2. Clickhouse + Flink (via JDBC, i.e. without Kafka). This is surprisingly fast, but has downsides: a) parallel partition processing: The proper sorting key for Clickhouse in the above example would be symbol, time. But in order to allow for parallel partition processing within Flink, the data should be sorted just by time, so all symbols come in at once, and not one after another. b) The DataStream API requires a lot of redundant schema definitions, in my experience has been rather error-prone and might not be suitable for a use case where two ordered batches of data should be processed on a common timestamp (I believe that watermarks solve this problem only for real-time streams).

A file format that works for sequential row-by-row processing and that can be persisted in sorted order by Spark might do the trick, but I'm not aware of any such format.