Introduced MergeTask and MutateTask

src/AggregateFunctions/ya.make

@@ -49,6 +49,7 @@ SRCS(
     AggregateFunctionSimpleLinearRegression.cpp
     AggregateFunctionSimpleState.cpp
     AggregateFunctionSingleValueOrNull.cpp
+    AggregateFunctionSparkbar.cpp
     AggregateFunctionState.cpp
     AggregateFunctionStatistics.cpp
     AggregateFunctionStatisticsSimple.cpp

src/Common/MemoryTracker.h

@@ -162,18 +162,18 @@ class MemoryTracker
     struct BlockerInThread
     {
     private:
-        BlockerInThread(const BlockerInThread &) = delete;
-        BlockerInThread & operator=(const BlockerInThread &) = delete;
-
         static thread_local uint64_t counter;
         static thread_local VariableContext level;
 
         VariableContext previous_level;
     public:
         /// level_ - block in level and above
-        BlockerInThread(VariableContext level_ = VariableContext::User);
+        explicit BlockerInThread(VariableContext level_ = VariableContext::User);
         ~BlockerInThread();
 
+        BlockerInThread(const BlockerInThread &) = delete;
+        BlockerInThread & operator=(const BlockerInThread &) = delete;
+
         static bool isBlocked(VariableContext current_level)
         {
             return counter > 0 && current_level >= level;
@@ -195,9 +195,6 @@ class MemoryTracker
     struct LockExceptionInThread
     {
     private:
-        LockExceptionInThread(const LockExceptionInThread &) = delete;
-        LockExceptionInThread & operator=(const LockExceptionInThread &) = delete;
-
         static thread_local uint64_t counter;
         static thread_local VariableContext level;
         static thread_local bool block_fault_injections;
@@ -207,9 +204,12 @@ class MemoryTracker
     public:
         /// level_ - block in level and above
         /// block_fault_injections_ - block in fault injection too
-        LockExceptionInThread(VariableContext level_ = VariableContext::User, bool block_fault_injections_ = true);
+        explicit LockExceptionInThread(VariableContext level_ = VariableContext::User, bool block_fault_injections_ = true);
         ~LockExceptionInThread();
 
+        LockExceptionInThread(const LockExceptionInThread &) = delete;
+        LockExceptionInThread & operator=(const LockExceptionInThread &) = delete;
+
         static bool isBlocked(VariableContext current_level, bool fault_injection)
         {
             return counter > 0 && current_level >= level && (!fault_injection || block_fault_injections);

src/Common/Stopwatch.h

@@ -4,6 +4,7 @@
 #include <common/types.h>
 
 #include <atomic>
+#include <memory>
 
 
 inline UInt64 clock_gettime_ns(clockid_t clock_type = CLOCK_MONOTONIC)
@@ -22,7 +23,7 @@ class Stopwatch
     /** CLOCK_MONOTONIC works relatively efficient (~15 million calls/sec) and doesn't lead to syscall.
       * Pass CLOCK_MONOTONIC_COARSE, if you need better performance with acceptable cost of several milliseconds of inaccuracy.
       */
-    Stopwatch(clockid_t clock_type_ = CLOCK_MONOTONIC) : clock_type(clock_type_) { start(); }
+    explicit Stopwatch(clockid_t clock_type_ = CLOCK_MONOTONIC) : clock_type(clock_type_) { start(); }
 
     void start()                       { start_ns = nanoseconds(); is_running = true; }
     void stop()                        { stop_ns = nanoseconds(); is_running = false; }
@@ -43,11 +44,13 @@ class Stopwatch
     UInt64 nanoseconds() const { return clock_gettime_ns(clock_type); }
 };
 
+using StopwatchUniquePtr = std::unique_ptr<Stopwatch>;
+
 
 class AtomicStopwatch
 {
 public:
-    AtomicStopwatch(clockid_t clock_type_ = CLOCK_MONOTONIC) : clock_type(clock_type_) { restart(); }
+    explicit AtomicStopwatch(clockid_t clock_type_ = CLOCK_MONOTONIC) : clock_type(clock_type_) { restart(); }
 
     void restart()                     { start_ns = nanoseconds(); }
     UInt64 elapsed() const             { return nanoseconds() - start_ns; }
@@ -78,11 +81,11 @@ class AtomicStopwatch
     {
         AtomicStopwatch * parent = nullptr;
 
-        Lock() {}
+        Lock() = default;
 
-        operator bool() const { return parent != nullptr; }
+        explicit operator bool() const { return parent != nullptr; }
 
-        Lock(AtomicStopwatch * parent_) : parent(parent_) {}
+        explicit Lock(AtomicStopwatch * parent_) : parent(parent_) {}
 
         Lock(Lock &&) = default;
 

src/Common/ThreadPool.cpp

@@ -3,6 +3,7 @@
 #include <Common/getNumberOfPhysicalCPUCores.h>
 
 #include <cassert>
+#include <iostream>
 #include <type_traits>
 
 #include <Poco/Util/Application.h>

src/Common/ya.make

@@ -43,7 +43,6 @@ SRCS(
     Epoll.cpp
     ErrorCodes.cpp
     Exception.cpp
-    ExternalLoaderStatus.cpp
     FieldVisitorDump.cpp
     FieldVisitorHash.cpp
     FieldVisitorSum.cpp

src/Compression/ya.make

@@ -35,11 +35,6 @@ SRCS(
     CompressionFactoryAdditions.cpp
     ICompressionCodec.cpp
     LZ4_decompress_faster.cpp
-    fuzzers/compressed_buffer_fuzzer.cpp
-    fuzzers/delta_decompress_fuzzer.cpp
-    fuzzers/double_delta_decompress_fuzzer.cpp
-    fuzzers/encrypted_decompress_fuzzer.cpp
-    fuzzers/lz4_decompress_fuzzer.cpp
     getCompressionCodecForFile.cpp
 
 )

src/Disks/IStoragePolicy.h

@@ -17,6 +17,7 @@ class IDisk;
 using DiskPtr = std::shared_ptr<IDisk>;
 using Disks = std::vector<DiskPtr>;
 class IReservation;
+using ReservationSharedPtr = std::shared_ptr<IReservation>;
 using ReservationPtr = std::unique_ptr<IReservation>;
 using Reservations = std::vector<ReservationPtr>;
 

src/IO/ya.make

@@ -61,7 +61,6 @@ SRCS(
     ReadBufferFromMemory.cpp
     ReadBufferFromPocoSocket.cpp
     ReadHelpers.cpp
-    ReadSettings.cpp
     SeekAvoidingReadBuffer.cpp
     SynchronousReader.cpp
     ThreadPoolReader.cpp

src/Interpreters/Context.h

@@ -282,6 +282,7 @@ class Context: public std::enable_shared_from_this<Context>
     /// A flag, used to distinguish between user query and internal query to a database engine (MaterializePostgreSQL).
     bool is_internal_query = false;
 
+
 public:
     // Top-level OpenTelemetry trace context for the query. Makes sense only for a query context.
     OpenTelemetryTraceContext query_trace_context;

src/Interpreters/ya.make

@@ -48,6 +48,7 @@ SRCS(
     DatabaseAndTableWithAlias.cpp
     DatabaseCatalog.cpp
     DictionaryReader.cpp
+    DuplicateOrderByVisitor.cpp
     EmbeddedDictionaries.cpp
     ExecuteScalarSubqueriesVisitor.cpp
     ExpressionActions.cpp

src/Storages/MergeTree/BackgroundProcessList.h

@@ -26,6 +26,8 @@ class BackgroundProcessListEntry
     BackgroundProcessListEntry(const BackgroundProcessListEntry &) = delete;
     BackgroundProcessListEntry & operator=(const BackgroundProcessListEntry &) = delete;
 
+    BackgroundProcessListEntry(BackgroundProcessListEntry &&) = default;
+
     BackgroundProcessListEntry(BackgroundProcessList<ListElement, Info> & list_, const typename container_t::iterator it_, const CurrentMetrics::Metric & metric)
         : list(list_), it{it_}, metric_increment{metric}
     {
@@ -58,7 +60,7 @@ class BackgroundProcessList
 
     CurrentMetrics::Metric metric;
 
-    BackgroundProcessList(const CurrentMetrics::Metric & metric_)
+    explicit BackgroundProcessList(const CurrentMetrics::Metric & metric_)
         : metric(metric_)
     {}
 public:
@@ -85,7 +87,7 @@ class BackgroundProcessList
 
     virtual void onEntryCreate(const Entry & /* entry */) {}
     virtual void onEntryDestroy(const Entry & /* entry */) {}
-    virtual inline ~BackgroundProcessList() {}
+    virtual inline ~BackgroundProcessList() = default;
 };
 
 }

src/Storages/MergeTree/ColumnSizeEstimator.h

@@ -0,0 +1,49 @@
+#pragma once
+
+#include "Storages/MergeTree/IMergeTreeDataPart.h"
+
+
+namespace DB
+{
+
+/* Allow to compute more accurate progress statistics */
+class ColumnSizeEstimator
+{
+    using ColumnToSize = MergeTreeDataPartInMemory::ColumnToSize;
+    ColumnToSize map;
+public:
+
+    /// Stores approximate size of columns in bytes
+    /// Exact values are not required since it used for relative values estimation (progress).
+    size_t sum_total = 0;
+    size_t sum_index_columns = 0;
+    size_t sum_ordinary_columns = 0;
+
+    ColumnSizeEstimator(ColumnToSize && map_, const Names & key_columns, const Names & ordinary_columns)
+        : map(std::move(map_))
+    {
+        for (const auto & name : key_columns)
+            if (!map.count(name)) map[name] = 0;
+        for (const auto & name : ordinary_columns)
+            if (!map.count(name)) map[name] = 0;
+
+        for (const auto & name : key_columns)
+            sum_index_columns += map.at(name);
+
+        for (const auto & name : ordinary_columns)
+            sum_ordinary_columns += map.at(name);
+
+        sum_total = std::max(static_cast<decltype(sum_index_columns)>(1), sum_index_columns + sum_ordinary_columns);
+    }
+
+    Float64 columnWeight(const String & column) const
+    {
+        return static_cast<Float64>(map.at(column)) / sum_total;
+    }
+
+    Float64 keyColumnsWeight() const
+    {
+        return static_cast<Float64>(sum_index_columns) / sum_total;
+    }
+};
+}

src/Storages/MergeTree/DataPartsExchange.cpp

@@ -580,9 +580,8 @@ MergeTreeData::MutableDataPartPtr Fetcher::downloadPartToMemory(
         new_projection_part->is_temp = false;
         new_projection_part->setColumns(block.getNamesAndTypesList());
         MergeTreePartition partition{};
-        IMergeTreeDataPart::MinMaxIndex minmax_idx{};
         new_projection_part->partition = std::move(partition);
-        new_projection_part->minmax_idx = std::move(minmax_idx);
+        new_projection_part->minmax_idx = std::make_shared<IMergeTreeDataPart::MinMaxIndex>();
 
         MergedBlockOutputStream part_out(
             new_projection_part,
@@ -608,7 +607,7 @@ MergeTreeData::MutableDataPartPtr Fetcher::downloadPartToMemory(
     new_data_part->uuid = part_uuid;
     new_data_part->is_temp = true;
     new_data_part->setColumns(block.getNamesAndTypesList());
-    new_data_part->minmax_idx.update(block, data.getMinMaxColumnsNames(metadata_snapshot->getPartitionKey()));
+    new_data_part->minmax_idx->update(block, data.getMinMaxColumnsNames(metadata_snapshot->getPartitionKey()));
     new_data_part->partition.create(metadata_snapshot, block, 0, context);
 
     MergedBlockOutputStream part_out(

src/Storages/MergeTree/FutureMergedMutatedPart.cpp

@@ -0,0 +1,92 @@
+#include "Storages/MergeTree/FutureMergedMutatedPart.h"
+
+
+namespace DB
+{
+
+namespace ErrorCodes
+{
+    extern const int LOGICAL_ERROR;
+}
+
+void FutureMergedMutatedPart::assign(MergeTreeData::DataPartsVector parts_)
+{
+    if (parts_.empty())
+        return;
+
+    size_t sum_rows = 0;
+    size_t sum_bytes_uncompressed = 0;
+    MergeTreeDataPartType future_part_type = MergeTreeDataPartType::UNKNOWN;
+    for (const auto & part : parts_)
+    {
+        sum_rows += part->rows_count;
+        sum_bytes_uncompressed += part->getTotalColumnsSize().data_uncompressed;
+        future_part_type = std::min(future_part_type, part->getType());
+    }
+
+    auto chosen_type = parts_.front()->storage.choosePartTypeOnDisk(sum_bytes_uncompressed, sum_rows);
+    future_part_type = std::min(future_part_type, chosen_type);
+    assign(std::move(parts_), future_part_type);
+}
+
+void FutureMergedMutatedPart::assign(MergeTreeData::DataPartsVector parts_, MergeTreeDataPartType future_part_type)
+{
+    if (parts_.empty())
+        return;
+
+    for (const MergeTreeData::DataPartPtr & part : parts_)
+    {
+        const MergeTreeData::DataPartPtr & first_part = parts_.front();
+
+        if (part->partition.value != first_part->partition.value)
+            throw Exception(
+                "Attempting to merge parts " + first_part->name + " and " + part->name + " that are in different partitions",
+                ErrorCodes::LOGICAL_ERROR);
+    }
+
+    parts = std::move(parts_);
+
+    UInt32 max_level = 0;
+    Int64 max_mutation = 0;
+    for (const auto & part : parts)
+    {
+        max_level = std::max(max_level, part->info.level);
+        max_mutation = std::max(max_mutation, part->info.mutation);
+    }
+
+    type = future_part_type;
+    part_info.partition_id = parts.front()->info.partition_id;
+    part_info.min_block = parts.front()->info.min_block;
+    part_info.max_block = parts.back()->info.max_block;
+    part_info.level = max_level + 1;
+    part_info.mutation = max_mutation;
+
+    if (parts.front()->storage.format_version < MERGE_TREE_DATA_MIN_FORMAT_VERSION_WITH_CUSTOM_PARTITIONING)
+    {
+        DayNum min_date = DayNum(std::numeric_limits<UInt16>::max());
+        DayNum max_date = DayNum(std::numeric_limits<UInt16>::min());
+        for (const auto & part : parts)
+        {
+            /// NOTE: getting min and max dates from part names (instead of part data) because we want
+            /// the merged part name be determined only by source part names.
+            /// It is simpler this way when the real min and max dates for the block range can change
+            /// (e.g. after an ALTER DELETE command).
+            DayNum part_min_date;
+            DayNum part_max_date;
+            MergeTreePartInfo::parseMinMaxDatesFromPartName(part->name, part_min_date, part_max_date);
+            min_date = std::min(min_date, part_min_date);
+            max_date = std::max(max_date, part_max_date);
+        }
+
+        name = part_info.getPartNameV0(min_date, max_date);
+    }
+    else
+        name = part_info.getPartName();
+}
+
+void FutureMergedMutatedPart::updatePath(const MergeTreeData & storage, const IReservation * reservation)
+{
+    path = storage.getFullPathOnDisk(reservation->getDisk()) + name + "/";
+}
+
+}