More parallel execution for queries with FINAL

[nickitat]

Changelog category (leave one):

• Performance Improvement

Changelog entry (a user-readable short description of the changes that goes to CHANGELOG.md):

Now we split data parts into layers and distribute them among threads instead of whole parts to make the execution of queries with FINAL more data-parallel.

an example of query pipeline

(Expression)
ExpressionTransform × 8
  (SettingQuotaAndLimits)
    (ReadFromMergeTree)
    ExpressionTransform × 8
      ReplacingSorted 4 → 1
        ExpressionTransform × 4
          FilterTransform × 4
          Description: filter values in [(4382720), +inf)
            MergeTreeInOrder × 4 0 → 1
              ReplacingSorted 4 → 1
                ExpressionTransform × 4
                  FilterTransform × 4
                  Description: filter values in [(3751936), (4382720))
                    MergeTreeInOrder × 4 0 → 1
                      ReplacingSorted 4 → 1
                        ExpressionTransform × 4
                          FilterTransform × 4
                          Description: filter values in [(3129344), (3751936))
                            MergeTreeInOrder × 4 0 → 1
                              ReplacingSorted 4 → 1
                                ExpressionTransform × 4
                                  FilterTransform × 4
                                  Description: filter values in [(2498560), (3129344))
                                    MergeTreeInOrder × 4 0 → 1
                                      ReplacingSorted 4 → 1
                                        ExpressionTransform × 4
                                          FilterTransform × 4
                                          Description: filter values in [(1875968), (2498560))
                                            MergeTreeInOrder × 4 0 → 1
                                              ReplacingSorted 4 → 1
                                                ExpressionTransform × 4
                                                  FilterTransform × 4
                                                  Description: filter values in [(1245184), (1875968))
                                                    MergeTreeInOrder × 4 0 → 1
                                                      ReplacingSorted 4 → 1
                                                        ExpressionTransform × 4
                                                          FilterTransform × 4
                                                          Description: filter values in [(622592), (1245184))
                                                            MergeTreeInOrder × 4 0 → 1
                                                              ReplacingSorted 4 → 1
                                                                ExpressionTransform × 4
                                                                  FilterTransform × 4
                                                                  Description: filter values in [-inf, (622592))
                                                                    MergeTreeInOrder × 4 0 → 1

perf tests

[nickitat]

@Mergifyio update

[mergify]

update

✅ Branch has been successfully updated

[KochetovNicolai]

Generally, it's good to write a comment to

• every class (both interface and implementation)
• all virtual methods
• and, actually, every method which purpose is not obvious from the first glance

[KochetovNicolai]

Looks like this iface has single implementation.
Do we actually need it? I think that, generally, you should not create an interfece if you don't plan to have > 1 implementations...

[KochetovNicolai]

Inheritance from std containers is suspicious. Maybe, composition?

[KochetovNicolai]

ok, but I would prefer for loop :)

[UnamedRus]

Hm, can we use it in order to lower amount of memory usage during GROUP BY column_from_order_by.

If each thread will have it's own range of column_from_order_by values, it does mean, that threads will not have all possible values in their own hash tables.

For example optimize_aggregation_in_order doesn't do such parallelization as it could be.

[KochetovNicolai]

Not clear - why this is needed?

[KochetovNicolai]

Maybe we can also add RangeEnd event to remove some ranges from event_queue

[jorisgio]

We have been running this code for 2 weeks, and from my perspective, it is very promising and delivering huge performance gain 🎉 . Here are some numbers (unfortunately i did not get any number before patch to compare, but i can tell you it was an order of magnitude slower).

311.49 billion rows, 13.44 TB (33.43 billion rows/s., 1.44 TB/s.)

And it seems to scale really well.

Mostly the two pain point i have are not directly related to this pr :

1. number of ranges directly depends on max_final_thread, it is like max_threads for normal queries. But for distributed queries, would love to have a way to tweak threads per shard based on data size, some setting like max_range_size so that it ajusts to data to process
2. Minor and not directly related it seems it cannot stack final with optimze_aggregation_in_order, and keep using the ranges in upper layer (but this can somehow be worked around with smarter queries)

[nickitat]

We have been running this code for 2 weeks, and from my perspective, it is very promising and delivering huge performance gain tada . Here are some numbers (unfortunately i did not get any number before patch to compare, but i can tell you it was an order of magnitude slower).

311.49 billion rows, 13.44 TB (33.43 billion rows/s., 1.44 TB/s.)

And it seems to scale really well.

Mostly the two pain point i have are not directly related to this pr :

1. number of ranges directly depends on max_final_thread, it is like max_threads for normal queries. But for distributed queries, would love to have a way to tweak threads per shard based on data size, some setting like max_range_size so that it ajusts to data to process
2. Minor and not directly related it seems it cannot stack final with optimze_aggregation_in_order, and keep using the ranges in upper layer (but this can somehow be worked around with smarter queries)

thank you for the feedback!
regarding your points:

1. each node will do the splitting on its own (in the best way the current implementation could provide), so in this sense, each node will adjust to the data it is looking into. what do you want to achieve by asking a node to split the data into more ranges than it has CPU cores?
2. we definitely want to integrate this splitting functionality with aggregation in order. without cross-thread merging, AIO should perform significantly better

[UnamedRus]

And it can make sense to integrate with parallel replicas processing: #26748

[jorisgio]

each node will do the splitting on its own (in the best way the current implementation could provide), so in this sense, each node will adjust to the data it is looking into. what do you want to achieve by asking a node to split the data into more ranges than it has CPU cores?

if you have a server with high core count, but many load. It makes sense to run with max_final_threads=16 to process many requests in parallel. But if there is some imbalance in data distribution, one server might have twice more data, and this could benefit of having max_final_threads=32 only there. So having a setting saying 'max_ranges_size' instead would achieve that, with max_final_threads as an upper bound ?

we definitely want to integrate this splitting functionality with aggregation in order. without cross-thread merging, AIO should perform significantly better

That would be great 💯 my question specifically though is do you plan to make it stack ? For cases like :

SELECT * FROM table FINAL WHERE somefilters LIMIT 1 BY some_prefix_of_pk

a good usecase to get first/last entry in time log data
Or stuff like

SELECT count(events) FROM table FINAL WHERE somefilters GROUP BY some_prefix_of_pk

In those case it makes a lot of sense to have GROUP BY operating on the same ranges of final ? though that is not totally trivial actually, because the ranges are computed for full primary key, not the prefix so some aggregation are not local, without using the prefix only to compute ranges.

[nickitat]

if you have a server with high core count, but many load. It makes sense to run with max_final_threads=16 to process many requests in parallel. But if there is some imbalance in data distribution, one server might have twice more data, and this could benefit of having max_final_threads=32 only there. So having a setting saying 'max_ranges_size' instead would achieve that, with max_final_threads as an upper bound ?

final won't run faster if you set max_final_threads to a value higher than amount of cores, because each final thread utilizes it's core (assuming data distributed evenly) and in practice it works as if it won't let any other thread to execute smth until it finishes. memory consumption also won't decrease, if you increase the number of final threads.

In those case it makes a lot of sense to have GROUP BY operating on the same ranges of final

yep, AIO should be able to use splitting regardless of presence of final in query

[nickitat]

@rschu1ze Robert, thank you for the review. I will make changes in a separate pr, to let this one to lend into the release.

[tavplubix]

@nickitat, you forgot to take a look at failed tests before merging