foreign_storage::ForeignDataWrapperFactory Class Reference

#include <ForeignDataWrapperFactory.h>

Static Public Member Functions
static std::unique_ptr < ForeignDataWrapper >	create (const std::string &data_wrapper_type, const int db_id, const ForeignTable *foreign_table)
static std::unique_ptr < UserMapping >	createUserMappingProxyIfApplicable (const int db_id, const int user_id, const std::string &file_path, const import_export::CopyParams &copy_params, const ForeignServer *server)
static std::unique_ptr < ForeignServer >	createForeignServerProxy (const int db_id, const int user_id, const std::string &file_path, const import_export::CopyParams &copy_params)
static std::unique_ptr < ForeignTable >	createForeignTableProxy (const int db_id, const TableDescriptor *table, const std::string &file_path, const import_export::CopyParams &copy_params, const ForeignServer *server)
static std::unique_ptr < ForeignDataWrapper >	createForImport (const std::string &data_wrapper_type, const int db_id, const ForeignTable *foreign_table, const UserMapping *user_mapping)
static std::unique_ptr < ForeignDataWrapper >	createForGeneralImport (const import_export::CopyParams &copy_params, const int db_id, const ForeignTable *foreign_table, const UserMapping *user_mapping)
static const ForeignDataWrapper *	createForValidation (const std::string &data_wrapper_type, const ForeignTable *foreign_table=nullptr)
static void	validateDataWrapperType (const std::string &data_wrapper_type)

Static Private Attributes
static std::map< std::string, std::unique_ptr < ForeignDataWrapper > >	validation_data_wrappers_

Detailed Description

Definition at line 107 of file ForeignDataWrapperFactory.h.

Member Function Documentation

std::unique_ptr< ForeignDataWrapper > foreign_storage::ForeignDataWrapperFactory::create ( const std::string &  data_wrapper_type, const int  db_id, const ForeignTable *  foreign_table  )

static

Creates an instance of a ForeignDataWrapper for the given data wrapper type using provided database and foreign table details.

Definition at line 343 of file ForeignDataWrapperFactory.cpp.

References foreign_storage::DataWrapperType::CSV, foreign_storage::DataWrapperType::INTERNAL_CATALOG, foreign_storage::DataWrapperType::INTERNAL_EXECUTOR_STATS, foreign_storage::DataWrapperType::INTERNAL_LOGS, foreign_storage::DataWrapperType::INTERNAL_MEMORY_STATS, foreign_storage::DataWrapperType::INTERNAL_ML_MODEL_METADATA, foreign_storage::DataWrapperType::INTERNAL_STORAGE_STATS, foreign_storage::DataWrapperType::PARQUET, foreign_storage::DataWrapperType::REGEX_PARSER, UNREACHABLE, and foreign_storage::CsvDataWrapper::validateAndGetIsS3Select().

                                       {
  std::unique_ptr<ForeignDataWrapper> data_wrapper;
  if (data_wrapper_type == DataWrapperType::CSV) {
    if (CsvDataWrapper::validateAndGetIsS3Select(foreign_table)) {
      UNREACHABLE();
    } else {
      data_wrapper = std::make_unique<CsvDataWrapper>(db_id, foreign_table);
    }
#ifdef ENABLE_IMPORT_PARQUET
  } else if (data_wrapper_type == DataWrapperType::PARQUET) {
    data_wrapper = std::make_unique<ParquetDataWrapper>(db_id, foreign_table);
#endif
  } else if (data_wrapper_type == DataWrapperType::REGEX_PARSER) {
    data_wrapper = std::make_unique<RegexParserDataWrapper>(db_id, foreign_table);
  } else if (data_wrapper_type == DataWrapperType::INTERNAL_CATALOG) {
    data_wrapper = std::make_unique<InternalCatalogDataWrapper>(db_id, foreign_table);
  } else if (data_wrapper_type == DataWrapperType::INTERNAL_EXECUTOR_STATS) {
    data_wrapper =
        std::make_unique<InternalExecutorStatsDataWrapper>(db_id, foreign_table);
  } else if (data_wrapper_type == DataWrapperType::INTERNAL_ML_MODEL_METADATA) {
    data_wrapper =
        std::make_unique<InternalMLModelMetadataDataWrapper>(db_id, foreign_table);
  } else if (data_wrapper_type == DataWrapperType::INTERNAL_MEMORY_STATS) {
    data_wrapper = std::make_unique<InternalMemoryStatsDataWrapper>(db_id, foreign_table);
  } else if (data_wrapper_type == DataWrapperType::INTERNAL_STORAGE_STATS) {
    data_wrapper =
        std::make_unique<InternalStorageStatsDataWrapper>(db_id, foreign_table);
  } else if (data_wrapper_type == DataWrapperType::INTERNAL_LOGS) {
    data_wrapper = std::make_unique<InternalLogsDataWrapper>(db_id, foreign_table);
  } else {
    throw std::runtime_error("Unsupported data wrapper");
  }
  return data_wrapper;
}

Here is the call graph for this function:

std::unique_ptr< ForeignServer > foreign_storage::ForeignDataWrapperFactory::createForeignServerProxy ( const int  db_id, const int  user_id, const std::string &  file_path, const import_export::CopyParams &  copy_params  )

static

Definition at line 184 of file ForeignDataWrapperFactory.cpp.

References CHECK, foreign_storage::DataWrapperType::CSV, shared::is_s3_uri(), foreign_storage::is_valid_source_type(), import_export::kDelimitedFile, import_export::kOdbc, import_export::kParquetFile, import_export::kRegexParsedFile, foreign_storage::AbstractFileStorageDataWrapper::LOCAL_FILE_STORAGE_TYPE, foreign_storage::DataWrapperType::PARQUET, foreign_storage::DataWrapperType::REGEX_PARSER, import_export::CopyParams::source_type, foreign_storage::AbstractFileStorageDataWrapper::STORAGE_TYPE_KEY, and UNREACHABLE.

Referenced by foreign_storage::create_proxy_fsi_objects().

                                                {
  CHECK(is_valid_source_type(copy_params));

  auto foreign_server = std::make_unique<foreign_storage::ForeignServer>();

  foreign_server->id = -1;
  foreign_server->user_id = user_id;
  if (copy_params.source_type == import_export::SourceType::kDelimitedFile) {
    foreign_server->data_wrapper_type = DataWrapperType::CSV;
  } else if (copy_params.source_type == import_export::SourceType::kRegexParsedFile) {
    foreign_server->data_wrapper_type = DataWrapperType::REGEX_PARSER;
#ifdef ENABLE_IMPORT_PARQUET
  } else if (copy_params.source_type == import_export::SourceType::kParquetFile) {
    foreign_server->data_wrapper_type = DataWrapperType::PARQUET;
#endif
  } else {
    UNREACHABLE();
  }
  foreign_server->name = "import_proxy_server";

  if (copy_params.source_type == import_export::SourceType::kOdbc) {
    throw std::runtime_error("ODBC storage not supported");
  } else if (shared::is_s3_uri(file_path)) {
    throw std::runtime_error("AWS storage not supported");
  } else {
    foreign_server->options[AbstractFileStorageDataWrapper::STORAGE_TYPE_KEY] =
        AbstractFileStorageDataWrapper::LOCAL_FILE_STORAGE_TYPE;
  }

  return foreign_server;
}

Here is the call graph for this function:

Here is the caller graph for this function:

std::unique_ptr< ForeignTable > foreign_storage::ForeignDataWrapperFactory::createForeignTableProxy ( const int  db_id, const TableDescriptor *  table, const std::string &  file_path, const import_export::CopyParams &  copy_params, const ForeignServer *  server  )

static

Definition at line 237 of file ForeignDataWrapperFactory.cpp.

References import_export::CopyParams::array_begin, import_export::CopyParams::array_delim, foreign_storage::CsvFileBufferParser::ARRAY_DELIMITER_KEY, import_export::CopyParams::array_end, foreign_storage::CsvFileBufferParser::ARRAY_MARKER_KEY, foreign_storage::bool_to_option_value(), import_export::CopyParams::buffer_size, foreign_storage::TextFileBufferParser::BUFFER_SIZE_KEY, CHECK, import_export::CopyParams::delimiter, foreign_storage::CsvFileBufferParser::DELIMITER_KEY, import_export::CopyParams::escape, foreign_storage::CsvFileBufferParser::ESCAPE_KEY, import_export::CopyParams::file_sort_order_by, foreign_storage::AbstractFileStorageDataWrapper::FILE_SORT_ORDER_BY_KEY, import_export::CopyParams::file_sort_regex, foreign_storage::AbstractFileStorageDataWrapper::FILE_SORT_REGEX_KEY, import_export::CopyParams::geo_explode_collections, foreign_storage::CsvFileBufferParser::GEO_EXPLODE_COLLECTIONS_KEY, import_export::CopyParams::geo_validate_geometry, foreign_storage::ForeignTable::GEO_VALIDATE_GEOMETRY_KEY, Catalog_Namespace::SysCatalog::getCatalog(), import_export::CopyParams::has_header, Catalog_Namespace::SysCatalog::instance(), shared::is_s3_uri(), foreign_storage::is_valid_source_type(), import_export::kAutoDetect, import_export::kDelimitedFile, import_export::kOdbc, import_export::kParquetFile, import_export::kRegexParsedFile, import_export::CopyParams::line_delim, foreign_storage::CsvFileBufferParser::LINE_DELIMITER_KEY, import_export::CopyParams::line_regex, foreign_storage::RegexFileBufferParser::LINE_REGEX_KEY, import_export::CopyParams::line_start_regex, foreign_storage::RegexFileBufferParser::LINE_START_REGEX_KEY, import_export::CopyParams::lonlat, foreign_storage::CsvFileBufferParser::LONLAT_KEY, import_export::CopyParams::null_str, foreign_storage::CsvFileBufferParser::NULLS_KEY, import_export::num_import_threads(), foreign_storage::OptionsContainer::options, import_export::CopyParams::quote, foreign_storage::CsvFileBufferParser::QUOTE_KEY, import_export::CopyParams::quoted, foreign_storage::CsvFileBufferParser::QUOTED_KEY, import_export::CopyParams::regex_path_filter, foreign_storage::AbstractFileStorageDataWrapper::REGEX_PATH_FILTER_KEY, foreign_storage::anonymous_namespace{ForeignDataWrapperFactory.cpp}::set_header_option(), import_export::CopyParams::source_srid, foreign_storage::CsvFileBufferParser::SOURCE_SRID_KEY, import_export::CopyParams::source_type, import_export::CopyParams::threads, foreign_storage::AbstractFileStorageDataWrapper::THREADS_KEY, to_string(), import_export::CopyParams::trim_spaces, and foreign_storage::CsvFileBufferParser::TRIM_SPACES_KEY.

Referenced by foreign_storage::create_proxy_fsi_objects().

                                 {
  CHECK(is_valid_source_type(copy_params));

  auto catalog = Catalog_Namespace::SysCatalog::instance().getCatalog(db_id);
  auto foreign_table = std::make_unique<ForeignTable>();

  *static_cast<TableDescriptor*>(foreign_table.get()) =
      *table;  // copy table related values

  CHECK(server);
  foreign_table->foreign_server = server;

  // enable geo validation in most/all source types
  if (copy_params.source_type == import_export::SourceType::kRegexParsedFile ||
      copy_params.source_type == import_export::SourceType::kDelimitedFile ||
      copy_params.source_type == import_export::SourceType::kParquetFile ||
      copy_params.source_type == import_export::SourceType::kOdbc) {
    foreign_table->options[ForeignTable::GEO_VALIDATE_GEOMETRY_KEY] =
        bool_to_option_value(copy_params.geo_validate_geometry);
  }

  // populate options for regex filtering of file-paths in supported data types
  if (copy_params.source_type == import_export::SourceType::kRegexParsedFile ||
      copy_params.source_type == import_export::SourceType::kDelimitedFile ||
      copy_params.source_type == import_export::SourceType::kParquetFile) {
    if (copy_params.regex_path_filter.has_value()) {
      foreign_table->options[AbstractFileStorageDataWrapper::REGEX_PATH_FILTER_KEY] =
          copy_params.regex_path_filter.value();
    }
    if (copy_params.file_sort_order_by.has_value()) {
      foreign_table->options[AbstractFileStorageDataWrapper::FILE_SORT_ORDER_BY_KEY] =
          copy_params.file_sort_order_by.value();
    }
    if (copy_params.file_sort_regex.has_value()) {
      foreign_table->options[AbstractFileStorageDataWrapper::FILE_SORT_REGEX_KEY] =
          copy_params.file_sort_regex.value();
    }
    foreign_table->options[AbstractFileStorageDataWrapper::THREADS_KEY] =
        std::to_string(import_export::num_import_threads(copy_params.threads));
  }

  if (copy_params.source_type == import_export::SourceType::kRegexParsedFile) {
    CHECK(!copy_params.line_regex.empty());
    foreign_table->options[RegexFileBufferParser::LINE_REGEX_KEY] =
        copy_params.line_regex;
    if (!copy_params.line_start_regex.empty()) {
      foreign_table->options[RegexFileBufferParser::LINE_START_REGEX_KEY] =
          copy_params.line_start_regex;
    }
    if (copy_params.has_header != import_export::ImportHeaderRow::kAutoDetect) {
      set_header_option(foreign_table->options, copy_params.has_header);
    }
  }

  // setup data source options based on various criteria
  if (copy_params.source_type == import_export::SourceType::kOdbc) {
    throw std::runtime_error("ODBC storage not supported");
  } else if (shared::is_s3_uri(copy_from_source)) {
    throw std::runtime_error("AWS storage not supported");
  } else {
    foreign_table->options["FILE_PATH"] = copy_from_source;
  }

  // for CSV import
  if (copy_params.source_type == import_export::SourceType::kDelimitedFile) {
    foreign_table->options[CsvFileBufferParser::DELIMITER_KEY] = copy_params.delimiter;
    foreign_table->options[CsvFileBufferParser::NULLS_KEY] = copy_params.null_str;
    set_header_option(foreign_table->options, copy_params.has_header);
    foreign_table->options[CsvFileBufferParser::QUOTED_KEY] =
        bool_to_option_value(copy_params.quoted);
    foreign_table->options[CsvFileBufferParser::QUOTE_KEY] = copy_params.quote;
    foreign_table->options[CsvFileBufferParser::ESCAPE_KEY] = copy_params.escape;
    foreign_table->options[CsvFileBufferParser::LINE_DELIMITER_KEY] =
        copy_params.line_delim;
    foreign_table->options[CsvFileBufferParser::ARRAY_DELIMITER_KEY] =
        copy_params.array_delim;
    const std::array<char, 3> array_marker{
        copy_params.array_begin, copy_params.array_end, 0};
    foreign_table->options[CsvFileBufferParser::ARRAY_MARKER_KEY] = array_marker.data();
    foreign_table->options[CsvFileBufferParser::LONLAT_KEY] =
        bool_to_option_value(copy_params.lonlat);
    if (copy_params.geo_explode_collections) {
      throw std::runtime_error(
          "geo_explode_collections is not yet supported for FSI CSV import");
    }
    foreign_table->options[CsvFileBufferParser::GEO_EXPLODE_COLLECTIONS_KEY] =
        bool_to_option_value(copy_params.geo_explode_collections);
    foreign_table->options[CsvFileBufferParser::SOURCE_SRID_KEY] =
        std::to_string(copy_params.source_srid);

    foreign_table->options[TextFileBufferParser::BUFFER_SIZE_KEY] =
        std::to_string(copy_params.buffer_size);

    foreign_table->options[CsvFileBufferParser::TRIM_SPACES_KEY] =
        bool_to_option_value(copy_params.trim_spaces);
  }

  foreign_table->initializeOptions();
  return foreign_table;
}

Here is the call graph for this function:

Here is the caller graph for this function:

std::unique_ptr< ForeignDataWrapper > foreign_storage::ForeignDataWrapperFactory::createForGeneralImport ( const import_export::CopyParams &  copy_params, const int  db_id, const ForeignTable *  foreign_table, const UserMapping *  user_mapping  )

static

Definition at line 135 of file ForeignDataWrapperFactory.cpp.

References CHECK, foreign_storage::DataWrapperType::CSV, anonymous_namespace{ForeignDataWrapperFactory.cpp}::get_data_wrapper_type(), anonymous_namespace{ForeignDataWrapperFactory.cpp}::is_valid_data_wrapper(), foreign_storage::DataWrapperType::PARQUET, and foreign_storage::DataWrapperType::REGEX_PARSER.

Referenced by import_export::ForeignDataImporter::importGeneral().

                                     {
  auto data_wrapper_type = get_data_wrapper_type(copy_params);
  CHECK(is_valid_data_wrapper(data_wrapper_type));

  if (data_wrapper_type == DataWrapperType::CSV) {
    return std::make_unique<CsvDataWrapper>(
        db_id, foreign_table, user_mapping, /*disable_cache=*/true);
  } else if (data_wrapper_type == DataWrapperType::REGEX_PARSER) {
    return std::make_unique<RegexParserDataWrapper>(
        db_id, foreign_table, user_mapping, true);
  }
#ifdef ENABLE_IMPORT_PARQUET
  else if (data_wrapper_type == DataWrapperType::PARQUET) {
    return std::make_unique<ParquetDataWrapper>(
        db_id, foreign_table, /*do_metadata_stats_validation=*/false);
  }
#endif

  return {};
}

Here is the call graph for this function:

Here is the caller graph for this function:

std::unique_ptr< ForeignDataWrapper > foreign_storage::ForeignDataWrapperFactory::createForImport ( const std::string &  data_wrapper_type, const int  db_id, const ForeignTable *  foreign_table, const UserMapping *  user_mapping  )

static

Create for the import use-case.

Definition at line 160 of file ForeignDataWrapperFactory.cpp.

References CHECK, and foreign_storage::DataWrapperType::PARQUET.

                                     {
#ifdef ENABLE_IMPORT_PARQUET
  // only supported for parquet import path currently
  CHECK(data_wrapper_type == DataWrapperType::PARQUET);
  return std::make_unique<ParquetImporter>(db_id, foreign_table, user_mapping);
#else
  return {};
#endif
}

const ForeignDataWrapper * foreign_storage::ForeignDataWrapperFactory::createForValidation ( const std::string &  data_wrapper_type, const ForeignTable *  foreign_table = nullptr  )

static

Creates an instance (or gets an existing instance) of an immutable ForeignDataWrapper to be used for validation purposes. Returned instance should not be used for any stateful operations, such as fetching foreign table data/metadata.

Definition at line 381 of file ForeignDataWrapperFactory.cpp.

References foreign_storage::DataWrapperType::CSV, foreign_storage::DataWrapperType::INTERNAL_CATALOG, foreign_storage::DataWrapperType::INTERNAL_EXECUTOR_STATS, foreign_storage::DataWrapperType::INTERNAL_LOGS, foreign_storage::DataWrapperType::INTERNAL_MEMORY_STATS, foreign_storage::DataWrapperType::INTERNAL_ML_MODEL_METADATA, foreign_storage::DataWrapperType::INTERNAL_STORAGE_STATS, foreign_storage::DataWrapperType::PARQUET, foreign_storage::DataWrapperType::REGEX_PARSER, UNREACHABLE, foreign_storage::CsvDataWrapper::validateAndGetIsS3Select(), and validation_data_wrappers_.

Referenced by foreign_storage::ForeignServer::validateStorageParameters().

                                       {
  bool is_s3_select_wrapper{false};
  std::string data_wrapper_type_key{data_wrapper_type};
  constexpr const char* S3_SELECT_WRAPPER_KEY = "CSV_S3_SELECT";
  if (foreign_table && data_wrapper_type == DataWrapperType::CSV &&
      CsvDataWrapper::validateAndGetIsS3Select(foreign_table)) {
    is_s3_select_wrapper = true;
    data_wrapper_type_key = S3_SELECT_WRAPPER_KEY;
  }

  auto [itr, is_new] = validation_data_wrappers_.emplace(data_wrapper_type_key, nullptr);
  if (is_new) {
    if (data_wrapper_type == DataWrapperType::CSV) {
      if (is_s3_select_wrapper) {
        UNREACHABLE();
      } else {
        itr->second = std::make_unique<CsvDataWrapper>();
      }
#ifdef ENABLE_IMPORT_PARQUET
    } else if (data_wrapper_type == DataWrapperType::PARQUET) {
      itr->second = std::make_unique<ParquetDataWrapper>();
#endif
    } else if (data_wrapper_type == DataWrapperType::REGEX_PARSER) {
      itr->second = std::make_unique<RegexParserDataWrapper>();
    } else if (data_wrapper_type == DataWrapperType::INTERNAL_CATALOG) {
      itr->second = std::make_unique<InternalCatalogDataWrapper>();
    } else if (data_wrapper_type == DataWrapperType::INTERNAL_EXECUTOR_STATS) {
      itr->second = std::make_unique<InternalExecutorStatsDataWrapper>();
    } else if (data_wrapper_type == DataWrapperType::INTERNAL_ML_MODEL_METADATA) {
      itr->second = std::make_unique<InternalMLModelMetadataDataWrapper>();
    } else if (data_wrapper_type == DataWrapperType::INTERNAL_MEMORY_STATS) {
      itr->second = std::make_unique<InternalMemoryStatsDataWrapper>();
    } else if (data_wrapper_type == DataWrapperType::INTERNAL_STORAGE_STATS) {
      itr->second = std::make_unique<InternalStorageStatsDataWrapper>();
    } else if (data_wrapper_type == DataWrapperType::INTERNAL_LOGS) {
      itr->second = std::make_unique<InternalLogsDataWrapper>();
    } else {
      UNREACHABLE();
    }
  }
  return itr->second.get();
}

Here is the call graph for this function:

Here is the caller graph for this function:

std::unique_ptr< UserMapping > foreign_storage::ForeignDataWrapperFactory::createUserMappingProxyIfApplicable ( const int  db_id, const int  user_id, const std::string &  file_path, const import_export::CopyParams &  copy_params, const ForeignServer *  server  )

static

Definition at line 175 of file ForeignDataWrapperFactory.cpp.

Referenced by foreign_storage::create_proxy_fsi_objects().

                                 {
  return {};
}

Here is the caller graph for this function:

void foreign_storage::ForeignDataWrapperFactory::validateDataWrapperType ( const std::string &  data_wrapper_type )

static

Checks that the given data wrapper type is valid.

Definition at line 426 of file ForeignDataWrapperFactory.cpp.

References shared::contains(), foreign_storage::DataWrapperType::INTERNAL_DATA_WRAPPERS, join(), foreign_storage::DataWrapperType::supported_data_wrapper_types, and run_benchmark_import::type.

Referenced by foreign_storage::ForeignServer::validate().

                                        {
  const auto& supported_wrapper_types = DataWrapperType::supported_data_wrapper_types;
  if (std::find(supported_wrapper_types.begin(),
                supported_wrapper_types.end(),
                data_wrapper_type) == supported_wrapper_types.end()) {
    std::vector<std::string_view> user_facing_wrapper_types;
    for (const auto& type : supported_wrapper_types) {
      if (!shared::contains(DataWrapperType::INTERNAL_DATA_WRAPPERS, type)) {
        user_facing_wrapper_types.emplace_back(type);
      }
    }
    throw std::runtime_error{"Invalid data wrapper type \"" + data_wrapper_type +
                             "\". Data wrapper type must be one of the following: " +
                             join(user_facing_wrapper_types, ", ") + "."};
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

std::map< std::string, std::unique_ptr< ForeignDataWrapper > > foreign_storage::ForeignDataWrapperFactory::validation_data_wrappers_

staticprivate

Definition at line 168 of file ForeignDataWrapperFactory.h.

Referenced by createForValidation().

---

The documentation for this class was generated from the following files:

• /home/jenkins-slave/workspace/core-os-doxygen/DataMgr/ForeignStorage/ForeignDataWrapperFactory.h
• /home/jenkins-slave/workspace/core-os-doxygen/DataMgr/ForeignStorage/ForeignDataWrapperFactory.cpp