be/src/io/file_factory.h

// 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/PlanNodes_types.h>
#include <gen_cpp/Types_types.h>
#include <glog/logging.h>
#include <stdint.h>

#include <map>
#include <memory>
#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;
};

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};

    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) {}
};

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;
    // modification time of this file.
    // 0 means unset.
    int64_t mtime = 0;
    // 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(RuntimeState* state,
                                                    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
    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);

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