Method maybeConsumeIndex

src/processor/operator/persistent/index_builder.cpp:57–103

}

void IndexBuilderGlobalQueues::maybeConsumeIndex(size_t index,
    NodeBatchInsertErrorHandler& errorHandler) {
    auto& pkIndex = nodeTable->getPKIndex()->cast<PrimaryKeyIndex>();
    if (!pkIndex.tryLockTypedIndex(index)) {
        return;
    }

    std::visit(
        [&](auto&& queues) {
            using T = std::decay_t<decltype(queues.type)>;
            auto lck = pkIndex.adoptLockOfTypedIndex(index);
            IndexBufferWithWarningData<T> bufferWithWarningData;
            while (queues.array[index].pop(bufferWithWarningData)) {
                auto& buffer = bufferWithWarningData.indexBuffer;
                auto& warningDataBuffer = bufferWithWarningData.warningDataBuffer;
                uint64_t insertBufferOffset = 0;
                while (insertBufferOffset < buffer.size()) {
                    auto numValuesInserted = pkIndex.appendWithIndexPosNoLock(transaction, buffer,
                        insertBufferOffset, index,
                        [&](offset_t offset) { return nodeTable->isVisible(transaction, offset); });
                    if (numValuesInserted < buffer.size() - insertBufferOffset) {
                        const auto& erroneousEntry = buffer[insertBufferOffset + numValuesInserted];
                        OptionalWarningSourceData erroneousEntryWarningData;
                        if (warningDataBuffer != nullptr) {
                            erroneousEntryWarningData =
                                (*warningDataBuffer)[insertBufferOffset + numValuesInserted];
                        }
                        errorHandler.handleError(
                            IndexBuilderError<T>{.message = ExceptionMessage::duplicatePKException(
                                                     TypeUtils::toString(erroneousEntry.first)),
                                .key = erroneousEntry.first,
                                .nodeID =
                                    nodeID_t{
                                        erroneousEntry.second,
                                        nodeTable->getTableID(),
                                    },
                                .kind = NodeCopyErrorKind::DUPLICATE_PK,
                                .warningData = erroneousEntryWarningData});
                        insertBufferOffset += 1; // skip the erroneous index then continue
                    }
                    insertBufferOffset += numValuesInserted;
                }
            }
            return;
        },
        std::move(queues));
}

IndexBuilderLocalBuffers::IndexBuilderLocalBuffers(IndexBuilderGlobalQueues& globalQueues)
    : globalQueues(&globalQueues) {