// 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.

#pragma once

#include <gen_cpp/PaloInternalService_types.h>

#include <gen_cpp/PlanNodes_types.h>

#include <gen_cpp/Types_types.h>

#include <glog/logging.h>

#include <stdint.h>

#include <map>

#include <memory>

#include <optional>

#include <string>

#include "common/factory_creator.h"

#include "common/status.h"

#include "io/fs/file_reader.h"

#include "io/fs/file_reader_writer_fwd.h"

#include "io/fs/file_system.h"

namespace doris {

namespace io {

class FileSystem;

class FileWriter;

struct FileSystemProperties {

    TFileType::type system_type;

    std::map<std::string, std::string> properties;

    THdfsParams hdfs_params;

    std::vector<TNetworkAddress> broker_addresses;

    // Typed native Azure access for FILE_S3, copied from TFileScanRangeParams. When set,

    // FileFactory builds the Azure client from it and ignores the S3-style Azure keys in

    // `properties`.

    std::optional<TAzureClientConfig> azure_client_config {};

};

// The typed native Azure access a scan carries, if FE sent one

// (TFileScanRangeParams.azure_client_config); every reader that builds FileSystemProperties from

// scan parameters copies it through this.

inline std::optional<TAzureClientConfig> azure_client_config_of(

        const TFileScanRangeParams& params) {

    return params.__isset.azure_client_config

                   ? std::optional<TAzureClientConfig>(params.azure_client_config)

                   : std::nullopt;

}

struct FSPropertiesRef {

    TFileType::type type;

    const std::map<std::string, std::string>* properties {nullptr};

    const THdfsParams* hdfs_params {nullptr};

    const std::vector<TNetworkAddress>* broker_addresses {nullptr};

    // Typed native Azure access for FILE_S3; see FileSystemProperties::azure_client_config.

    const TAzureClientConfig* azure_client_config {nullptr};

    FSPropertiesRef(TFileType::type type_) : type(type_) {}

    FSPropertiesRef(const FileSystemProperties& fs_properties)

            : type(fs_properties.system_type),

              properties(&fs_properties.properties),

              hdfs_params(&fs_properties.hdfs_params),

              broker_addresses(&fs_properties.broker_addresses),

              azure_client_config(fs_properties.azure_client_config.has_value()

                                          ? &*fs_properties.azure_client_config

                                          : nullptr) {}

};

struct FileDescription {

    std::string path;

    // length of the file in bytes.

    // -1 means unset.

    // If the file length is not set, the file length will be fetched from the file system.

    int64_t file_size = -1;

    int64_t range_start_offset = 0;

    int64_t range_size = -1;

    // modification time of this file.

    // 0 means unset.

    int64_t mtime = 0;

    // A correctness promise that the bytes at this path never change. This is only used when

    // mtime is unavailable: immutable table formats may still key process-global caches by

    // fs/path/size, while mutable files must not do so because the same path and size can refer to

    // different contents after an overwrite. Generic Hive/local files must keep the default false.

    bool is_immutable = false;

    // for hdfs, eg: hdfs://nameservices1/

    // because for a hive table, differenet partitions may have different

    // locations(or fs), so different files may have different fs.

    std::string fs_name;

    bool file_cache_admission = true;

};

} // namespace io

class ExecEnv;

class RuntimeProfile;

class RuntimeState;

class FileFactory {

    ENABLE_FACTORY_CREATOR(FileFactory);

public:

    static io::FileReaderOptions get_reader_options(const TQueryOptions& option,

                                                    const io::FileDescription& fd);

    /// Create a temporary FileSystem for accessing file corresponding to `file_description`

    /// FIXME(plat1ko): Declare the path formats supported by each file system

    static Result<io::FileSystemSPtr> create_fs(const io::FSPropertiesRef& fs_properties,

                                                const io::FileDescription& file_description);

    /// Create FileWriter without FS. A non-null azure_client_config selects typed Azure access

    /// for FILE_S3, as in create_fs.

    static Result<io::FileWriterPtr> create_file_writer(

            TFileType::type type, ExecEnv* env,

            const std::vector<TNetworkAddress>& broker_addresses,

            const std::map<std::string, std::string>& properties, const std::string& path,

            const io::FileWriterOptions& options,

            const TAzureClientConfig* azure_client_config = nullptr);

    /// Create FileReader without FS

    static Result<io::FileReaderSPtr> create_file_reader(

            const io::FileSystemProperties& system_properties,

            const io::FileDescription& file_description,

            const io::FileReaderOptions& reader_options, RuntimeProfile* profile = nullptr);

    // Create FileReader for stream load pipe

    static Status create_pipe_reader(const TUniqueId& load_id, io::FileReaderSPtr* file_reader,

                                     RuntimeState* runtime_state, bool need_schema);

    static Result<TFileType::type> convert_storage_type(TStorageBackendType::type type) {

        switch (type) {

        case TStorageBackendType::LOCAL:

            return TFileType::FILE_LOCAL;

        case TStorageBackendType::S3:

            return TFileType::FILE_S3;

        case TStorageBackendType::AZURE:

            return TFileType::FILE_S3;

        case TStorageBackendType::BROKER:

            return TFileType::FILE_BROKER;

        case TStorageBackendType::HDFS:

        case TStorageBackendType::JFS:

            return TFileType::FILE_HDFS;

        default:

            return ResultError(Status::FatalError("not match type to convert, from type:{}", type));

        }

    }

private:

    static std::string _get_fs_name(const io::FileDescription& file_description);

    /// Create FileReader without FS

    static Result<io::FileReaderSPtr> _create_file_reader_internal(

            const io::FileSystemProperties& system_properties,

            const io::FileDescription& file_description,

            const io::FileReaderOptions& reader_options, RuntimeProfile* profile = nullptr);

};

} // namespace doris