/*
 * 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 "GlutenDiskHDFS.h"
#include <ranges>
#include <Disks/ObjectStorages/CompactObjectStorageDiskTransaction.h>
#include <Common/QueryContext.h>
#include <Common/Throttler.h>

#if USE_HDFS

namespace local_engine
{
using namespace DB;

DiskTransactionPtr GlutenDiskHDFS::createTransaction()
{
    return std::make_shared<CompactObjectStorageDiskTransaction>(*this, QueryContext::globalContext()->getTempDataOnDisk());
}

void GlutenDiskHDFS::createDirectory(const String & path)
{
    DiskObjectStorage::createDirectory(path);
    hdfsCreateDirectory(hdfs_object_storage->getHDFSFS(), path.c_str());
}

void GlutenDiskHDFS::createDirectories(const String & path)
{
    DiskObjectStorage::createDirectories(path);
    auto * hdfs = hdfs_object_storage->getHDFSFS();
    fs::path p = "/" + path;
    std::vector<std::string> paths_created;
    while (hdfsExists(hdfs, p.c_str()) < 0)
    {
        paths_created.push_back(p);
        if (!p.has_parent_path())
            break;
        p = p.parent_path();
    }
    for (const auto & path_to_create : paths_created | std::views::reverse)
        hdfsCreateDirectory(hdfs, path_to_create.c_str());
}

void GlutenDiskHDFS::removeDirectory(const String & path)
{
    DiskObjectStorage::removeDirectory(path);
    String abs_path = "/" + path;
    hdfsDelete(hdfs_object_storage->getHDFSFS(), abs_path.c_str(), 1);
}

void GlutenDiskHDFS::removeRecursive(const String & path)
{
    DiskObjectStorage::removeRecursive(path);
    String abs_path = "/" + path;
    hdfsDelete(hdfs_object_storage->getHDFSFS(), abs_path.c_str(), 1);
}

DiskObjectStoragePtr GlutenDiskHDFS::createDiskObjectStorage()
{
    const auto config_prefix = "storage_configuration.disks." + name;
    return std::make_shared<GlutenDiskHDFS>(
        getName(),
        object_key_prefix,
        getMetadataStorage(),
        getObjectStorage(),
        QueryContext::globalContext()->getConfigRef(),
        config_prefix,
        object_storage_creator);
}

std::unique_ptr<DB::WriteBufferFromFileBase> GlutenDiskHDFS::writeFile(
    const String & path,
    size_t buf_size,
    DB::WriteMode mode,
    const DB::WriteSettings & settings)
{
    if (throttler)
        throttler->add(1);
    return DiskObjectStorage::writeFile(path, buf_size, mode, settings);
}
}
#endif