Merge

dbms/include/DB/Storages/MergeTree/MergeTreeBlockOutputStream.h

@@ -22,6 +22,8 @@ public:
 	
 	void write(const Block & block)
 	{
+		Poco::ScopedReadRWLock write_lock(storage.write_lock);
+		
 		storage.check(block, true);
 		
 		DateLUTSingleton & date_lut = DateLUTSingleton::instance();

dbms/include/DB/Storages/StorageMergeTree.h

@@ -7,6 +7,7 @@
 #include <DB/Interpreters/Context.h>
 #include <DB/Interpreters/ExpressionActions.h>
 #include <DB/Storages/IStorage.h>
+#include <Poco/RWLock.h>
 
 
 namespace DB
@@ -409,6 +410,15 @@ private:
 	DataParts data_parts;
 	Poco::FastMutex data_parts_mutex;
 
+	/** Взятие этого лока на запись, запрещает мердж */
+	Poco::RWLock merge_lock;
+
+	/** Взятие этого лока на запись, запрещает запись */
+	Poco::RWLock write_lock;
+
+	/** Взятие этого лока на запись, запрещает чтение */
+	Poco::RWLock read_lock;
+
 	StorageMergeTree(const String & path_, const String & name_, NamesAndTypesListPtr columns_,
 					const Context & context_,
 					ASTPtr & primary_expr_ast_,

dbms/src/Storages/StorageMergeTree.cpp

@@ -158,6 +158,8 @@ BlockInputStreams StorageMergeTree::read(
 	size_t max_block_size,
 	unsigned threads)
 {
+	Poco::ScopedReadRWLock lock(read_lock);
+	
 	check(column_names_to_return);
 	processed_stage = QueryProcessingStage::FetchColumns;
 	
@@ -783,6 +785,8 @@ void StorageMergeTree::mergeThread(bool while_can, bool aggressive)
 			clearOldParts();
 
 			{
+				Poco::ScopedReadRWLock lock(merge_lock);
+
 				/// К концу этого логического блока должен быть вызван деструктор, чтобы затем корректно определить удаленные куски
 				Poco::SharedPtr<CurrentlyMergingPartsTagger> what;