Method executeImpl

src/WorkerTasks/CloudUniqueMergeTreeMergeTask.cpp:165–366

}

void CloudUniqueMergeTreeMergeTask::executeImpl()
{
    Stopwatch watch;

    auto txn = getContext()->getCurrentTransaction();
    if (!txn)
        throw Exception("Transaction is not set", ErrorCodes::LOGICAL_ERROR);
    auto txn_id = txn->getTransactionID();

    auto lock = storage.lockForShare(RWLockImpl::NO_QUERY, storage.getSettings()->lock_acquire_timeout_for_background_operations);

    LOG_TRACE(log, "Begin to execute merge task {}", params.task_id);

    auto catalog = getContext()->getCnchCatalog();
    auto t1 = getContext()->getTimestamp();
    /// get and set src part's bitmap meta at t1
    curr_bitmap_metas = getDeleteBitmapMetas(*catalog, params.source_data_parts, t1);
    for (size_t i = 0; i < curr_bitmap_metas.size(); ++i)
    {
        params.source_data_parts[i]->setDeleteBitmapMeta(curr_bitmap_metas[i]);
        curr_bitmaps.push_back(params.source_data_parts[i]->getDeleteBitmap());
    }

    LOG_TRACE(log, "Prepared delete bitmap for source parts");

    /// merge src parts using delete bitmap at t1
    MergeTreeDataMerger merger(
        storage,
        params,
        getContext(),
        getManipulationListElement(),
        [&, this] {
            if (isCancelled())
                return true;

            /// TODO: refactor this
            auto last_touch_time = getManipulationListElement()->last_touch_time.load(std::memory_order_relaxed);
            if (UInt64(time(nullptr) - last_touch_time) > getContext()->getSettingsRef().cloud_task_auto_stop_timeout)
            {
                LOG_TRACE(
                    &Poco::Logger::get("CloudUniqueMergeTreeMergeTask"),
                    "Task {} doesn't receive heartbeat from server, stop it self.",
                    params.task_id);
                setCancelled();
            }
            return isCancelled();
        },
        /*build_rowid_mappings*/ true);
    /// data of temp part will be removed in dtor
    auto merged_part = merger.mergePartsToTemporaryPart();

    DeleteBitmapPtr merged_part_bitmap = std::make_shared<Roaring>();
    /// t2: convert any new deletes of src part into deletes on merged part
    updateDeleteBitmap(*catalog, merger, merged_part_bitmap);

    /// prepare parts and bitmaps to dump
    std::vector<ReservationPtr> reservations;
    IMutableMergeTreeDataPartsVector parts_to_dump;