#include <Importer.h>

Inheritance diagram for import_export::Importer:

Collaboration diagram for import_export::Importer:

Classes
struct	GeoFileLayerInfo

Public Types
enum	GeoFileLayerContents { GeoFileLayerContents::EMPTY, GeoFileLayerContents::GEO, GeoFileLayerContents::NON_GEO, GeoFileLayerContents::UNSUPPORTED_GEO }

Public Member Functions
	Importer (Catalog_Namespace::Catalog &c, const TableDescriptor *t, const std::string &f, const CopyParams &p)

	Importer (Loader *providedLoader, const std::string &f, const CopyParams &p)

	~Importer () override

ImportStatus	import (const Catalog_Namespace::SessionInfo *session_info)

ImportStatus	importDelimited (const std::string &file_path, const bool decompressed, const Catalog_Namespace::SessionInfo *session_info) override

ImportStatus	importGDAL (std::map< std::string, std::string > colname_to_src, const Catalog_Namespace::SessionInfo *session_info)

const CopyParams &	get_copy_params () const

const std::list< const ColumnDescriptor * > &	get_column_descs () const

void	load (const std::vector< std::unique_ptr< TypedImportBuffer >> &import_buffers, size_t row_count, const Catalog_Namespace::SessionInfo *session_info)

std::vector< std::vector < std::unique_ptr < TypedImportBuffer > > > &	get_import_buffers_vec ()

std::vector< std::unique_ptr < TypedImportBuffer > > &	get_import_buffers (int i)

const bool *	get_is_array () const

Catalog_Namespace::Catalog &	getCatalog ()

void	checkpoint (const std::vector< Catalog_Namespace::TableEpochInfo > &table_epochs)

auto	getLoader () const

Public Member Functions inherited from import_export::DataStreamSink
	DataStreamSink ()

	DataStreamSink (const CopyParams &copy_params, const std::string file_path)

virtual	~DataStreamSink ()

const CopyParams &	get_copy_params () const

void	import_compressed (std::vector< std::string > &file_paths, const Catalog_Namespace::SessionInfo *session_info)

Static Public Member Functions
static ImportStatus	get_import_status (const std::string &id)

static void	set_import_status (const std::string &id, const ImportStatus is)

static const std::list < ColumnDescriptor >	gdalToColumnDescriptors (const std::string &fileName, const std::string &geoColumnName, const CopyParams &copy_params)

static void	readMetadataSampleGDAL (const std::string &fileName, const std::string &geoColumnName, std::map< std::string, std::vector< std::string >> &metadata, int rowLimit, const CopyParams &copy_params)

static bool	gdalFileExists (const std::string &path, const CopyParams &copy_params)

static bool	gdalFileOrDirectoryExists (const std::string &path, const CopyParams &copy_params)

static std::vector< std::string >	gdalGetAllFilesInArchive (const std::string &archive_path, const CopyParams &copy_params)

static std::vector < GeoFileLayerInfo >	gdalGetLayersInGeoFile (const std::string &file_name, const CopyParams &copy_params)

static void	set_geo_physical_import_buffer (const Catalog_Namespace::Catalog &catalog, const ColumnDescriptor *cd, std::vector< std::unique_ptr< TypedImportBuffer >> &import_buffers, size_t &col_idx, std::vector< double > &coords, std::vector< double > &bounds, std::vector< int > &ring_sizes, std::vector< int > &poly_rings, int render_group)

static void	set_geo_physical_import_buffer_columnar (const Catalog_Namespace::Catalog &catalog, const ColumnDescriptor *cd, std::vector< std::unique_ptr< TypedImportBuffer >> &import_buffers, size_t &col_idx, std::vector< std::vector< double >> &coords_column, std::vector< std::vector< double >> &bounds_column, std::vector< std::vector< int >> &ring_sizes_column, std::vector< std::vector< int >> &poly_rings_column, std::vector< int > &render_groups_column)

Static Private Member Functions
static bool	gdalStatInternal (const std::string &path, const CopyParams &copy_params, bool also_dir)

static OGRDataSource *	openGDALDataset (const std::string &fileName, const CopyParams &copy_params)

static void	setGDALAuthorizationTokens (const CopyParams &copy_params)

Private Attributes
std::string	import_id

size_t	file_size

size_t	max_threads

char *	buffer [2]

std::vector< std::vector < std::unique_ptr < TypedImportBuffer > > >	import_buffers_vec

std::unique_ptr< Loader >	loader

std::unique_ptr< bool[]>	is_array_a

bool	interrupted

Static Private Attributes
static std::mutex	init_gdal_mutex

Additional Inherited Members
Protected Member Functions inherited from import_export::DataStreamSink
ImportStatus	archivePlumber (const Catalog_Namespace::SessionInfo *session_info)

Protected Attributes inherited from import_export::DataStreamSink
CopyParams	copy_params

const std::string	file_path

FILE *	p_file = nullptr

ImportStatus	import_status_

mapd_shared_mutex	import_mutex_

size_t	total_file_size {0}

std::vector< size_t >	file_offsets

std::mutex	file_offsets_mutex

Detailed Description

Definition at line 760 of file Importer.h.

Member Enumeration Documentation

enum import_export::Importer::GeoFileLayerContents

strong

Enumerator
EMPTY
GEO
NON_GEO
UNSUPPORTED_GEO

Definition at line 811 of file Importer.h.

811 { EMPTY, GEO, NON_GEO, UNSUPPORTED_GEO };

Constructor & Destructor Documentation

import_export::Importer::Importer	(	Catalog_Namespace::Catalog &	c,
		const TableDescriptor *	t,
		const std::string &	f,
		const CopyParams &	p
	)

Definition at line 171 of file Importer.cpp.

175 : Importer(new Loader(c, t), f, p) {}

import_export::Importer::Importer

Importer(Catalog_Namespace::Catalog &c, const TableDescriptor *t, const std::string &f, const CopyParams &p)

Definition: Importer.cpp:171

f

char * f

Definition: Parser_wnd_pregen.cpp:2080

import_export::Importer::Importer	(	Loader *	providedLoader,
		const std::string &	f,
		const CopyParams &	p
	)

Definition at line 177 of file Importer.cpp.

References buffer, import_export::DataStreamSink::file_path, file_size, i, import_id, is_array_a, kARRAY, loader, max_threads, and import_export::DataStreamSink::p_file.

     : DataStreamSink(p, f), loader(providedLoader) {
   import_id = boost::filesystem::path(file_path).filename().string();
   file_size = 0;
   max_threads = 0;
   p_file = nullptr;
   buffer[0] = nullptr;
   buffer[1] = nullptr;
   // we may be overallocating a little more memory here due to dropping phy cols.
   // it shouldn't be an issue because iteration of it is not supposed to go OOB.
   auto is_array = std::unique_ptr<bool[]>(new bool[loader->get_column_descs().size()]);
   int i = 0;
   bool has_array = false;
   // TODO: replace this ugly way of skipping phy cols once if isPhyGeo is defined
   int skip_physical_cols = 0;
   for (auto& p : loader->get_column_descs()) {
     // phy geo columns can't be in input file
     if (skip_physical_cols-- > 0) {
       continue;
     }
     // neither are rowid or $deleted$
     // note: columns can be added after rowid/$deleted$
     if (p->isVirtualCol || p->isDeletedCol) {
       continue;
     }
     skip_physical_cols = p->columnType.get_physical_cols();
     if (p->columnType.get_type() == kARRAY) {
       is_array.get()[i] = true;
       has_array = true;
     } else {
       is_array.get()[i] = false;
     }
     ++i;
   }
   if (has_array) {
     is_array_a = std::unique_ptr<bool[]>(is_array.release());
   } else {
     is_array_a = std::unique_ptr<bool[]>(nullptr);
   }
 }

import_export::Importer::~Importer ( )

override

Definition at line 218 of file Importer.cpp.

References buffer, and import_export::DataStreamSink::p_file.

                     {
   if (p_file != nullptr) {
     fclose(p_file);
   }
   if (buffer[0] != nullptr) {
     free(buffer[0]);
   }
   if (buffer[1] != nullptr) {
     free(buffer[1]);
   }
 }

Member Function Documentation

void import_export::Importer::checkpoint ( const std::vector< Catalog_Namespace::TableEpochInfo > & table_epochs )

Definition at line 3379 of file Importer.cpp.

References DEBUG_TIMING, Data_Namespace::DISK_LEVEL, logger::ERROR, measure< TimeT >::execution(), StorageType::FOREIGN_TABLE, import_buffers_vec, import_export::DataStreamSink::import_mutex_, import_export::DataStreamSink::import_status_, logger::INFO, import_export::ImportStatus::load_failed, import_export::ImportStatus::load_msg, loader, and LOG.

Referenced by importDelimited(), and importGDAL().

                                                                     {
   if (loader->getTableDesc()->storageType != StorageType::FOREIGN_TABLE) {
     mapd_lock_guard<mapd_shared_mutex> read_lock(import_mutex_);
     if (import_status_.load_failed) {
       // rollback to starting epoch - undo all the added records
       loader->setTableEpochs(table_epochs);
     } else {
       loader->checkpoint();
     }
   }
 
   if (loader->getTableDesc()->persistenceLevel ==
       Data_Namespace::MemoryLevel::DISK_LEVEL) {  // only checkpoint disk-resident
                                                   // tables
     auto ms = measure<>::execution([&]() {
       mapd_lock_guard<mapd_shared_mutex> write_lock(import_mutex_);
       if (!import_status_.load_failed) {
         for (auto& p : import_buffers_vec[0]) {
           if (!p->stringDictCheckpoint()) {
             LOG(ERROR) << "Checkpointing Dictionary for Column "
                        << p->getColumnDesc()->columnName << " failed.";
             import_status_.load_failed = true;
             import_status_.load_msg = "Dictionary checkpoint failed";
             break;
           }
         }
       }
     });
     if (DEBUG_TIMING) {
       LOG(INFO) << "Dictionary Checkpointing took " << (double)ms / 1000.0 << " Seconds."
                 << std::endl;
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool import_export::Importer::gdalFileExists	(	const std::string &	path,
		const CopyParams &	copy_params
	)

static

Definition at line 4735 of file Importer.cpp.

References gdalStatInternal().

Referenced by DBHandler::check_geospatial_files(), DBHandler::detect_column_types(), DBHandler::get_all_files_in_archive(), DBHandler::get_first_geo_file_in_archive(), DBHandler::get_layers_in_geo_file(), and DBHandler::import_geo_table().

                                                                                   {
   return gdalStatInternal(path, copy_params, false);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool import_export::Importer::gdalFileOrDirectoryExists	(	const std::string &	path,
		const CopyParams &	copy_params
	)

static

Definition at line 4740 of file Importer.cpp.

References gdalStatInternal().

Referenced by DBHandler::detect_column_types(), DBHandler::get_layers_in_geo_file(), and DBHandler::import_geo_table().

                                                                         {
   return gdalStatInternal(path, copy_params, true);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

std::vector< std::string > import_export::Importer::gdalGetAllFilesInArchive	(	const std::string &	archive_path,
		const CopyParams &	copy_params
	)

static

Definition at line 4812 of file Importer.cpp.

References import_export::gdalGatherFilesInArchiveRecursive(), Geospatial::GDAL::init(), and setGDALAuthorizationTokens().

Referenced by anonymous_namespace{DBHandler.cpp}::find_first_geo_file_in_archive(), and DBHandler::get_all_files_in_archive().

                                    {
   // lazy init GDAL
   Geospatial::GDAL::init();
 
   // set authorization tokens
   setGDALAuthorizationTokens(copy_params);
 
   // prepare to gather files
   std::vector<std::string> files;
 
   // gather the files recursively
   gdalGatherFilesInArchiveRecursive(archive_path, files);
 
   // convert to relative paths inside archive
   for (auto& file : files) {
     file.erase(0, archive_path.size() + 1);  // remove archive_path and the slash
   }
 
   // done
   return files;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

std::vector< Importer::GeoFileLayerInfo > import_export::Importer::gdalGetLayersInGeoFile	(	const std::string &	file_name,
		const CopyParams &	copy_params
	)

static

Definition at line 4837 of file Importer.cpp.

References CHECK, EMPTY, GEO, import_export::CopyParams::geo_explode_collections, Geospatial::GDAL::init(), NON_GEO, openGDALDataset(), setGDALAuthorizationTokens(), and UNSUPPORTED_GEO.

Referenced by DBHandler::get_layers_in_geo_file(), and DBHandler::import_geo_table().

                                    {
   // lazy init GDAL
   Geospatial::GDAL::init();
 
   // set authorization tokens
   setGDALAuthorizationTokens(copy_params);
 
   // prepare to gather layer info
   std::vector<GeoFileLayerInfo> layer_info;
 
   // open the data set
   OGRDataSourceUqPtr poDS(openGDALDataset(file_name, copy_params));
   if (poDS == nullptr) {
     throw std::runtime_error("openGDALDataset Error: Unable to open geo file " +
                              file_name);
   }
 
   // enumerate the layers
   for (auto&& poLayer : poDS->GetLayers()) {
     GeoFileLayerContents contents = GeoFileLayerContents::EMPTY;
     // prepare to read this layer
     poLayer->ResetReading();
     // skip layer if empty
     if (poLayer->GetFeatureCount() > 0) {
       // get first feature
       OGRFeatureUqPtr first_feature(poLayer->GetNextFeature());
       CHECK(first_feature);
       // check feature for geometry
       const OGRGeometry* geometry = first_feature->GetGeometryRef();
       if (!geometry) {
         // layer has no geometry
         contents = GeoFileLayerContents::NON_GEO;
       } else {
         // check the geometry type
         const OGRwkbGeometryType geometry_type = geometry->getGeometryType();
         switch (wkbFlatten(geometry_type)) {
           case wkbPoint:
           case wkbLineString:
           case wkbPolygon:
           case wkbMultiPolygon:
             // layer has supported geo
             contents = GeoFileLayerContents::GEO;
             break;
           case wkbMultiPoint:
           case wkbMultiLineString:
             // supported if geo_explode_collections is specified
             contents = copy_params.geo_explode_collections
                            ? GeoFileLayerContents::GEO
                            : GeoFileLayerContents::UNSUPPORTED_GEO;
             break;
           default:
             // layer has unsupported geometry
             contents = GeoFileLayerContents::UNSUPPORTED_GEO;
             break;
         }
       }
     }
     // store info for this layer
     layer_info.emplace_back(poLayer->GetName(), contents);
   }
 
   // done
   return layer_info;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool import_export::Importer::gdalStatInternal	(	const std::string &	path,
		const CopyParams &	copy_params,
		bool	also_dir
	)

staticprivate

Definition at line 4702 of file Importer.cpp.

References Geospatial::GDAL::init(), run_benchmark_import::result, and setGDALAuthorizationTokens().

Referenced by gdalFileExists(), and gdalFileOrDirectoryExists().

                                                {
   // lazy init GDAL
   Geospatial::GDAL::init();
 
   // set authorization tokens
   setGDALAuthorizationTokens(copy_params);
 
 #if (GDAL_VERSION_MAJOR > 2) || (GDAL_VERSION_MAJOR == 2 && GDAL_VERSION_MINOR >= 3)
   // clear GDAL stat cache
   // without this, file existence will be cached, even if authentication changes
   // supposed to be available from GDAL 2.2.1 but our CentOS build disagrees
   VSICurlClearCache();
 #endif
 
   // stat path
   VSIStatBufL sb;
   int result = VSIStatExL(path.c_str(), &sb, VSI_STAT_EXISTS_FLAG);
   if (result < 0) {
     return false;
   }
 
   // exists?
   if (also_dir && (VSI_ISREG(sb.st_mode) || VSI_ISDIR(sb.st_mode))) {
     return true;
   } else if (VSI_ISREG(sb.st_mode)) {
     return true;
   }
   return false;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

const std::list< ColumnDescriptor > import_export::Importer::gdalToColumnDescriptors	(	const std::string &	fileName,
		const std::string &	geoColumnName,
		const CopyParams &	copy_params
	)

static

Definition at line 4612 of file Importer.cpp.

References CHECK, ColumnDescriptor::columnName, ColumnDescriptor::columnType, import_export::CopyParams::geo_coords_comp_param, import_export::CopyParams::geo_coords_encoding, import_export::CopyParams::geo_coords_srid, import_export::CopyParams::geo_coords_type, import_export::CopyParams::geo_explode_collections, import_export::CopyParams::geo_layer_name, import_export::anonymous_namespace{Importer.cpp}::getLayerWithSpecifiedName(), kARRAY, kENCODING_DICT, kMULTIPOLYGON, kPOLYGON, kTEXT, import_export::ogr_to_type(), openGDALDataset(), import_export::PROMOTE_POLYGON_TO_MULTIPOLYGON, SQLTypeInfo::set_comp_param(), SQLTypeInfo::set_compression(), SQLTypeInfo::set_fixed_size(), SQLTypeInfo::set_input_srid(), SQLTypeInfo::set_output_srid(), SQLTypeInfo::set_subtype(), SQLTypeInfo::set_type(), and ColumnDescriptor::sourceName.

Referenced by DBHandler::detect_column_types().

                                    {
   std::list<ColumnDescriptor> cds;
 
   OGRDataSourceUqPtr poDS(openGDALDataset(file_name, copy_params));
   if (poDS == nullptr) {
     throw std::runtime_error("openGDALDataset Error: Unable to open geo file " +
                              file_name);
   }
 
   OGRLayer& layer =
       getLayerWithSpecifiedName(copy_params.geo_layer_name, poDS, file_name);
 
   layer.ResetReading();
   // TODO(andrewseidl): support multiple features
   OGRFeatureUqPtr poFeature(layer.GetNextFeature());
   if (poFeature == nullptr) {
     throw std::runtime_error("No features found in " + file_name);
   }
   // get fields as regular columns
   OGRFeatureDefn* poFDefn = layer.GetLayerDefn();
   CHECK(poFDefn);
   int iField;
   for (iField = 0; iField < poFDefn->GetFieldCount(); iField++) {
     OGRFieldDefn* poFieldDefn = poFDefn->GetFieldDefn(iField);
     auto typePair = ogr_to_type(poFieldDefn->GetType());
     ColumnDescriptor cd;
     cd.columnName = poFieldDefn->GetNameRef();
     cd.sourceName = poFieldDefn->GetNameRef();
     SQLTypeInfo ti;
     if (typePair.second) {
       ti.set_type(kARRAY);
       ti.set_subtype(typePair.first);
     } else {
       ti.set_type(typePair.first);
     }
     if (typePair.first == kTEXT) {
       ti.set_compression(kENCODING_DICT);
       ti.set_comp_param(32);
     }
     ti.set_fixed_size();
     cd.columnType = ti;
     cds.push_back(cd);
   }
   // get geo column, if any
   OGRGeometry* poGeometry = poFeature->GetGeometryRef();
   if (poGeometry) {
     ColumnDescriptor cd;
     cd.columnName = geo_column_name;
     cd.sourceName = geo_column_name;
 
     // get GDAL type
     auto ogr_type = wkbFlatten(poGeometry->getGeometryType());
 
     // if exploding, override any collection type to child type
     if (copy_params.geo_explode_collections) {
       if (ogr_type == wkbMultiPolygon) {
         ogr_type = wkbPolygon;
       } else if (ogr_type == wkbMultiLineString) {
         ogr_type = wkbLineString;
       } else if (ogr_type == wkbMultiPoint) {
         ogr_type = wkbPoint;
       }
     }
 
     // convert to internal type
     SQLTypes geoType = ogr_to_type(ogr_type);
 
     // for now, we promote POLYGON to MULTIPOLYGON (unless exploding)
     if (PROMOTE_POLYGON_TO_MULTIPOLYGON && !copy_params.geo_explode_collections) {
       geoType = (geoType == kPOLYGON) ? kMULTIPOLYGON : geoType;
     }
 
     // build full internal type
     SQLTypeInfo ti;
     ti.set_type(geoType);
     ti.set_subtype(copy_params.geo_coords_type);
     ti.set_input_srid(copy_params.geo_coords_srid);
     ti.set_output_srid(copy_params.geo_coords_srid);
     ti.set_compression(copy_params.geo_coords_encoding);
     ti.set_comp_param(copy_params.geo_coords_comp_param);
     cd.columnType = ti;
 
     cds.push_back(cd);
   }
   return cds;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

const std::list<const ColumnDescriptor*>& import_export::Importer::get_column_descs ( ) const

inline

Definition at line 776 of file Importer.h.

References loader.

Referenced by import_export::import_thread_delimited(), and import_export::import_thread_shapefile().

                                                                    {
     return loader->get_column_descs();
   }

Here is the caller graph for this function:

const CopyParams& import_export::Importer::get_copy_params ( ) const

inline

Definition at line 775 of file Importer.h.

References import_export::DataStreamSink::copy_params.

Referenced by import_export::import_thread_delimited(), and import_export::import_thread_shapefile().

775 { return copy_params; }

import_export::DataStreamSink::copy_params

CopyParams copy_params

Definition: Importer.h:674

Here is the caller graph for this function:

std::vector<std::unique_ptr<TypedImportBuffer> >& import_export::Importer::get_import_buffers ( int i )

inline

Definition at line 785 of file Importer.h.

References i, and import_buffers_vec.

Referenced by import_export::import_thread_delimited(), and import_export::import_thread_shapefile().

                                                                          {
     return import_buffers_vec[i];
   }

Here is the caller graph for this function:

std::vector<std::vector<std::unique_ptr<TypedImportBuffer> > >& import_export::Importer::get_import_buffers_vec ( )

inline

Definition at line 782 of file Importer.h.

References import_buffers_vec.

                                                                                    {
     return import_buffers_vec;
   }

ImportStatus import_export::Importer::get_import_status ( const std::string & id )

static

Definition at line 230 of file Importer.cpp.

References import_export::import_status_map, and import_export::status_mutex.

Referenced by DBHandler::import_table_status().

                                                                    {
   mapd_shared_lock<mapd_shared_mutex> read_lock(status_mutex);
   return import_status_map.at(import_id);
 }

Here is the caller graph for this function:

const bool* import_export::Importer::get_is_array ( ) const

inline

Definition at line 788 of file Importer.h.

References is_array_a.

Referenced by import_export::import_thread_delimited().

788 { return is_array_a.get(); }

import_export::Importer::is_array_a

std::unique_ptr< bool[]> is_array_a

Definition: Importer.h:858

Here is the caller graph for this function:

Catalog_Namespace::Catalog& import_export::Importer::getCatalog ( )

inline

Definition at line 821 of file Importer.h.

References loader.

Referenced by import_export::TypedImportBuffer::convert_arrow_val_to_import_buffer(), and import_export::import_thread_delimited().

821 { return loader->getCatalog(); }

import_export::Importer::loader

std::unique_ptr< Loader > loader

Definition: Importer.h:857

Here is the caller graph for this function:

auto import_export::Importer::getLoader ( ) const

inline

Definition at line 843 of file Importer.h.

References loader.

843 { return loader.get(); }

import_export::Importer::loader

std::unique_ptr< Loader > loader

Definition: Importer.h:857

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

Definition at line 4110 of file Importer.cpp.

References import_export::DataStreamSink::archivePlumber().

                                                                               {
   return DataStreamSink::archivePlumber(session_info);
 }

Here is the call graph for this function:

ImportStatus import_export::Importer::importDelimited	(	const std::string &	file_path,
		const bool	decompressed,
		const Catalog_Namespace::SessionInfo *	session_info
	)

overridevirtual

Implements import_export::DataStreamSink.

Definition at line 4114 of file Importer.cpp.

References import_export::CopyParams::buffer_size, CHECK, checkpoint(), ColumnDescriptor::columnId, ColumnDescriptor::columnName, ColumnDescriptor::columnType, import_export::DataStreamSink::copy_params, logger::ERROR, import_export::DataStreamSink::file_offsets, import_export::DataStreamSink::file_offsets_mutex, file_size, import_export::delimited_parser::find_row_end_pos(), omnisci::fopen(), g_max_import_threads, import_export::CopyParams::geo_assign_render_groups, Catalog_Namespace::SessionInfo::get_session_id(), SQLTypeInfo::get_type(), Executor::getExecutor(), i, import_buffers_vec, import_id, import_export::DataStreamSink::import_mutex_, import_export::DataStreamSink::import_status_, import_export::import_thread_delimited(), kMULTIPOLYGON, kPOLYGON, import_export::ImportStatus::load_failed, import_export::ImportStatus::load_msg, loader, LOG, import_export::CopyParams::max_reject, max_threads, import_export::DataStreamSink::p_file, import_export::ImportStatus::rows_completed, import_export::ImportStatus::rows_estimated, import_export::ImportStatus::rows_rejected, set_import_status(), logger::thread_id(), import_export::CopyParams::threads, import_export::DataStreamSink::total_file_size, Executor::UNITARY_EXECUTOR_ID, and VLOG.

                                                       {
   set_import_status(import_id, import_status_);
   auto query_session = session_info ? session_info->get_session_id() : "";
 
   if (!p_file) {
     p_file = fopen(file_path.c_str(), "rb");
   }
   if (!p_file) {
     throw std::runtime_error("failed to open file '" + file_path +
                              "': " + strerror(errno));
   }
 
   if (!decompressed) {
     (void)fseek(p_file, 0, SEEK_END);
     file_size = ftell(p_file);
   }
 
   if (copy_params.threads == 0) {
     max_threads = std::min(static_cast<size_t>(sysconf(_SC_NPROCESSORS_CONF)),
                            g_max_import_threads);
   } else {
     max_threads = static_cast<size_t>(copy_params.threads);
   }
   VLOG(1) << "Delimited import # threads: " << max_threads;
 
   // deal with small files
   size_t alloc_size = copy_params.buffer_size;
   if (!decompressed && file_size < alloc_size) {
     alloc_size = file_size;
   }
 
   for (size_t i = 0; i < max_threads; i++) {
     import_buffers_vec.emplace_back();
     for (const auto cd : loader->get_column_descs()) {
       import_buffers_vec[i].emplace_back(
           std::make_unique<TypedImportBuffer>(cd, loader->getStringDict(cd)));
     }
   }
 
   auto scratch_buffer = std::make_unique<char[]>(alloc_size);
   size_t current_pos = 0;
   size_t end_pos;
   size_t begin_pos = 0;
 
   (void)fseek(p_file, current_pos, SEEK_SET);
   size_t size =
       fread(reinterpret_cast<void*>(scratch_buffer.get()), 1, alloc_size, p_file);
 
   // make render group analyzers for each poly column
   ColumnIdToRenderGroupAnalyzerMapType columnIdToRenderGroupAnalyzerMap;
   if (copy_params.geo_assign_render_groups) {
     auto columnDescriptors = loader->getCatalog().getAllColumnMetadataForTable(
         loader->getTableDesc()->tableId, false, false, false);
     for (auto cd : columnDescriptors) {
       SQLTypes ct = cd->columnType.get_type();
       if (ct == kPOLYGON || ct == kMULTIPOLYGON) {
         auto rga = std::make_shared<RenderGroupAnalyzer>();
         rga->seedFromExistingTableContents(loader, cd->columnName);
         columnIdToRenderGroupAnalyzerMap[cd->columnId] = rga;
       }
     }
   }
 
   ChunkKey chunkKey = {loader->getCatalog().getCurrentDB().dbId,
                        loader->getTableDesc()->tableId};
   auto table_epochs = loader->getTableEpochs();
   auto executor = Executor::getExecutor(Executor::UNITARY_EXECUTOR_ID).get();
   {
     std::list<std::future<ImportStatus>> threads;
 
     // use a stack to track thread_ids which must not overlap among threads
     // because thread_id is used to index import_buffers_vec[]
     std::stack<size_t> stack_thread_ids;
     for (size_t i = 0; i < max_threads; i++) {
       stack_thread_ids.push(i);
     }
     // added for true row index on error
     size_t first_row_index_this_buffer = 0;
 
     while (size > 0) {
       unsigned int num_rows_this_buffer = 0;
       CHECK(scratch_buffer);
       end_pos = delimited_parser::find_row_end_pos(alloc_size,
                                                    scratch_buffer,
                                                    size,
                                                    copy_params,
                                                    first_row_index_this_buffer,
                                                    num_rows_this_buffer,
                                                    p_file);
 
       // unput residual
       int nresidual = size - end_pos;
       std::unique_ptr<char[]> unbuf;
       if (nresidual > 0) {
         unbuf = std::make_unique<char[]>(nresidual);
         memcpy(unbuf.get(), scratch_buffer.get() + end_pos, nresidual);
       }
 
       // get a thread_id not in use
       auto thread_id = stack_thread_ids.top();
       stack_thread_ids.pop();
       // LOG(INFO) << " stack_thread_ids.pop " << thread_id << std::endl;
 
       threads.push_back(std::async(std::launch::async,
                                    import_thread_delimited,
                                    thread_id,
                                    this,
                                    std::move(scratch_buffer),
                                    begin_pos,
                                    end_pos,
                                    end_pos,
                                    columnIdToRenderGroupAnalyzerMap,
                                    first_row_index_this_buffer,
                                    session_info,
                                    executor));
 
       first_row_index_this_buffer += num_rows_this_buffer;
 
       current_pos += end_pos;
       scratch_buffer = std::make_unique<char[]>(alloc_size);
       CHECK(scratch_buffer);
       memcpy(scratch_buffer.get(), unbuf.get(), nresidual);
       size = nresidual +
              fread(scratch_buffer.get() + nresidual, 1, alloc_size - nresidual, p_file);
 
       begin_pos = 0;
       while (threads.size() > 0) {
         int nready = 0;
         for (std::list<std::future<ImportStatus>>::iterator it = threads.begin();
              it != threads.end();) {
           auto& p = *it;
           std::chrono::milliseconds span(0);
           if (p.wait_for(span) == std::future_status::ready) {
             auto ret_import_status = p.get();
             {
               mapd_lock_guard<mapd_shared_mutex> write_lock(import_mutex_);
               import_status_ += ret_import_status;
               if (ret_import_status.load_failed) {
                 set_import_status(import_id, import_status_);
               }
             }
             // sum up current total file offsets
             size_t total_file_offset{0};
             if (decompressed) {
               std::unique_lock<std::mutex> lock(file_offsets_mutex);
               for (const auto file_offset : file_offsets) {
                 total_file_offset += file_offset;
               }
             }
             // estimate number of rows per current total file offset
             if (decompressed ? total_file_offset : current_pos) {
               import_status_.rows_estimated =
                   (decompressed ? (float)total_file_size / total_file_offset
                                 : (float)file_size / current_pos) *
                   import_status_.rows_completed;
             }
             VLOG(3) << "rows_completed " << import_status_.rows_completed
                     << ", rows_estimated " << import_status_.rows_estimated
                     << ", total_file_size " << total_file_size << ", total_file_offset "
                     << total_file_offset;
             set_import_status(import_id, import_status_);
             // recall thread_id for reuse
             stack_thread_ids.push(ret_import_status.thread_id);
             threads.erase(it++);
             ++nready;
           } else {
             ++it;
           }
         }
 
         if (nready == 0) {
           std::this_thread::yield();
         }
 
         // on eof, wait all threads to finish
         if (0 == size) {
           continue;
         }
 
         // keep reading if any free thread slot
         // this is one of the major difference from old threading model !!
         if (threads.size() < max_threads) {
           break;
         }
         mapd_shared_lock<mapd_shared_mutex> read_lock(import_mutex_);
         if (import_status_.load_failed) {
           break;
         }
       }
       mapd_unique_lock<mapd_shared_mutex> write_lock(import_mutex_);
       if (import_status_.rows_rejected > copy_params.max_reject) {
         import_status_.load_failed = true;
         // todo use better message
         import_status_.load_msg = "Maximum rows rejected exceeded. Halting load";
         LOG(ERROR) << "Maximum rows rejected exceeded. Halting load";
         break;
       }
       if (import_status_.load_failed) {
         LOG(ERROR) << "Load failed, the issue was: " + import_status_.load_msg;
         break;
       }
     }
 
     // join dangling threads in case of LOG(ERROR) above
     for (auto& p : threads) {
       p.wait();
     }
   }
 
   checkpoint(table_epochs);
 
   fclose(p_file);
   p_file = nullptr;
   return import_status_;
 }

Here is the call graph for this function:

ImportStatus import_export::Importer::importGDAL	(	std::map< std::string, std::string >	colname_to_src,
		const Catalog_Namespace::SessionInfo *	session_info
	)

Definition at line 4904 of file Importer.cpp.

References CHECK, CHECK_EQ, checkpoint(), ColumnDescriptor::columnId, ColumnDescriptor::columnName, ColumnDescriptor::columnType, import_export::DataStreamSink::copy_params, logger::ERROR, g_enable_non_kernel_time_query_interrupt, g_max_import_threads, import_export::CopyParams::geo_assign_render_groups, import_export::CopyParams::geo_coords_srid, import_export::CopyParams::geo_layer_name, Catalog_Namespace::SessionInfo::get_session_id(), SQLTypeInfo::get_type(), Executor::getExecutor(), import_export::anonymous_namespace{Importer.cpp}::getLayerWithSpecifiedName(), i, import_buffers_vec, import_id, import_export::DataStreamSink::import_mutex_, import_export::DataStreamSink::import_status_, import_export::import_thread_shapefile(), kMULTIPOLYGON, kPOLYGON, import_export::ImportStatus::load_failed, import_export::ImportStatus::load_msg, loader, LOG, import_export::CopyParams::max_reject, max_threads, openGDALDataset(), import_export::ImportStatus::rows_completed, import_export::ImportStatus::rows_estimated, import_export::ImportStatus::rows_rejected, set_import_status(), logger::thread_id(), import_export::CopyParams::threads, toString(), Executor::UNITARY_EXECUTOR_ID, and VLOG.

Referenced by QueryRunner::ImportDriver::importGeoTable().

                                                                                     {
   // initial status
   set_import_status(import_id, import_status_);
   OGRDataSourceUqPtr poDS(openGDALDataset(file_path, copy_params));
   if (poDS == nullptr) {
     throw std::runtime_error("openGDALDataset Error: Unable to open geo file " +
                              file_path);
   }
 
   OGRLayer& layer =
       getLayerWithSpecifiedName(copy_params.geo_layer_name, poDS, file_path);
 
   // get the number of features in this layer
   size_t numFeatures = layer.GetFeatureCount();
 
   // build map of metadata field (additional columns) name to index
   // use shared_ptr since we need to pass it to the worker
   FieldNameToIndexMapType fieldNameToIndexMap;
   OGRFeatureDefn* poFDefn = layer.GetLayerDefn();
   CHECK(poFDefn);
   size_t numFields = poFDefn->GetFieldCount();
   for (size_t iField = 0; iField < numFields; iField++) {
     OGRFieldDefn* poFieldDefn = poFDefn->GetFieldDefn(iField);
     fieldNameToIndexMap.emplace(std::make_pair(poFieldDefn->GetNameRef(), iField));
   }
 
   // the geographic spatial reference we want to put everything in
   OGRSpatialReferenceUqPtr poGeographicSR(new OGRSpatialReference());
   poGeographicSR->importFromEPSG(copy_params.geo_coords_srid);
 
 #if GDAL_VERSION_MAJOR >= 3
   // GDAL 3.x (really Proj.4 6.x) now enforces lat, lon order
   // this results in X and Y being transposed for angle-based
   // coordinate systems. This restores the previous behavior.
   poGeographicSR->SetAxisMappingStrategy(OAMS_TRADITIONAL_GIS_ORDER);
 #endif
 
 #if DISABLE_MULTI_THREADED_SHAPEFILE_IMPORT
   // just one "thread"
   max_threads = 1;
 #else
   // how many threads to use
   if (copy_params.threads == 0) {
     max_threads = std::min(static_cast<size_t>(sysconf(_SC_NPROCESSORS_CONF)),
                            g_max_import_threads);
   } else {
     max_threads = copy_params.threads;
   }
 #endif
 
   VLOG(1) << "GDAL import # threads: " << max_threads;
 
   // import geo table is specifically handled in both DBHandler and QueryRunner
   // that is separate path against a normal SQL execution
   // so we here explicitly enroll the import session to allow interruption
   // while importing geo table
   auto query_session = session_info ? session_info->get_session_id() : "";
   auto query_submitted_time = ::toString(std::chrono::system_clock::now());
   auto executor = Executor::getExecutor(Executor::UNITARY_EXECUTOR_ID);
   if (g_enable_non_kernel_time_query_interrupt && !query_session.empty()) {
     executor->enrollQuerySession(query_session,
                                  "Import Geo Table",
                                  query_submitted_time,
                                  Executor::UNITARY_EXECUTOR_ID,
                                  QuerySessionStatus::QueryStatus::RUNNING);
   }
 
   ScopeGuard clearInterruptStatus = [executor, &query_session, &query_submitted_time] {
     // reset the runtime query interrupt status
     if (g_enable_non_kernel_time_query_interrupt && !query_session.empty()) {
       executor->clearQuerySessionStatus(query_session, query_submitted_time, false);
     }
   };
 
   // make an import buffer for each thread
   CHECK_EQ(import_buffers_vec.size(), 0u);
   import_buffers_vec.resize(max_threads);
   for (size_t i = 0; i < max_threads; i++) {
     for (const auto cd : loader->get_column_descs()) {
       import_buffers_vec[i].emplace_back(
           new TypedImportBuffer(cd, loader->getStringDict(cd)));
     }
   }
 
   // make render group analyzers for each poly column
   ColumnIdToRenderGroupAnalyzerMapType columnIdToRenderGroupAnalyzerMap;
   if (copy_params.geo_assign_render_groups) {
     auto columnDescriptors = loader->getCatalog().getAllColumnMetadataForTable(
         loader->getTableDesc()->tableId, false, false, false);
     for (auto cd : columnDescriptors) {
       SQLTypes ct = cd->columnType.get_type();
       if (ct == kPOLYGON || ct == kMULTIPOLYGON) {
         auto rga = std::make_shared<RenderGroupAnalyzer>();
         rga->seedFromExistingTableContents(loader, cd->columnName);
         columnIdToRenderGroupAnalyzerMap[cd->columnId] = rga;
       }
     }
   }
 
 #if !DISABLE_MULTI_THREADED_SHAPEFILE_IMPORT
   // threads
   std::list<std::future<ImportStatus>> threads;
 
   // use a stack to track thread_ids which must not overlap among threads
   // because thread_id is used to index import_buffers_vec[]
   std::stack<size_t> stack_thread_ids;
   for (size_t i = 0; i < max_threads; i++) {
     stack_thread_ids.push(i);
   }
 #endif
 
   // checkpoint the table
   auto table_epochs = loader->getTableEpochs();
 
   // reset the layer
   layer.ResetReading();
 
   static const size_t MAX_FEATURES_PER_CHUNK = 1000;
 
   // make a features buffer for each thread
   std::vector<FeaturePtrVector> features(max_threads);
 
   // for each feature...
   size_t firstFeatureThisChunk = 0;
   while (firstFeatureThisChunk < numFeatures) {
     // how many features this chunk
     size_t numFeaturesThisChunk =
         std::min(MAX_FEATURES_PER_CHUNK, numFeatures - firstFeatureThisChunk);
 
 // get a thread_id not in use
 #if DISABLE_MULTI_THREADED_SHAPEFILE_IMPORT
     size_t thread_id = 0;
 #else
     auto thread_id = stack_thread_ids.top();
     stack_thread_ids.pop();
     CHECK(thread_id < max_threads);
 #endif
 
     // fill features buffer for new thread
     for (size_t i = 0; i < numFeaturesThisChunk; i++) {
       features[thread_id].emplace_back(layer.GetNextFeature());
     }
 
 #if DISABLE_MULTI_THREADED_SHAPEFILE_IMPORT
     // call worker function directly
     auto ret_import_status = import_thread_shapefile(0,
                                                      this,
                                                      poGeographicSR.get(),
                                                      std::move(features[thread_id]),
                                                      firstFeatureThisChunk,
                                                      numFeaturesThisChunk,
                                                      fieldNameToIndexMap,
                                                      columnNameToSourceNameMap,
                                                      columnIdToRenderGroupAnalyzerMap,
                                                      session_info,
                                                      executor.get());
     import_status += ret_import_status;
     import_status.rows_estimated = ((float)firstFeatureThisChunk / (float)numFeatures) *
                                    import_status.rows_completed;
     set_import_status(import_id, import_status);
 #else
     // fire up that thread to import this geometry
     threads.push_back(std::async(std::launch::async,
                                  import_thread_shapefile,
                                  thread_id,
                                  this,
                                  poGeographicSR.get(),
                                  std::move(features[thread_id]),
                                  firstFeatureThisChunk,
                                  numFeaturesThisChunk,
                                  fieldNameToIndexMap,
                                  columnNameToSourceNameMap,
                                  columnIdToRenderGroupAnalyzerMap,
                                  session_info,
                                  executor.get()));
 
     // let the threads run
     while (threads.size() > 0) {
       int nready = 0;
       for (std::list<std::future<ImportStatus>>::iterator it = threads.begin();
            it != threads.end();) {
         auto& p = *it;
         std::chrono::milliseconds span(
             0);  //(std::distance(it, threads.end()) == 1? 1: 0);
         if (p.wait_for(span) == std::future_status::ready) {
           auto ret_import_status = p.get();
           {
             mapd_lock_guard<mapd_shared_mutex> write_lock(import_mutex_);
             import_status_ += ret_import_status;
             import_status_.rows_estimated =
                 ((float)firstFeatureThisChunk / (float)numFeatures) *
                 import_status_.rows_completed;
             set_import_status(import_id, import_status_);
             if (import_status_.load_failed) {
               break;
             }
           }
           // recall thread_id for reuse
           stack_thread_ids.push(ret_import_status.thread_id);
 
           threads.erase(it++);
           ++nready;
         } else {
           ++it;
         }
       }
 
       if (nready == 0) {
         std::this_thread::yield();
       }
 
       // keep reading if any free thread slot
       // this is one of the major difference from old threading model !!
       if (threads.size() < max_threads) {
         break;
       }
     }
 #endif
 
     // out of rows?
 
     mapd_unique_lock<mapd_shared_mutex> write_lock(import_mutex_);
     if (import_status_.rows_rejected > copy_params.max_reject) {
       import_status_.load_failed = true;
       // todo use better message
       import_status_.load_msg = "Maximum rows rejected exceeded. Halting load";
       LOG(ERROR) << "Maximum rows rejected exceeded. Halting load";
       break;
     }
     if (import_status_.load_failed) {
       LOG(ERROR) << "A call to the Loader failed in GDAL, Please review the logs for "
                     "more details";
       break;
     }
 
     firstFeatureThisChunk += numFeaturesThisChunk;
   }
 
 #if !DISABLE_MULTI_THREADED_SHAPEFILE_IMPORT
   // wait for any remaining threads
   if (threads.size()) {
     for (auto& p : threads) {
       // wait for the thread
       p.wait();
       // get the result and update the final import status
       auto ret_import_status = p.get();
       import_status_ += ret_import_status;
       import_status_.rows_estimated = import_status_.rows_completed;
       set_import_status(import_id, import_status_);
     }
   }
 #endif
 
   checkpoint(table_epochs);
 
   return import_status_;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void import_export::Importer::load	(	const std::vector< std::unique_ptr< TypedImportBuffer >> &	import_buffers,
		size_t	row_count,
		const Catalog_Namespace::SessionInfo *	session_info
	)

Definition at line 3369 of file Importer.cpp.

References import_export::DataStreamSink::import_mutex_, import_export::DataStreamSink::import_status_, import_export::ImportStatus::load_failed, import_export::ImportStatus::load_msg, and loader.

Referenced by import_export::import_thread_delimited(), and import_export::import_thread_shapefile().

                                                                       {
   if (!loader->loadNoCheckpoint(import_buffers, row_count, session_info)) {
     mapd_lock_guard<mapd_shared_mutex> write_lock(import_mutex_);
     import_status_.load_failed = true;
     import_status_.load_msg = loader->getErrorMessage();
   }
 }

Here is the caller graph for this function:

OGRDataSource * import_export::Importer::openGDALDataset	(	const std::string &	fileName,
		const CopyParams &	copy_params
	)

staticprivate

Definition at line 4422 of file Importer.cpp.

References logger::ERROR, logger::INFO, Geospatial::GDAL::init(), LOG, and setGDALAuthorizationTokens().

Referenced by gdalGetLayersInGeoFile(), gdalToColumnDescriptors(), importGDAL(), and readMetadataSampleGDAL().

                                                                         {
   // lazy init GDAL
   Geospatial::GDAL::init();
 
   // set authorization tokens
   setGDALAuthorizationTokens(copy_params);
 
   // open the file
   OGRDataSource* poDS;
 #if GDAL_VERSION_MAJOR == 1
   poDS = (OGRDataSource*)OGRSFDriverRegistrar::Open(file_name.c_str(), false);
 #else
   poDS = (OGRDataSource*)GDALOpenEx(
       file_name.c_str(), GDAL_OF_VECTOR, nullptr, nullptr, nullptr);
   if (poDS == nullptr) {
     poDS = (OGRDataSource*)GDALOpenEx(
         file_name.c_str(), GDAL_OF_READONLY | GDAL_OF_VECTOR, nullptr, nullptr, nullptr);
     if (poDS) {
       LOG(INFO) << "openGDALDataset had to open as read-only";
     }
   }
 #endif
   if (poDS == nullptr) {
     LOG(ERROR) << "openGDALDataset Error: " << CPLGetLastErrorMsg();
   }
   // NOTE(adb): If extending this function, refactor to ensure any errors will not
   // result in a memory leak if GDAL successfully opened the input dataset.
   return poDS;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void import_export::Importer::readMetadataSampleGDAL	(	const std::string &	fileName,
		const std::string &	geoColumnName,
		std::map< std::string, std::vector< std::string >> &	metadata,
		int	rowLimit,
		const CopyParams &	copy_params
	)

static

Definition at line 4478 of file Importer.cpp.

References CHECK, import_export::CopyParams::geo_explode_collections, import_export::CopyParams::geo_layer_name, import_export::anonymous_namespace{Importer.cpp}::getLayerWithSpecifiedName(), i, and openGDALDataset().

Referenced by DBHandler::detect_column_types().

                                    {
   OGRDataSourceUqPtr poDS(openGDALDataset(file_name, copy_params));
   if (poDS == nullptr) {
     throw std::runtime_error("openGDALDataset Error: Unable to open geo file " +
                              file_name);
   }
 
   OGRLayer& layer =
       getLayerWithSpecifiedName(copy_params.geo_layer_name, poDS, file_name);
 
   OGRFeatureDefn* poFDefn = layer.GetLayerDefn();
   CHECK(poFDefn);
 
   // typeof GetFeatureCount() is different between GDAL 1.x (int32_t) and 2.x (int64_t)
   auto nFeats = layer.GetFeatureCount();
   size_t numFeatures =
       std::max(static_cast<decltype(nFeats)>(0),
                std::min(static_cast<decltype(nFeats)>(rowLimit), nFeats));
   for (auto iField = 0; iField < poFDefn->GetFieldCount(); iField++) {
     OGRFieldDefn* poFieldDefn = poFDefn->GetFieldDefn(iField);
     // FIXME(andrewseidl): change this to the faster one used by readVerticesFromGDAL
     metadata.emplace(poFieldDefn->GetNameRef(), std::vector<std::string>(numFeatures));
   }
   metadata.emplace(geo_column_name, std::vector<std::string>(numFeatures));
   layer.ResetReading();
   size_t iFeature = 0;
   while (iFeature < numFeatures) {
     OGRFeatureUqPtr poFeature(layer.GetNextFeature());
     if (!poFeature) {
       break;
     }
 
     OGRGeometry* poGeometry = poFeature->GetGeometryRef();
     if (poGeometry != nullptr) {
       // validate geom type (again?)
       switch (wkbFlatten(poGeometry->getGeometryType())) {
         case wkbPoint:
         case wkbLineString:
         case wkbPolygon:
         case wkbMultiPolygon:
           break;
         case wkbMultiPoint:
         case wkbMultiLineString:
           // supported if geo_explode_collections is specified
           if (!copy_params.geo_explode_collections) {
             throw std::runtime_error("Unsupported geometry type: " +
                                      std::string(poGeometry->getGeometryName()));
           }
           break;
         default:
           throw std::runtime_error("Unsupported geometry type: " +
                                    std::string(poGeometry->getGeometryName()));
       }
 
       // populate metadata for regular fields
       for (auto i : metadata) {
         auto iField = poFeature->GetFieldIndex(i.first.c_str());
         if (iField >= 0) {  // geom is -1
           metadata[i.first].at(iFeature) =
               std::string(poFeature->GetFieldAsString(iField));
         }
       }
 
       // populate metadata for geo column with WKT string
       char* wkts = nullptr;
       poGeometry->exportToWkt(&wkts);
       CHECK(wkts);
       metadata[geo_column_name].at(iFeature) = wkts;
       CPLFree(wkts);
     }
     iFeature++;
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void import_export::Importer::set_geo_physical_import_buffer	(	const Catalog_Namespace::Catalog &	catalog,
		const ColumnDescriptor *	cd,
		std::vector< std::unique_ptr< TypedImportBuffer >> &	import_buffers,
		size_t &	col_idx,
		std::vector< double > &	coords,
		std::vector< double > &	bounds,
		std::vector< int > &	ring_sizes,
		std::vector< int > &	poly_rings,
		int	render_group
	)

static

Definition at line 1459 of file Importer.cpp.

References ColumnDescriptor::columnId, ColumnDescriptor::columnType, Geospatial::compress_coords(), SQLTypeInfo::get_type(), Catalog_Namespace::Catalog::getMetadataForColumn(), Geospatial::is_null_point(), kLINESTRING, kMULTIPOLYGON, kPOINT, kPOLYGON, NULL_ARRAY_DOUBLE, NULL_DOUBLE, and ColumnDescriptor::tableId.

Referenced by import_export::TypedImportBuffer::convert_arrow_val_to_import_buffer(), Parser::AddColumnStmt::execute(), import_export::fill_missing_columns(), import_export::import_thread_delimited(), DBHandler::load_table(), and foreign_storage::csv_file_buffer_parser::process_geo_column().

                       {
   const auto col_ti = cd->columnType;
   const auto col_type = col_ti.get_type();
   auto columnId = cd->columnId;
   auto cd_coords = catalog.getMetadataForColumn(cd->tableId, ++columnId);
   bool is_null_geo = false;
   bool is_null_point = false;
   if (!col_ti.get_notnull()) {
     // Check for NULL geo
     if (col_type == kPOINT && (coords.empty() || coords[0] == NULL_ARRAY_DOUBLE)) {
       is_null_point = true;
       coords.clear();
     }
     is_null_geo = coords.empty();
     if (is_null_point) {
       coords.push_back(NULL_ARRAY_DOUBLE);
       coords.push_back(NULL_DOUBLE);
       // Treating POINT coords as notnull, need to store actual encoding
       // [un]compressed+[not]null
       is_null_geo = false;
     }
   }
   TDatum tdd_coords;
   // Get the raw data representing [optionally compressed] non-NULL geo's coords.
   // One exception - NULL POINT geo: coords need to be processed to encode nullness
   // in a fixlen array, compressed and uncompressed.
   if (!is_null_geo) {
     std::vector<uint8_t> compressed_coords = Geospatial::compress_coords(coords, col_ti);
     tdd_coords.val.arr_val.reserve(compressed_coords.size());
     for (auto cc : compressed_coords) {
       tdd_coords.val.arr_val.emplace_back();
       tdd_coords.val.arr_val.back().val.int_val = cc;
     }
   }
   tdd_coords.is_null = is_null_geo;
   import_buffers[col_idx++]->add_value(cd_coords, tdd_coords, false);
 
   if (col_type == kPOLYGON || col_type == kMULTIPOLYGON) {
     // Create ring_sizes array value and add it to the physical column
     auto cd_ring_sizes = catalog.getMetadataForColumn(cd->tableId, ++columnId);
     TDatum tdd_ring_sizes;
     tdd_ring_sizes.val.arr_val.reserve(ring_sizes.size());
     if (!is_null_geo) {
       for (auto ring_size : ring_sizes) {
         tdd_ring_sizes.val.arr_val.emplace_back();
         tdd_ring_sizes.val.arr_val.back().val.int_val = ring_size;
       }
     }
     tdd_ring_sizes.is_null = is_null_geo;
     import_buffers[col_idx++]->add_value(cd_ring_sizes, tdd_ring_sizes, false);
   }
 
   if (col_type == kMULTIPOLYGON) {
     // Create poly_rings array value and add it to the physical column
     auto cd_poly_rings = catalog.getMetadataForColumn(cd->tableId, ++columnId);
     TDatum tdd_poly_rings;
     tdd_poly_rings.val.arr_val.reserve(poly_rings.size());
     if (!is_null_geo) {
       for (auto num_rings : poly_rings) {
         tdd_poly_rings.val.arr_val.emplace_back();
         tdd_poly_rings.val.arr_val.back().val.int_val = num_rings;
       }
     }
     tdd_poly_rings.is_null = is_null_geo;
     import_buffers[col_idx++]->add_value(cd_poly_rings, tdd_poly_rings, false);
   }
 
   if (col_type == kLINESTRING || col_type == kPOLYGON || col_type == kMULTIPOLYGON) {
     auto cd_bounds = catalog.getMetadataForColumn(cd->tableId, ++columnId);
     TDatum tdd_bounds;
     tdd_bounds.val.arr_val.reserve(bounds.size());
     if (!is_null_geo) {
       for (auto b : bounds) {
         tdd_bounds.val.arr_val.emplace_back();
         tdd_bounds.val.arr_val.back().val.real_val = b;
       }
     }
     tdd_bounds.is_null = is_null_geo;
     import_buffers[col_idx++]->add_value(cd_bounds, tdd_bounds, false);
   }
 
   if (col_type == kPOLYGON || col_type == kMULTIPOLYGON) {
     // Create render_group value and add it to the physical column
     auto cd_render_group = catalog.getMetadataForColumn(cd->tableId, ++columnId);
     TDatum td_render_group;
     td_render_group.val.int_val = render_group;
     td_render_group.is_null = is_null_geo;
     import_buffers[col_idx++]->add_value(cd_render_group, td_render_group, false);
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void import_export::Importer::set_geo_physical_import_buffer_columnar	(	const Catalog_Namespace::Catalog &	catalog,
		const ColumnDescriptor *	cd,
		std::vector< std::unique_ptr< TypedImportBuffer >> &	import_buffers,
		size_t &	col_idx,
		std::vector< std::vector< double >> &	coords_column,
		std::vector< std::vector< double >> &	bounds_column,
		std::vector< std::vector< int >> &	ring_sizes_column,
		std::vector< std::vector< int >> &	poly_rings_column,
		std::vector< int > &	render_groups_column
	)

static

Definition at line 1559 of file Importer.cpp.

References CHECK, ColumnDescriptor::columnId, ColumnDescriptor::columnType, Geospatial::compress_coords(), SQLTypeInfo::get_type(), Catalog_Namespace::Catalog::getMetadataForColumn(), Geospatial::is_null_point(), kLINESTRING, kMULTIPOLYGON, kPOINT, kPOLYGON, NULL_ARRAY_DOUBLE, NULL_DOUBLE, and ColumnDescriptor::tableId.

Referenced by DBHandler::load_table_binary_columnar_internal().

                                           {
   const auto col_ti = cd->columnType;
   const auto col_type = col_ti.get_type();
   auto columnId = cd->columnId;
 
   auto coords_row_count = coords_column.size();
   auto cd_coords = catalog.getMetadataForColumn(cd->tableId, ++columnId);
   for (auto& coords : coords_column) {
     bool is_null_geo = false;
     bool is_null_point = false;
     if (!col_ti.get_notnull()) {
       // Check for NULL geo
       if (col_type == kPOINT && (coords.empty() || coords[0] == NULL_ARRAY_DOUBLE)) {
         is_null_point = true;
         coords.clear();
       }
       is_null_geo = coords.empty();
       if (is_null_point) {
         coords.push_back(NULL_ARRAY_DOUBLE);
         coords.push_back(NULL_DOUBLE);
         // Treating POINT coords as notnull, need to store actual encoding
         // [un]compressed+[not]null
         is_null_geo = false;
       }
     }
     std::vector<TDatum> td_coords_data;
     if (!is_null_geo) {
       std::vector<uint8_t> compressed_coords =
           Geospatial::compress_coords(coords, col_ti);
       for (auto const& cc : compressed_coords) {
         TDatum td_byte;
         td_byte.val.int_val = cc;
         td_coords_data.push_back(td_byte);
       }
     }
     TDatum tdd_coords;
     tdd_coords.val.arr_val = td_coords_data;
     tdd_coords.is_null = is_null_geo;
     import_buffers[col_idx]->add_value(cd_coords, tdd_coords, false);
   }
   col_idx++;
 
   if (col_type == kPOLYGON || col_type == kMULTIPOLYGON) {
     if (ring_sizes_column.size() != coords_row_count) {
       CHECK(false) << "Geometry import columnar: ring sizes column size mismatch";
     }
     // Create ring_sizes array value and add it to the physical column
     auto cd_ring_sizes = catalog.getMetadataForColumn(cd->tableId, ++columnId);
     for (auto const& ring_sizes : ring_sizes_column) {
       bool is_null_geo = false;
       if (!col_ti.get_notnull()) {
         // Check for NULL geo
         is_null_geo = ring_sizes.empty();
       }
       std::vector<TDatum> td_ring_sizes;
       for (auto const& ring_size : ring_sizes) {
         TDatum td_ring_size;
         td_ring_size.val.int_val = ring_size;
         td_ring_sizes.push_back(td_ring_size);
       }
       TDatum tdd_ring_sizes;
       tdd_ring_sizes.val.arr_val = td_ring_sizes;
       tdd_ring_sizes.is_null = is_null_geo;
       import_buffers[col_idx]->add_value(cd_ring_sizes, tdd_ring_sizes, false);
     }
     col_idx++;
   }
 
   if (col_type == kMULTIPOLYGON) {
     if (poly_rings_column.size() != coords_row_count) {
       CHECK(false) << "Geometry import columnar: poly rings column size mismatch";
     }
     // Create poly_rings array value and add it to the physical column
     auto cd_poly_rings = catalog.getMetadataForColumn(cd->tableId, ++columnId);
     for (auto const& poly_rings : poly_rings_column) {
       bool is_null_geo = false;
       if (!col_ti.get_notnull()) {
         // Check for NULL geo
         is_null_geo = poly_rings.empty();
       }
       std::vector<TDatum> td_poly_rings;
       for (auto const& num_rings : poly_rings) {
         TDatum td_num_rings;
         td_num_rings.val.int_val = num_rings;
         td_poly_rings.push_back(td_num_rings);
       }
       TDatum tdd_poly_rings;
       tdd_poly_rings.val.arr_val = td_poly_rings;
       tdd_poly_rings.is_null = is_null_geo;
       import_buffers[col_idx]->add_value(cd_poly_rings, tdd_poly_rings, false);
     }
     col_idx++;
   }
 
   if (col_type == kLINESTRING || col_type == kPOLYGON || col_type == kMULTIPOLYGON) {
     if (bounds_column.size() != coords_row_count) {
       CHECK(false) << "Geometry import columnar: bounds column size mismatch";
     }
     auto cd_bounds = catalog.getMetadataForColumn(cd->tableId, ++columnId);
     for (auto const& bounds : bounds_column) {
       bool is_null_geo = false;
       if (!col_ti.get_notnull()) {
         // Check for NULL geo
         is_null_geo = (bounds.empty() || bounds[0] == NULL_ARRAY_DOUBLE);
       }
       std::vector<TDatum> td_bounds_data;
       for (auto const& b : bounds) {
         TDatum td_double;
         td_double.val.real_val = b;
         td_bounds_data.push_back(td_double);
       }
       TDatum tdd_bounds;
       tdd_bounds.val.arr_val = td_bounds_data;
       tdd_bounds.is_null = is_null_geo;
       import_buffers[col_idx]->add_value(cd_bounds, tdd_bounds, false);
     }
     col_idx++;
   }
 
   if (col_type == kPOLYGON || col_type == kMULTIPOLYGON) {
     // Create render_group value and add it to the physical column
     auto cd_render_group = catalog.getMetadataForColumn(cd->tableId, ++columnId);
     for (auto const& render_group : render_groups_column) {
       TDatum td_render_group;
       td_render_group.val.int_val = render_group;
       td_render_group.is_null = false;
       import_buffers[col_idx]->add_value(cd_render_group, td_render_group, false);
     }
     col_idx++;
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void import_export::Importer::set_import_status	(	const std::string &	id,
		const ImportStatus	is
	)

static

Definition at line 235 of file Importer.cpp.

References import_export::ImportStatus::elapsed, import_export::ImportStatus::end, import_id, import_export::import_status_map, import_export::ImportStatus::start, and import_export::status_mutex.

Referenced by importDelimited(), and importGDAL().

                                                                             {
   mapd_lock_guard<mapd_shared_mutex> write_lock(status_mutex);
   is.end = std::chrono::steady_clock::now();
   is.elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(is.end - is.start);
   import_status_map[import_id] = is;
 }

Here is the caller graph for this function:

void import_export::Importer::setGDALAuthorizationTokens ( const CopyParams & copy_params )

staticprivate

Definition at line 4352 of file Importer.cpp.

References logger::INFO, LOG, import_export::CopyParams::s3_access_key, import_export::CopyParams::s3_endpoint, import_export::CopyParams::s3_region, and import_export::CopyParams::s3_secret_key.

Referenced by gdalGetAllFilesInArchive(), gdalGetLayersInGeoFile(), gdalStatInternal(), and openGDALDataset().

                                                                        {
   // for now we only support S3
   // @TODO generalize CopyParams to have a dictionary of GDAL tokens
   // only set if non-empty to allow GDAL defaults to persist
   // explicitly clear if empty to revert to default and not reuse a previous session's
   // keys
   if (copy_params.s3_region.size()) {
 #if DEBUG_AWS_AUTHENTICATION
     LOG(INFO) << "GDAL: Setting AWS_REGION to '" << copy_params.s3_region << "'";
 #endif
     CPLSetConfigOption("AWS_REGION", copy_params.s3_region.c_str());
   } else {
 #if DEBUG_AWS_AUTHENTICATION
     LOG(INFO) << "GDAL: Clearing AWS_REGION";
 #endif
     CPLSetConfigOption("AWS_REGION", nullptr);
   }
   if (copy_params.s3_endpoint.size()) {
 #if DEBUG_AWS_AUTHENTICATION
     LOG(INFO) << "GDAL: Setting AWS_S3_ENDPOINT to '" << copy_params.s3_endpoint << "'";
 #endif
     CPLSetConfigOption("AWS_S3_ENDPOINT", copy_params.s3_endpoint.c_str());
   } else {
 #if DEBUG_AWS_AUTHENTICATION
     LOG(INFO) << "GDAL: Clearing AWS_S3_ENDPOINT";
 #endif
     CPLSetConfigOption("AWS_S3_ENDPOINT", nullptr);
   }
   if (copy_params.s3_access_key.size()) {
 #if DEBUG_AWS_AUTHENTICATION
     LOG(INFO) << "GDAL: Setting AWS_ACCESS_KEY_ID to '" << copy_params.s3_access_key
               << "'";
 #endif
     CPLSetConfigOption("AWS_ACCESS_KEY_ID", copy_params.s3_access_key.c_str());
   } else {
 #if DEBUG_AWS_AUTHENTICATION
     LOG(INFO) << "GDAL: Clearing AWS_ACCESS_KEY_ID";
 #endif
     CPLSetConfigOption("AWS_ACCESS_KEY_ID", nullptr);
   }
   if (copy_params.s3_secret_key.size()) {
 #if DEBUG_AWS_AUTHENTICATION
     LOG(INFO) << "GDAL: Setting AWS_SECRET_ACCESS_KEY to '" << copy_params.s3_secret_key
               << "'";
 #endif
     CPLSetConfigOption("AWS_SECRET_ACCESS_KEY", copy_params.s3_secret_key.c_str());
   } else {
 #if DEBUG_AWS_AUTHENTICATION
     LOG(INFO) << "GDAL: Clearing AWS_SECRET_ACCESS_KEY";
 #endif
     CPLSetConfigOption("AWS_SECRET_ACCESS_KEY", nullptr);
   }
 
 #if (GDAL_VERSION_MAJOR > 2) || (GDAL_VERSION_MAJOR == 2 && GDAL_VERSION_MINOR >= 3)
   // if we haven't set keys, we need to disable signed access
   if (copy_params.s3_access_key.size() || copy_params.s3_secret_key.size()) {
 #if DEBUG_AWS_AUTHENTICATION
     LOG(INFO) << "GDAL: Clearing AWS_NO_SIGN_REQUEST";
 #endif
     CPLSetConfigOption("AWS_NO_SIGN_REQUEST", nullptr);
   } else {
 #if DEBUG_AWS_AUTHENTICATION
     LOG(INFO) << "GDAL: Setting AWS_NO_SIGN_REQUEST to 'YES'";
 #endif
     CPLSetConfigOption("AWS_NO_SIGN_REQUEST", "YES");
   }
 #endif
 }

Here is the caller graph for this function:

Member Data Documentation

char* import_export::Importer::buffer[2]

private

Definition at line 855 of file Importer.h.

Referenced by Importer(), and ~Importer().

size_t import_export::Importer::file_size

private

Definition at line 853 of file Importer.h.

Referenced by importDelimited(), and Importer().

std::vector<std::vector<std::unique_ptr<TypedImportBuffer> > > import_export::Importer::import_buffers_vec

private

Definition at line 856 of file Importer.h.

Referenced by checkpoint(), get_import_buffers(), get_import_buffers_vec(), importDelimited(), and importGDAL().

std::string import_export::Importer::import_id

private

Definition at line 852 of file Importer.h.

Referenced by importDelimited(), Importer(), importGDAL(), omnisci.thrift.OmniSci.import_table_status_args::read(), set_import_status(), and omnisci.thrift.OmniSci.import_table_status_args::write().

std::mutex import_export::Importer::init_gdal_mutex

staticprivate

Definition at line 859 of file Importer.h.

bool import_export::Importer::interrupted

private

Definition at line 860 of file Importer.h.

std::unique_ptr<bool[]> import_export::Importer::is_array_a

private

Definition at line 858 of file Importer.h.

Referenced by get_is_array(), and Importer().

std::unique_ptr<Loader> import_export::Importer::loader

private

Definition at line 857 of file Importer.h.

Referenced by checkpoint(), get_column_descs(), getCatalog(), getLoader(), importDelimited(), Importer(), importGDAL(), and load().

size_t import_export::Importer::max_threads

private

Definition at line 854 of file Importer.h.

Referenced by importDelimited(), Importer(), and importGDAL().

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

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

Classes

Public Types

Public Member Functions

Static Public Member Functions

Static Private Member Functions

Private Attributes

Static Private Attributes

Additional Inherited Members

Detailed Description

Member Enumeration Documentation

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation