// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you under the Apache License, Version 2.0 (the

// "License"); you may not use this file except in compliance

// with the License.  You may obtain a copy of the License at

//

//   http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing,

// software distributed under the License is distributed on an

// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

// KIND, either express or implied.  See the License for the

// specific language governing permissions and limitations

// under the License.

#include "exec/pipeline/pipeline.h"

#include <memory>

#include <string>

#include <utility>

#include "exec/exchange/local_exchange_source_operator.h"

#include "exec/operator/operator.h"

#include "exec/pipeline/pipeline_fragment_context.h"

#include "exec/pipeline/pipeline_task.h"

namespace doris {

void Pipeline::_init_profile() {

    auto s = fmt::format("Pipeline (pipeline id={})", _pipeline_id);

    _pipeline_profile = std::make_unique<RuntimeProfile>(std::move(s));

}

bool Pipeline::need_to_local_exchange(const DataDistribution target_data_distribution,

                                      const int idx) const {

    if (!target_data_distribution.need_local_exchange()) {

        return false;

    }

    // If serial operator exists after `idx`-th operator, we should not improve parallelism.

    if (std::any_of(_operators.begin() + idx, _operators.end(),

                    [&](OperatorPtr op) -> bool { return op->is_serial_operator(); })) {

        return false;

    }

    // If all operators are serial and sink is not serial, we should improve parallelism for sink.

    if (std::all_of(_operators.begin(), _operators.end(),

                    [&](OperatorPtr op) -> bool { return op->is_serial_operator(); })) {

        if (!_sink->is_serial_operator()) {

            return true;

        }

    } else if (std::any_of(_operators.begin(), _operators.end(),

                           [&](OperatorPtr op) -> bool { return op->is_serial_operator(); })) {

        // If non-serial operators exist, we should improve parallelism for those.

        return true;

    }

    if (auto local_exchange_source =

                std::dynamic_pointer_cast<LocalExchangeSourceOperatorX>(_operators.front());

        local_exchange_source && is_hash_exchange(target_data_distribution.distribution_type)) {

        const auto source_exchange_type = local_exchange_source->exchange_type();

        if (source_exchange_type != TLocalPartitionType::NOOP &&

            !is_hash_exchange(source_exchange_type)) {

            return true;

        }

    }

    if (!is_hash_exchange(target_data_distribution.distribution_type)) {

        // Always do local exchange if non-hash-partition exchanger is required.

        // For example, `PASSTHROUGH` exchanger is always required to distribute data evenly.

        return true;

    } else if (_operators.front()->is_serial_operator()) {

        DCHECK(std::all_of(_operators.begin(), _operators.end(),

                           [&](OperatorPtr op) -> bool { return op->is_serial_operator(); }) &&

               _sink->is_serial_operator())

                << debug_string();

        // All operators and sink are serial in this path.

        return false;

    } else {

        return _data_distribution.distribution_type != target_data_distribution.distribution_type &&

               !(is_hash_exchange(_data_distribution.distribution_type) &&

                 is_hash_exchange(target_data_distribution.distribution_type));

    }

}

Status Pipeline::add_operator(OperatorPtr& op, const int parallelism) {

    if (parallelism > 0 && op->is_serial_operator()) {

        set_num_tasks(parallelism);

    }

    op->set_parallel_tasks(num_tasks());

    _operators.emplace_back(op);

    if (op->is_source()) {

        std::reverse(_operators.begin(), _operators.end());

    }

    return Status::OK();

}

Status Pipeline::prepare(RuntimeState* state) {

    RETURN_IF_ERROR(_operators.back()->prepare(state));

    RETURN_IF_ERROR(_sink->prepare(state));

    _name.append(std::to_string(id()));

    _name.push_back('-');

    for (auto& op : _operators) {

        _name.append(std::to_string(op->node_id()));

        _name.append(op->get_name());

    }

    _name.push_back('-');

    _name.append(std::to_string(_sink->node_id()));

    _name.append(_sink->get_name());

    return Status::OK();

}

Status Pipeline::set_sink(DataSinkOperatorPtr& sink) {

    if (_sink) {

        return Status::InternalError("set sink twice");

    }

    if (!sink->is_sink()) {

        return Status::InternalError("should set a sink operator but {}", typeid(sink).name());

    }

    _sink = sink;

    return Status::OK();

}

void Pipeline::make_all_runnable(PipelineId wake_by) {

    DBUG_EXECUTE_IF("Pipeline::make_all_runnable.sleep", {

        auto pipeline_id = DebugPoints::instance()->get_debug_param_or_default<int32_t>(

                "Pipeline::make_all_runnable.sleep", "pipeline_id", -1);

        if (pipeline_id == id()) {

            LOG(WARNING) << "Pipeline::make_all_runnable.sleep sleep 10s";

            sleep(10);

        }

    });

    if (_sink->count_down_destination()) {

        for (auto* task : _tasks) {

            if (task) {

                task->set_wake_up_early(wake_by);

                task->unblock_all_dependencies();

            }

        }

    }

}

} // namespace doris