thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-09-20 08:40:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

check if block is empty after async insert retries

Browse Source
This commit is contained in:
Han Fei 2023-09-29 15:11:15 +02:00
parent 2a0ec41d85
commit e57217ea9c
1 changed files with 6 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/Storages/MergeTree/ReplicatedMergeTreeSink.cpp
View File

@ -321,7 +321,10 @@ void ReplicatedMergeTreeSinkImpl<async_insert>::consume(Chunk chunk)
if constexpr (async_insert) if constexpr (async_insert)
{ {
block_id = AsyncInsertBlockInfo::getHashesForBlocks(unmerged_block.has_value() ? *unmerged_block : current_block, temp_part.part->info.partition_id); block_id = AsyncInsertBlockInfo::getHashesForBlocks(unmerged_block ? *unmerged_block : current_block, temp_part.part->info.partition_id);
if (unmerged_block)
LOG_TRACE(log, "async insert part, part id {}, block id {}, offsets {}, size {}", temp_part.part->info.partition_id, toString(block_id), toString(unmerged_block->offsets), unmerged_block->offsets.size());
else
LOG_TRACE(log, "async insert part, part id {}, block id {}, offsets {}, size {}", temp_part.part->info.partition_id, toString(block_id), toString(current_block.offsets), current_block.offsets.size()); LOG_TRACE(log, "async insert part, part id {}, block id {}, offsets {}, size {}", temp_part.part->info.partition_id, toString(block_id), toString(current_block.offsets), current_block.offsets.size());
} }
else else
@ -461,7 +464,7 @@ void ReplicatedMergeTreeSinkImpl<true>::finishDelayedChunk(const ZooKeeperWithFa
LOG_DEBUG(log, "Found duplicate block IDs: {}, retry times {}", toString(conflict_block_ids), retry_times); LOG_DEBUG(log, "Found duplicate block IDs: {}, retry times {}", toString(conflict_block_ids), retry_times);
/// partition clean conflict /// partition clean conflict
partition.filterBlockDuplicate(conflict_block_ids, false); partition.filterBlockDuplicate(conflict_block_ids, false);
if (partition.block_id.empty()) if (partition.block_with_partition.block.rows() == 0)
break; break;
partition.block_with_partition.partition = std::move(partition.temp_part.part->partition.value); partition.block_with_partition.partition = std::move(partition.temp_part.part->partition.value);
/// partition.temp_part is already finalized, no need to call cancel /// partition.temp_part is already finalized, no need to call cancel