// 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 "runtime/plugin/cloud_plugin_downloader.h"

#include <fmt/format.h>

#include "cloud/cloud_storage_engine.h"

#include "io/fs/local_file_system.h"

#include "io/fs/remote_file_system.h"

#include "runtime/exec_env.h"

namespace doris {

// Use 10MB buffer for all downloads - same as cloud_warm_up_manager

static constexpr size_t DOWNLOAD_BUFFER_SIZE = 10 * 1024 * 1024; // 10MB

// Static mutex definition for synchronizing downloads

std::mutex CloudPluginDownloader::_download_mutex;

Status CloudPluginDownloader::download_from_cloud(PluginType type, const std::string& name,

const std::string& local_path,

std::string* result_path) {

// Use lock_guard to synchronize concurrent downloads

std::lock_guard<std::mutex> lock(_download_mutex);

if (name.empty()) {

return Status::InvalidArgument("Plugin name cannot be empty");

}

CloudPluginDownloader downloader;

// 1. Get FileSystem

io::RemoteFileSystemSPtr filesystem;

RETURN_IF_ERROR(downloader._get_cloud_filesystem(&filesystem));

// 2. Build remote plugin path

std::string remote_path;

RETURN_IF_ERROR(downloader._build_plugin_path(type, name, &remote_path));

LOG(INFO) << "Downloading plugin: " << name << " -> " << local_path;

// 3. Prepare local environment

RETURN_IF_ERROR(downloader._prepare_local_path(local_path));

// 4. Download remote file to local

RETURN_IF_ERROR(downloader._download_remote_file(filesystem, remote_path, local_path));

*result_path = local_path;

LOG(INFO) << "Successfully downloaded plugin: " << name << " to " << local_path;

return Status::OK();

}

Status CloudPluginDownloader::_build_plugin_path(PluginType type, const std::string& name,

std::string* path) {

std::string type_name;

switch (type) {

case PluginType::JDBC_DRIVERS:

type_name = "jdbc_drivers";

break;

case PluginType::JAVA_UDF:

type_name = "java_udf";

break;

default:

return Status::InvalidArgument("Unsupported plugin type: {}", static_cast<int>(type));

}

*path = fmt::format("plugins/{}/{}", type_name, name);

return Status::OK();

}

Status CloudPluginDownloader::_get_cloud_filesystem(io::RemoteFileSystemSPtr* filesystem) {

BaseStorageEngine& base_engine = ExecEnv::GetInstance()->storage_engine();

CloudStorageEngine* cloud_engine = dynamic_cast<CloudStorageEngine*>(&base_engine);

if (!cloud_engine) {

return Status::NotFound("CloudStorageEngine not found, not in cloud mode");

}

*filesystem = cloud_engine->latest_fs();

if (!*filesystem) {

return Status::NotFound("No latest filesystem available in cloud mode");

}

return Status::OK();

}

Status CloudPluginDownloader::_prepare_local_path(const std::string& local_path) {

// Remove existing file if present

bool exists = false;

RETURN_IF_ERROR(io::global_local_filesystem()->exists(local_path, &exists));

if (exists) {

RETURN_IF_ERROR(io::global_local_filesystem()->delete_file(local_path));

LOG(INFO) << "Removed existing file: " << local_path;

}

// Ensure local directory exists

std::string dir_path = local_path.substr(0, local_path.find_last_of('/'));

if (!dir_path.empty()) {

RETURN_IF_ERROR(io::global_local_filesystem()->create_directory(dir_path));

}

return Status::OK();

}

Status CloudPluginDownloader::_download_remote_file(io::RemoteFileSystemSPtr filesystem,

const std::string& remote_path,

const std::string& local_path) {

// Open remote file for reading

io::FileReaderSPtr remote_reader;

io::FileReaderOptions opts;

RETURN_IF_ERROR(filesystem->open_file(remote_path, &remote_reader, &opts));

// Get file size

int64_t file_size;

RETURN_IF_ERROR(filesystem->file_size(remote_path, &file_size));

// Create local file writer

io::FileWriterPtr local_writer;

RETURN_IF_ERROR(io::global_local_filesystem()->create_file(local_path, &local_writer));

auto buffer = std::make_unique<char[]>(DOWNLOAD_BUFFER_SIZE);

size_t total_read = 0;

while (total_read < file_size) {

size_t to_read =

std::min(DOWNLOAD_BUFFER_SIZE, static_cast<size_t>(file_size - total_read));

size_t bytes_read;

RETURN_IF_ERROR(remote_reader->read_at(total_read, {buffer.get(), to_read}, &bytes_read));

RETURN_IF_ERROR(local_writer->append({buffer.get(), bytes_read}));

total_read += bytes_read;

}

RETURN_IF_ERROR(local_writer->close());

return Status::OK();

}

} // namespace doris