2016-01-17 05:22:22 +00:00
# include <DB/Storages/StorageReplicatedMergeTree.h>
# include <DB/Storages/MergeTree/AbandonableLockInZooKeeper.h>
# include <DB/Storages/MergeTree/ReplicatedMergeTreeQuorumEntry.h>
# include <DB/Storages/MergeTree/ReplicatedMergeTreeBlockOutputStream.h>
# include <DB/DataStreams/IBlockOutputStream.h>
# include <DB/IO/Operators.h>
namespace DB
{
namespace ErrorCodes
{
extern const int TOO_LESS_LIVE_REPLICAS ;
extern const int UNSATISFIED_QUORUM_FOR_PREVIOUS_WRITE ;
extern const int CHECKSUM_DOESNT_MATCH ;
extern const int UNEXPECTED_ZOOKEEPER_ERROR ;
extern const int NO_ZOOKEEPER ;
}
ReplicatedMergeTreeBlockOutputStream : : ReplicatedMergeTreeBlockOutputStream (
StorageReplicatedMergeTree & storage_ , const String & insert_id_ , size_t quorum_ )
: storage ( storage_ ) , insert_id ( insert_id_ ) , quorum ( quorum_ ) ,
log ( & Logger : : get ( storage . data . getLogName ( ) + " (Replicated OutputStream) " ) )
{
/// Значение кворума 1 имеет такой же смысл, как если он отключён.
if ( quorum = = 1 )
quorum = 0 ;
}
void ReplicatedMergeTreeBlockOutputStream : : writePrefix ( )
{
/// TODO Можно ли здесь не блокировать структуру таблицы?
storage . data . delayInsertIfNeeded ( & storage . restarting_thread - > getWakeupEvent ( ) ) ;
}
/// Позволяет проверить, что сессия в ZooKeeper ещё жива.
static void assertSessionIsNotExpired ( zkutil : : ZooKeeperPtr & zookeeper )
{
2016-01-17 08:12:48 +00:00
if ( ! zookeeper )
throw Exception ( " No ZooKeeper session. " , ErrorCodes : : NO_ZOOKEEPER ) ;
2016-01-17 05:22:22 +00:00
if ( zookeeper - > expired ( ) )
throw Exception ( " ZooKeeper session has been expired. " , ErrorCodes : : NO_ZOOKEEPER ) ;
}
void ReplicatedMergeTreeBlockOutputStream : : write ( const Block & block )
{
auto zookeeper = storage . getZooKeeper ( ) ;
assertSessionIsNotExpired ( zookeeper ) ;
/** Если запись с
* а т а к ж е ч т о д л я в с е х п р е д ы д у щ и х к у с к о в , д л я к о т о р ы х т р е б у е т с я к в о р у м , э т о т к в о р у м д о с т и г н у т .
*/
String quorum_status_path = storage . zookeeper_path + " /quorum/status " ;
if ( quorum )
{
/// Список живых реплик. В с е
auto live_replicas = zookeeper - > getChildren ( storage . zookeeper_path + " /leader_election " ) ;
if ( live_replicas . size ( ) < quorum )
throw Exception ( " Number of alive replicas ( "
+ toString ( live_replicas . size ( ) ) + " ) is less than requested quorum ( " + toString ( quorum ) + " ). " ,
ErrorCodes : : TOO_LESS_LIVE_REPLICAS ) ;
/** Достигнут ли кворум для последнего куска, для которого нужен кворум?
* З а п и с ь в с е х к у с к о в с в к л ю ч е н н ы м к в о р у м о м л и н е й н о у п о р я д о ч е н а .
* Э т о з н а ч и т , ч т о в л ю б о й м о м е н т в р е м е н и м о ж е т б ы т ь т о л ь к о о д и н к у с о к ,
* д л я к о т о р о г о н у ж е н , н о е щ ё н е д о с т и г н у т к в о р у м .
* И н ф о р м а ц и я о т а к о м к у с к е б у д е т р а с п о л о ж е н а в н о д е / quorum / status .
* Е с л и к в о р у м д о с т и г н у т , т о н о д а у д а л я е т с я .
*/
String quorum_status ;
bool quorum_unsatisfied = zookeeper - > tryGet ( quorum_status_path , quorum_status ) ;
if ( quorum_unsatisfied )
throw Exception ( " Quorum for previous write has not been satisfied yet. Status: " + quorum_status , ErrorCodes : : UNSATISFIED_QUORUM_FOR_PREVIOUS_WRITE ) ;
/// О б е
}
auto part_blocks = storage . writer . splitBlockIntoParts ( block ) ;
for ( auto & current_block : part_blocks )
{
assertSessionIsNotExpired ( zookeeper ) ;
+ + block_index ;
String block_id = insert_id . empty ( ) ? " " : insert_id + " __ " + toString ( block_index ) ;
String month_name = toString ( DateLUT : : instance ( ) . toNumYYYYMMDD ( DayNum_t ( current_block . min_date ) ) / 100 ) ;
AbandonableLockInZooKeeper block_number_lock = storage . allocateBlockNumber ( month_name ) ;
Int64 part_number = block_number_lock . getNumber ( ) ;
MergeTreeData : : MutableDataPartPtr part = storage . writer . writeTempPart ( current_block , part_number ) ;
String part_name = ActiveDataPartSet : : getPartName ( part - > left_date , part - > right_date , part - > left , part - > right , part - > level ) ;
/// Хэш от данных.
SipHash hash ;
part - > checksums . summaryDataChecksum ( hash ) ;
union
{
char bytes [ 16 ] ;
UInt64 lo , hi ;
} hash_value ;
hash . get128 ( hash_value . bytes ) ;
String checksum ( hash_value . bytes , 16 ) ;
/// Если в запросе не указан ID, возьмем в качестве ID хеш от данных. Т о
/// NOTE: Если такая дедупликация не нужна, можно вместо этого оставлять block_id пустым.
/// Можно для этого сделать настройку или синтаксис в запросе (например, ID=null).
if ( block_id . empty ( ) )
{
block_id = toString ( hash_value . lo ) + " _ " + toString ( hash_value . hi ) ;
if ( block_id . empty ( ) )
throw Exception ( " Logical error: block_id is empty. " , ErrorCodes : : LOGICAL_ERROR ) ;
}
LOG_DEBUG ( log , " Wrote block " < < part_number < < " with ID " < < block_id < < " , " < < current_block . block . rows ( ) < < " rows " ) ;
StorageReplicatedMergeTree : : LogEntry log_entry ;
log_entry . type = StorageReplicatedMergeTree : : LogEntry : : GET_PART ;
log_entry . create_time = time ( 0 ) ;
log_entry . source_replica = storage . replica_name ;
log_entry . new_part_name = part_name ;
log_entry . quorum = quorum ;
log_entry . block_id = block_id ;
/// Одновременно добавим информацию о
/// Информация о
zkutil : : Ops ops ;
auto acl = zookeeper - > getDefaultACL ( ) ;
ops . push_back (
new zkutil : : Op : : Create (
storage . zookeeper_path + " /blocks/ " + block_id ,
" " ,
acl ,
zkutil : : CreateMode : : Persistent ) ) ;
ops . push_back (
new zkutil : : Op : : Create (
storage . zookeeper_path + " /blocks/ " + block_id + " /checksum " ,
checksum ,
acl ,
zkutil : : CreateMode : : Persistent ) ) ;
ops . push_back (
new zkutil : : Op : : Create (
storage . zookeeper_path + " /blocks/ " + block_id + " /number " ,
toString ( part_number ) ,
acl ,
zkutil : : CreateMode : : Persistent ) ) ;
/// Информация о
storage . checkPartAndAddToZooKeeper ( part , ops , part_name ) ;
/// Лог репликации.
ops . push_back ( new zkutil : : Op : : Create (
storage . zookeeper_path + " /log/log- " ,
log_entry . toString ( ) ,
acl ,
zkutil : : CreateMode : : PersistentSequential ) ) ;
/// Удаление информации о
block_number_lock . getUnlockOps ( ops ) ;
/** Если нужен кворум - создание узла, в котором отслеживается кворум.
* ( Е с л и т а к о й у з е л у ж е с у щ е с т в у е т - з н а ч и т к т о - т о у с п е л о д н о в р е м е н н о с д е л а т ь д р у г у ю к в о р у м н у ю з а п и с ь , н о д л я н е ё к в о р у м е щ ё н е д о с т и г н у т .
* Д е л а т ь в э т о в р е м я с л е д у ю щ у ю к в о р у м н у ю з а п и с ь н е л ь з я . )
*/
if ( quorum )
{
ReplicatedMergeTreeQuorumEntry quorum_entry ;
quorum_entry . part_name = part_name ;
quorum_entry . required_number_of_replicas = quorum ;
quorum_entry . replicas . insert ( storage . replica_name ) ;
/** В о
* К о г д а д р у г и е р е п л и к и б у д у т п о л у ч а т ь э т о т к у с о к ( о б ы ч н ы м с п о с о б о м , о б р а б а т ы в а я л о г р е п л и к а ц и и ) ,
* о н и б у д у т д о б а в л я т ь с е б я в с о д е р ж и м о е э т о г о у з л а .
* К о г д а в н ё м б у д е т и н ф о р м а ц и я о quorum к о л и ч е с т в е р е п л и к , э т о т у з е л у д а л я е т с я ,
* ч т о г о в о р и т о т о м , ч т о к в о р у м д о с т и г н у т .
*/
ops . push_back (
new zkutil : : Op : : Create (
quorum_status_path ,
quorum_entry . toString ( ) ,
acl ,
zkutil : : CreateMode : : Persistent ) ) ;
}
MergeTreeData : : Transaction transaction ; /// Если не получится добавить кусок в ZK, снова уберем е г о
storage . data . renameTempPartAndAdd ( part , nullptr , & transaction ) ;
try
{
auto code = zookeeper - > tryMulti ( ops ) ;
if ( code = = ZOK )
{
transaction . commit ( ) ;
storage . merge_selecting_event . set ( ) ;
}
else if ( code = = ZNODEEXISTS )
{
/// Если блок с е г о
String expected_checksum ;
if ( ! block_id . empty ( ) & & zookeeper - > tryGet (
storage . zookeeper_path + " /blocks/ " + block_id + " /checksum " , expected_checksum ) )
{
LOG_INFO ( log , " Block with ID " < < block_id < < " already exists; ignoring it (removing part " < < part - > name < < " ) " ) ;
/// Если данные отличались от тех, что были вставлены ранее с
if ( expected_checksum ! = checksum )
{
if ( ! insert_id . empty ( ) )
throw Exception ( " Attempt to insert block with same ID but different checksum " , ErrorCodes : : CHECKSUM_DOESNT_MATCH ) ;
else
throw Exception ( " Logical error: got ZNODEEXISTS while inserting data, block ID is derived from checksum but checksum doesn't match " , ErrorCodes : : LOGICAL_ERROR ) ;
}
transaction . rollback ( ) ;
}
else if ( zookeeper - > exists ( quorum_status_path ) )
{
transaction . rollback ( ) ;
throw Exception ( " Another quorum insert has been already started " , ErrorCodes : : UNSATISFIED_QUORUM_FOR_PREVIOUS_WRITE ) ;
}
else
{
/// Сюда можем попасть также, если узел с
throw Exception ( " Unexpected ZNODEEXISTS while adding block " + toString ( part_number ) + " with ID " + block_id + " : "
+ zkutil : : ZooKeeper : : error2string ( code ) , ErrorCodes : : UNEXPECTED_ZOOKEEPER_ERROR ) ;
}
}
else
{
throw Exception ( " Unexpected error while adding block " + toString ( part_number ) + " with ID " + block_id + " : "
+ zkutil : : ZooKeeper : : error2string ( code ) , ErrorCodes : : UNEXPECTED_ZOOKEEPER_ERROR ) ;
}
}
catch ( const zkutil : : KeeperException & e )
{
/** Если потерялось соединение, и мы не знаем, применились ли изменения, нельзя удалять локальный кусок:
* е с л и и з м е н е н и я п р и м е н и л и с ь , в / blocks / п о я в и л с я в с т а в л е н н ы й б л о к , и е г о н е л ь з я б у д е т в с т а в и т ь с н о в а .
*/
if ( e . code = = ZOPERATIONTIMEOUT | |
e . code = = ZCONNECTIONLOSS )
{
transaction . commit ( ) ;
storage . enqueuePartForCheck ( part - > name ) ;
}
throw ;
}
if ( quorum )
{
/// Дожидаемся достижения кворума. TODO Настраиваемый таймаут.
LOG_TRACE ( log , " Waiting for quorum " ) ;
zookeeper - > waitForDisappear ( quorum_status_path ) ;
LOG_TRACE ( log , " Quorum satisfied " ) ;
}
}
}
}
