**Describe what's wrong**

We had an issue with `INSERT`s into ClickHouse throwing an error:

```
2021-11-29 18:26:13.177 ESTError message from worker: ru.yandex.clickhouse.except.ClickHouseUnknownException: ClickHouse exception, code: 1002, host: <clickhouse_host>, port: 8123; Code: 252. DB::Exception: Too many parts (301). Merges are processing significantly slower than inserts: while pushing to view default.metrics_shard_stat_10m_view (aa99702d-bd2c-4aca-aa99-702dbd2c6aca): while pushing to view default.metrics_shard_stat_1m_view (c2e4400c-c18e-46ae-82e4-400cc18ef6ae). (TOO_MANY_PARTS) (version 21.11.3.6 (official build))
```

Upon investigation it appeared that there were a large amount of unmerged parts for the `metrics_shard_stat_10m` table. This seems to be due to this error thrown during a `deserialize` of an `AggregateFunction` `quantilesTDigest()`. This was also causing the node to have a very high CPU usage.

**Does it reproduce on recent release?**

We're using ClickHouse server version ` 21.11.3.6`

**Table Schema**

`metrics_shard_stat_1m`:
```
CREATE TABLE default.metrics_shard_stat_1m ON CLUSTER testcluster
(
    metric_id                  UUID,
    date                       Date DEFAULT toDate(bucket) CODEC (DoubleDelta),
    bucket                     DateTime CODEC (DoubleDelta, LZ4),
    s_avg                      AggregateFunction(avg, Float64),
    s_count                    AggregateFunction(count, Float64),
    sum                        SimpleAggregateFunction(sum, Float64) CODEC (Gorilla),
    sum2                       SimpleAggregateFunction(sum, Float64) CODEC (Gorilla),
    min                        SimpleAggregateFunction(min, Float64) CODEC (Gorilla),
    max                        SimpleAggregateFunction(max, Float64) CODEC (Gorilla),
    first                      SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    last                       SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    s_linear_regression        AggregateFunction(simpleLinearRegression, UInt32, Float64),
    s_variance                 AggregateFunction(varSamp, Float64),
    s_quantiles                AggregateFunction(quantilesTDigest(0.10, 0.25, 0.50, 0.75, 0.90, 0.95, 0.99),
                                   Float64),
    s_avg_nozero               AggregateFunction(avg, Float64),
    s_count_nozero             AggregateFunction(count, Float64),
    sum_nozero                 SimpleAggregateFunction(sum, Float64) CODEC (Gorilla),
    min_nozero                 SimpleAggregateFunction(min, Float64) CODEC (Gorilla),
    max_nozero                 SimpleAggregateFunction(max, Float64) CODEC (Gorilla),
    first_nozero               SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    last_nozero                SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    s_linear_regression_nozero AggregateFunction(simpleLinearRegression, UInt32, Float64),
    s_variance_nozero          AggregateFunction(varSamp, Float64),
    s_quantiles_nozero         AggregateFunction(quantilesTDigest(0.10, 0.25, 0.50, 0.75, 0.90, 0.95, 0.99),
                                   Float64)
) ENGINE = ReplicatedAggregatingMergeTree() PARTITION BY toYYYYMM(bucket)
      ORDER BY (metric_id, date, bucket) SETTINGS index_granularity = 64;

CREATE MATERIALIZED VIEW default.metrics_shard_stat_1m_view ON CLUSTER testcluster TO default.metrics_shard_stat_1m AS
SELECT metric_id,
       date,
       toStartOfMinute(toDateTime64(timestamp, 3))                                         as bucket,
       avgState(value)                                                                     as s_avg,
       countState(value)                                                                   as s_count,
       sum(value)                                                                          as sum,
       sum(value * value)                                                                  as sum2,
       min(value)                                                                          as min,
       max(value)                                                                          as max,
       any(value)                                                                          as first,
       anyLast(value)                                                                      as last,
       quantilesTDigestState(0.10, 0.25, 0.50, 0.75, 0.90, 0.95, 0.99)(value)              as s_quantiles,
       simpleLinearRegressionState(toUnixTimestamp(timestamp), value)                      as s_linear_regression,
       varSampState(value)                                                                 as s_variance,
       avgStateIf(value, value > 0)                                                        as s_avg_nozero,
       countStateIf(value, value > 0)                                                      as s_count_nozero,
       sumIf(value, value > 0)                                                             as sum_nozero,
       minIf(value, value > 0)                                                             as min_nozero,
       maxIf(value, value > 0)                                                             as max_nozero,
       anyIf(value, value > 0)                                                             as first_nozero,
       anyLastIf(value, value > 0)                                                         as last_nozero,
       quantilesTDigestStateIf(0.10, 0.25, 0.50, 0.75, 0.90, 0.95, 0.99)(value, value > 0) as s_quantiles_nozero,
       simpleLinearRegressionStateIf(toUnixTimestamp(timestamp), value, value >
                                                                        0)                 as s_linear_regression_nozero,
       varSampStateIf(value, value > 0)                                                    as s_variance_nozero
FROM default.metrics_shard
GROUP BY metric_id,
         date,
         bucket;

CREATE TABLE default.metrics_stat_1m ON CLUSTER testcluster
(
    metric_id                  UUID,
    date                       Date DEFAULT toDate(bucket) CODEC (DoubleDelta),
    bucket                     DateTime CODEC (DoubleDelta, LZ4),
    s_avg                      AggregateFunction(avg, Float64),
    s_count                    AggregateFunction(count, Float64),
    sum                        SimpleAggregateFunction(sum, Float64) CODEC (Gorilla),
    sum2                       SimpleAggregateFunction(sum, Float64) CODEC (Gorilla),
    min                        SimpleAggregateFunction(min, Float64) CODEC (Gorilla),
    max                        SimpleAggregateFunction(max, Float64) CODEC (Gorilla),
    first                      SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    last                       SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    s_quantiles                AggregateFunction(quantilesTDigest(0.10, 0.25, 0.50, 0.75, 0.90, 0.95, 0.99),
                                   Float64),
    s_linear_regression        AggregateFunction(simpleLinearRegression, UInt32, Float64),
    s_variance                 AggregateFunction(varSamp, Float64),
    s_avg_nozero               AggregateFunction(avg, Float64),
    s_count_nozero             AggregateFunction(count, Float64),
    sum_nozero                 SimpleAggregateFunction(sum, Float64) CODEC (Gorilla),
    min_nozero                 SimpleAggregateFunction(min, Float64) CODEC (Gorilla),
    max_nozero                 SimpleAggregateFunction(max, Float64) CODEC (Gorilla),
    first_nozero               SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    last_nozero                SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    s_quantiles_nozero         AggregateFunction(quantilesTDigest(0.10, 0.25, 0.50, 0.75, 0.90, 0.95, 0.99),
                                   Float64),
    s_linear_regression_nozero AggregateFunction(simpleLinearRegression, UInt32, Float64),
    s_variance_nozero          AggregateFunction(varSamp, Float64)
) ENGINE = Distributed(
           testcluster,
           default,
           metrics_shard_stat_1m,
           farmFingerprint64(metric_id)
    );
```

`metrics_shard_stat_10m`:
```
CREATE TABLE default.metrics_shard_stat_10m ON cluster testcluster
(
    metric_id                  UUID,
    date                       Date DEFAULT toDate(bucket) CODEC (DoubleDelta),
    bucket                     DateTime CODEC (DoubleDelta, LZ4),
    s_avg                      AggregateFunction(avg, Float64),
    s_count                    AggregateFunction(count, Float64),
    sum                        SimpleAggregateFunction(sum, Float64) CODEC (Gorilla),
    sum2                       SimpleAggregateFunction(sum, Float64) CODEC (Gorilla),
    min                        SimpleAggregateFunction(min, Float64) CODEC (Gorilla),
    max                        SimpleAggregateFunction(max, Float64) CODEC (Gorilla),
    first                      SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    last                       SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    s_linear_regression        AggregateFunction(simpleLinearRegression, UInt32, Float64),
    s_variance                 AggregateFunction(varSamp, Float64),
    s_quantiles                AggregateFunction(quantilesTDigest(0.10, 0.25, 0.50, 0.75, 0.90, 0.95, 0.99),
                                   Float64),
    s_avg_nozero               AggregateFunction(avg, Float64),
    s_count_nozero             AggregateFunction(count, Float64),
    sum_nozero                 SimpleAggregateFunction(sum, Float64) CODEC (Gorilla),
    min_nozero                 SimpleAggregateFunction(min, Float64) CODEC (Gorilla),
    max_nozero                 SimpleAggregateFunction(max, Float64) CODEC (Gorilla),
    first_nozero               SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    last_nozero                SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    s_linear_regression_nozero AggregateFunction(simpleLinearRegression, UInt32, Float64),
    s_variance_nozero          AggregateFunction(varSamp, Float64),
    s_quantiles_nozero         AggregateFunction(quantilesTDigest(0.10, 0.25, 0.50, 0.75, 0.90, 0.95, 0.99),
                                   Float64)
) ENGINE = ReplicatedAggregatingMergeTree() PARTITION BY toYYYYMM(bucket)
      ORDER BY (metric_id, date, bucket) SETTINGS index_granularity = 64;

CREATE MATERIALIZED VIEW default.metrics_shard_stat_10m_view ON CLUSTER testcluster TO default.metrics_shard_stat_10m AS
SELECT metric_id,
       date,
       toStartOfTenMinutes(toDateTime64(bucket, 3))                                             as bucket,
       avgMergeState(s_avg)                                                                     as s_avg,
       countMergeState(s_count)                                                                 as s_count,
       sum(sum)                                                                                 as sum,
       sum(sum2)                                                                                as sum2,
       min(min)                                                                                 as min,
       max(max)                                                                                 as max,
       any(first)                                                                               as first,
       anyLast(last)                                                                            as last,
       quantilesTDigestMergeState(0.10, 0.25, 0.50, 0.75, 0.90, 0.95, 0.99)(s_quantiles)        as s_quantiles,
       simpleLinearRegressionMergeState(s_linear_regression)                                    as s_linear_regression,
       varSampMergeState(s_variance)                                                            as s_variance,
       avgMergeState(s_avg_nozero)                                                              as s_avg_nozero,
       countMergeState(s_count_nozero)                                                          as s_count_nozero,
       sum(sum_nozero)                                                                          as sum_nozero,
       min(min_nozero)                                                                          as min_nozero,
       max(max_nozero)                                                                          as max_nozero,
       any(first_nozero)                                                                        as first_nozero,
       anyLast(last_nozero)                                                                     as last_nozero,
       quantilesTDigestMergeState(0.10, 0.25, 0.50, 0.75, 0.90, 0.95, 0.99)(s_quantiles_nozero) as s_quantiles_nozero,
       simpleLinearRegressionMergeState(s_linear_regression_nozero)                             as s_linear_regression_nozero,
       varSampMergeState(s_variance_nozero)                                                     as s_variance_nozero
FROM default.metrics_shard_stat_1m
GROUP BY metric_id,
         date,
         bucket;

CREATE TABLE default.metrics_stat_10m ON CLUSTER testcluster
(
    metric_id                  UUID,
    date                       Date DEFAULT toDate(bucket) CODEC (DoubleDelta),
    bucket                     DateTime CODEC (DoubleDelta, LZ4),
    s_avg                      AggregateFunction(avg, Float64),
    s_count                    AggregateFunction(count, Float64),
    sum                        SimpleAggregateFunction(sum, Float64) CODEC (Gorilla),
    sum2                       SimpleAggregateFunction(sum, Float64) CODEC (Gorilla),
    min                        SimpleAggregateFunction(min, Float64) CODEC (Gorilla),
    max                        SimpleAggregateFunction(max, Float64) CODEC (Gorilla),
    first                      SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    last                       SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    s_quantiles                AggregateFunction(quantilesTDigest(0.10, 0.25, 0.50, 0.75, 0.90, 0.95, 0.99),
                                   Float64),
    s_linear_regression        AggregateFunction(simpleLinearRegression, UInt32, Float64),
    s_variance                 AggregateFunction(varSamp, Float64),
    s_avg_nozero               AggregateFunction(avg, Float64),
    s_count_nozero             AggregateFunction(count, Float64),
    sum_nozero                 SimpleAggregateFunction(sum, Float64) CODEC (Gorilla),
    min_nozero                 SimpleAggregateFunction(min, Float64) CODEC (Gorilla),
    max_nozero                 SimpleAggregateFunction(max, Float64) CODEC (Gorilla),
    first_nozero               SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    last_nozero                SimpleAggregateFunction(any, Float64) CODEC (Gorilla),
    s_quantiles_nozero         AggregateFunction(quantilesTDigest(0.10, 0.25, 0.50, 0.75, 0.90, 0.95, 0.99),
                                   Float64),
    s_linear_regression_nozero AggregateFunction(simpleLinearRegression, UInt32, Float64),
    s_variance_nozero          AggregateFunction(varSamp, Float64)
) ENGINE = Distributed(
           testcluster,
           default,
           metrics_shard_stat_10m,
           farmFingerprint64(metric_id)
    );
```

**Expected behavior**

`quantilesTDigest()` function should work or throw a meaningful without causing part merges to be blocked forever.

**Error message and/or stacktrace**

```
2021.12.01 03:58:30.375438 [ 62 ] {} <Error> void DB::MergeTreeBackgroundExecutor<DB::MergeMutateRuntimeQueue>::routine(DB::TaskRuntimeDataPtr) [Queue = DB::MergeMutateRuntimeQueue]: Code: 27. DB::Exception: Invalid centroid 2.000000:-nan: (while reading column s_quantiles): (while reading from part /var/lib/clickhouse/store/29e/29ec85cb-e183-4946-a9ec-85cbe1837946/202111_5921759_5923975_485/ from mark 5 with max_rows_to_read = 64): While executing MergeTreeSequentialSource. (CANNOT_PARSE_INPUT_ASSERTION_FAILED), Stack trace (when copying this message, always include the lines below):
0. DB::Exception::Exception(std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, int, bool) @ 0x9b63054 in /usr/bin/clickhouse
1. DB::QuantileTDigest<double>::deserialize(DB::ReadBuffer&) @ 0xa51b433 in /usr/bin/clickhouse
2. DB::SerializationAggregateFunction::deserializeBinaryBulk(DB::IColumn&, DB::ReadBuffer&, unsigned long, double) const @ 0x11cd71fa in /usr/bin/clickhouse
3. DB::ISerialization::deserializeBinaryBulkWithMultipleStreams(COW<DB::IColumn>::immutable_ptr<DB::IColumn>&, unsigned long, DB::ISerialization::DeserializeBinaryBulkSettings&, std::__1::shared_ptr<DB::ISerialization::DeserializeBinaryBulkState>&, std::__1::unordered_map<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> >, COW<DB::IColumn>::immutable_ptr<DB::IColumn>, std::__1::hash<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > >, std::__1::equal_to<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > >, std::__1::allocator<std::__1::pair<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const, COW<DB::IColumn>::immutable_ptr<DB::IColumn> > > >*) const @ 0x11cd4435 in /usr/bin/clickhouse
4. DB::MergeTreeReaderWide::readData(DB::NameAndTypePair const&, COW<DB::IColumn>::immutable_ptr<DB::IColumn>&, unsigned long, bool, unsigned long, unsigned long, std::__1::unordered_map<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> >, COW<DB::IColumn>::immutable_ptr<DB::IColumn>, std::__1::hash<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > >, std::__1::equal_to<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > >, std::__1::allocator<std::__1::pair<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const, COW<DB::IColumn>::immutable_ptr<DB::IColumn> > > >&, bool) @ 0x12e854a2 in /usr/bin/clickhouse
5. DB::MergeTreeReaderWide::readRows(unsigned long, unsigned long, bool, unsigned long, std::__1::vector<COW<DB::IColumn>::immutable_ptr<DB::IColumn>, std::__1::allocator<COW<DB::IColumn>::immutable_ptr<DB::IColumn> > >&) @ 0x12e84279 in /usr/bin/clickhouse
6. DB::MergeTreeSequentialSource::generate() @ 0x12e88377 in /usr/bin/clickhouse
7. DB::ISource::tryGenerate() @ 0x131156b5 in /usr/bin/clickhouse
8. DB::ISource::work() @ 0x1311527a in /usr/bin/clickhouse
9. DB::SourceWithProgress::work() @ 0x13321062 in /usr/bin/clickhouse
10. ? @ 0x13130b1b in /usr/bin/clickhouse
11. DB::PipelineExecutor::executeStepImpl(unsigned long, unsigned long, std::__1::atomic<bool>*) @ 0x1312cad1 in /usr/bin/clickhouse
12. DB::PipelineExecutor::executeStep(std::__1::atomic<bool>*) @ 0x1312b2e5 in /usr/bin/clickhouse
13. DB::PullingPipelineExecutor::pull(DB::Chunk&) @ 0x13139d4b in /usr/bin/clickhouse
14. DB::PullingPipelineExecutor::pull(DB::Block&) @ 0x13139fec in /usr/bin/clickhouse
15. DB::MergeTask::ExecuteAndFinalizeHorizontalPart::executeImpl() @ 0x12d3d52b in /usr/bin/clickhouse
16. DB::MergeTask::ExecuteAndFinalizeHorizontalPart::execute() @ 0x12d3d48b in /usr/bin/clickhouse
17. DB::MergeTask::execute() @ 0x12d4217a in /usr/bin/clickhouse
18. DB::MergePlainMergeTreeTask::executeStep() @ 0x12fbceec in /usr/bin/clickhouse
19. DB::MergeTreeBackgroundExecutor<DB::MergeMutateRuntimeQueue>::routine(std::__1::shared_ptr<DB::TaskRuntimeData>) @ 0x12d50bdd in /usr/bin/clickhouse
20. DB::MergeTreeBackgroundExecutor<DB::MergeMutateRuntimeQueue>::threadFunction() @ 0x12d5167a in /usr/bin/clickhouse
21. ThreadPoolImpl<ThreadFromGlobalPool>::worker(std::__1::__list_iterator<ThreadFromGlobalPool, void*>) @ 0x9ba7d0a in /usr/bin/clickhouse
22. ThreadFromGlobalPool::ThreadFromGlobalPool<void ThreadPoolImpl<ThreadFromGlobalPool>::scheduleImpl<void>(std::__1::function<void ()>, int, std::__1::optional<unsigned long>)::'lambda0'()>(void&&, void ThreadPoolImpl<ThreadFromGlobalPool>::scheduleImpl<void>(std::__1::function<void ()>, int, std::__1::optional<unsigned long>)::'lambda0'()&&...)::'lambda'()::operator()() @ 0x9ba9b27 in /usr/bin/clickhouse
23. ThreadPoolImpl<std::__1::thread>::worker(std::__1::__list_iterator<std::__1::thread, void*>) @ 0x9ba5117 in /usr/bin/clickhouse
24. ? @ 0x9ba8b1d in /usr/bin/clickhouse
25. start_thread @ 0x9609 in /usr/lib/x86_64-linux-gnu/libpthread-2.31.so
26. clone @ 0x122293 in /usr/lib/x86_64-linux-gnu/libc-2.31.so
 (version 21.11.3.6 (official build))
 ```
 
 **System Settings**
 ```
 connect_timeout_with_failover_ms,1000,1,Connection timeout for selecting first healthy replica.,,,0,Milliseconds
load_balancing,random,1,Which replicas (among healthy replicas) to preferably send a query to (on the first attempt) for distributed processing.,,,0,LoadBalancing
distributed_aggregation_memory_efficient,1,1,Is the memory-saving mode of distributed aggregation enabled.,,,0,Bool
log_queries,1,1,Log requests and write the log to the system table.,,,0,Bool
max_memory_usage,10000000000,1,Maximum memory usage for processing of single query. Zero means unlimited.,,,0,UInt64
parallel_view_processing,1,1,Enables pushing to attached views concurrently instead of sequentially.,,,0,Bool
default_database_engine,Ordinary,1,Default database engine.,,,0,DefaultDatabaseEngine
```

if (l != r)
                        *l = *r;
                    l_mean = l->mean;
                    l_count = l->count;
                }
                ++r;
            }
            count = sum + l_count; 
            
            centroids.resize(l - centroids.begin() + 1);
            unmerged = 0;
        }
        
        compressBrute();
    }
    /** Adds to the digest a change in `x` with a weight of `cnt` (default 1)
      */
    void add(T x, UInt64 cnt = 1)
    {
        auto vx = static_cast<Value>(x);
        if (cnt == 0 || std::isnan(vx))
            return; 
        addCentroid(Centroid{vx, static_cast<Count>(cnt)});
    }
    void merge(const QuantileTDigest & other)
    {
        for (const auto & c : other.centroids)
            addCentroid(c);
    }
    void serialize(WriteBuffer & buf)