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

#include <new>
#include <ostream>
#include <utility>
#include <vector>

#include "common/logging.h"
#include "core/block/block.h"
#include "exprs/vexpr_context.h"
#include "format/table/jdbc_jni_reader.h"
#include "runtime/descriptors.h"
#include "runtime/runtime_profile.h"
#include "runtime/runtime_state.h"
#include "util/jdbc_utils.h"

namespace doris {

JdbcScanner::JdbcScanner(RuntimeState* state, doris::JDBCScanLocalState* local_state, int64_t limit,
                         const TupleId& tuple_id, const std::string& query_string,
                         TOdbcTableType::type table_type, bool is_tvf, RuntimeProfile* profile)
        : Scanner(state, local_state, limit, profile),
          _jdbc_eos(false),
          _tuple_id(tuple_id),
          _query_string(query_string),
          _tuple_desc(nullptr),
          _table_type(table_type),
          _is_tvf(is_tvf) {
    _has_prepared = false;
}

std::map<std::string, std::string> JdbcScanner::_build_jdbc_params(
        const TupleDescriptor* tuple_desc) {
    const JdbcTableDescriptor* jdbc_table =
            static_cast<const JdbcTableDescriptor*>(tuple_desc->table_desc());

    std::map<std::string, std::string> params;
    params["jdbc_url"] = jdbc_table->jdbc_url();
    params["jdbc_user"] = jdbc_table->jdbc_user();
    params["jdbc_password"] = jdbc_table->jdbc_passwd();
    params["jdbc_driver_class"] = jdbc_table->jdbc_driver_class();
    // Resolve jdbc_driver_url to absolute file:// URL
    // FE sends just the JAR filename; we need to resolve it to a full path.
    std::string driver_url;
    auto resolve_st = JdbcUtils::resolve_driver_url(jdbc_table->jdbc_driver_url(), &driver_url);
    if (!resolve_st.ok()) {
        LOG(WARNING) << "Failed to resolve JDBC driver URL: " << resolve_st.to_string();
        driver_url = jdbc_table->jdbc_driver_url();
    }
    params["jdbc_driver_url"] = driver_url;
    params["jdbc_driver_checksum"] = jdbc_table->jdbc_driver_checksum();
    params["query_sql"] = _query_string;
    params["catalog_id"] = std::to_string(jdbc_table->jdbc_catalog_id());
    params["table_type"] = _odbc_table_type_to_string(_table_type);
    params["connection_pool_min_size"] = std::to_string(jdbc_table->connection_pool_min_size());
    params["connection_pool_max_size"] = std::to_string(jdbc_table->connection_pool_max_size());
    params["connection_pool_max_wait_time"] =
            std::to_string(jdbc_table->connection_pool_max_wait_time());
    params["connection_pool_max_life_time"] =
            std::to_string(jdbc_table->connection_pool_max_life_time());
    params["connection_pool_keep_alive"] =
            jdbc_table->connection_pool_keep_alive() ? "true" : "false";
    return params;
}

Status JdbcScanner::init(RuntimeState* state, const VExprContextSPtrs& conjuncts) {
    VLOG_CRITICAL << "JdbcScanner::init";
    RETURN_IF_ERROR(Scanner::init(state, conjuncts));

    if (state == nullptr) {
        return Status::InternalError("input pointer is NULL of JdbcScanner::init.");
    }

    // get tuple desc
    _tuple_desc = state->desc_tbl().get_tuple_descriptor(_tuple_id);
    if (_tuple_desc == nullptr) {
        return Status::InternalError("Failed to get tuple descriptor.");
    }

    // get jdbc table info
    const JdbcTableDescriptor* jdbc_table =
            static_cast<const JdbcTableDescriptor*>(_tuple_desc->table_desc());
    if (jdbc_table == nullptr) {
        return Status::InternalError("jdbc table pointer is NULL of JdbcScanner::init.");
    }

    _local_state->scanner_profile()->add_info_string("JdbcDriverClass",
                                                     jdbc_table->jdbc_driver_class());
    _local_state->scanner_profile()->add_info_string("JdbcDriverUrl",
                                                     jdbc_table->jdbc_driver_url());
    _local_state->scanner_profile()->add_info_string("JdbcUrl", jdbc_table->jdbc_url());
    _local_state->scanner_profile()->add_info_string("QuerySql", _query_string);

    // Build reader params from tuple descriptor
    auto jdbc_params = _build_jdbc_params(_tuple_desc);

    // Pass _tuple_desc->slots() directly. JniReader stores _file_slot_descs as a reference,
    // so we must pass a vector whose lifetime outlives the reader (i.e., _tuple_desc->slots()).
    // Previously a local vector was passed, causing a dangling reference after init() returned.
    _jni_reader = JdbcJniReader::create_unique(_tuple_desc->slots(), state, _profile, jdbc_params);

    return Status::OK();
}

Status JdbcScanner::_open_impl(RuntimeState* state) {
    VLOG_CRITICAL << "JdbcScanner::open";
    if (state == nullptr) {
        return Status::InternalError("input pointer is NULL of JdbcScanner::open.");
    }

    if (!_has_prepared) {
        return Status::InternalError("used before initialize of JdbcScanner::open.");
    }
    RETURN_IF_CANCELLED(state);
    RETURN_IF_ERROR(Scanner::_open_impl(state));
    RETURN_IF_ERROR(_jni_reader->init_reader());
    return Status::OK();
}

Status JdbcScanner::_get_block_impl(RuntimeState* state, Block* block, bool* eof) {
    VLOG_CRITICAL << "JdbcScanner::_get_block_impl";
    if (nullptr == state || nullptr == block || nullptr == eof) {
        return Status::InternalError("input is NULL pointer");
    }

    if (!_has_prepared) {
        return Status::InternalError("used before initialize of JdbcScanner::get_next.");
    }

    if (_jdbc_eos) {
        *eof = true;
        return Status::OK();
    }

    // only empty block should be here
    DCHECK(block->rows() == 0);

    do {
        RETURN_IF_CANCELLED(state);

        size_t read_rows = 0;
        bool reader_eof = false;
        RETURN_IF_ERROR(_jni_reader->get_next_block(block, &read_rows, &reader_eof));

        if (reader_eof) {
            _jdbc_eos = true;
            if (block->rows() == 0) {
                *eof = true;
            }
            break;
        }

        VLOG_ROW << "JdbcScanner output rows: " << block->rows();
    } while (block->rows() == 0 && !(*eof));
    return Status::OK();
}

Status JdbcScanner::close(RuntimeState* state) {
    if (!_try_close()) {
        return Status::OK();
    }
    RETURN_IF_ERROR(Scanner::close(state));
    if (_jni_reader) {
        RETURN_IF_ERROR(_jni_reader->close());
    }
    return Status::OK();
}
} // namespace doris