【ClickHouse为什么这么快？】MergeTree 表存储引擎图文实例详解

 CREATE TABLE IF NOT EXISTS mergetree_sample_table

 (
 name     String,
 price    UInt64,
 shop_id  UInt64,
 quantity UInt64,
 p_date   DateTime
 )
 ENGINE =MergeTree()
 partition by p_date
         order by (name,shop_id,p_date)
 SETTINGS index_granularity =2;

 INSERT INTO mergetree_sample_table
 VALUES ('Apple',2,1,40,now()) ('Apple',2,3,35,now()) ('Apple',3,2,45,now()) ('Apple',3,4,35,now()) ('Orange',1,2,40,now()) ('Orange',3,3,50,now()) ('Banana',2,1,25,now()) ('Banana',2,2,55,now());
 

 SELECT t.*

       FROM mydb.mergetree_sample_table t
       LIMIT 501
 

 ├── 1638121099_1_1_0

 │   ├── checksums.txt
 │   ├── columns.txt
 │   ├── count.txt
 │   ├── data.bin
 │   ├── data.mrk3
 │   ├── default_compression_codec.txt
 │   ├── minmax_p_date.idx
 │   ├── partition.dat
 │   └── primary.idx
 ├── detached
 └── format_version.txt
 2 directories, 10 files
 

$cat default_compression_codec.txt 

CODEC(LZ4)% 
$cat columns.txt 

columns format version: 1
5 columns:
`name` String
`price` UInt64
`shop_id` UInt64
`quantity` UInt64
`p_date` DateTime
$cat count.txt 

8%         

 CREATE TABLE IF NOT EXISTS replacingmergetree_test
 (
 ID          String,
 Name        String,
 DateOfBirth Date
 )
 ENGINE =ReplacingMergeTree()
 PARTITION BY ID
         ORDER BY (ID,DateOfBirth)
 SETTINGS
             index_granularity =1024;

 INSERT INTO replacingmergetree_test

 VALUES ('a1','Jim','1995-05-01'),
 ('a1','Jim','1995-05-01'),
 ('a1','Jim','1995-05-02'),
 ('a2','Jil','1995-06-01'),
 ('a2','Jil','1995-06-01'),
 ('a2','Jil','1995-06-02');
 

  desc replacingmergetree_test

 DESCRIBE TABLE  replacingmergetree_test
 Query id: 59ab6932-9912-4d86-b97d-7782a8e11f65
 ┌─name────────┬─type───┬─default_type─┬─default_expression─┬─comment─┬─codec_expression─┬─ttl_expression─┐
 │ ID          │ String │              │                    │        │                  │                │
 │ Name        │ String │              │                    │        │                  │                │
 │ DateOfBirth │ Date  │              │                    │        │                  │                │
 └─────────────┴────────┴──────────────┴────────────────────┴─────────┴──────────────────┴────────────────┘
 3 rows in set. Elapsed: 0.001 sec. 
 
 SELECT *

 FROM replacingmergetree_test
 Query id: 082de655-db24-4679-ab10-8d431320fae2
 ┌─ID─┬─Name─┬─DateOfBirth─┐
 │ a2 │ Jil  │  1995-06-01 │
 │ a2 │ Jil  │  1995-06-02 │
 └────┴──────┴─────────────┘
 ┌─ID─┬─Name─┬─DateOfBirth─┐
 │ a1 │ Jim  │  1995-05-01 │
 │ a1 │ Jim  │  1995-05-02 │
 └────┴──────┴─────────────┘
 4 rows in set. Elapsed: 0.003 sec. 
 

OPTIMIZE TABLE replacingmergetree_test FINAL;

 #pragma once

 #include <deque>
 #include <functional>
 #include <atomic>
 #include <mutex>
 #include <future>
 #include <condition_variable>
 #include <set>
 #include <iostream>
 #include <boost/circular_buffer.hpp>
 #include <base/shared_ptr_helper.h>
 #include <base/logger_useful.h>
 #include <Common/ThreadPool.h>
 #include <Common/Stopwatch.h>
 #include <Storages/MergeTree/IExecutableTask.h>
 namespace DB
 {
 namespace ErrorCodes
 {
     extern const int LOGICAL_ERROR;
 }
 struct TaskRuntimeData;
 using TaskRuntimeDataPtr = std::shared_ptr<TaskRuntimeData>;
 /**
  * Has RAII class to determine how many tasks are waiting for the execution and executing at the moment.
  * Also has some flags and primitives to wait for current task to be executed.
  */
 struct TaskRuntimeData
 {
     TaskRuntimeData(ExecutableTaskPtr && task_, CurrentMetrics::Metric metric_)
         : task(std::move(task_))
         , increment(std::move(metric_))
     {}
     ExecutableTaskPtr task;
     CurrentMetrics::Increment increment;
     std::atomic_bool is_currently_deleting{false};
     /// Actually autoreset=false is needed only for unit test
     /// where multiple threads could remove tasks corresponding to the same storage
     /// This scenario in not possible in reality.
     Poco::Event is_done{/*autoreset=*/false};
     /// This is equal to task->getPriority() not to do useless virtual calls in comparator
     UInt64 priority{0};
     /// By default priority queue will have max element at top
     static bool comparePtrByPriority(const TaskRuntimeDataPtr & lhs, const TaskRuntimeDataPtr & rhs)
     {
         return lhs->priority > rhs->priority;
     }
 };
 class OrdinaryRuntimeQueue
 {
 public:
     TaskRuntimeDataPtr pop()
     {
         auto result = std::move(queue.front());
         queue.pop_front();
         return result;
     }
     void push(TaskRuntimeDataPtr item) { queue.push_back(std::move(item));}
     void remove(StorageID id)
     {
         auto it = std::remove_if(queue.begin(), queue.end(),
             [&] (auto item) -> bool { return item->task->getStorageID() == id; });
         queue.erase(it, queue.end());
     }
     void setCapacity(size_t count) { queue.set_capacity(count); }
     bool empty() { return queue.empty(); }
 private:
     boost::circular_buffer<TaskRuntimeDataPtr> queue{0};
 };
 /// Uses a heap to pop a task with minimal priority
 class MergeMutateRuntimeQueue
 {
 public:
     TaskRuntimeDataPtr pop()
     {
         std::pop_heap(buffer.begin(), buffer.end(), TaskRuntimeData::comparePtrByPriority);
         auto result = std::move(buffer.back());
         buffer.pop_back();
         return result;
     }
     void push(TaskRuntimeDataPtr item)
     {
         item->priority = item->task->getPriority();
         buffer.push_back(std::move(item));
         std::push_heap(buffer.begin(), buffer.end(), TaskRuntimeData::comparePtrByPriority);
     }
     void remove(StorageID id)
     {
         auto it = std::remove_if(buffer.begin(), buffer.end(),
             [&] (auto item) -> bool { return item->task->getStorageID() == id; });
         buffer.erase(it, buffer.end());
         std::make_heap(buffer.begin(), buffer.end(), TaskRuntimeData::comparePtrByPriority);
     }
     void setCapacity(size_t count) { buffer.reserve(count); }
     bool empty() { return buffer.empty(); }
 private:
     std::vector<TaskRuntimeDataPtr> buffer{};
 };
 /**
  *  Executor for a background MergeTree related operations such as merges, mutations, fetches an so on.
  *  It can execute only successors of ExecutableTask interface.
  *  Which is a self-written coroutine. It suspends, when returns true from executeStep() method.
  *
  *  There are two queues of a tasks: pending (main queue for all the tasks) and active (currently executing).
  *  Pending queue is needed since the number of tasks will be more than thread to execute.
  *  Pending tasks are tasks that successfully scheduled to an executor or tasks that have some extra steps to execute.
  *  There is an invariant, that task may occur only in one of these queue. It can occur in both queues only in critical sections.
  *
  *  Pending:                                              Active:
  *
  *  |s| |s| |s| |s| |s| |s| |s| |s| |s| |s|               |s|
  *  |s| |s| |s| |s| |s| |s| |s| |s| |s|                   |s|
  *  |s| |s|     |s|     |s| |s|     |s|                   |s|
  *      |s|             |s| |s|                           |s|
  *      |s|                 |s|
  *                          |s|
  *
  *  Each task is simply a sequence of steps. Heavier tasks have longer sequences.
  *  When a step of a task is executed, we move tasks to pending queue. And take another from the queue's head.
  *  With these architecture all small merges / mutations will be executed faster, than bigger ones.
  *
  *  We use boost::circular_buffer as a container for queues not to do any allocations.
  *
  *  Another nuisance that we faces with is than background operations always interact with an associated Storage.
  *  So, when a Storage want to shutdown, it must wait until all its background operaions are finished.
  */
 template <class Queue>
 class MergeTreeBackgroundExecutor final : public shared_ptr_helper<MergeTreeBackgroundExecutor<Queue>>
 {
 public:
     MergeTreeBackgroundExecutor(
         String name_,
         size_t threads_count_,
         size_t max_tasks_count_,
         CurrentMetrics::Metric metric_)
         : name(name_)
         , threads_count(threads_count_)
         , max_tasks_count(max_tasks_count_)
         , metric(metric_)
     {
         if (max_tasks_count == 0)
             throw Exception(ErrorCodes::LOGICAL_ERROR, "Task count for MergeTreeBackgroundExecutor must not be zero");
         pending.setCapacity(max_tasks_count);
         active.set_capacity(max_tasks_count);
         pool.setMaxThreads(std::max(1UL, threads_count));
         pool.setMaxFreeThreads(std::max(1UL, threads_count));
         pool.setQueueSize(std::max(1UL, threads_count));
         for (size_t number = 0; number < threads_count; ++number)
             pool.scheduleOrThrowOnError([this] { threadFunction(); });
     }
     ~MergeTreeBackgroundExecutor()
     {
         wait();
     }
     bool trySchedule(ExecutableTaskPtr task);
     void removeTasksCorrespondingToStorage(StorageID id);
     void wait();
 private:
     String name;
     size_t threads_count{0};
     size_t max_tasks_count{0};
     CurrentMetrics::Metric metric;
     void routine(TaskRuntimeDataPtr item);
     void threadFunction();
     /// Initially it will be empty
     Queue pending{}; // 等待队列
     boost::circular_buffer<TaskRuntimeDataPtr> active{0}; // 执行队列
     std::mutex mutex; // 互斥锁
     std::condition_variable has_tasks;
     std::atomic_bool shutdown{false};
     ThreadPool pool;
 };
 extern template class MergeTreeBackgroundExecutor<MergeMutateRuntimeQueue>;
 extern template class MergeTreeBackgroundExecutor<OrdinaryRuntimeQueue>;
 using MergeMutateBackgroundExecutor = MergeTreeBackgroundExecutor<MergeMutateRuntimeQueue>;
 using OrdinaryBackgroundExecutor = MergeTreeBackgroundExecutor<OrdinaryRuntimeQueue>;
 }
 

 #include <Storages/MergeTree/MergeTreeBackgroundExecutor.h>

 #include <algorithm>
 #include <Common/setThreadName.h>
 #include <Storages/MergeTree/BackgroundJobsAssignee.h>
 namespace DB
 {
 template <class Queue>
 void MergeTreeBackgroundExecutor<Queue>::wait() // 等待执行
 {
     {
         std::lock_guard lock(mutex);
         shutdown = true;
         has_tasks.notify_all();
     }
     pool.wait();
 }
 template <class Queue>
 // 调度任务执行
 bool MergeTreeBackgroundExecutor<Queue>::trySchedule(ExecutableTaskPtr task)
 {
     std::lock_guard lock(mutex); // 上锁
     if (shutdown)
         return false;
     auto & value = CurrentMetrics::values[metric];
     if (value.load() >= static_cast<int64_t>(max_tasks_count))
         return false;
     pending.push(std::make_shared<TaskRuntimeData>(std::move(task), metric));
     has_tasks.notify_one();
     return true;
 }
 template <class Queue>
 void MergeTreeBackgroundExecutor<Queue>::removeTasksCorrespondingToStorage(StorageID id)
 {
     std::vector<TaskRuntimeDataPtr> tasks_to_wait;
     {
         std::lock_guard lock(mutex); // 上锁
         /// Erase storage related tasks from pending and select active tasks to wait for
         pending.remove(id);
         /// Copy items to wait for their completion
         std::copy_if(active.begin(), active.end(), std::back_inserter(tasks_to_wait),
             [&] (auto item) -> bool { return item->task->getStorageID() == id; });
         for (auto & item : tasks_to_wait)
             item->is_currently_deleting = true;
     }
     /// Wait for each task to be executed
     for (auto & item : tasks_to_wait)
     {
         item->is_done.wait();
         item.reset();
     }
 }
 template <class Queue>
 void MergeTreeBackgroundExecutor<Queue>::routine(TaskRuntimeDataPtr item)
 {
     DENY_ALLOCATIONS_IN_SCOPE;
     /// All operations with queues are considered no to do any allocations
     auto erase_from_active = [this, item]
     {
         active.erase(std::remove(active.begin(), active.end(), item), active.end());
     };
     bool need_execute_again = false;
     try
     {
         ALLOW_ALLOCATIONS_IN_SCOPE;
         need_execute_again = item->task->executeStep();
     }
     catch (...)
     {
         tryLogCurrentException(__PRETTY_FUNCTION__);
     }
     if (need_execute_again)
     {
         std::lock_guard guard(mutex);  // 上锁
         if (item->is_currently_deleting)
         {
             erase_from_active();
             /// This is significant to order the destructors.
             item->task.reset();
             item->is_done.set();
             item = nullptr;
             return;
         }
         /// After the `guard` destruction `item` has to be in moved from state
         /// Not to own the object it points to.
         /// Otherwise the destruction of the task won't be ordered with the destruction of the
         /// storage.
         pending.push(std::move(item));
         erase_from_active();
         has_tasks.notify_one();
         item = nullptr;
         return;
     }
     {
         std::lock_guard guard(mutex);  // 上锁
         erase_from_active();
         has_tasks.notify_one();
         try
         {
             ALLOW_ALLOCATIONS_IN_SCOPE;
             /// In a situation of a lack of memory this method can throw an exception,
             /// because it may interact somehow with BackgroundSchedulePool, which may allocate memory
             /// But it is rather safe, because we have try...catch block here, and another one in ThreadPool.
             item->task->onCompleted();
         }
         catch (...)
         {
             tryLogCurrentException(__PRETTY_FUNCTION__);
         }
         /// We have to call reset() under a lock, otherwise a race is possible.
         /// Imagine, that task is finally completed (last execution returned false),
         /// we removed the task from both queues, but still have pointer.
         /// The thread that shutdowns storage will scan queues in order to find some tasks to wait for, but will find nothing.
         /// So, the destructor of a task and the destructor of a storage will be executed concurrently.
         item->task.reset();
         item->is_done.set();
         item = nullptr;
     }
 }
 

 template <class Queue>
 void MergeTreeBackgroundExecutor<Queue>::threadFunction()
 {
     setThreadName(name.c_str());
     DENY_ALLOCATIONS_IN_SCOPE;
     while (true) // MergeTreeBackgroundExecutor 常驻线程池 
     {
         try
         {
             TaskRuntimeDataPtr item;
             {
                 std::unique_lock lock(mutex);   // 上锁
                 has_tasks.wait(lock, [this](){ return !pending.empty() || shutdown; });
                 if (shutdown)
                     break;
                 item = std::move(pending.pop());
                 active.push_back(item);
             }
             routine(std::move(item));
         }
         catch (...)
         {
             tryLogCurrentException(__PRETTY_FUNCTION__);
         }
     }
 }
 template class MergeTreeBackgroundExecutor<MergeMutateRuntimeQueue>;
 template class MergeTreeBackgroundExecutor<OrdinaryRuntimeQueue>;
 }
 

 
#pragma once
 #include 
 #include 
 namespace DB
 {
 struct ASTTablesInSelectQueryElement;
 struct StorageID;
 /** SELECT query
 */
 class ASTSelectQuery :public IAST
 {
 public:
 enum class Expression : uint8_t
 {
 WITH,
 SELECT,
 TABLES,
 PREWHERE,
 WHERE,
 GROUP_BY,
 HAVING,
 WINDOW,
 ORDER_BY,
 LIMIT_BY_OFFSET,
 LIMIT_BY_LENGTH,
 LIMIT_BY,
 LIMIT_OFFSET,
 LIMIT_LENGTH,
 SETTINGS
     };
 static String expressionToString(Expression expr)
 {
 switch (expr)
 {
 case Expression::WITH:
 return "WITH";
 case Expression::SELECT:
 return "SELECT";
 case Expression::TABLES:
 return "TABLES";
 case Expression::PREWHERE:
 return "PREWHERE";
 case Expression::WHERE:
 return "WHERE";
 case Expression::GROUP_BY:
 return "GROUP BY";
 case Expression::HAVING:
 return "HAVING";
 case Expression::WINDOW:
 return "WINDOW";
 case Expression::ORDER_BY:
 return "ORDER BY";
 case Expression::LIMIT_BY_OFFSET:
 return "LIMIT BY OFFSET";
 case Expression::LIMIT_BY_LENGTH:
 return "LIMIT BY LENGTH";
 case Expression::LIMIT_BY:
 return "LIMIT BY";
 case Expression::LIMIT_OFFSET:
 return "LIMIT OFFSET";
 case Expression::LIMIT_LENGTH:
 return "LIMIT LENGTH";
 case Expression::SETTINGS:
 return "SETTINGS";
 }
 return "";
 }
 /** Get the text that identifies this element. */
     String getID(char)const override {return "SelectQuery"; }
 ASTPtr clone()const override;
 bool distinct =false;
 bool group_by_with_totals =false;
 bool group_by_with_rollup =false;
 bool group_by_with_cube =false;
 bool group_by_with_constant_keys =false;
 bool limit_with_ties =false;
 ASTPtr & refSelect()    {return getExpression(Expression::SELECT); }
 ASTPtr & refTables()    {return getExpression(Expression::TABLES); }
 ASTPtr & refPrewhere()  {return getExpression(Expression::PREWHERE); }
 ASTPtr & refWhere()     {return getExpression(Expression::WHERE); }
 ASTPtr & refHaving()    {return getExpression(Expression::HAVING); }
 const ASTPtr with()const {return getExpression(Expression::WITH); }
 const ASTPtr select()const {return getExpression(Expression::SELECT); }
 const ASTPtr tables()const {return getExpression(Expression::TABLES); }
 const ASTPtr prewhere()const {return getExpression(Expression::PREWHERE); }
 const ASTPtr where()const {return getExpression(Expression::WHERE); }
 const ASTPtr groupBy()const {return getExpression(Expression::GROUP_BY); }
 const ASTPtr having()const {return getExpression(Expression::HAVING); }
 const ASTPtr window()const {return getExpression(Expression::WINDOW); }
 const ASTPtr orderBy()const {return getExpression(Expression::ORDER_BY); }
 const ASTPtr limitByOffset()const {return getExpression(Expression::LIMIT_BY_OFFSET); }
 const ASTPtr limitByLength()const {return getExpression(Expression::LIMIT_BY_LENGTH); }
 const ASTPtr limitBy()const {return getExpression(Expression::LIMIT_BY); }
 const ASTPtr limitOffset()const {return getExpression(Expression::LIMIT_OFFSET); }
 const ASTPtr limitLength()const {return getExpression(Expression::LIMIT_LENGTH); }
 const ASTPtr settings()const {return getExpression(Expression::SETTINGS); }
 bool hasFiltration()const {return where() || prewhere() || having(); }
 /// Set/Reset/Remove expression.
     void setExpression(Expression expr, ASTPtr && ast);
 ASTPtr getExpression(Expression expr,bool clone =false)const
     {
 auto it = positions.find(expr);
 if (it != positions.end())
 return clone ? children[it->second]->clone() : children[it->second];
 return {};
 }
 /// Compatibility with old parser of tables list. TODO remove
     ASTPtr sampleSize()const;
 ASTPtr sampleOffset()const;
 std::pair arrayJoinExpressionList()const;
 const ASTTablesInSelectQueryElement * join()const;
 bool final()const;
 bool withFill()const;
 void replaceDatabaseAndTable(const String & database_name,const String & table_name);
 void replaceDatabaseAndTable(const StorageID & table_id);
 void addTableFunction(ASTPtr & table_function_ptr);
 void updateTreeHashImpl(SipHash & hash_state)const override;
 void setFinal();
 const char * getQueryKindString()const override {return "Select"; }
 protected:
 void formatImpl(const FormatSettings & settings, FormatState & state, FormatStateStacked frame)const override;
 private:
 std::unordered_mappositions;
 ASTPtr & getExpression(Expression expr);
 };
 }