#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) override

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

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

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)

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

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 bool is_raster, 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, const bool force_null=false)

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)

Private Member Functions
ImportStatus	importGDALGeo (const std::map< std::string, std::string > &colname_to_src, const Catalog_Namespace::SessionInfo *session_info)

ImportStatus	importGDALRaster (const Catalog_Namespace::SessionInfo *session_info)

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

static Geospatial::GDAL::DataSourceUqPtr	openGDALDataSource (const std::string &fileName, const CopyParams &copy_params)

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

static const std::list < ColumnDescriptor >	gdalToColumnDescriptorsRaster (const std::string &fileName, const std::string &geoColumnName, 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

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_

heavyai::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 784 of file Importer.h.

Member Enumeration Documentation

enum import_export::Importer::GeoFileLayerContents

strong

Enumerator
EMPTY
GEO
NON_GEO
UNSUPPORTED_GEO

Definition at line 837 of file Importer.h.

837 { 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 172 of file Importer.cpp.

176 : 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:172

f

torch::Tensor f(torch::Tensor x, torch::Tensor W_target, torch::Tensor b_target)

Definition: TestTorchTableFunctions.cpp:103

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

Definition at line 178 of file Importer.cpp.

References buffer, import_export::DataStreamSink::file_path, file_size, 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 219 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 3524 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(), importGDALGeo(), and importGDALRaster().

                                                                     {
   if (loader->getTableDesc()->storageType != StorageType::FOREIGN_TABLE) {
     heavyai::lock_guard<heavyai::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([&]() {
       heavyai::lock_guard<heavyai::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 5053 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::importGeoTableSingle().

                                                                                   {
   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 5058 of file Importer.cpp.

References gdalStatInternal().

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

                                                                         {
   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 5130 of file Importer.cpp.

References import_export::gdalGatherFilesInArchiveRecursive(), Geospatial::GDAL::init(), import_export::CopyParams::s3_access_key, import_export::CopyParams::s3_endpoint, import_export::CopyParams::s3_region, import_export::CopyParams::s3_secret_key, import_export::CopyParams::s3_session_token, and Geospatial::GDAL::setAuthorizationTokens().

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();
   Geospatial::GDAL::setAuthorizationTokens(copy_params.s3_region,
                                            copy_params.s3_endpoint,
                                            copy_params.s3_access_key,
                                            copy_params.s3_secret_key,
                                            copy_params.s3_session_token);
 
   // 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 5157 of file Importer.cpp.

References CHECK, EMPTY, GEO, Geospatial::GDAL::init(), NON_GEO, openGDALDataSource(), import_export::CopyParams::s3_access_key, import_export::CopyParams::s3_endpoint, import_export::CopyParams::s3_region, import_export::CopyParams::s3_secret_key, import_export::CopyParams::s3_session_token, Geospatial::GDAL::setAuthorizationTokens(), and UNSUPPORTED_GEO.

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

                                    {
   // lazy init GDAL
   Geospatial::GDAL::init();
   Geospatial::GDAL::setAuthorizationTokens(copy_params.s3_region,
                                            copy_params.s3_endpoint,
                                            copy_params.s3_access_key,
                                            copy_params.s3_secret_key,
                                            copy_params.s3_session_token);
 
   // prepare to gather layer info
   std::vector<GeoFileLayerInfo> layer_info;
 
   // open the data set
   Geospatial::GDAL::DataSourceUqPtr poDS(openGDALDataSource(file_name, copy_params));
   if (poDS == nullptr) {
     throw std::runtime_error("openGDALDataSource 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) {
       // first read layer geo type
       auto ogr_type = wkbFlatten(poLayer->GetGeomType());
       if (ogr_type == wkbUnknown) {
         // layer geo type unknown, so try reading from the first feature
         Geospatial::GDAL::FeatureUqPtr first_feature(poLayer->GetNextFeature());
         CHECK(first_feature);
         auto const* ogr_geometry = first_feature->GetGeometryRef();
         if (ogr_geometry) {
           ogr_type = wkbFlatten(ogr_geometry->getGeometryType());
         } else {
           ogr_type = wkbNone;
         }
       }
       switch (ogr_type) {
         case wkbNone:
           // no geo
           contents = GeoFileLayerContents::NON_GEO;
           break;
         case wkbPoint:
         case wkbMultiPoint:
         case wkbLineString:
         case wkbMultiLineString:
         case wkbPolygon:
         case wkbMultiPolygon:
           // layer has supported geo
           contents = GeoFileLayerContents::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 5018 of file Importer.cpp.

References Geospatial::GDAL::init(), run_benchmark_import::result, import_export::CopyParams::s3_access_key, import_export::CopyParams::s3_endpoint, import_export::CopyParams::s3_region, import_export::CopyParams::s3_secret_key, import_export::CopyParams::s3_session_token, and Geospatial::GDAL::setAuthorizationTokens().

Referenced by gdalFileExists(), and gdalFileOrDirectoryExists().

                                                {
   // lazy init GDAL
   Geospatial::GDAL::init();
   Geospatial::GDAL::setAuthorizationTokens(copy_params.s3_region,
                                            copy_params.s3_endpoint,
                                            copy_params.s3_access_key,
                                            copy_params.s3_secret_key,
                                            copy_params.s3_session_token);
 
 #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 bool	is_raster,
		const std::string &	geoColumnName,
		const CopyParams &	copy_params
	)

static

Definition at line 4820 of file Importer.cpp.

References gdalToColumnDescriptorsGeo(), and gdalToColumnDescriptorsRaster().

Referenced by DBHandler::detect_column_types().

                                    {
   if (is_raster) {
     return gdalToColumnDescriptorsRaster(file_name, geo_column_name, copy_params);
   }
   return gdalToColumnDescriptorsGeo(file_name, geo_column_name, copy_params);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

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

staticprivate

Definition at line 4904 of file Importer.cpp.

References import_export::CopyParams::add_metadata_columns, 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, kLINESTRING, kMULTILINESTRING, kMULTIPOINT, kMULTIPOLYGON, kPOINT, kPOLYGON, kTEXT, import_export::anonymous_namespace{Importer.cpp}::ogr_to_type(), openGDALDataSource(), import_export::parse_add_metadata_columns(), import_export::PROMOTE_LINESTRING_TO_MULTILINESTRING, import_export::PROMOTE_POINT_TO_MULTIPOINT, 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 gdalToColumnDescriptors().

                                    {
   std::list<ColumnDescriptor> cds;
 
   Geospatial::GDAL::DataSourceUqPtr poDS(openGDALDataSource(file_name, copy_params));
   if (poDS == nullptr) {
     throw std::runtime_error("openGDALDataSource Error: Unable to open geo file " +
                              file_name);
   }
   if (poDS->GetLayerCount() == 0) {
     throw std::runtime_error("gdalToColumnDescriptors Error: Geo file " + file_name +
                              " has no layers");
   }
 
   OGRLayer& layer =
       getLayerWithSpecifiedName(copy_params.geo_layer_name, poDS, file_name);
 
   layer.ResetReading();
   // TODO(andrewseidl): support multiple features
   Geospatial::GDAL::FeatureUqPtr 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(0);
     }
     ti.set_fixed_size();
     cd.columnType = ti;
     cds.push_back(cd);
   }
   // try getting the geo column type from the layer
   auto ogr_type = wkbFlatten(layer.GetGeomType());
   if (ogr_type == wkbUnknown) {
     // layer geo type unknown, so try the feature (that we already got)
     auto const* ogr_geometry = poFeature->GetGeometryRef();
     if (ogr_geometry) {
       ogr_type = wkbFlatten(ogr_geometry->getGeometryType());
     }
   }
   // do we have a geo column?
   if (ogr_type != wkbNone) {
     ColumnDescriptor cd;
     cd.columnName = geo_column_name;
     cd.sourceName = geo_column_name;
 
     // 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
     // this will throw if the type is unsupported
     SQLTypes geoType = ogr_to_type(ogr_type);
 
     // promote column type? (unless exploding)
     if (!copy_params.geo_explode_collections) {
       if (PROMOTE_POINT_TO_MULTIPOINT && geoType == kPOINT) {
         geoType = kMULTIPOINT;
       } else if (PROMOTE_LINESTRING_TO_MULTILINESTRING && geoType == kLINESTRING) {
         geoType = kMULTILINESTRING;
       } else if (PROMOTE_POLYGON_TO_MULTIPOLYGON && geoType == kPOLYGON) {
         geoType = kMULTIPOLYGON;
       }
     }
 
     // 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);
   }
 
   // metadata columns?
   auto metadata_column_infos =
       parse_add_metadata_columns(copy_params.add_metadata_columns, file_name);
   for (auto& mci : metadata_column_infos) {
     cds.push_back(std::move(mci.column_descriptor));
   }
 
   return cds;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

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

staticprivate

Definition at line 4832 of file Importer.cpp.

Referenced by gdalToColumnDescriptors().

                                    {
   // lazy init GDAL
   Geospatial::GDAL::init();
   Geospatial::GDAL::setAuthorizationTokens(copy_params.s3_region,
                                            copy_params.s3_endpoint,
                                            copy_params.s3_access_key,
                                            copy_params.s3_secret_key,
                                            copy_params.s3_session_token);
 
   // prepare for metadata column
   auto metadata_column_infos =
       parse_add_metadata_columns(copy_params.add_metadata_columns, file_name);
 
   // create a raster importer and do the detect
   RasterImporter raster_importer;
   raster_importer.detect(
       file_name,
       copy_params.raster_import_bands,
       copy_params.raster_import_dimensions,
       convert_raster_point_type(copy_params.raster_point_type),
       convert_raster_point_transform(copy_params.raster_point_transform),
       copy_params.raster_point_compute_angle,
       false,
       metadata_column_infos);
 
   // prepare to capture column descriptors
   std::list<ColumnDescriptor> cds;
 
   // get the point column info
   auto const point_names_and_sql_types = raster_importer.getPointNamesAndSQLTypes();
 
   // create the columns for the point in the specified type
   for (auto const& [col_name, sql_type] : point_names_and_sql_types) {
     ColumnDescriptor cd;
     cd.columnName = cd.sourceName = col_name;
     cd.columnType.set_type(sql_type);
     // hardwire other POINT attributes for now
     if (sql_type == kPOINT) {
       cd.columnType.set_subtype(kGEOMETRY);
       cd.columnType.set_input_srid(4326);
       cd.columnType.set_output_srid(4326);
       cd.columnType.set_compression(kENCODING_GEOINT);
       cd.columnType.set_comp_param(32);
     }
     cds.push_back(cd);
   }
 
   // get the names and types for the band column(s)
   auto const band_names_and_types = raster_importer.getBandNamesAndSQLTypes();
 
   // add column descriptors for each band
   for (auto const& [band_name, sql_type] : band_names_and_types) {
     ColumnDescriptor cd;
     cd.columnName = cd.sourceName = band_name;
     cd.columnType.set_type(sql_type);
     cd.columnType.set_fixed_size();
     cds.push_back(cd);
   }
 
   // metadata columns?
   for (auto& mci : metadata_column_infos) {
     cds.push_back(std::move(mci.column_descriptor));
   }
 
   // return the results
   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 801 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 800 of file Importer.h.

References import_export::DataStreamSink::copy_params.

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

800 { return copy_params; }

import_export::DataStreamSink::copy_params

CopyParams copy_params

Definition: Importer.h:719

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 810 of file Importer.h.

References 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 807 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 231 of file Importer.cpp.

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

Referenced by DBHandler::import_table_status().

                                                                    {
   heavyai::shared_lock<heavyai::shared_mutex> read_lock(status_mutex);
   auto it = import_status_map.find(import_id);
   if (it == import_status_map.end()) {
     throw std::runtime_error("Import status not found for id: " + import_id);
   }
   return it->second;
 }

Here is the caller graph for this function:

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

inline

Definition at line 813 of file Importer.h.

References is_array_a.

Referenced by import_export::import_thread_delimited().

813 { return is_array_a.get(); }

import_export::Importer::is_array_a

std::unique_ptr< bool[]> is_array_a

Definition: Importer.h:901

Here is the caller graph for this function:

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

inline

Definition at line 847 of file Importer.h.

References loader.

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

                                          {
     return loader->getCatalog();
   }

Here is the caller graph for this function:

auto import_export::Importer::getLoader ( ) const

inline

Definition at line 870 of file Importer.h.

References loader.

                          {
     return loader.get();
   }

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

overridevirtual

Implements import_export::AbstractImporter.

Definition at line 4361 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 4365 of file Importer.cpp.

References threading_serial::async(), CHECK, checkpoint(), logger::ERROR, import_export::DataStreamSink::file_offsets, import_export::DataStreamSink::file_offsets_mutex, file_size, import_export::delimited_parser::find_row_end_pos(), heavyai::fopen(), Catalog_Namespace::SessionInfo::get_session_id(), Executor::getExecutor(), import_buffers_vec, import_id, import_export::DataStreamSink::import_mutex_, import_export::DataStreamSink::import_status_, import_export::import_thread_delimited(), logger::INFO, import_export::ImportStatus::load_failed, import_export::ImportStatus::load_msg, loader, LOG, max_threads, import_export::num_import_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::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);
   }
 
   max_threads = num_import_threads(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);
 
   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,
                                    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();
             {
               heavyai::lock_guard<heavyai::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;
             }
 
             LOG(INFO) << "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;
         }
         heavyai::shared_lock<heavyai::shared_mutex> read_lock(import_mutex_);
         if (import_status_.load_failed) {
           break;
         }
       }
       heavyai::unique_lock<heavyai::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	(	const std::map< std::string, std::string > &	colname_to_src,
		const Catalog_Namespace::SessionInfo *	session_info,
		const bool	is_raster
	)

Definition at line 5226 of file Importer.cpp.

References importGDALGeo(), and importGDALRaster().

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

                           {
   if (is_raster) {
     return importGDALRaster(session_info);
   }
   return importGDALGeo(columnNameToSourceNameMap, session_info);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

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

private

Definition at line 5236 of file Importer.cpp.

References threading_serial::async(), CHECK, CHECK_EQ, checkpoint(), logger::ERROR, g_enable_non_kernel_time_query_interrupt, Catalog_Namespace::SessionInfo::get_session_id(), Executor::getExecutor(), import_export::anonymous_namespace{Importer.cpp}::getLayerWithSpecifiedName(), import_buffers_vec, import_id, import_export::DataStreamSink::import_mutex_, import_export::DataStreamSink::import_status_, import_export::import_thread_shapefile(), import_export::ImportStatus::load_failed, import_export::ImportStatus::load_msg, loader, LOG, max_threads, import_export::num_import_threads(), openGDALDataSource(), import_export::parse_add_metadata_columns(), import_export::ImportStatus::rows_completed, import_export::ImportStatus::rows_estimated, import_export::ImportStatus::rows_rejected, set_import_status(), logger::thread_id(), toString(), Executor::UNITARY_EXECUTOR_ID, and VLOG.

Referenced by importGDAL().

                                                       {
   // initial status
   set_import_status(import_id, import_status_);
   Geospatial::GDAL::DataSourceUqPtr poDS(openGDALDataSource(file_path, copy_params));
   if (poDS == nullptr) {
     throw std::runtime_error("openGDALDataSource 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
   Geospatial::GDAL::SpatialReferenceUqPtr 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
   max_threads = num_import_threads(copy_params.threads);
 #endif
   VLOG(1) << "GDAL import # threads: " << max_threads;
 
   // metadata columns?
   auto const metadata_column_infos =
       parse_add_metadata_columns(copy_params.add_metadata_columns, file_path);
 
   // 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);
   auto is_session_already_registered = false;
   {
     heavyai::shared_lock<heavyai::shared_mutex> session_read_lock(
         executor->getSessionLock());
     is_session_already_registered =
         executor->checkIsQuerySessionEnrolled(query_session, session_read_lock);
   }
   if (g_enable_non_kernel_time_query_interrupt && !query_session.empty() &&
       !is_session_already_registered) {
     executor->enrollQuerySession(query_session,
                                  "IMPORT_GEO_TABLE",
                                  query_submitted_time,
                                  Executor::UNITARY_EXECUTOR_ID,
                                  QuerySessionStatus::QueryStatus::RUNNING_IMPORTER);
   }
   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);
     }
   };
 
   // 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)));
     }
   }
 
 #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);
 
   // make one of these for each thread, based on the first feature's SR
   std::vector<std::unique_ptr<OGRCoordinateTransformation>> coordinate_transformations(
       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());
     }
 
     // construct a coordinate transformation for each thread, if needed
     // some features may not have geometry, so look for the first one that does
     if (coordinate_transformations[thread_id] == nullptr) {
       for (auto const& feature : features[thread_id]) {
         auto const* geometry = feature->GetGeometryRef();
         if (geometry) {
           auto const* geometry_sr = geometry->getSpatialReference();
           // if the SR is non-null and non-empty and different from what we want
           // we need to make a reusable CoordinateTransformation
           if (geometry_sr &&
 #if GDAL_VERSION_MAJOR >= 3
               !geometry_sr->IsEmpty() &&
 #endif
               !geometry_sr->IsSame(poGeographicSR.get())) {
             // validate the SR before trying to use it
             if (geometry_sr->Validate() != OGRERR_NONE) {
               throw std::runtime_error("Incoming geo has invalid Spatial Reference");
             }
             // create the OGRCoordinateTransformation that will be used for
             // all the features in this chunk
             coordinate_transformations[thread_id].reset(
                 OGRCreateCoordinateTransformation(geometry_sr, poGeographicSR.get()));
             if (coordinate_transformations[thread_id] == nullptr) {
               throw std::runtime_error(
                   "Failed to create a GDAL CoordinateTransformation for incoming geo");
             }
           }
           // once we find at least one geometry with an SR, we're done
           break;
         }
       }
     }
 
 #if DISABLE_MULTI_THREADED_SHAPEFILE_IMPORT
     // call worker function directly
     auto ret_import_status =
         import_thread_shapefile(0,
                                 this,
                                 coordinate_transformations[thread_id].get(),
                                 std::move(features[thread_id]),
                                 firstFeatureThisChunk,
                                 numFeaturesThisChunk,
                                 fieldNameToIndexMap,
                                 columnNameToSourceNameMap,
                                 session_info,
                                 executor.get(),
                                 metadata_column_infos);
     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,
                                  coordinate_transformations[thread_id].get(),
                                  std::move(features[thread_id]),
                                  firstFeatureThisChunk,
                                  numFeaturesThisChunk,
                                  fieldNameToIndexMap,
                                  columnNameToSourceNameMap,
                                  session_info,
                                  executor.get(),
                                  metadata_column_infos));
 
     // 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();
           {
             heavyai::lock_guard<heavyai::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?
 
     heavyai::unique_lock<heavyai::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:

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

private

Definition at line 5524 of file Importer.cpp.

References threading_serial::async(), CHECK, CHECK_EQ, anonymous_namespace{Importer.cpp}::check_session_interrupted(), checkpoint(), ColumnDescriptor::columnName, ColumnDescriptor::columnType, Geospatial::compress_coords(), import_export::anonymous_namespace{Importer.cpp}::convert_raster_point_transform(), import_export::anonymous_namespace{Importer.cpp}::convert_raster_point_type(), import_export::RasterImporter::detect(), Executor::ERR_INTERRUPTED, logger::ERROR, f(), g_enable_non_kernel_time_query_interrupt, SQLTypeInfo::get_output_srid(), Catalog_Namespace::SessionInfo::get_session_id(), SQLTypeInfo::get_subtype(), SQLTypeInfo::get_type(), import_export::RasterImporter::getBandNamesAndSQLTypes(), import_export::RasterImporter::getBandNullValue(), import_export::RasterImporter::getBandsHeight(), import_export::RasterImporter::getBandsWidth(), Executor::getExecutor(), import_export::RasterImporter::getNumBands(), import_export::RasterImporter::getPointNamesAndSQLTypes(), import_export::RasterImporter::getProjectedPixelCoords(), import_export::RasterImporter::getRawPixels(), import_export::RasterImporter::import(), import_buffers_vec, import_id, import_export::DataStreamSink::import_status_, logger::INFO, ColumnDescriptor::isGeoPhyCol, kARRAY, kBIGINT, kDOUBLE, kFLOAT, kINT, kMaxRasterScanlinesPerThread, kNULLT, kPOINT, kSMALLINT, kTINYINT, load(), import_export::ImportStatus::load_failed, import_export::ImportStatus::load_msg, loader, LOG, max_threads, NULL_DOUBLE, NULL_FLOAT, NULL_INT, NULL_SMALLINT, import_export::num_import_threads(), import_export::parse_add_metadata_columns(), import_export::ImportStatus::rows_completed, import_export::ImportStatus::rows_estimated, import_export::ImportStatus::rows_rejected, set_import_status(), logger::thread_id(), timer_start(), TIMER_STOP, to_string(), toString(), Executor::UNITARY_EXECUTOR_ID, UNLIKELY, and VLOG.

Referenced by importGDAL().

                                                       {
   // initial status
   set_import_status(import_id, import_status_);
 
   // metadata columns?
   auto const metadata_column_infos =
       parse_add_metadata_columns(copy_params.add_metadata_columns, file_path);
 
   // create a raster importer and do the detect
   RasterImporter raster_importer;
   raster_importer.detect(
       file_path,
       copy_params.raster_import_bands,
       copy_params.raster_import_dimensions,
       convert_raster_point_type(copy_params.raster_point_type),
       convert_raster_point_transform(copy_params.raster_point_transform),
       copy_params.raster_point_compute_angle,
       true,
       metadata_column_infos);
 
   // get the table columns and count actual columns
   auto const& column_descs = loader->get_column_descs();
   uint32_t num_table_cols{0u};
   for (auto const* cd : column_descs) {
     if (!cd->isGeoPhyCol) {
       num_table_cols++;
     }
   }
 
   // how many bands do we have?
   auto num_bands = raster_importer.getNumBands();
 
   // get point columns info
   auto const point_names_and_sql_types = raster_importer.getPointNamesAndSQLTypes();
 
   // validate that the table column count matches
   auto num_expected_cols = num_bands;
   num_expected_cols += point_names_and_sql_types.size();
   num_expected_cols += metadata_column_infos.size();
   if (num_expected_cols != num_table_cols) {
     throw std::runtime_error(
         "Raster Import aborted. Band/Column count mismatch (file requires " +
         std::to_string(num_expected_cols) + ", table has " +
         std::to_string(num_table_cols) + ")");
   }
 
   // validate the point column names and types
   // if we're importing the coords as a POINT, then the first column
   // must be a POINT (two physical columns, POINT and TINYINT[])
   // if we're not, the first two columns must be the matching type
   // optionally followed by an angle column
   auto cd_itr = column_descs.begin();
   for (auto const& [col_name, sql_type] : point_names_and_sql_types) {
     if (sql_type == kPOINT) {
       // POINT column
       {
         auto const* cd = *cd_itr++;
         if (cd->columnName != col_name) {
           throw std::runtime_error("Column '" + cd->columnName +
                                    "' overridden name invalid (must be '" + col_name +
                                    "')");
         }
         auto const cd_type = cd->columnType.get_type();
         if (cd_type != kPOINT) {
           throw std::runtime_error("Column '" + cd->columnName +
                                    "' overridden type invalid (must be POINT)");
         }
         if (cd->columnType.get_output_srid() != 4326) {
           throw std::runtime_error("Column '" + cd->columnName +
                                    "' overridden SRID invalid (must be 4326)");
         }
       }
       // TINYINT[] coords sub-column
       {
         // if the above is true, this must be true
         auto const* cd = *cd_itr++;
         CHECK(cd->columnType.get_type() == kARRAY);
         CHECK(cd->columnType.get_subtype() == kTINYINT);
       }
     } else {
       // column of the matching name and type
       auto const* cd = *cd_itr++;
       if (cd->columnName != col_name) {
         throw std::runtime_error("Column '" + cd->columnName +
                                  "' overridden name invalid (must be '" + col_name +
                                  "')");
       }
       auto const cd_type = cd->columnType.get_type();
       if (cd_type != sql_type) {
         throw std::runtime_error("Column '" + cd->columnName +
                                  "' overridden type invalid (must be " +
                                  to_string(sql_type) + ")");
       }
     }
   }
 
   // validate the band column types
   // any Immerse overriding to other types will currently be rejected
   auto const band_names_and_types = raster_importer.getBandNamesAndSQLTypes();
   if (band_names_and_types.size() != num_bands) {
     throw std::runtime_error("Column/Band count mismatch when validating types");
   }
   for (uint32_t i = 0; i < num_bands; i++) {
     auto const* cd = *cd_itr++;
     auto const cd_type = cd->columnType.get_type();
     auto const sql_type = band_names_and_types[i].second;
     if (cd_type != sql_type) {
       throw std::runtime_error("Band Column '" + cd->columnName +
                                "' overridden type invalid (must be " +
                                to_string(sql_type) + ")");
     }
   }
 
   // validate metadata column
   for (auto const& mci : metadata_column_infos) {
     auto const* cd = *cd_itr++;
     if (mci.column_descriptor.columnName != cd->columnName) {
       throw std::runtime_error("Metadata Column '" + cd->columnName +
                                "' overridden name invalid (must be '" +
                                mci.column_descriptor.columnName + "')");
     }
     auto const cd_type = cd->columnType.get_type();
     auto const md_type = mci.column_descriptor.columnType.get_type();
     if (cd_type != md_type) {
       throw std::runtime_error("Metadata Column '" + cd->columnName +
                                "' overridden type invalid (must be " +
                                to_string(md_type) + ")");
     }
   }
 
   // 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);
   auto is_session_already_registered = false;
   {
     heavyai::shared_lock<heavyai::shared_mutex> session_read_lock(
         executor->getSessionLock());
     is_session_already_registered =
         executor->checkIsQuerySessionEnrolled(query_session, session_read_lock);
   }
   if (g_enable_non_kernel_time_query_interrupt && !query_session.empty() &&
       !is_session_already_registered) {
     executor->enrollQuerySession(query_session,
                                  "IMPORT_GEO_TABLE",
                                  query_submitted_time,
                                  Executor::UNITARY_EXECUTOR_ID,
                                  QuerySessionStatus::QueryStatus::RUNNING_IMPORTER);
   }
   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);
     }
   };
 
   // how many threads are we gonna use?
   max_threads = num_import_threads(copy_params.threads);
   VLOG(1) << "GDAL import # threads: " << max_threads;
 
   if (copy_params.raster_scanlines_per_thread < 0) {
     throw std::runtime_error("Invalid CopyParams.raster_scanlines_per_thread! (" +
                              std::to_string(copy_params.raster_scanlines_per_thread) +
                              ")");
   }
   const int max_scanlines_per_thread =
       copy_params.raster_scanlines_per_thread == 0
           ? kMaxRasterScanlinesPerThread
           : std::min(copy_params.raster_scanlines_per_thread,
                      kMaxRasterScanlinesPerThread);
   VLOG(1) << "Raster Importer: Max scanlines per thread: " << max_scanlines_per_thread;
 
   // prepare to checkpoint the table
   auto table_epochs = loader->getTableEpochs();
 
   // start wall clock
   auto wall_timer = timer_start();
 
   // start the import
   raster_importer.import(
       max_threads,
       copy_params.threads == 0);  // NOTE: `max_threads` may change after this call
 
   // 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 (auto const& cd : loader->get_column_descs()) {
       import_buffers_vec[i].emplace_back(
           new TypedImportBuffer(cd, loader->getStringDict(cd)));
     }
   }
 
   // status and times
   using ThreadReturn = std::tuple<ImportStatus, std::array<float, 3>>;
 
   // get the band dimensions
   auto const band_size_x = raster_importer.getBandsWidth();
   auto const band_size_y = raster_importer.getBandsHeight();
 
   // allocate raw pixel buffers per thread
   std::vector<RasterImporter::RawPixels> raw_pixel_bytes_per_thread(max_threads);
   for (size_t i = 0; i < max_threads; i++) {
     raw_pixel_bytes_per_thread[i].resize(band_size_x * max_scanlines_per_thread *
                                          sizeof(double));
   }
 
   // just the sql type of the first point column (if any)
   auto const point_sql_type = point_names_and_sql_types.size()
                                   ? point_names_and_sql_types.begin()->second
                                   : kNULLT;
 
   // lambda for importing to raw data buffers (threadable)
   auto import_rows =
       [&](const size_t thread_idx, const int y_start, const int y_end) -> ThreadReturn {
     // this threads's import buffers
     auto& import_buffers = import_buffers_vec[thread_idx];
 
     // this thread's raw pixel bytes
     auto& raw_pixel_bytes = raw_pixel_bytes_per_thread[thread_idx];
 
     // clear the buffers
     for (auto& col_buffer : import_buffers) {
       col_buffer->clear();
     }
 
     // prepare to iterate columns
     auto col_itr = column_descs.begin();
     int col_idx{0};
 
     float proj_s{0.0f};
     if (point_sql_type != kNULLT) {
       // the first two columns (either lon/lat or POINT/coords)
       auto const* cd_col0 = *col_itr++;
       auto const* cd_col1 = *col_itr++;
       auto const* cd_angle =
           copy_params.raster_point_compute_angle ? *col_itr++ : nullptr;
 
       // compute and add x and y
       auto proj_timer = timer_start();
       for (int y = y_start; y < y_end; y++) {
         // get projected pixel coords for this scan-line
         auto const coords = raster_importer.getProjectedPixelCoords(thread_idx, y);
 
         // add to buffers
         for (int x = 0; x < band_size_x; x++) {
           // this point and angle
           auto const& [dx, dy, angle] = coords[x];
 
           // store the point
           switch (point_sql_type) {
             case kPOINT: {
               // add empty value to POINT buffer
               TDatum td_point;
               import_buffers[0]->add_value(cd_col0, td_point, false);
 
               // convert lon/lat to bytes (compressed or not) and add to POINT coords
               // buffer
               auto const compressed_coords =
                   Geospatial::compress_coords({dx, dy}, cd_col0->columnType);
               std::vector<TDatum> td_coords_data;
               for (auto const& cc : compressed_coords) {
                 TDatum td_byte;
                 td_byte.val.int_val = cc;
                 td_coords_data.push_back(td_byte);
               }
               TDatum td_coords;
               td_coords.val.arr_val = td_coords_data;
               td_coords.is_null = false;
               import_buffers[1]->add_value(cd_col1, td_coords, false);
             } break;
             case kFLOAT:
             case kDOUBLE: {
               TDatum td;
               td.is_null = false;
               td.val.real_val = dx;
               import_buffers[0]->add_value(cd_col0, td, false);
               td.val.real_val = dy;
               import_buffers[1]->add_value(cd_col1, td, false);
             } break;
             case kSMALLINT:
             case kINT: {
               TDatum td;
               td.is_null = false;
               td.val.int_val = static_cast<int64_t>(x);
               import_buffers[0]->add_value(cd_col0, td, false);
               td.val.int_val = static_cast<int64_t>(y);
               import_buffers[1]->add_value(cd_col1, td, false);
             } break;
             default:
               CHECK(false);
           }
 
           // angle?
           if (copy_params.raster_point_compute_angle) {
             CHECK(cd_angle);
             TDatum td;
             td.is_null = false;
             td.val.real_val = static_cast<double>(angle);
             import_buffers[2]->add_value(cd_angle, td, false);
           }
         }
       }
       proj_s = TIMER_STOP(proj_timer);
       col_idx += (copy_params.raster_point_compute_angle ? 3 : 2);
     }
 
     // prepare to accumulate read and conv times
     float read_s{0.0f};
     float conv_s{0.0f};
 
     // y_end is one past the actual end, so don't add 1
     auto const num_rows = y_end - y_start;
     auto const num_elems = band_size_x * num_rows;
 
     ImportStatus thread_import_status;
 
     bool read_block_failed = false;
 
     // for each band/column
     for (uint32_t band_idx = 0; band_idx < num_bands; band_idx++) {
       // the corresponding column
       auto const* cd_band = *col_itr;
       CHECK(cd_band);
 
       // data type to read as
       auto const cd_type = cd_band->columnType.get_type();
 
       // read the scanlines (will do a data type conversion if necessary)
       try {
         auto read_timer = timer_start();
         raster_importer.getRawPixels(
             thread_idx, band_idx, y_start, num_rows, cd_type, raw_pixel_bytes);
         read_s += TIMER_STOP(read_timer);
       } catch (std::runtime_error& e) {
         // report error
         LOG(ERROR) << e.what();
         // abort this block
         read_block_failed = true;
         break;
       }
 
       // null value?
       auto const [null_value, null_value_valid] =
           raster_importer.getBandNullValue(band_idx);
 
       // copy to this thread's import buffers
       // convert any nulls we find
       auto conv_timer = timer_start();
       TDatum td;
       switch (cd_type) {
         case kSMALLINT: {
           const int16_t* values =
               reinterpret_cast<const int16_t*>(raw_pixel_bytes.data());
           for (int idx = 0; idx < num_elems; idx++) {
             auto const& value = values[idx];
             if (null_value_valid && value == static_cast<int16_t>(null_value)) {
               td.is_null = true;
               td.val.int_val = NULL_SMALLINT;
             } else {
               td.is_null = false;
               td.val.int_val = static_cast<int64_t>(value);
             }
             import_buffers[col_idx]->add_value(cd_band, td, false);
           }
         } break;
         case kINT: {
           const int32_t* values =
               reinterpret_cast<const int32_t*>(raw_pixel_bytes.data());
           for (int idx = 0; idx < num_elems; idx++) {
             auto const& value = values[idx];
             if (null_value_valid && value == static_cast<int32_t>(null_value)) {
               td.is_null = true;
               td.val.int_val = NULL_INT;
             } else {
               td.is_null = false;
               td.val.int_val = static_cast<int64_t>(value);
             }
             import_buffers[col_idx]->add_value(cd_band, td, false);
           }
         } break;
         case kBIGINT: {
           const uint32_t* values =
               reinterpret_cast<const uint32_t*>(raw_pixel_bytes.data());
           for (int idx = 0; idx < num_elems; idx++) {
             auto const& value = values[idx];
             if (null_value_valid && value == static_cast<uint32_t>(null_value)) {
               td.is_null = true;
               td.val.int_val = NULL_INT;
             } else {
               td.is_null = false;
               td.val.int_val = static_cast<int64_t>(value);
             }
             import_buffers[col_idx]->add_value(cd_band, td, false);
           }
         } break;
         case kFLOAT: {
           const float* values = reinterpret_cast<const float*>(raw_pixel_bytes.data());
           for (int idx = 0; idx < num_elems; idx++) {
             auto const& value = values[idx];
             if (null_value_valid && value == static_cast<float>(null_value)) {
               td.is_null = true;
               td.val.real_val = NULL_FLOAT;
             } else {
               td.is_null = false;
               td.val.real_val = static_cast<double>(value);
             }
             import_buffers[col_idx]->add_value(cd_band, td, false);
           }
         } break;
         case kDOUBLE: {
           const double* values = reinterpret_cast<const double*>(raw_pixel_bytes.data());
           for (int idx = 0; idx < num_elems; idx++) {
             auto const& value = values[idx];
             if (null_value_valid && value == null_value) {
               td.is_null = true;
               td.val.real_val = NULL_DOUBLE;
             } else {
               td.is_null = false;
               td.val.real_val = value;
             }
             import_buffers[col_idx]->add_value(cd_band, td, false);
           }
         } break;
         default:
           CHECK(false);
       }
       conv_s += TIMER_STOP(conv_timer);
 
       // next column
       col_idx++;
       col_itr++;
     }
 
     if (read_block_failed) {
       // discard block data
       for (auto& col_buffer : import_buffers) {
         col_buffer->clear();
       }
       thread_import_status.rows_rejected += num_elems;
     } else {
       // metadata columns?
       for (auto const& mci : metadata_column_infos) {
         auto const* cd_band = *col_itr++;
         CHECK(cd_band);
         for (int i = 0; i < num_elems; i++) {
           import_buffers[col_idx]->add_value(cd_band, mci.value, false, copy_params);
         }
         col_idx++;
       }
       thread_import_status.rows_estimated = num_elems;
       thread_import_status.rows_completed = num_elems;
     }
 
     // done
     return {std::move(thread_import_status), {proj_s, read_s, conv_s}};
   };
 
   // time the phases
   float total_proj_s{0.0f};
   float total_read_s{0.0f};
   float total_conv_s{0.0f};
   float total_load_s{0.0f};
 
   const int min_scanlines_per_thread = 8;
   const int max_scanlines_per_block = max_scanlines_per_thread * max_threads;
   for (int block_y = 0; block_y < band_size_y;
        block_y += (max_threads * max_scanlines_per_thread)) {
     using Future = std::future<ThreadReturn>;
     std::vector<Future> futures;
     const int scanlines_in_block =
         std::min(band_size_y - block_y, max_scanlines_per_block);
     const int pixels_in_block = scanlines_in_block * band_size_x;
     const int block_max_scanlines_per_thread =
         std::max((scanlines_in_block + static_cast<int>(max_threads) - 1) /
                      static_cast<int>(max_threads),
                  min_scanlines_per_thread);
     VLOG(1) << "Raster Importer: scanlines_in_block: " << scanlines_in_block
             << ", block_max_scanlines_per_thread:  " << block_max_scanlines_per_thread;
 
     std::vector<size_t> rows_per_thread;
     auto block_wall_timer = timer_start();
     // run max_threads scanlines at once
     for (size_t thread_id = 0; thread_id < max_threads; thread_id++) {
       const int y_start = block_y + thread_id * block_max_scanlines_per_thread;
       if (y_start < band_size_y) {
         const int y_end = std::min(y_start + block_max_scanlines_per_thread, band_size_y);
         if (y_start < y_end) {
           rows_per_thread.emplace_back((y_end - y_start) * band_size_x);
           futures.emplace_back(
               std::async(std::launch::async, import_rows, thread_id, y_start, y_end));
         }
       }
     }
 
     // wait for the threads to finish and
     // accumulate the results and times
     float proj_s{0.0f}, read_s{0.0f}, conv_s{0.0f}, load_s{0.0f};
     size_t thread_idx = 0;
     for (auto& future : futures) {
       auto const [import_status, times] = future.get();
       import_status_ += import_status;
       proj_s += times[0];
       read_s += times[1];
       conv_s += times[2];
       // We load the data in thread order so we can get deterministic row-major raster
       // ordering
       // Todo: We should consider invoking the load on another thread in a ping-pong
       // fashion so we can simultaneously read the next batch of data
       auto thread_load_timer = timer_start();
       // only try to load this thread's data if valid
       if (import_status.rows_rejected == 0) {
         load(import_buffers_vec[thread_idx], rows_per_thread[thread_idx], session_info);
       }
       load_s += TIMER_STOP(thread_load_timer);
       ++thread_idx;
     }
 
     // average times over all threads (except for load which is single-threaded)
     total_proj_s += (proj_s / float(futures.size()));
     total_read_s += (read_s / float(futures.size()));
     total_conv_s += (conv_s / float(futures.size()));
     total_load_s += load_s;
 
     // update the status
     set_import_status(import_id, import_status_);
 
     // more debug
     auto const block_wall_s = TIMER_STOP(block_wall_timer);
     auto const scanlines_per_second = scanlines_in_block / block_wall_s;
     auto const rows_per_second = pixels_in_block / block_wall_s;
     LOG(INFO) << "Raster Importer: Loaded " << scanlines_in_block
               << " scanlines starting at " << block_y << " out of " << band_size_y
               << " in " << block_wall_s << "s at " << scanlines_per_second
               << " scanlines/s and " << rows_per_second << " rows/s";
 
     // check for interrupt
     if (UNLIKELY(check_session_interrupted(query_session, executor.get()))) {
       import_status_.load_failed = true;
       import_status_.load_msg = "Raster Import interrupted";
       throw QueryExecutionError(Executor::ERR_INTERRUPTED);
     }
 
     // hit max_reject?
     if (import_status_.rows_rejected > copy_params.max_reject) {
       break;
     }
   }
 
   // checkpoint
   auto checkpoint_timer = timer_start();
   checkpoint(table_epochs);
   auto const checkpoint_s = TIMER_STOP(checkpoint_timer);
 
   // stop wall clock
   auto const total_wall_s = TIMER_STOP(wall_timer);
 
   // report
   auto const total_scanlines_per_second = float(band_size_y) / total_wall_s;
   auto const total_rows_per_second =
       float(band_size_x) * float(band_size_y) / total_wall_s;
   LOG(INFO) << "Raster Importer: Imported "
             << static_cast<uint64_t>(band_size_x) * static_cast<uint64_t>(band_size_y)
             << " rows";
   LOG(INFO) << "Raster Importer: Total Import Time " << total_wall_s << "s at "
             << total_scanlines_per_second << " scanlines/s and " << total_rows_per_second
             << " rows/s";
 
   // if we hit max_reject, throw an exception now to report the error and abort any
   // multi-file loop
   if (import_status_.rows_rejected > copy_params.max_reject) {
     std::string msg = "Raster Importer: Import aborted after failing to read " +
                       std::to_string(import_status_.rows_rejected) +
                       " rows/pixels (limit " + std::to_string(copy_params.max_reject) +
                       ")";
     import_status_.load_msg = msg;
     set_import_status(import_id, import_status_);
     throw std::runtime_error(msg);
   }
 
   // phase times (with proportions)
   auto proj_pct = float(int(total_proj_s / total_wall_s * 1000.0f) * 0.1f);
   auto read_pct = float(int(total_read_s / total_wall_s * 1000.0f) * 0.1f);
   auto conv_pct = float(int(total_conv_s / total_wall_s * 1000.0f) * 0.1f);
   auto load_pct = float(int(total_load_s / total_wall_s * 1000.0f) * 0.1f);
   auto cpnt_pct = float(int(checkpoint_s / total_wall_s * 1000.0f) * 0.1f);
 
   VLOG(1) << "Raster Importer: Import timing breakdown:";
   VLOG(1) << "  Project    " << total_proj_s << "s (" << proj_pct << "%)";
   VLOG(1) << "  Read       " << total_read_s << "s (" << read_pct << "%)";
   VLOG(1) << "  Convert    " << total_conv_s << "s (" << conv_pct << "%)";
   VLOG(1) << "  Load       " << total_load_s << "s (" << load_pct << "%)";
   VLOG(1) << "  Checkpoint " << checkpoint_s << "s (" << cpnt_pct << "%)";
 
   // all done
   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 3514 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(), import_export::import_thread_shapefile(), and importGDALRaster().

                                                                       {
   if (!loader->loadNoCheckpoint(import_buffers, row_count, session_info)) {
     heavyai::lock_guard<heavyai::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:

Geospatial::GDAL::DataSourceUqPtr import_export::Importer::openGDALDataSource	(	const std::string &	fileName,
		const CopyParams &	copy_params
	)

staticprivate

Definition at line 4583 of file Importer.cpp.

References Geospatial::GDAL::init(), import_export::kGeoFile, Geospatial::GDAL::openDataSource(), import_export::CopyParams::s3_access_key, import_export::CopyParams::s3_endpoint, import_export::CopyParams::s3_region, import_export::CopyParams::s3_secret_key, import_export::CopyParams::s3_session_token, Geospatial::GDAL::setAuthorizationTokens(), import_export::CopyParams::source_type, and to_string().

Referenced by gdalGetLayersInGeoFile(), gdalToColumnDescriptorsGeo(), importGDALGeo(), and readMetadataSampleGDAL().

                                    {
   Geospatial::GDAL::init();
   Geospatial::GDAL::setAuthorizationTokens(copy_params.s3_region,
                                            copy_params.s3_endpoint,
                                            copy_params.s3_access_key,
                                            copy_params.s3_secret_key,
                                            copy_params.s3_session_token);
   if (copy_params.source_type != import_export::SourceType::kGeoFile) {
     throw std::runtime_error("Unexpected CopyParams.source_type (" +
                              std::to_string(static_cast<int>(copy_params.source_type)) +
                              ")");
   }
   return Geospatial::GDAL::openDataSource(file_name, import_export::SourceType::kGeoFile);
 }

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 4625 of file Importer.cpp.

References import_export::CopyParams::add_metadata_columns, CHECK, import_export::CopyParams::geo_layer_name, import_export::anonymous_namespace{Importer.cpp}::getLayerWithSpecifiedName(), openGDALDataSource(), and import_export::parse_add_metadata_columns().

Referenced by DBHandler::detect_column_types().

                                    {
   Geospatial::GDAL::DataSourceUqPtr datasource(
       openGDALDataSource(file_name, copy_params));
   if (datasource == nullptr) {
     throw std::runtime_error("openGDALDataSource Error: Unable to open geo file " +
                              file_name);
   }
 
   OGRLayer& layer =
       getLayerWithSpecifiedName(copy_params.geo_layer_name, datasource, file_name);
 
   auto const* feature_defn = layer.GetLayerDefn();
   CHECK(feature_defn);
 
   // metadata columns?
   auto const metadata_column_infos =
       parse_add_metadata_columns(copy_params.add_metadata_columns, file_name);
 
   // get limited feature count
   // typeof GetFeatureCount() is different between GDAL 1.x (int32_t) and 2.x (int64_t)
   auto const feature_count = static_cast<uint64_t>(layer.GetFeatureCount());
   auto const num_features = std::min(static_cast<uint64_t>(row_limit), feature_count);
 
   // prepare sample data map
   for (int field_index = 0; field_index < feature_defn->GetFieldCount(); field_index++) {
     auto const* column_name = feature_defn->GetFieldDefn(field_index)->GetNameRef();
     CHECK(column_name);
     sample_data[column_name] = {};
   }
   sample_data[geo_column_name] = {};
   for (auto const& mci : metadata_column_infos) {
     sample_data[mci.column_descriptor.columnName] = {};
   }
 
   // prepare to read
   layer.ResetReading();
 
   // read features (up to limited count)
   uint64_t feature_index{0u};
   while (feature_index < num_features) {
     // get (and take ownership of) feature
     Geospatial::GDAL::FeatureUqPtr feature(layer.GetNextFeature());
     if (!feature) {
       break;
     }
 
     // get feature geometry
     auto const* geometry = feature->GetGeometryRef();
     if (geometry == nullptr) {
       break;
     }
 
     // validate geom type (again?)
     switch (wkbFlatten(geometry->getGeometryType())) {
       case wkbPoint:
       case wkbMultiPoint:
       case wkbLineString:
       case wkbMultiLineString:
       case wkbPolygon:
       case wkbMultiPolygon:
         break;
       default:
         throw std::runtime_error("Unsupported geometry type: " +
                                  std::string(geometry->getGeometryName()));
     }
 
     // populate sample data for regular field columns
     for (int field_index = 0; field_index < feature->GetFieldCount(); field_index++) {
       auto const* column_name = feature_defn->GetFieldDefn(field_index)->GetNameRef();
       sample_data[column_name].push_back(feature->GetFieldAsString(field_index));
     }
 
     // populate sample data for metadata columns?
     for (auto const& mci : metadata_column_infos) {
       sample_data[mci.column_descriptor.columnName].push_back(mci.value);
     }
 
     // populate sample data for geo column with WKT string
     char* wkts = nullptr;
     geometry->exportToWkt(&wkts);
     CHECK(wkts);
     sample_data[geo_column_name].push_back(wkts);
     CPLFree(wkts);
 
     // next feature
     feature_index++;
   }
 }

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,
		const bool	force_null = `false`
	)

static

Definition at line 1636 of file Importer.cpp.

References ColumnDescriptor::columnId, ColumnDescriptor::columnType, Geospatial::compress_coords(), SQLTypeInfo::get_type(), Catalog_Namespace::Catalog::getMetadataForColumn(), Geospatial::is_null_point(), kLINESTRING, kMULTILINESTRING, kMULTIPOINT, 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(), foreign_storage::TextFileBufferParser::processGeoColumn(), and foreign_storage::TextFileBufferParser::processInvalidGeoColumn().

                            {
   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;
     }
   }
   if (force_null) {
     is_null_geo = true;
   }
   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 == kMULTILINESTRING || col_type == kPOLYGON || col_type == kMULTIPOLYGON) {
     // Create [linest]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 == kMULTILINESTRING || col_type == kPOLYGON ||
       col_type == kMULTIPOLYGON || col_type == kMULTIPOINT) {
     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);
   }
 }

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
	)

static

Definition at line 1731 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, kMULTILINESTRING, kMULTIPOINT, kMULTIPOLYGON, kPOINT, kPOLYGON, NULL_ARRAY_DOUBLE, NULL_DOUBLE, and ColumnDescriptor::tableId.

Referenced by DBHandler::fillGeoColumns().

                                                   {
   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 == kMULTILINESTRING || 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 [linest[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 == kMULTILINESTRING || col_type == kPOLYGON ||
       col_type == kMULTIPOLYGON || col_type == kMULTIPOINT) {
     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++;
   }
 }

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 240 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(), importGDALGeo(), importGDALRaster(), import_export::ForeignDataImporter::importGeneralS3(), and anonymous_namespace{ForeignDataImporter.cpp}::load_foreign_data_buffers().

                                                                             {
   heavyai::lock_guard<heavyai::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:

Member Data Documentation

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

private

Definition at line 898 of file Importer.h.

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

size_t import_export::Importer::file_size

private

Definition at line 896 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 899 of file Importer.h.

Referenced by checkpoint(), get_import_buffers(), get_import_buffers_vec(), importDelimited(), importGDALGeo(), and importGDALRaster().

std::string import_export::Importer::import_id

private

Definition at line 895 of file Importer.h.

Referenced by importDelimited(), Importer(), importGDALGeo(), importGDALRaster(), heavydb.thrift.Heavy.import_table_status_args::read(), set_import_status(), and heavydb.thrift.Heavy.import_table_status_args::write().

std::mutex import_export::Importer::init_gdal_mutex

staticprivate

Definition at line 902 of file Importer.h.

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

private

Definition at line 901 of file Importer.h.

Referenced by get_is_array(), and Importer().

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

private

Definition at line 900 of file Importer.h.

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

size_t import_export::Importer::max_threads

private

Definition at line 897 of file Importer.h.

Referenced by importDelimited(), Importer(), importGDALGeo(), and importGDALRaster().

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

Private 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