Method run

engine/src/execution_kernels/BatchWindowFunctionProcessing.cpp:250–316

}

kstatus ComputeWindowKernel::run() {
    CodeTimer timer;

    int self_node_idx = context->getNodeIndex(ral::communication::CommunicationData::getInstance().getSelfNode());
    int num_nodes = context->getTotalNodes();

    bool is_first_batch = true;
    bool is_last_batch = false;
    bool is_first_node = self_node_idx == 0;
    bool is_last_node = self_node_idx == num_nodes - 1;

    std::unique_ptr<ral::cache::CacheData> cache_data = this->input_cache()->pullCacheData();

    if (this->remove_overlap){
        while (cache_data != nullptr ){
            std::vector<std::unique_ptr <ral::cache::CacheData> > inputs;
            inputs.push_back(std::move(cache_data));

            cache_data = this->input_cache()->pullCacheData();
            if (cache_data == nullptr){
                is_last_batch = true;
            }

            std::map<std::string, std::string> task_args;
            task_args[TASK_ARG_REMOVE_PRECEDING_OVERLAP] = !(is_first_batch && is_first_node) ? TRUE : FALSE;
            task_args[TASK_ARG_REMOVE_FOLLOWING_OVERLAP] = !(is_last_batch && is_last_node) ? TRUE : FALSE;

            ral::execution::executor::get_instance()->add_task(
                    std::move(inputs),
                    this->output_cache(),
                    this, 
                    task_args);

            is_first_batch = false;
        }
    } else {
        while (cache_data != nullptr ){
            std::vector<std::unique_ptr <ral::cache::CacheData> > inputs;
            inputs.push_back(std::move(cache_data));

            ral::execution::executor::get_instance()->add_task(
                    std::move(inputs),
                    this->output_cache(),
                    this);

            cache_data = this->input_cache()->pullCacheData();
        }
    }

    

    std::unique_lock<std::mutex> lock(kernel_mutex);
    kernel_cv.wait(lock,[this]{
        return this->tasks.empty();
    });

    if (logger != nullptr) {
        logger->debug("{query_id}|{step}|{substep}|{info}|{duration}|kernel_id|{kernel_id}||",