// 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 "exprs/function_context.h"

#include <iostream>

#include <utility>

// Be careful what this includes since this needs to be linked into UDF's

// binary. For example, it would be unfortunate if they had a random dependency

// on libhdfs.

#include "common/cast_set.h"

#include "core/data_type/data_type.h"

#include "core/string_ref.h"

#include "runtime/runtime_state.h"

namespace doris {

static const int MAX_WARNINGS = 1000;

std::unique_ptr<doris::FunctionContext> FunctionContext::create_context(

        RuntimeState* state, const DataTypePtr& return_type,

        const std::vector<DataTypePtr>& arg_types) {

    auto ctx = std::unique_ptr<doris::FunctionContext>(new doris::FunctionContext());

    ctx->_state = state;

    ctx->_return_type = return_type;

    ctx->_arg_types = arg_types;

    ctx->_num_warnings = 0;

    ctx->_thread_local_fn_state = nullptr;

    ctx->_fragment_local_fn_state = nullptr;

    return ctx;

}

void FunctionContext::set_constant_cols(

        const std::vector<std::shared_ptr<doris::ColumnPtrWrapper>>& constant_cols) {

    _constant_cols = constant_cols;

}

std::unique_ptr<FunctionContext> FunctionContext::clone() {

    auto new_context = create_context(_state, _return_type, _arg_types);

    new_context->_constant_cols = _constant_cols;

    new_context->_fragment_local_fn_state = _fragment_local_fn_state;

    new_context->_check_overflow_for_decimal = _check_overflow_for_decimal;

    new_context->_enable_strict_mode = _enable_strict_mode;

    new_context->_string_as_jsonb_string = _string_as_jsonb_string;

    new_context->_jsonb_string_as_string = _jsonb_string_as_string;

    return new_context;

}

void FunctionContext::set_function_state(FunctionStateScope scope, std::shared_ptr<void> ptr) {

    switch (scope) {

    case THREAD_LOCAL:

        _thread_local_fn_state = std::move(ptr);

        break;

    case FRAGMENT_LOCAL:

        _fragment_local_fn_state = std::move(ptr);

        break;

    default:

        std::stringstream ss;

        ss << "Unknown FunctionStateScope: " << scope;

        set_error(ss.str().c_str());

    }

}

void FunctionContext::set_error(const char* error_msg) {

    if (_error_msg.empty()) {

        _error_msg = error_msg;

        std::stringstream ss;

        ss << "UDF ERROR: " << error_msg;

        if (_state != nullptr) {

            _state->cancel(Status::InternalError(ss.str()));

        }

    }

}

bool FunctionContext::add_warning(const char* warning_msg) {

    if (_num_warnings++ >= MAX_WARNINGS) {

        return false;

    }

    std::stringstream ss;

    ss << "UDF WARNING: " << warning_msg;

    if (_state != nullptr) {

        return _state->log_error(ss.str());

    } else {

        std::cerr << ss.str() << std::endl;

        return true;

    }

}

const DataTypePtr FunctionContext::get_arg_type(int arg_idx) const {

    if (arg_idx < 0 || arg_idx >= _arg_types.size()) {

        return nullptr;

    }

    return _arg_types[arg_idx];

}

bool FunctionContext::is_col_constant(int i) const {

    if (i < 0 || i >= _constant_cols.size()) {

        return false;

    }

    return _constant_cols[i] != nullptr;

}

doris::ColumnPtrWrapper* FunctionContext::get_constant_col(int i) const {

    if (i < 0 || i >= _constant_cols.size()) {

        return nullptr;

    }

    return _constant_cols[i].get();

}

int FunctionContext::get_num_args() const {

    return cast_set<int>(_arg_types.size());

}

const DataTypePtr FunctionContext::get_return_type() const {

    return _return_type;

}

void* FunctionContext::get_function_state(FunctionStateScope scope) const {

    switch (scope) {

    case THREAD_LOCAL:

        return _thread_local_fn_state.get();

    case FRAGMENT_LOCAL:

        return _fragment_local_fn_state.get();

    default:

        // TODO: signal error somehow

        return nullptr;

    }

}

StringRef FunctionContext::create_temp_string_val(int64_t len) {

    _string_result.resize(len);

    return StringRef((uint8_t*)_string_result.c_str(), len);

}

} // namespace doris