// 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 "io/fs/hdfs_file_system.h"

#include <errno.h>

#include <fcntl.h>

#include <gen_cpp/PlanNodes_types.h>

#include <algorithm>

#include <filesystem>

#include <map>

#include <mutex>

#include <ostream>

#include <unordered_map>

#include <utility>

#include "common/config.h"

#include "common/status.h"

#include "io/fs/err_utils.h"

#include "io/fs/hdfs/hdfs_mgr.h"

#include "io/fs/hdfs_file_reader.h"

#include "io/fs/hdfs_file_writer.h"

#include "io/fs/local_file_system.h"

#include "io/hdfs_builder.h"

#include "io/hdfs_util.h"

#include "runtime/exec_env.h"

#include "util/obj_lru_cache.h"

#include "util/slice.h"

namespace doris::io {

#ifndef CHECK_HDFS_HANDLER

#define CHECK_HDFS_HANDLER(handler)                        \

    if (!handler) {                                        \

        return Status::IOError("init Hdfs handler error"); \

    }

#endif

Result<std::shared_ptr<HdfsFileSystem>> HdfsFileSystem::create(

        const std::map<std::string, std::string>& properties, std::string fs_name, std::string id,

        RuntimeProfile* profile, std::string root_path) {

    return HdfsFileSystem::create(parse_properties(properties), std::move(fs_name), std::move(id),

                                  profile, std::move(root_path));

}

Result<std::shared_ptr<HdfsFileSystem>> HdfsFileSystem::create(const THdfsParams& hdfs_params,

                                                               std::string fs_name, std::string id,

                                                               RuntimeProfile* profile,

                                                               std::string root_path) {

#ifdef USE_HADOOP_HDFS

    if (!config::enable_java_support) {

        return ResultError(Status::InternalError(

                "hdfs file system is not enabled, you can change be config enable_java_support to "

                "true."));

    }

#endif

    std::shared_ptr<HdfsFileSystem> fs(new HdfsFileSystem(

            hdfs_params, std::move(fs_name), std::move(id), profile, std::move(root_path)));

    RETURN_IF_ERROR_RESULT(fs->init());

    return fs;

}

HdfsFileSystem::HdfsFileSystem(const THdfsParams& hdfs_params, std::string fs_name, std::string id,

                               RuntimeProfile* profile, std::string root_path)

        : RemoteFileSystem(std::move(root_path), std::move(id), FileSystemType::HDFS),

          _hdfs_params(hdfs_params),

          _fs_name(std::move(fs_name)),

          _profile(profile) {

    if (_fs_name.empty()) {

        _fs_name = hdfs_params.fs_name;

    }

}

HdfsFileSystem::~HdfsFileSystem() = default;

Status HdfsFileSystem::init() {

    RETURN_IF_ERROR(ExecEnv::GetInstance()->hdfs_mgr()->get_or_create_fs(_hdfs_params, _fs_name,

                                                                         &_fs_handler));

    if (!_fs_handler) {

        return Status::InternalError("failed to init Hdfs handler with, please check hdfs params.");

    }

    return Status::OK();

}

Status HdfsFileSystem::create_file_impl(const Path& file, FileWriterPtr* writer,

                                        const FileWriterOptions* opts) {

    auto res = io::HdfsFileWriter::create(file, _fs_handler, _fs_name, opts);

    if (res.has_value()) {

        *writer = std::move(res).value();

        return Status::OK();

    } else {

        return std::move(res).error();

    }

}

Status HdfsFileSystem::open_file_internal(const Path& file, FileReaderSPtr* reader,

                                          const FileReaderOptions& opts) {

    CHECK_HDFS_HANDLER(_fs_handler);

    *reader =

            DORIS_TRY(HdfsFileReader::create(file, _fs_handler->hdfs_fs, _fs_name, opts, _profile));

    return Status::OK();

}

Status HdfsFileSystem::create_directory_impl(const Path& dir, bool failed_if_exists) {

    CHECK_HDFS_HANDLER(_fs_handler);

    Path real_path = convert_path(dir, _fs_name);

    int res = hdfsCreateDirectory(_fs_handler->hdfs_fs, real_path.string().c_str());

    if (res == -1) {

        return Status::IOError("failed to create directory {}: {}", dir.native(), hdfs_error());

    }

    return Status::OK();

}

Status HdfsFileSystem::delete_file_impl(const Path& file) {

    return delete_internal(file, 0);

}

Status HdfsFileSystem::delete_directory_impl(const Path& dir) {

    return delete_internal(dir, 1);

}

Status HdfsFileSystem::batch_delete_impl(const std::vector<Path>& files) {

    for (auto& file : files) {

        RETURN_IF_ERROR(delete_file_impl(file));

    }

    return Status::OK();

}

Status HdfsFileSystem::delete_internal(const Path& path, int is_recursive) {

    bool exists = true;

    RETURN_IF_ERROR(exists_impl(path, &exists));

    if (!exists) {

        return Status::OK();

    }

    CHECK_HDFS_HANDLER(_fs_handler);

    Path real_path = convert_path(path, _fs_name);

    int res = hdfsDelete(_fs_handler->hdfs_fs, real_path.string().c_str(), is_recursive);

    if (res == -1) {

        return Status::IOError("failed to delete directory {}: {}", path.native(), hdfs_error());

    }

    return Status::OK();

}

Status HdfsFileSystem::exists_impl(const Path& path, bool* res) const {

    CHECK_HDFS_HANDLER(_fs_handler);

    Path real_path = convert_path(path, _fs_name);

    int is_exists = hdfsExists(_fs_handler->hdfs_fs, real_path.string().c_str());

#ifdef USE_HADOOP_HDFS

    // when calling hdfsExists() and return non-zero code,

    // if errno is ENOENT, which means the file does not exist.

    // if errno is not ENOENT, which means it encounter other error, should return.

    // NOTE: not for libhdfs3 since it only runs on MaxOS, don't have to support it.

    //

    // See details:

    //  https://github.com/apache/hadoop/blob/5cda162a804fb0cfc2a5ac0058ab407662c5fb00/

    //  hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfs/hdfs.c#L1923-L1924

    if (is_exists != 0 && errno != ENOENT) {

        char* root_cause = hdfsGetLastExceptionRootCause();

        return Status::IOError("failed to check path existence {}: {}", path.native(),

                               (root_cause ? root_cause : "unknown"));

    }

#endif

    *res = (is_exists == 0);

    return Status::OK();

}

Status HdfsFileSystem::file_size_impl(const Path& path, int64_t* file_size) const {

    CHECK_HDFS_HANDLER(_fs_handler);

    Path real_path = convert_path(path, _fs_name);

    hdfsFileInfo* file_info = hdfsGetPathInfo(_fs_handler->hdfs_fs, real_path.string().c_str());

    if (file_info == nullptr) {

        return Status::IOError("failed to get file size of {}: {}", path.native(), hdfs_error());

    }

    *file_size = file_info->mSize;

    hdfsFreeFileInfo(file_info, 1);

    return Status::OK();

}

Status HdfsFileSystem::list_impl(const Path& path, bool only_file, std::vector<FileInfo>* files,

                                 bool* exists) {

    RETURN_IF_ERROR(exists_impl(path, exists));

    if (!(*exists)) {

        return Status::OK();

    }

    CHECK_HDFS_HANDLER(_fs_handler);

    Path real_path = convert_path(path, _fs_name);

    int numEntries = 0;

    hdfsFileInfo* hdfs_file_info =

            hdfsListDirectory(_fs_handler->hdfs_fs, real_path.c_str(), &numEntries);

    if (hdfs_file_info == nullptr) {

        return Status::IOError("failed to list files/directors {}: {}", path.native(),

                               hdfs_error());

    }

    for (int idx = 0; idx < numEntries; ++idx) {

        auto& file = hdfs_file_info[idx];

        if (only_file && file.mKind == kObjectKindDirectory) {

            continue;

        }

        auto& file_info = files->emplace_back();

        std::string_view fname(file.mName);

        fname.remove_prefix(fname.rfind('/') + 1);

        file_info.file_name = fname;

        file_info.file_size = file.mSize;

        file_info.is_file = (file.mKind != kObjectKindDirectory);

    }

    hdfsFreeFileInfo(hdfs_file_info, numEntries);

    return Status::OK();

}

Status HdfsFileSystem::rename_impl(const Path& orig_name, const Path& new_name) {

    Path normal_orig_name = convert_path(orig_name, _fs_name);

    Path normal_new_name = convert_path(new_name, _fs_name);

    int ret = hdfsRename(_fs_handler->hdfs_fs, normal_orig_name.c_str(), normal_new_name.c_str());

    if (ret == 0) {

        LOG(INFO) << "finished to rename file. orig: " << normal_orig_name

                  << ", new: " << normal_new_name;

        return Status::OK();

    } else {

        return Status::IOError("fail to rename from {} to {}: {}", normal_orig_name.native(),

                               normal_new_name.native(), hdfs_error());

    }

    return Status::OK();

}

Status HdfsFileSystem::upload_impl(const Path& local_file, const Path& remote_file) {

    // 1. open local file for read

    FileSystemSPtr local_fs = global_local_filesystem();

    FileReaderSPtr local_reader = nullptr;

    RETURN_IF_ERROR(local_fs->open_file(local_file, &local_reader));

    int64_t file_len = local_reader->size();

    if (file_len == -1) {

        return Status::IOError("failed to get size of file: {}", local_file.string());

    }

    // 2. open remote file for write

    FileWriterPtr hdfs_writer = nullptr;

    RETURN_IF_ERROR(create_file_impl(remote_file, &hdfs_writer, nullptr));

    constexpr size_t buf_sz = 1024 * 1024;

    char read_buf[buf_sz];

    size_t left_len = file_len;

    size_t read_offset = 0;

    size_t bytes_read = 0;

    while (left_len > 0) {

        size_t read_len = left_len > buf_sz ? buf_sz : left_len;

        RETURN_IF_ERROR(local_reader->read_at(read_offset, {read_buf, read_len}, &bytes_read));

        RETURN_IF_ERROR(hdfs_writer->append({read_buf, read_len}));

        read_offset += read_len;

        left_len -= read_len;

    }

    return hdfs_writer->close();

}

Status HdfsFileSystem::batch_upload_impl(const std::vector<Path>& local_files,

                                         const std::vector<Path>& remote_files) {

    DCHECK(local_files.size() == remote_files.size());

    for (int i = 0; i < local_files.size(); ++i) {

        RETURN_IF_ERROR(upload_impl(local_files[i], remote_files[i]));

    }

    return Status::OK();

}

Status HdfsFileSystem::download_impl(const Path& remote_file, const Path& local_file) {

    // 1. open remote file for read

    FileReaderSPtr hdfs_reader = nullptr;

    RETURN_IF_ERROR(open_file_internal(remote_file, &hdfs_reader, FileReaderOptions::DEFAULT));

    // 2. remove the existing local file if exist

    if (std::filesystem::remove(local_file)) {

        LOG(INFO) << "remove the previously exist local file: " << local_file;

    }

    // 3. open local file for write

    FileSystemSPtr local_fs = global_local_filesystem();

    FileWriterPtr local_writer = nullptr;

    RETURN_IF_ERROR(local_fs->create_file(local_file, &local_writer));

    // 4. read remote and write to local

    LOG(INFO) << "read remote file: " << remote_file << " to local: " << local_file;

    constexpr size_t buf_sz = 1024 * 1024;

    std::unique_ptr<char[]> read_buf(new char[buf_sz]);

    size_t cur_offset = 0;

    while (true) {

        size_t read_len = 0;

        Slice file_slice(read_buf.get(), buf_sz);

        RETURN_IF_ERROR(hdfs_reader->read_at(cur_offset, file_slice, &read_len));

        cur_offset += read_len;

        if (read_len == 0) {

            break;

        }

        RETURN_IF_ERROR(local_writer->append({read_buf.get(), read_len}));

    }

    return local_writer->close();

}

} // namespace doris::io