#include #include #include #include namespace DB { Block MergeTreeBlockOutputStream::getHeader() const { return metadata_snapshot->getSampleBlock(); } void MergeTreeBlockOutputStream::write(const Block & block) { storage.delayInsertOrThrowIfNeeded(); auto part_blocks = storage.writer.splitBlockIntoParts(block, max_parts_per_block, metadata_snapshot); for (auto & current_block : part_blocks) { Stopwatch watch; MergeTreeData::MutableDataPartPtr part = storage.writer.writeTempPart(current_block, metadata_snapshot); storage.renameTempPartAndAdd(part, &storage.increment); PartLog::addNewPart(storage.global_context, part, watch.elapsed()); if (storage.merging_mutating_task_handle) { /// Initiate async merge - it will be done if it's good time for merge and if there are space in 'background_pool'. storage.merging_mutating_task_handle->signalReadyToRun(); } } } }