Repository: milvus-io/milvus
File: internal/core/src/segcore/storagev2translator/ManifestGroupTranslator.cpp (C++)

Code before:

    LOG_INFO(
        "[StorageV2] translator {} submits {} batch tasks for manifest column "
        "group {}",
        key_,
        load_futures.size(),
        column_group_index_);

    // Pop loop — convert each cell immediately, no ArrowTable accumulation
    std::unordered_map<milvus::cachinglayer::cid_t,
                       std::unique_ptr<milvus::GroupChunk>>
        completed_cells;
    completed_cells.reserve(cids.size());

    try {
        std::shared_ptr<milvus::segcore::CellLoadResult> cell_data;
        while (channel->pop(cell_data)) {
            CheckCancellation(
                ctx, segment_id_, "ManifestGroupTranslator::get_cells()");
            completed_cells[cell_data->cid] =
                load_group_chunk(cell_data->tables, cell_data->cid);
        }
    } catch (std::exception& ex) {
        // make sure all futures are handled, but still throw pop & load ex
        try {
            storage::WaitAllFutures(load_futures);
        } catch (...) {
            LOG_WARN("WaitAllFutures exception swallowed");
        }
        throw;
    }

    storage::WaitAllFutures(load_futures);

    for (auto cid : cids) {
        auto it = completed_cells.find(cid);
        AssertInfo(
            it != completed_cells.end(),
            fmt::format(
                "[StorageV2] translator {} cell {} not loaded", key_, cid));
        cells.emplace_back(cid, std::move(it->second));
    }

    return cells;
}

std::unique_ptr<milvus::GroupChunk>
ManifestGroupTranslator::load_group_chunk(
    const std::vector<std::shared_ptr<arrow::Table>>& tables,
    const milvus::cachinglayer::cid_t cid) {
    assert(!tables.empty());

Review comment (bug):
Why does the catch block call `WaitAllFutures` here? All lambda captures are by-value (`shared_ptr`), so nothing dangles if `get_cells()` throws — the tasks will complete on their own in the thread pool and release resources naturally.

As-is, this can deadlock with the bounded channel: if the channel is full when `load_group_chunk()` throws, producers are blocked on `channel->push()`, and `WaitAllFutures` waits for those producers — classic circular wait. Same issue exists in `GroupChunkTranslator::get_cells()`.

Code after:

    LOG_INFO(
        "[StorageV2] translator {} submits {} batch tasks for manifest column "
        "group {}",
        key_,
        load_futures.size(),
        column_group_index_);

    // Pop loop — convert each cell immediately, no ArrowTable accumulation
    std::unordered_map<milvus::cachinglayer::cid_t,
                       std::unique_ptr<milvus::GroupChunk>>
        completed_cells;
    completed_cells.reserve(cids.size());

    try {
        std::shared_ptr<milvus::segcore::CellLoadResult> cell_data;
        while (channel->pop(cell_data)) {
            CheckCancellation(
                ctx, segment_id_, "ManifestGroupTranslator::get_cells()");
            completed_cells[cell_data->cid] =
                load_group_chunk(cell_data->tables, cell_data->cid);
        }
    } catch (...) {
        // Drain the channel to unblock producers that may be stuck on push()
        // to a full bounded channel. Without draining, producers block forever
        // and their task_guard (which calls channel->close()) never executes.
        std::shared_ptr<milvus::segcore::CellLoadResult> discard;
        try {
            while (channel->pop(discard)) {
            }
        } catch (...) {
            LOG_WARN("drain channel exception swallowed");
        }
        throw;
    }

    storage::WaitAllFutures(load_futures);

    for (auto cid : cids) {
        auto it = completed_cells.find(cid);
        AssertInfo(
            it != completed_cells.end(),
            fmt::format(
                "[StorageV2] translator {} cell {} not loaded", key_, cid));
        cells.emplace_back(cid, std::move(it->second));
    }

    return cells;
}

std::unique_ptr<milvus::GroupChunk>