#include <ForeignDataImporter.h>

Inheritance diagram for import_export::ForeignDataImporter:

Collaboration diagram for import_export::ForeignDataImporter:

Public Member Functions
	ForeignDataImporter (const std::string &file_path, const CopyParams &copy_params, const TableDescriptor *table)

ImportStatus	import (const Catalog_Namespace::SessionInfo *session_info) override

Public Member Functions inherited from import_export::AbstractImporter
virtual	~AbstractImporter ()=default

Static Public Member Functions
static void	setDefaultImportPath (const std::string &base_path)

Static Public Attributes
static int32_t	proxy_foreign_table_fragment_size_ = 0

Protected Attributes
std::unique_ptr < Fragmenter_Namespace::InsertDataLoader::InsertConnector >	connector_

Private Member Functions
void	finalize (const Catalog_Namespace::SessionInfo &parent_session_info, ImportStatus &import_status, const std::vector< std::pair< const ColumnDescriptor , StringDictionary >> &string_dictionaries)

void	finalize (const Catalog_Namespace::SessionInfo &parent_session_info, ImportStatus &import_status, const int32_t table_id)

ImportStatus	importGeneral (const Catalog_Namespace::SessionInfo *session_info)

ImportStatus	importGeneral (const Catalog_Namespace::SessionInfo *session_info, const std::string &copy_from_source, const CopyParams &copy_params)

ImportStatus	importGeneralS3 (const Catalog_Namespace::SessionInfo *session_info)

Private Attributes
std::string	copy_from_source_

CopyParams	copy_params_

const TableDescriptor *	table_

Static Private Attributes
static std::string	default_import_path_

Detailed Description

Definition at line 26 of file ForeignDataImporter.h.

Constructor & Destructor Documentation

import_export::ForeignDataImporter::ForeignDataImporter	(	const std::string &	file_path,
		const CopyParams &	copy_params,
		const TableDescriptor *	table
	)

Definition at line 407 of file ForeignDataImporter.cpp.

References connector_.

     : copy_from_source_(copy_from_source), copy_params_(copy_params), table_(table) {
   connector_ = std::make_unique<Fragmenter_Namespace::LocalInsertConnector>();
 }

Member Function Documentation

void import_export::ForeignDataImporter::finalize	(	const Catalog_Namespace::SessionInfo &	parent_session_info,
		ImportStatus &	import_status,
		const std::vector< std::pair< const ColumnDescriptor , StringDictionary >> &	string_dictionaries
	)

private

Definition at line 414 of file ForeignDataImporter.cpp.

References connector_, DEBUG_TIMER, Data_Namespace::DISK_LEVEL, logger::ERROR, import_export::ImportStatus::load_failed, import_export::ImportStatus::load_msg, LOG, TableDescriptor::persistenceLevel, table_, and TableDescriptor::tableId.

Referenced by finalize(), and importGeneral().

                              {
   if (table_->persistenceLevel ==
       Data_Namespace::MemoryLevel::DISK_LEVEL) {  // only checkpoint disk-resident
                                                   // tables
     if (!import_status.load_failed) {
       auto timer = DEBUG_TIMER("Dictionary Checkpointing");
       for (const auto& [column_desciptor, string_dictionary] : string_dictionaries) {
         if (!string_dictionary->checkpoint()) {
           LOG(ERROR) << "Checkpointing Dictionary for Column "
                      << column_desciptor->columnName << " failed.";
           import_status.load_failed = true;
           import_status.load_msg = "Dictionary checkpoint failed";
           break;
         }
       }
     }
   }
   if (import_status.load_failed) {
     connector_->rollback(parent_session_info, table_->tableId);
   } else {
     connector_->checkpoint(parent_session_info, table_->tableId);
   }
 }

Here is the caller graph for this function:

void import_export::ForeignDataImporter::finalize	(	const Catalog_Namespace::SessionInfo &	parent_session_info,
		ImportStatus &	import_status,
		const int32_t	table_id
	)

private

Definition at line 442 of file ForeignDataImporter.cpp.

References finalize(), Catalog_Namespace::Catalog::getAllColumnMetadataForTable(), and Catalog_Namespace::SessionInfo::getCatalog().

                             {
   auto& catalog = parent_session_info.getCatalog();
 
   auto logical_columns =
       catalog.getAllColumnMetadataForTable(table_id, false, false, false);
 
   std::vector<std::pair<const ColumnDescriptor*, StringDictionary*>> string_dictionaries;
   for (const auto& column_descriptor : logical_columns) {
     if (!column_descriptor->columnType.is_dict_encoded_string()) {
       continue;
     }
     auto dict_descriptor =
         catalog.getMetadataForDict(column_descriptor->columnType.get_comp_param(), true);
     string_dictionaries.emplace_back(column_descriptor,
                                      dict_descriptor->stringDict.get());
   }
 
   finalize(parent_session_info, import_status, string_dictionaries);
 }

Here is the call graph for this function:

ImportStatus import_export::ForeignDataImporter::import ( const Catalog_Namespace::SessionInfo * session_info )

overridevirtual

Implements import_export::AbstractImporter.

Definition at line 931 of file ForeignDataImporter.cpp.

References importGeneral(), importGeneralS3(), and shared::is_s3_uri().

                                                       {
   if (shared::is_s3_uri(copy_from_source_)) {
     return importGeneralS3(session_info);
   }
   return importGeneral(session_info);
 }

Here is the call graph for this function:

ImportStatus import_export::ForeignDataImporter::importGeneral ( const Catalog_Namespace::SessionInfo * session_info )

private

Definition at line 587 of file ForeignDataImporter.cpp.

References copy_from_source_, and copy_params_.

Referenced by import(), and importGeneralS3().

                                                       {
   return importGeneral(session_info, copy_from_source_, copy_params_);
 }

Here is the caller graph for this function:

ImportStatus import_export::ForeignDataImporter::importGeneral	(	const Catalog_Namespace::SessionInfo *	session_info,
		const std::string &	copy_from_source,
		const CopyParams &	copy_params
	)

private

Definition at line 514 of file ForeignDataImporter.cpp.

References CHECK, CHECK_GE, CHUNK_KEY_FRAGMENT_IDX, connector_, foreign_storage::create_proxy_fsi_objects(), foreign_storage::ForeignDataWrapperFactory::createForGeneralImport(), finalize(), Catalog_Namespace::SessionInfo::get_currentUser(), import_export::anonymous_namespace{ForeignDataImporter.cpp}::get_proxy_foreign_table_fragment_size(), Catalog_Namespace::SessionInfo::getCatalog(), anonymous_namespace{ForeignDataImporter.cpp}::import_foreign_data(), logger::INFO, foreign_storage::is_valid_source_type(), LOG, import_export::CopyParams::max_import_batch_row_count, anonymous_namespace{ForeignDataImporter.cpp}::metadata_scan(), table_, TableDescriptor::tableId, TableDescriptor::tableName, TableDescriptor::userId, and anonymous_namespace{ForeignDataImporter.cpp}::validate_copy_params().

                                    {
   auto& catalog = session_info->getCatalog();
 
   CHECK(foreign_storage::is_valid_source_type(copy_params));
 
   // validate copy params before import in order to print user friendly messages
   validate_copy_params(copy_params);
 
   ImportStatus import_status;
   {
     auto& current_user = session_info->get_currentUser();
     auto [server, user_mapping, foreign_table] =
         foreign_storage::create_proxy_fsi_objects(copy_from_source,
                                                   copy_params,
                                                   catalog.getDatabaseId(),
                                                   table_,
                                                   current_user.userId);
 
     // maximum number of fragments buffered in memory at any one time, affects
     // `maxFragRows` heuristic below
     const size_t maximum_num_fragments_buffered = 3;
     // set fragment size for proxy foreign table during import
     foreign_table->maxFragRows =
         get_proxy_foreign_table_fragment_size(maximum_num_fragments_buffered,
                                               copy_params.max_import_batch_row_count,
                                               *session_info,
                                               table_->tableId);
 
     // log for debugging purposes
     LOG(INFO) << "Import fragment row count is " << foreign_table->maxFragRows
               << " for table " << table_->tableName;
 
     auto data_wrapper =
         foreign_storage::ForeignDataWrapperFactory::createForGeneralImport(
             copy_params,
             catalog.getDatabaseId(),
             foreign_table.get(),
             user_mapping.get());
 
     int32_t max_fragment_id = std::numeric_limits<int32_t>::max();
     if (!data_wrapper->isLazyFragmentFetchingEnabled()) {
       ChunkMetadataVector metadata_vector =
           metadata_scan(data_wrapper.get(), foreign_table.get());
       if (metadata_vector.empty()) {  // an empty data source
         return {};
       }
       max_fragment_id = 0;
       for (const auto& [key, _] : metadata_vector) {
         max_fragment_id = std::max(max_fragment_id, key[CHUNK_KEY_FRAGMENT_IDX]);
       }
       CHECK_GE(max_fragment_id, 0);
     }
 
     import_status = import_foreign_data(max_fragment_id,
                                         connector_.get(),
                                         catalog,
                                         table_,
                                         data_wrapper.get(),
                                         session_info,
                                         copy_params,
                                         copy_from_source,
                                         maximum_num_fragments_buffered);
 
   }  // this scope ensures that fsi proxy objects are destroyed prior to checkpoint
 
   finalize(*session_info, import_status, table_->tableId);
 
   return import_status;
 }

Here is the call graph for this function:

ImportStatus import_export::ForeignDataImporter::importGeneralS3 ( const Catalog_Namespace::SessionInfo * session_info )

private

Definition at line 597 of file ForeignDataImporter.cpp.

Referenced by import().

                                                       {
   CHECK(shared::is_s3_uri(copy_from_source_));
 
   if (!(copy_params_.source_type == SourceType::kDelimitedFile ||
 #if ENABLE_IMPORT_PARQUET
         copy_params_.source_type == SourceType::kParquetFile ||
 #endif
         copy_params_.source_type == SourceType::kRegexParsedFile)) {
     throw std::runtime_error("Attempting to load S3 resource '" + copy_from_source_ +
                              "' for unsupported 'source_type' (must be 'DELIMITED_FILE'"
 #if ENABLE_IMPORT_PARQUET
                              ", 'PARQUET_FILE'"
 #endif
                              " or 'REGEX_PARSED_FILE'");
   }
 
   const shared::FilePathOptions options{copy_params_.regex_path_filter,
                                         copy_params_.file_sort_order_by,
                                         copy_params_.file_sort_regex};
   shared::validate_sort_options(options);
 
 #ifdef HAVE_AWS_S3
 
   auto uuid = boost::uuids::random_generator()();
   std::string base_path = "s3-import-" + boost::uuids::to_string(uuid);
   auto import_path = std::filesystem::path(default_import_path_) / base_path;
 
   // Ensure files & dirs are cleaned up, regardless of outcome
   ScopeGuard cleanup_guard = [&] {
     if (std::filesystem::exists(import_path)) {
       std::filesystem::remove_all(import_path);
     }
   };
 
   auto s3_archive = std::make_unique<S3Archive>(copy_from_source_,
                                                 copy_params_.s3_access_key,
                                                 copy_params_.s3_secret_key,
                                                 copy_params_.s3_session_token,
                                                 copy_params_.s3_region,
                                                 copy_params_.s3_endpoint,
                                                 copy_params_.plain_text,
                                                 copy_params_.regex_path_filter,
                                                 copy_params_.file_sort_order_by,
                                                 copy_params_.file_sort_regex,
                                                 import_path);
   s3_archive->init_for_read();
 
   const auto bucket_name = s3_archive->url_part(4);
 
   auto object_keys = s3_archive->get_objkeys();
   std::vector<DownloadedObjectToProcess> objects_to_process(object_keys.size());
   size_t object_count = 0;
   for (const auto& objkey : object_keys) {
     auto& object = objects_to_process[object_count++];
     object.object_key = objkey;
     object.is_downloaded = false;
   }
 
   ImportStatus aggregate_import_status;
   const int num_download_threads = copy_params_.s3_max_concurrent_downloads;
 
   std::mutex communication_mutex;
   bool continue_downloading = true;
   bool download_exception_occured = false;
 
   std::condition_variable files_download_condition;
 
   auto is_downloading_finished = [&] {
     std::unique_lock communication_lock(communication_mutex);
     return !continue_downloading || download_exception_occured;
   };
 
   std::function<void(const std::vector<size_t>&)> download_objects =
       [&](const std::vector<size_t>& partition) {
         for (const auto& index : partition) {
           DownloadedObjectToProcess& object = objects_to_process[index];
           const std::string& obj_key = object.object_key;
           if (is_downloading_finished()) {
             return;
           }
           std::exception_ptr eptr;  // unused
           std::string local_file_path;
           std::string exception_what;
           bool exception_occured = false;
 
           try {
             local_file_path = s3_archive->land(obj_key,
                                                eptr,
                                                false,
                                                /*allow_named_pipe_use=*/false,
                                                /*track_file_path=*/false);
           } catch (const std::exception& e) {
             exception_what = e.what();
             exception_occured = true;
           }
 
           if (is_downloading_finished()) {
             return;
           }
           if (exception_occured) {
             {
               std::unique_lock communication_lock(communication_mutex);
               download_exception_occured = true;
             }
             files_download_condition.notify_all();
             throw std::runtime_error("failed to fetch s3 object: '" + obj_key +
                                      "': " + exception_what);
           }
 
           object.download_file_path = local_file_path;
           object.is_downloaded =
               true;  // this variable is atomic and therefore acts as a lock, it must be
                      // set last to ensure no data race
 
           files_download_condition.notify_all();
         }
       };
 
   std::function<void()> import_local_files = [&]() {
     for (size_t object_index = 0; object_index < object_count;) {
       {
         std::unique_lock communication_lock(communication_mutex);
         files_download_condition.wait(
             communication_lock,
             [&download_exception_occured, object_index, &objects_to_process]() {
               return objects_to_process[object_index].is_downloaded ||
                      download_exception_occured;
             });
         if (download_exception_occured) {  // do not wait for object index if a download
                                            // error has occured
           return;
         }
       }
 
       // find largest range of files to import
       size_t end_object_index = object_count;
       for (size_t i = object_index + 1; i < object_count; ++i) {
         if (!objects_to_process[i].is_downloaded) {
           end_object_index = i;
           break;
         }
       }
 
       ImportStatus local_import_status;
       std::string local_import_dir;
       try {
         CopyParams local_copy_params;
         std::tie(local_import_dir, local_copy_params) = get_local_copy_source_and_params(
             copy_params_, objects_to_process, object_index, end_object_index);
         local_import_status =
             importGeneral(session_info, local_import_dir, local_copy_params);
         // clean up temporary files
         std::filesystem::remove_all(local_import_dir);
       } catch (const std::exception& except) {
         // replace all occurences of file names with the object keys for
         // users
         std::string what = except.what();
 
         for (size_t i = object_index; i < end_object_index; ++i) {
           auto& object = objects_to_process[i];
           what = boost::regex_replace(what,
                                       boost::regex{object.import_file_path},
                                       bucket_name + "/" + object.object_key);
         }
         {
           std::unique_lock communication_lock(communication_mutex);
           continue_downloading = false;
         }
         // clean up temporary files
         std::filesystem::remove_all(local_import_dir);
         throw std::runtime_error(what);
       }
       aggregate_import_status += local_import_status;
       import_export::Importer::set_import_status(copy_from_source_,
                                                  aggregate_import_status);
       if (aggregate_import_status.load_failed) {
         {
           std::unique_lock communication_lock(communication_mutex);
           continue_downloading = false;
         }
         return;
       }
 
       object_index =
           end_object_index;  // all objects in range [object_index,end_object_index)
                              // correctly imported at this point in excecution, move onto
                              // next range
     }
   };
 
   std::vector<size_t> partition_range(object_count);
   std::iota(partition_range.begin(), partition_range.end(), 0);
   auto download_futures = foreign_storage::create_futures_for_workers(
       partition_range, num_download_threads, download_objects);
 
   auto import_future = std::async(std::launch::async, import_local_files);
 
   for (auto& future : download_futures) {
     future.wait();
   }
   import_future.get();  // may throw an exception
 
   // get any remaining exceptions
   for (auto& future : download_futures) {
     future.get();
   }
   return aggregate_import_status;
 
 #else
   throw std::runtime_error("AWS S3 support not available");
 
   return {};
 #endif
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void import_export::ForeignDataImporter::setDefaultImportPath ( const std::string & base_path )

static

Definition at line 592 of file ForeignDataImporter.cpp.

References default_import_path_, and shared::kDefaultImportDirName.

Referenced by CommandLineOptions::validate().

                                                                          {
   auto data_dir_path = boost::filesystem::canonical(base_path);
   default_import_path_ = (data_dir_path / shared::kDefaultImportDirName).string();
 }

Here is the caller graph for this function:

Member Data Documentation

std::unique_ptr<Fragmenter_Namespace::InsertDataLoader::InsertConnector> import_export::ForeignDataImporter::connector_

protected

Definition at line 43 of file ForeignDataImporter.h.

Referenced by finalize(), ForeignDataImporter(), and importGeneral().

std::string import_export::ForeignDataImporter::copy_from_source_

private

Definition at line 66 of file ForeignDataImporter.h.

Referenced by importGeneral(), and importGeneralS3().

CopyParams import_export::ForeignDataImporter::copy_params_

private

Definition at line 67 of file ForeignDataImporter.h.

Referenced by importGeneral(), and importGeneralS3().

std::string import_export::ForeignDataImporter::default_import_path_

inlinestaticprivate

Definition at line 69 of file ForeignDataImporter.h.

Referenced by importGeneralS3(), and setDefaultImportPath().

int32_t import_export::ForeignDataImporter::proxy_foreign_table_fragment_size_ = 0

static

Definition at line 40 of file ForeignDataImporter.h.

Referenced by import_export::anonymous_namespace{ForeignDataImporter.cpp}::get_proxy_foreign_table_fragment_size().

const TableDescriptor* import_export::ForeignDataImporter::table_

private

Definition at line 68 of file ForeignDataImporter.h.

Referenced by finalize(), and importGeneral().

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

/home/jenkins-slave/workspace/core-os-doxygen/ImportExport/ForeignDataImporter.h
/home/jenkins-slave/workspace/core-os-doxygen/ImportExport/ForeignDataImporter.cpp

Public Member Functions

Static Public Member Functions

Static Public Attributes

Protected Attributes

Private Member Functions

Private Attributes

Static Private Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation