import_export Namespace Reference

Namespaces
	anonymous_namespace{ExpressionParser.cpp}
	anonymous_namespace{ForeignDataImporter.cpp}
	anonymous_namespace{Importer.cpp}
	anonymous_namespace{QueryExporterCSV.cpp}
	anonymous_namespace{QueryExporterGDAL.cpp}
	anonymous_namespace{RasterImporter.cpp}
	delimited_parser

Classes
class	AbstractImporter
struct	CopyParams
class	ExpressionParser
class	ForeignDataImporter
class	ImportBatchResult
struct	GeoImportException
class	ColumnNotGeoError
struct	BadRowsTracker
class	ImporterUtils
class	TypedImportBuffer
class	Loader
struct	ImportStatus
class	DataStreamSink
class	Detector
class	Importer
struct	MetadataColumnInfo
class	QueryExporter
class	QueryExporterCSV
class	QueryExporterGDAL
class	GCPTransformer
class	RasterImporter

Typedefs
using	FieldNameToIndexMapType = std::map< std::string, size_t >
using	ColumnNameToSourceNameMapType = std::map< std::string, std::string >
using	FeaturePtrVector = std::vector< Geospatial::GDAL::FeatureUqPtr >
using	ArraySliceRange = std::pair< size_t, size_t >
using	MetadataColumnInfos = std::vector< MetadataColumnInfo >

Enumerations
enum	ImportHeaderRow { ImportHeaderRow::kAutoDetect, ImportHeaderRow::kNoHeader, ImportHeaderRow::kHasHeader }
enum	RasterPointType {   RasterPointType::kNone, RasterPointType::kAuto, RasterPointType::kSmallInt, RasterPointType::kInt,   RasterPointType::kFloat, RasterPointType::kDouble, RasterPointType::kPoint }
enum	RasterPointTransform { RasterPointTransform::kNone, RasterPointTransform::kAuto, RasterPointTransform::kFile, RasterPointTransform::kWorld }
enum	SourceType {   SourceType::kUnknown, SourceType::kUnsupported, SourceType::kDelimitedFile, SourceType::kGeoFile,   SourceType::kRasterFile, SourceType::kParquetFile, SourceType::kOdbc, SourceType::kRegexParsedFile }

Functions
static const std::string	trim_space (const char *field, const size_t len)
Datum	NullArrayDatum (SQLTypeInfo &ti)
ArrayDatum	StringToArray (const std::string &s, const SQLTypeInfo &ti, const CopyParams &copy_params)
ArrayDatum	NullArray (const SQLTypeInfo &ti)
void	addBinaryStringArray (const TDatum &datum, std::vector< std::string > &string_vec)
Datum	TDatumToDatum (const TDatum &datum, SQLTypeInfo &ti)
ArrayDatum	TDatumToArrayDatum (const TDatum &datum, const SQLTypeInfo &ti)
bool	importGeoFromLonLat (double lon, double lat, std::vector< double > &coords, std::vector< double > &bounds, SQLTypeInfo &ti)
static ImportStatus	import_thread_delimited (int thread_id, Importer *importer, std::unique_ptr< char[]> scratch_buffer, size_t begin_pos, size_t end_pos, size_t total_size, size_t first_row_index_this_buffer, const Catalog_Namespace::SessionInfo *session_info, Executor *executor)
static ImportStatus	import_thread_shapefile (int thread_id, Importer *importer, OGRCoordinateTransformation *coordinate_transformation, const FeaturePtrVector &features, size_t firstFeature, size_t numFeatures, const FieldNameToIndexMapType &fieldNameToIndexMap, const ColumnNameToSourceNameMapType &columnNameToSourceNameMap, const Catalog_Namespace::SessionInfo *session_info, Executor *executor, const MetadataColumnInfos &metadata_column_infos)
template<class T >
bool	try_cast (const std::string &str)
void	gdalGatherFilesInArchiveRecursive (const std::string &archive_path, std::vector< std::string > &files)
std::vector< std::unique_ptr < TypedImportBuffer > >	setup_column_loaders (const TableDescriptor *td, Loader *loader)
std::vector< std::unique_ptr < TypedImportBuffer > >	fill_missing_columns (const Catalog_Namespace::Catalog *cat, Fragmenter_Namespace::InsertData &insert_data)
std::unique_ptr< AbstractImporter >	create_importer (Catalog_Namespace::Catalog &catalog, const TableDescriptor *td, const std::string &copy_from_source, const import_export::CopyParams &copy_params)
MetadataColumnInfos	parse_add_metadata_columns (const std::string &add_metadata_columns, const std::string &file_path)
size_t	num_import_threads (const int32_t copy_params_threads)

Variables
static constexpr size_t	kImportFileBufferSize = (1 << 23)
static constexpr size_t	max_import_buffer_resize_byte_size = 1024 * 1024 * 1024
static constexpr bool	PROMOTE_POINT_TO_MULTIPOINT = false
static constexpr bool	PROMOTE_LINESTRING_TO_MULTILINESTRING = false
static constexpr bool	PROMOTE_POLYGON_TO_MULTIPOLYGON = true
static heavyai::shared_mutex	status_mutex
static std::map< std::string, ImportStatus >	import_status_map
static const size_t	kImportRowLimit = 10000

Typedef Documentation

using import_export::ArraySliceRange = typedef std::pair<size_t, size_t>

Definition at line 74 of file Importer.h.

using import_export::ColumnNameToSourceNameMapType = typedef std::map<std::string, std::string>

Definition at line 151 of file Importer.cpp.

using import_export::FeaturePtrVector = typedef std::vector<Geospatial::GDAL::FeatureUqPtr>

Definition at line 152 of file Importer.cpp.

using import_export::FieldNameToIndexMapType = typedef std::map<std::string, size_t>

Definition at line 150 of file Importer.cpp.

using import_export::MetadataColumnInfos = typedef std::vector<MetadataColumnInfo>

Definition at line 36 of file MetadataColumn.h.

Enumeration Type Documentation

enum import_export::ImportHeaderRow

strong

Enumerator
kAutoDetect
kNoHeader
kHasHeader

Definition at line 39 of file CopyParams.h.

{ kAutoDetect, kNoHeader, kHasHeader };

enum import_export::RasterPointTransform

strong

Enumerator
kNone
kAuto
kFile
kWorld

Definition at line 41 of file CopyParams.h.

{ kNone, kAuto, kFile, kWorld };

enum import_export::RasterPointType

strong

Enumerator
kNone
kAuto
kSmallInt
kInt
kFloat
kDouble
kPoint

Definition at line 40 of file CopyParams.h.

{ kNone, kAuto, kSmallInt, kInt, kFloat, kDouble, kPoint };

enum import_export::SourceType

strong

Enumerator
kUnknown
kUnsupported
kDelimitedFile
kGeoFile
kRasterFile
kParquetFile
kOdbc
kRegexParsedFile

Definition at line 27 of file SourceType.h.

                      {
  kUnknown,
  kUnsupported,
  kDelimitedFile,
  kGeoFile,
  kRasterFile,
  kParquetFile,
  kOdbc,
  kRegexParsedFile,
};

Function Documentation

void import_export::addBinaryStringArray	(	const TDatum &	datum,
		std::vector< std::string > &	string_vec
	)

Definition at line 415 of file Importer.cpp.

Referenced by import_export::TypedImportBuffer::add_value().

                                                                                 {
  const auto& arr = datum.val.arr_val;
  for (const auto& elem_datum : arr) {
    string_vec.push_back(elem_datum.val.str_val);
  }
}

Here is the caller graph for this function:

std::unique_ptr< AbstractImporter > import_export::create_importer	(	Catalog_Namespace::Catalog &	catalog,
		const TableDescriptor *	td,
		const std::string &	copy_from_source,
		const import_export::CopyParams &	copy_params
	)

Definition at line 6211 of file Importer.cpp.

References g_enable_fsi_regex_import, g_enable_legacy_delimited_import, kDelimitedFile, kParquetFile, kRegexParsedFile, and import_export::CopyParams::source_type.

Referenced by Parser::CopyTableStmt::execute(), and DBHandler::import_table().

                                                {
  if (copy_params.source_type == import_export::SourceType::kParquetFile) {
#ifdef ENABLE_IMPORT_PARQUET
    if (!g_enable_legacy_parquet_import) {
      return std::make_unique<import_export::ForeignDataImporter>(
          copy_from_source, copy_params, td);
    }
#else
    throw std::runtime_error("Parquet not supported!");
#endif
  }

  if (copy_params.source_type == import_export::SourceType::kDelimitedFile &&
      !g_enable_legacy_delimited_import) {
    return std::make_unique<import_export::ForeignDataImporter>(
        copy_from_source, copy_params, td);
  }

  if (copy_params.source_type == import_export::SourceType::kRegexParsedFile) {
    if (g_enable_fsi_regex_import) {
      return std::make_unique<import_export::ForeignDataImporter>(
          copy_from_source, copy_params, td);
    } else {
      throw std::runtime_error(
          "Regex parsed import only supported using 'fsi-regex-import' flag");
    }
  }

  return std::make_unique<import_export::Importer>(
      catalog, td, copy_from_source, copy_params);
}

Here is the caller graph for this function:

std::vector< std::unique_ptr< TypedImportBuffer > > import_export::fill_missing_columns	(	const Catalog_Namespace::Catalog *	cat,
		Fragmenter_Namespace::InsertData &	insert_data
	)

Definition at line 6141 of file Importer.cpp.

References anonymous_namespace{Utm.h}::a, CHECK, ColumnDescriptor::columnId, Fragmenter_Namespace::InsertData::columnIds, ColumnDescriptor::columnName, ColumnDescriptor::columnType, Fragmenter_Namespace::InsertData::data, ColumnDescriptor::default_value, SQLTypeInfo::get_comp_param(), SQLTypeInfo::get_compression(), import_export::TypedImportBuffer::get_data_block_pointers(), SQLTypeInfo::get_physical_cols(), SQLTypeInfo::get_subtype(), SQLTypeInfo::get_type(), Catalog_Namespace::Catalog::getAllColumnMetadataForTable(), Geospatial::GeoTypesFactory::getGeoColumns(), Catalog_Namespace::Catalog::getMetadataForDict(), Fragmenter_Namespace::InsertData::is_default, SQLTypeInfo::is_geometry(), IS_STRING, kARRAY, kENCODING_DICT, import_export::Importer::set_geo_physical_import_buffer(), gpu_enabled::sort(), and Fragmenter_Namespace::InsertData::tableId.

Referenced by RelAlgExecutor::executeSimpleInsert(), DBHandler::insert_data(), and Parser::InsertIntoTableAsSelectStmt::populateData().

                                                 {
  std::vector<std::unique_ptr<import_export::TypedImportBuffer>> defaults_buffers;
  if (insert_data.is_default.size() == 0) {
    insert_data.is_default.resize(insert_data.columnIds.size(), false);
  }
  CHECK(insert_data.is_default.size() == insert_data.is_default.size());
  auto cds = cat->getAllColumnMetadataForTable(insert_data.tableId, false, false, true);
  if (cds.size() == insert_data.columnIds.size()) {
    // all columns specified
    return defaults_buffers;
  }
  for (auto cd : cds) {
    if (std::find(insert_data.columnIds.begin(),
                  insert_data.columnIds.end(),
                  cd->columnId) == insert_data.columnIds.end()) {
      StringDictionary* dict = nullptr;
      if (cd->columnType.get_type() == kARRAY &&
          IS_STRING(cd->columnType.get_subtype()) && !cd->default_value.has_value()) {
        throw std::runtime_error("Cannot omit column \"" + cd->columnName +
                                 "\": omitting TEXT arrays is not supported yet");
      }
      if (cd->columnType.get_compression() == kENCODING_DICT) {
        dict = cat->getMetadataForDict(cd->columnType.get_comp_param())->stringDict.get();
      }
      defaults_buffers.emplace_back(std::make_unique<TypedImportBuffer>(cd, dict));
    }
  }
  // put buffers in order to fill geo sub-columns properly
  std::sort(defaults_buffers.begin(),
            defaults_buffers.end(),
            [](decltype(defaults_buffers[0])& a, decltype(defaults_buffers[0])& b) {
              return a->getColumnDesc()->columnId < b->getColumnDesc()->columnId;
            });
  for (size_t i = 0; i < defaults_buffers.size(); ++i) {
    auto cd = defaults_buffers[i]->getColumnDesc();
    std::string default_value = cd->default_value.value_or("NULL");
    defaults_buffers[i]->add_value(
        cd, default_value, !cd->default_value.has_value(), import_export::CopyParams());
    if (cd->columnType.is_geometry()) {
      std::vector<double> coords, bounds;
      std::vector<int> ring_sizes, poly_rings;
      SQLTypeInfo tinfo{cd->columnType};
      const bool validate_with_geos_if_available = false;
      CHECK(Geospatial::GeoTypesFactory::getGeoColumns(default_value,
                                                       tinfo,
                                                       coords,
                                                       bounds,
                                                       ring_sizes,
                                                       poly_rings,
                                                       validate_with_geos_if_available));
      // set physical columns starting with the following ID
      auto next_col = i + 1;
      import_export::Importer::set_geo_physical_import_buffer(
          *cat, cd, defaults_buffers, next_col, coords, bounds, ring_sizes, poly_rings);
      // skip physical columns filled with the call above
      i += cd->columnType.get_physical_cols();
    }
  }
  auto data = import_export::TypedImportBuffer::get_data_block_pointers(defaults_buffers);
  CHECK(data.size() == defaults_buffers.size());
  for (size_t i = 0; i < defaults_buffers.size(); ++i) {
    insert_data.data.push_back(data[i]);
    insert_data.columnIds.push_back(defaults_buffers[i]->getColumnDesc()->columnId);
    insert_data.is_default.push_back(true);
  }
  return defaults_buffers;
}

Here is the call graph for this function:

Here is the caller graph for this function:

void import_export::gdalGatherFilesInArchiveRecursive	(	const std::string &	archive_path,
		std::vector< std::string > &	files
	)

Definition at line 5063 of file Importer.cpp.

References LOG, run_benchmark_import::result, and logger::WARNING.

Referenced by import_export::Importer::gdalGetAllFilesInArchive().

                                                                    {
  // prepare to gather subdirectories
  std::vector<std::string> subdirectories;

  // get entries
  char** entries = VSIReadDir(archive_path.c_str());
  if (!entries) {
    LOG(WARNING) << "Failed to get file listing at archive: " << archive_path;
    return;
  }

  // force scope
  {
    // request clean-up
    ScopeGuard entries_guard = [&] { CSLDestroy(entries); };

    // check all the entries
    int index = 0;
    while (true) {
      // get next entry, or drop out if there isn't one
      char* entry_c = entries[index++];
      if (!entry_c) {
        break;
      }
      std::string entry(entry_c);

      // ignore '.' and '..'
      if (entry == "." || entry == "..") {
        continue;
      }

      // build the full path
      std::string entry_path = archive_path + std::string("/") + entry;

      // is it a file or a sub-folder
      VSIStatBufL sb;
      int result = VSIStatExL(entry_path.c_str(), &sb, VSI_STAT_NATURE_FLAG);
      if (result < 0) {
        break;
      }

      if (VSI_ISDIR(sb.st_mode)) {
        // a directory that ends with .gdb could be a Geodatabase bundle
        // arguably dangerous to decide this purely by name, but any further
        // validation would be very complex especially at this scope
        if (boost::iends_with(entry_path, ".gdb")) {
          // add the directory as if it was a file and don't recurse into it
          files.push_back(entry_path);
        } else {
          // add subdirectory to be recursed into
          subdirectories.push_back(entry_path);
        }
      } else {
        // add this file
        files.push_back(entry_path);
      }
    }
  }

  // recurse into each subdirectories we found
  for (const auto& subdirectory : subdirectories) {
    gdalGatherFilesInArchiveRecursive(subdirectory, files);
  }
}

Here is the caller graph for this function:

static ImportStatus import_export::import_thread_delimited ( int  thread_id, Importer *  importer, std::unique_ptr< char[]>  scratch_buffer, size_t  begin_pos, size_t  end_pos, size_t  total_size, size_t  first_row_index_this_buffer, const Catalog_Namespace::SessionInfo *  session_info, Executor *  executor  )

static

Definition at line 1986 of file Importer.cpp.

References CHECK, CHECK_LT, anonymous_namespace{Importer.cpp}::check_session_interrupted(), Geospatial::GeoTypesFactory::createOGRGeometry(), DEBUG_TIMING, logger::ERROR, measure< TimeT >::execution(), import_export::anonymous_namespace{Importer.cpp}::explode_collections_step1(), import_export::anonymous_namespace{Importer.cpp}::explode_collections_step2(), import_export::delimited_parser::find_beginning(), import_export::CopyParams::geo_explode_collections, geo_promoted_type_match(), import_export::CopyParams::geo_validate_geometry, import_export::Importer::get_column_descs(), import_export::Importer::get_copy_params(), import_export::Importer::get_import_buffers(), import_export::Importer::get_is_array(), import_export::delimited_parser::get_row(), Catalog_Namespace::SessionInfo::get_session_id(), import_export::Importer::getCatalog(), Geospatial::GeoTypesFactory::getGeoColumns(), Geospatial::GeoTypesFactory::getNullGeoColumns(), importGeoFromLonLat(), logger::INFO, IS_GEO, is_null(), ImportHelpers::is_null_datum(), kDelimitedFile, kMULTIPOINT, kPOINT, import_export::Importer::load(), import_export::ImportStatus::load_failed, import_export::ImportStatus::load_msg, LOG, import_export::CopyParams::lonlat, import_export::CopyParams::max_reject, import_export::CopyParams::null_str, shared::printContainer(), import_export::ImportStatus::rows_completed, import_export::ImportStatus::rows_rejected, import_export::Importer::set_geo_physical_import_buffer(), import_export::CopyParams::source_srid, import_export::CopyParams::source_type, sv_strip(), gpu_enabled::swap(), import_export::ImportStatus::thread_id, logger::thread_id(), to_string(), import_export::CopyParams::trim_spaces, and UNLIKELY.

Referenced by import_export::Importer::importDelimited().

                        {
  ImportStatus thread_import_status;
  int64_t total_get_row_time_us = 0;
  int64_t total_str_to_val_time_us = 0;
  auto query_session = session_info ? session_info->get_session_id() : "";
  CHECK(scratch_buffer);
  auto buffer = scratch_buffer.get();
  auto load_ms = measure<>::execution([]() {});

  thread_import_status.thread_id = thread_id;

  auto ms = measure<>::execution([&]() {
    const CopyParams& copy_params = importer->get_copy_params();
    const std::list<const ColumnDescriptor*>& col_descs = importer->get_column_descs();
    size_t begin =
        delimited_parser::find_beginning(buffer, begin_pos, end_pos, copy_params);
    const char* thread_buf = buffer + begin_pos + begin;
    const char* thread_buf_end = buffer + end_pos;
    const char* buf_end = buffer + total_size;
    bool try_single_thread = false;
    std::vector<std::unique_ptr<TypedImportBuffer>>& import_buffers =
        importer->get_import_buffers(thread_id);
    auto us = measure<std::chrono::microseconds>::execution([&]() {});
    int phys_cols = 0;
    int point_cols = 0;
    for (const auto cd : col_descs) {
      const auto& col_ti = cd->columnType;
      phys_cols += col_ti.get_physical_cols();
      if (cd->columnType.get_type() == kPOINT ||
          cd->columnType.get_type() == kMULTIPOINT) {
        point_cols++;
      }
    }
    auto num_cols = col_descs.size() - phys_cols;
    for (const auto& p : import_buffers) {
      p->clear();
    }
    std::vector<std::string_view> row;
    size_t row_index_plus_one = 0;
    for (const char* p = thread_buf; p < thread_buf_end; p++) {
      row.clear();
      std::vector<std::unique_ptr<char[]>>
          tmp_buffers;  // holds string w/ removed escape chars, etc
      row_index_plus_one++;
      if (DEBUG_TIMING) {
        us = measure<std::chrono::microseconds>::execution([&]() {
          p = import_export::delimited_parser::get_row(p,
                                                       thread_buf_end,
                                                       buf_end,
                                                       copy_params,
                                                       importer->get_is_array(),
                                                       row,
                                                       tmp_buffers,
                                                       try_single_thread,
                                                       true);
        });
        total_get_row_time_us += us;
      } else {
        p = import_export::delimited_parser::get_row(p,
                                                     thread_buf_end,
                                                     buf_end,
                                                     copy_params,
                                                     importer->get_is_array(),
                                                     row,
                                                     tmp_buffers,
                                                     try_single_thread,
                                                     true);
      }
      // Each POINT could consume two separate coords instead of a single WKT
      if (row.size() < num_cols || (num_cols + point_cols) < row.size()) {
        thread_import_status.rows_rejected++;
        LOG(ERROR) << "Incorrect Row (expected " << num_cols << " columns, has "
                   << row.size() << "): " << shared::printContainer(row);
        if (thread_import_status.rows_rejected > copy_params.max_reject) {
          break;
        }
        continue;
      }

      //
      // lambda for importing a row (perhaps multiple times if exploding a collection)
      //

      auto execute_import_row = [&](OGRGeometry* import_geometry) {
        size_t import_idx = 0;
        size_t col_idx = 0;
        try {
          for (auto cd_it = col_descs.begin(); cd_it != col_descs.end(); cd_it++) {
            auto cd = *cd_it;
            const auto& col_ti = cd->columnType;

            bool is_null =
                ImportHelpers::is_null_datum(row[import_idx], copy_params.null_str);
            // Note: default copy_params.null_str is "\N", but everyone uses "NULL".
            // So initially nullness may be missed and not passed to add_value,
            // which then might also check and still decide it's actually a NULL, e.g.
            // if kINT doesn't start with a digit or a '-' then it's considered NULL.
            // So "NULL" is not recognized as NULL but then it's not recognized as
            // a valid kINT, so it's a NULL after all.
            // Checking for "NULL" here too, as a widely accepted notation for NULL.

            // Treating empty as NULL
            if (!cd->columnType.is_string() && row[import_idx].empty()) {
              is_null = true;
            }
            if (!cd->columnType.is_string() && !copy_params.trim_spaces) {
              // everything but strings should be always trimmed
              row[import_idx] = sv_strip(row[import_idx]);
            }

            if (col_ti.get_physical_cols() == 0) {
              // not geo

              import_buffers[col_idx]->add_value(
                  cd, row[import_idx], is_null, copy_params);

              // next
              ++import_idx;
              ++col_idx;
            } else {
              // geo

              // store null string in the base column
              import_buffers[col_idx]->add_value(
                  cd, copy_params.null_str, true, copy_params);

              // WKT from string we're not storing
              auto const& geo_string = row[import_idx];

              // next
              ++import_idx;
              ++col_idx;

              SQLTypes col_type = col_ti.get_type();
              CHECK(IS_GEO(col_type));

              std::vector<double> coords;
              std::vector<double> bounds;
              std::vector<int> ring_sizes;
              std::vector<int> poly_rings;

              // if this is a POINT or MULTIPOINT column, and the field is not null, and
              // looks like a scalar numeric value (and not a hex blob) attempt to import
              // two columns as lon/lat (or lat/lon)
              if ((col_type == kPOINT || col_type == kMULTIPOINT) && !is_null &&
                  geo_string.size() > 0 &&
                  (geo_string[0] == '.' || isdigit(geo_string[0]) ||
                   geo_string[0] == '-') &&
                  geo_string.find_first_of("ABCDEFabcdef") == std::string::npos) {
                double lon = std::atof(std::string(geo_string).c_str());
                double lat = NAN;
                auto lat_str = row[import_idx];
                ++import_idx;
                if (lat_str.size() > 0 &&
                    (lat_str[0] == '.' || isdigit(lat_str[0]) || lat_str[0] == '-')) {
                  lat = std::atof(std::string(lat_str).c_str());
                }
                // Swap coordinates if this table uses a reverse order: lat/lon
                if (!copy_params.lonlat) {
                  std::swap(lat, lon);
                }
                // TODO: should check if POINT/MULTIPOINT column should have been declared
                // with SRID WGS 84, EPSG 4326 ? if (col_ti.get_dimension() != 4326) {
                //  throw std::runtime_error("POINT column " + cd->columnName + " is
                //  not WGS84, cannot insert lon/lat");
                // }
                SQLTypeInfo import_ti{col_ti};
                if (copy_params.source_type ==
                        import_export::SourceType::kDelimitedFile &&
                    import_ti.get_output_srid() == 4326) {
                  auto srid0 = copy_params.source_srid;
                  if (srid0 > 0) {
                    // srid0 -> 4326 transform is requested on import
                    import_ti.set_input_srid(srid0);
                  }
                }
                if (!importGeoFromLonLat(lon, lat, coords, bounds, import_ti)) {
                  throw std::runtime_error(
                      "Cannot read lon/lat to insert into POINT/MULTIPOINT column " +
                      cd->columnName);
                }
              } else {
                // import it
                SQLTypeInfo import_ti{col_ti};
                if (copy_params.source_type ==
                        import_export::SourceType::kDelimitedFile &&
                    import_ti.get_output_srid() == 4326) {
                  auto srid0 = copy_params.source_srid;
                  if (srid0 > 0) {
                    // srid0 -> 4326 transform is requested on import
                    import_ti.set_input_srid(srid0);
                  }
                }
                if (is_null) {
                  if (col_ti.get_notnull()) {
                    throw std::runtime_error("NULL geo for column " + cd->columnName);
                  }
                  Geospatial::GeoTypesFactory::getNullGeoColumns(
                      import_ti, coords, bounds, ring_sizes, poly_rings);
                } else {
                  if (import_geometry) {
                    // geometry already exploded
                    if (!Geospatial::GeoTypesFactory::getGeoColumns(
                            import_geometry,
                            import_ti,
                            coords,
                            bounds,
                            ring_sizes,
                            poly_rings,
                            copy_params.geo_validate_geometry)) {
                      std::string msg =
                          "Failed to extract valid geometry from exploded row " +
                          std::to_string(first_row_index_this_buffer +
                                         row_index_plus_one) +
                          " for column " + cd->columnName;
                      throw std::runtime_error(msg);
                    }
                  } else {
                    // extract geometry directly from WKT
                    if (!Geospatial::GeoTypesFactory::getGeoColumns(
                            std::string(geo_string),
                            import_ti,
                            coords,
                            bounds,
                            ring_sizes,
                            poly_rings,
                            copy_params.geo_validate_geometry)) {
                      std::string msg = "Failed to extract valid geometry from row " +
                                        std::to_string(first_row_index_this_buffer +
                                                       row_index_plus_one) +
                                        " for column " + cd->columnName;
                      throw std::runtime_error(msg);
                    }
                  }

                  // validate types
                  if (!geo_promoted_type_match(import_ti.get_type(), col_type)) {
                    throw std::runtime_error(
                        "Imported geometry doesn't match the type of column " +
                        cd->columnName);
                  }
                }
              }

              // import extracted geo
              Importer::set_geo_physical_import_buffer(importer->getCatalog(),
                                                       cd,
                                                       import_buffers,
                                                       col_idx,
                                                       coords,
                                                       bounds,
                                                       ring_sizes,
                                                       poly_rings);

              // skip remaining physical columns
              for (int i = 0; i < cd->columnType.get_physical_cols(); ++i) {
                ++cd_it;
              }
            }
          }
          if (UNLIKELY((thread_import_status.rows_completed & 0xFFFF) == 0 &&
                       check_session_interrupted(query_session, executor))) {
            thread_import_status.load_failed = true;
            thread_import_status.load_msg =
                "Table load was cancelled via Query Interrupt";
            return;
          }
          thread_import_status.rows_completed++;
        } catch (const std::exception& e) {
          for (size_t col_idx_to_pop = 0; col_idx_to_pop < col_idx; ++col_idx_to_pop) {
            import_buffers[col_idx_to_pop]->pop_value();
          }
          thread_import_status.rows_rejected++;
          LOG(ERROR) << "Input exception thrown: " << e.what()
                     << ". Row discarded. Data: " << shared::printContainer(row);
          if (thread_import_status.rows_rejected > copy_params.max_reject) {
            LOG(ERROR) << "Load was cancelled due to max reject rows being reached";
            thread_import_status.load_failed = true;
            thread_import_status.load_msg =
                "Load was cancelled due to max reject rows being reached";
          }
        }
      };  // End of lambda

      if (copy_params.geo_explode_collections) {
        // explode and import
        auto const [collection_col_idx, collection_child_type, collection_col_name] =
            explode_collections_step1(col_descs);
        // pull out the collection WKT or WKB hex
        CHECK_LT(collection_col_idx, (int)row.size()) << "column index out of range";
        auto const& collection_geo_string = row[collection_col_idx];
        // convert to OGR
        OGRGeometry* ogr_geometry = nullptr;
        ScopeGuard destroy_ogr_geometry = [&] {
          if (ogr_geometry) {
            OGRGeometryFactory::destroyGeometry(ogr_geometry);
          }
        };
        ogr_geometry = Geospatial::GeoTypesFactory::createOGRGeometry(
            std::string(collection_geo_string), copy_params.geo_validate_geometry);
        // do the explode and import
        us = explode_collections_step2(ogr_geometry,
                                       collection_child_type,
                                       collection_col_name,
                                       first_row_index_this_buffer + row_index_plus_one,
                                       execute_import_row);
      } else {
        // import non-collection row just once
        us = measure<std::chrono::microseconds>::execution(
            [&] { execute_import_row(nullptr); });
      }

      if (thread_import_status.load_failed) {
        break;
      }
    }  // end thread
    total_str_to_val_time_us += us;
    if (!thread_import_status.load_failed && thread_import_status.rows_completed > 0) {
      load_ms = measure<>::execution([&]() {
        importer->load(import_buffers, thread_import_status.rows_completed, session_info);
      });
    }
  });  // end execution

  if (DEBUG_TIMING && !thread_import_status.load_failed &&
      thread_import_status.rows_completed > 0) {
    LOG(INFO) << "Thread" << std::this_thread::get_id() << ":"
              << thread_import_status.rows_completed << " rows inserted in "
              << (double)ms / 1000.0 << "sec, Insert Time: " << (double)load_ms / 1000.0
              << "sec, get_row: " << (double)total_get_row_time_us / 1000000.0
              << "sec, str_to_val: " << (double)total_str_to_val_time_us / 1000000.0
              << "sec" << std::endl;
  }

  return thread_import_status;
}

Here is the call graph for this function:

Here is the caller graph for this function:

static ImportStatus import_export::import_thread_shapefile ( int  thread_id, Importer *  importer, OGRCoordinateTransformation *  coordinate_transformation, const FeaturePtrVector &  features, size_t  firstFeature, size_t  numFeatures, const FieldNameToIndexMapType &  fieldNameToIndexMap, const ColumnNameToSourceNameMapType &  columnNameToSourceNameMap, const Catalog_Namespace::SessionInfo *  session_info, Executor *  executor, const MetadataColumnInfos &  metadata_column_infos  )

static

Definition at line 2338 of file Importer.cpp.

References CHECK, anonymous_namespace{Importer.cpp}::check_session_interrupted(), Geospatial::compress_coords(), DEBUG_TIMING, Executor::ERR_INTERRUPTED, logger::ERROR, import_export::anonymous_namespace{Importer.cpp}::explode_collections_step1(), import_export::anonymous_namespace{Importer.cpp}::explode_collections_step2(), import_export::CopyParams::geo_explode_collections, geo_promoted_type_match(), import_export::CopyParams::geo_validate_geometry, import_export::Importer::get_column_descs(), import_export::Importer::get_copy_params(), import_export::Importer::get_import_buffers(), Catalog_Namespace::SessionInfo::get_session_id(), QueryExecutionError::getErrorCode(), Geospatial::GeoTypesFactory::getGeoColumns(), Geospatial::GeoTypesFactory::getNullGeoColumns(), logger::INFO, kLINESTRING, kMULTILINESTRING, kMULTIPOINT, kMULTIPOLYGON, kPOLYGON, import_export::Importer::load(), import_export::ImportStatus::load_failed, import_export::ImportStatus::load_msg, LOG, import_export::CopyParams::null_str, import_export::ImportStatus::rows_completed, import_export::ImportStatus::rows_rejected, generate_TableFunctionsFactory_init::separator, import_export::ImportStatus::thread_id, logger::thread_id(), timer_start(), TIMER_STOP, to_string(), and UNLIKELY.

Referenced by import_export::Importer::importGDALGeo().

                                                      {
  ImportStatus thread_import_status;
  const CopyParams& copy_params = importer->get_copy_params();
  const std::list<const ColumnDescriptor*>& col_descs = importer->get_column_descs();
  std::vector<std::unique_ptr<TypedImportBuffer>>& import_buffers =
      importer->get_import_buffers(thread_id);
  auto query_session = session_info ? session_info->get_session_id() : "";
  for (const auto& p : import_buffers) {
    p->clear();
  }

  auto convert_timer = timer_start();

  // for all the features in this chunk...
  for (size_t iFeature = 0; iFeature < numFeatures; iFeature++) {
    // ignore null features
    if (!features[iFeature]) {
      continue;
    }

    // get this feature's geometry
    // for geodatabase, we need to consider features with no geometry
    // as we still want to create a table, even if it has no geo column
    OGRGeometry* pGeometry = features[iFeature]->GetGeometryRef();
    if (pGeometry && coordinate_transformation) {
      pGeometry->transform(coordinate_transformation);
    }

    //
    // lambda for importing a feature (perhaps multiple times if exploding a collection)
    //

    auto execute_import_feature = [&](OGRGeometry* import_geometry) {
      size_t col_idx = 0;
      try {
        if (UNLIKELY((thread_import_status.rows_completed & 0xFFFF) == 0 &&
                     check_session_interrupted(query_session, executor))) {
          thread_import_status.load_failed = true;
          thread_import_status.load_msg = "Table load was cancelled via Query Interrupt";
          throw QueryExecutionError(Executor::ERR_INTERRUPTED);
        }

        uint32_t field_column_count{0u};
        uint32_t metadata_column_count{0u};

        for (auto cd_it = col_descs.begin(); cd_it != col_descs.end(); cd_it++) {
          auto cd = *cd_it;

          // is this a geo column?
          const auto& col_ti = cd->columnType;
          if (col_ti.is_geometry()) {
            // Note that this assumes there is one and only one geo column in the
            // table. Currently, the importer only supports reading a single
            // geospatial feature from an input shapefile / geojson file, but this
            // code will need to be modified if that changes
            SQLTypes col_type = col_ti.get_type();

            // store null string in the base column
            import_buffers[col_idx]->add_value(
                cd, copy_params.null_str, true, copy_params);
            ++col_idx;

            // the data we now need to extract for the other columns
            std::vector<double> coords;
            std::vector<double> bounds;
            std::vector<int> ring_sizes;
            std::vector<int> poly_rings;

            // extract it
            SQLTypeInfo import_ti{col_ti};
            bool is_null_geo = !import_geometry;
            if (is_null_geo) {
              if (col_ti.get_notnull()) {
                throw std::runtime_error("NULL geo for column " + cd->columnName);
              }
              Geospatial::GeoTypesFactory::getNullGeoColumns(
                  import_ti, coords, bounds, ring_sizes, poly_rings);
            } else {
              if (!Geospatial::GeoTypesFactory::getGeoColumns(
                      import_geometry,
                      import_ti,
                      coords,
                      bounds,
                      ring_sizes,
                      poly_rings,
                      copy_params.geo_validate_geometry)) {
                std::string msg = "Failed to extract valid geometry from feature " +
                                  std::to_string(firstFeature + iFeature + 1) +
                                  " for column " + cd->columnName;
                throw std::runtime_error(msg);
              }

              // validate types
              if (!geo_promoted_type_match(import_ti.get_type(), col_type)) {
                throw std::runtime_error(
                    "Imported geometry doesn't match the type of column " +
                    cd->columnName);
              }
            }

            // create coords array value and add it to the physical column
            ++cd_it;
            auto cd_coords = *cd_it;
            std::vector<TDatum> td_coord_data;
            if (!is_null_geo) {
              std::vector<uint8_t> compressed_coords =
                  Geospatial::compress_coords(coords, col_ti);
              for (auto cc : compressed_coords) {
                TDatum td_byte;
                td_byte.val.int_val = cc;
                td_coord_data.push_back(td_byte);
              }
            }
            TDatum tdd_coords;
            tdd_coords.val.arr_val = td_coord_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) {
              // Create [linest]ring_sizes array value and add it to the physical column
              ++cd_it;
              auto cd_ring_sizes = *cd_it;
              std::vector<TDatum> td_ring_sizes;
              if (!is_null_geo) {
                for (auto 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) {
              // Create poly_rings array value and add it to the physical column
              ++cd_it;
              auto cd_poly_rings = *cd_it;
              std::vector<TDatum> td_poly_rings;
              if (!is_null_geo) {
                for (auto 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) {
              // Create bounds array value and add it to the physical column
              ++cd_it;
              auto cd_bounds = *cd_it;
              std::vector<TDatum> td_bounds_data;
              if (!is_null_geo) {
                for (auto 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;
            }
          } else if (field_column_count < fieldNameToIndexMap.size()) {
            //
            // field column
            //
            auto const cit = columnNameToSourceNameMap.find(cd->columnName);
            CHECK(cit != columnNameToSourceNameMap.end());
            auto const& field_name = cit->second;

            auto const fit = fieldNameToIndexMap.find(field_name);
            if (fit == fieldNameToIndexMap.end()) {
              throw ColumnNotGeoError(cd->columnName);
            }

            auto const& field_index = fit->second;
            CHECK(field_index < fieldNameToIndexMap.size());

            auto const& feature = features[iFeature];

            auto field_defn = feature->GetFieldDefnRef(field_index);
            CHECK(field_defn);

            // OGRFeature::GetFieldAsString() can only return 80 characters
            // so for array columns, we are obliged to fetch the actual values
            // and construct the concatenated string ourselves

            std::string value_string;
            int array_index = 0, array_size = 0;

            auto stringify_numeric_list = [&](auto* values) {
              value_string = "{";
              while (array_index < array_size) {
                auto separator = (array_index > 0) ? "," : "";
                value_string += separator + std::to_string(values[array_index]);
                array_index++;
              }
              value_string += "}";
            };

            auto field_type = field_defn->GetType();
            switch (field_type) {
              case OFTInteger:
              case OFTInteger64:
              case OFTReal:
              case OFTString:
              case OFTBinary:
              case OFTDate:
              case OFTTime:
              case OFTDateTime: {
                value_string = feature->GetFieldAsString(field_index);
              } break;
              case OFTIntegerList: {
                auto* values = feature->GetFieldAsIntegerList(field_index, &array_size);
                stringify_numeric_list(values);
              } break;
              case OFTInteger64List: {
                auto* values = feature->GetFieldAsInteger64List(field_index, &array_size);
                stringify_numeric_list(values);
              } break;
              case OFTRealList: {
                auto* values = feature->GetFieldAsDoubleList(field_index, &array_size);
                stringify_numeric_list(values);
              } break;
              case OFTStringList: {
                auto** array_of_strings = feature->GetFieldAsStringList(field_index);
                value_string = "{";
                if (array_of_strings) {
                  while (auto* this_string = array_of_strings[array_index]) {
                    auto separator = (array_index > 0) ? "," : "";
                    value_string += separator + std::string(this_string);
                    array_index++;
                  }
                }
                value_string += "}";
              } break;
              default:
                throw std::runtime_error("Unsupported geo file field type (" +
                                         std::to_string(static_cast<int>(field_type)) +
                                         ")");
            }

            import_buffers[col_idx]->add_value(cd, value_string, false, copy_params);
            ++col_idx;
            field_column_count++;
          } else if (metadata_column_count < metadata_column_infos.size()) {
            //
            // metadata column
            //
            auto const& mci = metadata_column_infos[metadata_column_count];
            if (mci.column_descriptor.columnName != cd->columnName) {
              throw std::runtime_error("Metadata column name mismatch");
            }
            import_buffers[col_idx]->add_value(cd, mci.value, false, copy_params);
            ++col_idx;
            metadata_column_count++;
          } else {
            throw std::runtime_error("Column count mismatch");
          }
        }
        thread_import_status.rows_completed++;
      } catch (QueryExecutionError& e) {
        if (e.getErrorCode() == Executor::ERR_INTERRUPTED) {
          throw e;
        }
      } catch (ColumnNotGeoError& e) {
        LOG(ERROR) << "Input exception thrown: " << e.what() << ". Aborting import.";
        throw std::runtime_error(e.what());
      } catch (const std::exception& e) {
        for (size_t col_idx_to_pop = 0; col_idx_to_pop < col_idx; ++col_idx_to_pop) {
          import_buffers[col_idx_to_pop]->pop_value();
        }
        thread_import_status.rows_rejected++;
        LOG(ERROR) << "Input exception thrown: " << e.what() << ". Row discarded.";
      }
    };

    if (pGeometry && copy_params.geo_explode_collections) {
      // explode and import
      auto const [collection_idx_type_name, collection_child_type, collection_col_name] =
          explode_collections_step1(col_descs);
      explode_collections_step2(pGeometry,
                                collection_child_type,
                                collection_col_name,
                                firstFeature + iFeature + 1,
                                execute_import_feature);
    } else {
      // import non-collection or null feature just once
      execute_import_feature(pGeometry);
    }
  }  // end features

  float convert_s = TIMER_STOP(convert_timer);

  float load_s = 0.0f;
  if (thread_import_status.rows_completed > 0) {
    auto load_timer = timer_start();
    importer->load(import_buffers, thread_import_status.rows_completed, session_info);
    load_s = TIMER_STOP(load_timer);
  }

  if (DEBUG_TIMING && thread_import_status.rows_completed > 0) {
    LOG(INFO) << "DEBUG:      Process " << convert_s << "s";
    LOG(INFO) << "DEBUG:      Load " << load_s << "s";
    LOG(INFO) << "DEBUG:      Total " << (convert_s + load_s) << "s";
  }

  thread_import_status.thread_id = thread_id;

  return thread_import_status;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bool import_export::importGeoFromLonLat	(	double	lon,
		double	lat,
		std::vector< double > &	coords,
		std::vector< double > &	bounds,
		SQLTypeInfo &	ti
	)

Definition at line 1608 of file Importer.cpp.

References CHECK_EQ, Geospatial::GeoPoint::getColumns(), and SQLTypeInfo::transforms().

Referenced by import_thread_delimited().

                                          {
  if (std::isinf(lat) || std::isnan(lat) || std::isinf(lon) || std::isnan(lon)) {
    return false;
  }
  if (ti.transforms()) {
    Geospatial::GeoPoint pt{std::vector<double>{lon, lat}};
    if (!pt.transform(ti)) {
      return false;
    }
    pt.getColumns(coords);
  } else {
    coords.push_back(lon);
    coords.push_back(lat);
  }
  // in case of promotion to MULTIPOINT
  CHECK_EQ(coords.size(), 2u);
  bounds.reserve(4);
  bounds.push_back(coords[0]);
  bounds.push_back(coords[1]);
  bounds.push_back(coords[0]);
  bounds.push_back(coords[1]);
  return true;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ArrayDatum import_export::NullArray

(

const SQLTypeInfo &

ti

)

Definition at line 371 of file Importer.cpp.

References append_datum(), CHECK, checked_malloc(), SQLTypeInfo::get_elem_type(), SQLTypeInfo::get_size(), NullArrayDatum(), and NullDatum().

Referenced by import_export::TypedImportBuffer::add_value(), import_export::TypedImportBuffer::add_values(), import_export::TypedImportBuffer::addDefaultValues(), import_export::ImporterUtils::composeNullArray(), and TDatumToArrayDatum().

                                            {
  SQLTypeInfo elem_ti = ti.get_elem_type();
  auto len = ti.get_size();

  if (len > 0) {
    // Compose a NULL fixlen array
    int8_t* buf = (int8_t*)checked_malloc(len);
    // First scalar is a NULL_ARRAY sentinel
    Datum d = NullArrayDatum(elem_ti);
    int8_t* p = append_datum(buf, d, elem_ti);
    CHECK(p);
    // Rest is filled with normal NULL sentinels
    Datum d0 = NullDatum(elem_ti);
    while ((p - buf) < len) {
      p = append_datum(p, d0, elem_ti);
      CHECK(p);
    }
    CHECK((p - buf) == len);
    return ArrayDatum(len, buf, true);
  }
  // NULL varlen array
  return ArrayDatum(0, NULL, true);
}

Here is the call graph for this function:

Here is the caller graph for this function:

Datum import_export::NullArrayDatum

(

SQLTypeInfo &

ti

)

Definition at line 273 of file Importer.cpp.

References Datum::bigintval, Datum::boolval, Datum::doubleval, Datum::floatval, import_export::anonymous_namespace{Importer.cpp}::get_type_for_datum(), inline_fixed_encoding_null_array_val(), Datum::intval, kBIGINT, kBOOLEAN, kDATE, kDOUBLE, kFLOAT, kINT, kLINESTRING, kMULTILINESTRING, kMULTIPOINT, kMULTIPOLYGON, kPOINT, kPOLYGON, kSMALLINT, kTIME, kTIMESTAMP, kTINYINT, NULL_ARRAY_DOUBLE, NULL_ARRAY_FLOAT, Datum::smallintval, Datum::tinyintval, and run_benchmark_import::type.

Referenced by NullArray().

                                      {
  Datum d;
  const auto type = get_type_for_datum(ti);
  switch (type) {
    case kBOOLEAN:
      d.boolval = inline_fixed_encoding_null_array_val(ti);
      break;
    case kBIGINT:
      d.bigintval = inline_fixed_encoding_null_array_val(ti);
      break;
    case kINT:
      d.intval = inline_fixed_encoding_null_array_val(ti);
      break;
    case kSMALLINT:
      d.smallintval = inline_fixed_encoding_null_array_val(ti);
      break;
    case kTINYINT:
      d.tinyintval = inline_fixed_encoding_null_array_val(ti);
      break;
    case kFLOAT:
      d.floatval = NULL_ARRAY_FLOAT;
      break;
    case kDOUBLE:
      d.doubleval = NULL_ARRAY_DOUBLE;
      break;
    case kTIME:
    case kTIMESTAMP:
    case kDATE:
      d.bigintval = inline_fixed_encoding_null_array_val(ti);
      break;
    case kPOINT:
    case kMULTIPOINT:
    case kLINESTRING:
    case kMULTILINESTRING:
    case kPOLYGON:
    case kMULTIPOLYGON:
      throw std::runtime_error("Internal error: geometry type in NullArrayDatum.");
    default:
      throw std::runtime_error("Internal error: invalid type in NullArrayDatum.");
  }
  return d;
}

Here is the call graph for this function:

Here is the caller graph for this function:

size_t import_export::num_import_threads ( const int32_t  copy_params_threads )

inline

Definition at line 31 of file thread_count.h.

References g_max_import_threads.

Referenced by foreign_storage::ForeignDataWrapperFactory::createForeignTableProxy(), foreign_storage::get_num_threads(), import_export::Importer::importDelimited(), import_export::Importer::importGDALGeo(), and import_export::Importer::importGDALRaster().

                                                                    {
  if (copy_params_threads > 0) {
    return static_cast<size_t>(copy_params_threads);
  }
  return std::min(static_cast<size_t>(std::thread::hardware_concurrency()),
                  g_max_import_threads);
}

Here is the caller graph for this function:

MetadataColumnInfos import_export::parse_add_metadata_columns	(	const std::string &	add_metadata_columns,
		const std::string &	file_path
	)

Definition at line 35 of file MetadataColumn.cpp.

References ColumnDescriptor::columnName, ColumnDescriptor::columnType, import_export::ExpressionParser::evalAsString(), IS_INTEGER, join(), kBIGINT, kDATE, kDOUBLE, kENCODING_DICT, kFLOAT, kINT, kNULLT, kSMALLINT, kTEXT, kTIME, kTIMESTAMP, kTINYINT, run_benchmark_import::parser, SQLTypeInfo::set_comp_param(), SQLTypeInfo::set_compression(), SQLTypeInfo::set_fixed_size(), SQLTypeInfo::set_type(), import_export::ExpressionParser::setExpression(), import_export::ExpressionParser::setStringConstant(), ColumnDescriptor::sourceName, split(), strip(), to_lower(), and to_string().

Referenced by import_export::Importer::gdalToColumnDescriptorsGeo(), import_export::Importer::gdalToColumnDescriptorsRaster(), import_export::Importer::importGDALGeo(), import_export::Importer::importGDALRaster(), and import_export::Importer::readMetadataSampleGDAL().

                                                                           {
  //
  // each string is "column_name,column_type,expression"
  //
  // column_type can be:
  //   tinyint
  //   smallint
  //   int
  //   bigint
  //   float
  //   double
  //   date
  //   time
  //   timestamp
  //   text
  //
  // expression can be in terms of:
  //   filename
  //   filedir
  //   filepath
  //   etc.
  //

  // anything to do?
  if (add_metadata_columns.length() == 0u) {
    return {};
  }

  // split by ";"
  // @TODO(se) is this safe?
  // probably won't appear in a file name/path or a date/time string
  std::vector<std::string> add_metadata_column_strings;
  boost::split(add_metadata_column_strings, add_metadata_columns, boost::is_any_of(";"));
  if (add_metadata_column_strings.size() == 0u) {
    return {};
  }

  ExpressionParser parser;

  // known string constants
  auto const fn = boost::filesystem::path(file_path).filename().string();
  auto const fd = boost::filesystem::path(file_path).parent_path().string();
  auto const fp = file_path;
  parser.setStringConstant("filename", fn);
  parser.setStringConstant("filedir", fd);
  parser.setStringConstant("filepath", fp);

  MetadataColumnInfos metadata_column_infos;

  // for each requested column...
  for (auto const& add_metadata_column_string : add_metadata_column_strings) {
    // strip
    auto const add_metadata_column = strip(add_metadata_column_string);

    // tokenize and extract
    std::vector<std::string> tokens;
    boost::split(tokens, add_metadata_column, boost::is_any_of(","));
    if (tokens.size() < 3u) {
      throw std::runtime_error("Invalid metadata column info '" + add_metadata_column +
                               "' (must be of the form 'name,type,expression')");
    }
    auto token_itr = tokens.begin();
    auto const column_name = strip(*token_itr++);
    auto const data_type = strip(to_lower(*token_itr++));
    tokens.erase(tokens.begin(), token_itr);
    auto const expression = strip(boost::join(tokens, ","));

    // get column type
    SQLTypes sql_type{kNULLT};
    double range_min{0.0}, range_max{0.0};
    if (data_type == "tinyint") {
      sql_type = kTINYINT;
      range_min = static_cast<double>(std::numeric_limits<int8_t>::min());
      range_max = static_cast<double>(std::numeric_limits<int8_t>::max());
    } else if (data_type == "smallint") {
      sql_type = kSMALLINT;
      range_min = static_cast<double>(std::numeric_limits<int16_t>::min());
      range_max = static_cast<double>(std::numeric_limits<int16_t>::max());
    } else if (data_type == "int") {
      sql_type = kINT;
      range_min = static_cast<double>(std::numeric_limits<int32_t>::min());
      range_max = static_cast<double>(std::numeric_limits<int32_t>::max());
    } else if (data_type == "bigint") {
      sql_type = kBIGINT;
      range_min = static_cast<double>(std::numeric_limits<int64_t>::min());
      range_max = static_cast<double>(std::numeric_limits<int64_t>::max());
    } else if (data_type == "float") {
      sql_type = kFLOAT;
      range_min = static_cast<double>(std::numeric_limits<float>::min());
      range_max = static_cast<double>(std::numeric_limits<float>::max());
    } else if (data_type == "double") {
      sql_type = kDOUBLE;
      range_min = static_cast<double>(std::numeric_limits<double>::min());
      range_max = static_cast<double>(std::numeric_limits<double>::max());
    } else if (data_type == "date") {
      sql_type = kDATE;
    } else if (data_type == "time") {
      sql_type = kTIME;
    } else if (data_type == "timestamp") {
      sql_type = kTIMESTAMP;
    } else if (data_type == "text") {
      sql_type = kTEXT;
    } else {
      throw std::runtime_error("Invalid metadata column data type '" + data_type +
                               "' for column '" + column_name + "'");
    }

    // set expression with force cast back to string
    parser.setExpression("str(" + expression + ")");

    // evaluate
    auto value = parser.evalAsString();

    // validate date/time/timestamp value now
    // @TODO(se) do we need to provide for non-zero dimension?
    try {
      if (sql_type == kDATE) {
        dateTimeParse<kDATE>(value, 0);
      } else if (sql_type == kTIME) {
        dateTimeParse<kTIME>(value, 0);
      } else if (sql_type == kTIMESTAMP) {
        dateTimeParse<kTIMESTAMP>(value, 0);
      }
    } catch (std::runtime_error& e) {
      throw std::runtime_error("Invalid metadata column " + to_string(sql_type) +
                               " value '" + value + "' for column '" + column_name + "'");
    }

    // validate int/float/double
    try {
      if (IS_INTEGER(sql_type) || sql_type == kFLOAT || sql_type == kDOUBLE) {
        size_t num_chars{0u};
        auto const v = static_cast<double>(std::stod(value, &num_chars));
        if (v < range_min || v > range_max) {
          throw std::out_of_range(to_string(sql_type));
        }
        if (num_chars == 0u) {
          throw std::invalid_argument("empty value");
        }
      }
    } catch (std::invalid_argument& e) {
      throw std::runtime_error("Invalid metadata column " + to_string(sql_type) +
                               " value '" + value + "' for column '" + column_name +
                               "' (" + e.what() + ")");
    } catch (std::out_of_range& e) {
      throw std::runtime_error("Out-of-range metadata column " + to_string(sql_type) +
                               " value '" + value + "' for column '" + column_name +
                               "' (" + e.what() + ")");
    }

    // build column descriptor
    ColumnDescriptor cd;
    cd.columnName = cd.sourceName = column_name;
    cd.columnType.set_type(sql_type);
    cd.columnType.set_fixed_size();
    if (sql_type == kTEXT) {
      cd.columnType.set_compression(kENCODING_DICT);
      cd.columnType.set_comp_param(0);
    }

    // add to result
    metadata_column_infos.push_back({std::move(cd), std::move(value)});
  }

  // done
  return metadata_column_infos;
}

Here is the call graph for this function:

Here is the caller graph for this function:

std::vector< std::unique_ptr< TypedImportBuffer > > import_export::setup_column_loaders	(	const TableDescriptor *	td,
		Loader *	loader
	)

Definition at line 6126 of file Importer.cpp.

References CHECK, import_export::Loader::get_column_descs(), and import_export::Loader::getStringDict().

Referenced by DBHandler::prepare_loader_generic().

                    {
  CHECK(td);
  auto col_descs = loader->get_column_descs();

  std::vector<std::unique_ptr<TypedImportBuffer>> import_buffers;
  for (auto cd : col_descs) {
    import_buffers.emplace_back(
        std::make_unique<TypedImportBuffer>(cd, loader->getStringDict(cd)));
  }

  return import_buffers;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ArrayDatum import_export::StringToArray	(	const std::string &	s,
		const SQLTypeInfo &	ti,
		const CopyParams &	copy_params
	)

Definition at line 316 of file Importer.cpp.

References append_datum(), import_export::CopyParams::array_begin, import_export::CopyParams::array_delim, import_export::CopyParams::array_end, CHECK, checked_malloc(), SQLTypeInfo::get_elem_type(), SQLTypeInfo::get_size(), is_null(), SQLTypeInfo::is_number(), SQLTypeInfo::is_string(), SQLTypeInfo::is_time(), LOG, import_export::CopyParams::null_str, NullDatum(), StringToDatum(), trim_space(), and logger::WARNING.

Referenced by import_export::TypedImportBuffer::add_value(), and import_export::TypedImportBuffer::addDefaultValues().

                                                        {
  SQLTypeInfo elem_ti = ti.get_elem_type();
  if (s == copy_params.null_str || s == "NULL" || s.empty()) {
    return ArrayDatum(0, NULL, true);
  }
  if (s[0] != copy_params.array_begin || s[s.size() - 1] != copy_params.array_end) {
    LOG(WARNING) << "Malformed array: " << s;
    return ArrayDatum(0, NULL, true);
  }
  std::vector<std::string> elem_strs;
  size_t last = 1;
  for (size_t i = s.find(copy_params.array_delim, 1); i != std::string::npos;
       i = s.find(copy_params.array_delim, last)) {
    elem_strs.push_back(s.substr(last, i - last));
    last = i + 1;
  }
  if (last + 1 <= s.size()) {
    elem_strs.push_back(s.substr(last, s.size() - 1 - last));
  }
  if (elem_strs.size() == 1) {
    auto str = elem_strs.front();
    auto str_trimmed = trim_space(str.c_str(), str.length());
    if (str_trimmed == "") {
      elem_strs.clear();  // Empty array
    }
  }
  if (!elem_ti.is_string()) {
    size_t len = elem_strs.size() * elem_ti.get_size();
    std::unique_ptr<int8_t, FreeDeleter> buf(
        reinterpret_cast<int8_t*>(checked_malloc(len)));
    int8_t* p = buf.get();
    for (auto& es : elem_strs) {
      auto e = trim_space(es.c_str(), es.length());
      bool is_null = (e == copy_params.null_str) || e == "NULL";
      if (!elem_ti.is_string() && e == "") {
        is_null = true;
      }
      if (elem_ti.is_number() || elem_ti.is_time()) {
        if (!isdigit(e[0]) && e[0] != '-') {
          is_null = true;
        }
      }
      Datum d = is_null ? NullDatum(elem_ti) : StringToDatum(e, elem_ti);
      p = append_datum(p, d, elem_ti);
      CHECK(p);
    }
    return ArrayDatum(len, buf.release(), false);
  }
  // must not be called for array of strings
  CHECK(false);
  return ArrayDatum(0, NULL, true);
}

Here is the call graph for this function:

Here is the caller graph for this function:

ArrayDatum import_export::TDatumToArrayDatum	(	const TDatum &	datum,
		const SQLTypeInfo &	ti
	)

Definition at line 469 of file Importer.cpp.

References append_datum(), CHECK, checked_malloc(), SQLTypeInfo::get_elem_type(), SQLTypeInfo::get_size(), SQLTypeInfo::is_string(), NullArray(), and TDatumToDatum().

Referenced by import_export::TypedImportBuffer::add_value().

                                                                          {
  SQLTypeInfo elem_ti = ti.get_elem_type();

  CHECK(!elem_ti.is_string());

  if (datum.is_null) {
    return NullArray(ti);
  }

  size_t len = datum.val.arr_val.size() * elem_ti.get_size();
  int8_t* buf = (int8_t*)checked_malloc(len);
  int8_t* p = buf;
  for (auto& e : datum.val.arr_val) {
    p = append_datum(p, TDatumToDatum(e, elem_ti), elem_ti);
    CHECK(p);
  }

  return ArrayDatum(len, buf, false);
}

Here is the call graph for this function:

Here is the caller graph for this function:

Datum import_export::TDatumToDatum	(	const TDatum &	datum,
		SQLTypeInfo &	ti
	)

Definition at line 422 of file Importer.cpp.

References Datum::bigintval, Datum::boolval, decimal_to_int_type(), Datum::doubleval, Datum::floatval, SQLTypeInfo::get_type(), inline_fixed_encoding_null_val(), Datum::intval, SQLTypeInfo::is_decimal(), kBIGINT, kBOOLEAN, kDATE, kDOUBLE, kFLOAT, kINT, kLINESTRING, kMULTILINESTRING, kMULTIPOINT, kMULTIPOLYGON, kPOINT, kPOLYGON, kSMALLINT, kTIME, kTIMESTAMP, kTINYINT, NULL_DOUBLE, NULL_FLOAT, Datum::smallintval, Datum::tinyintval, and run_benchmark_import::type.

Referenced by TDatumToArrayDatum().

                                                          {
  Datum d;
  const auto type = ti.is_decimal() ? decimal_to_int_type(ti) : ti.get_type();
  switch (type) {
    case kBOOLEAN:
      d.boolval = datum.is_null ? inline_fixed_encoding_null_val(ti) : datum.val.int_val;
      break;
    case kBIGINT:
      d.bigintval =
          datum.is_null ? inline_fixed_encoding_null_val(ti) : datum.val.int_val;
      break;
    case kINT:
      d.intval = datum.is_null ? inline_fixed_encoding_null_val(ti) : datum.val.int_val;
      break;
    case kSMALLINT:
      d.smallintval =
          datum.is_null ? inline_fixed_encoding_null_val(ti) : datum.val.int_val;
      break;
    case kTINYINT:
      d.tinyintval =
          datum.is_null ? inline_fixed_encoding_null_val(ti) : datum.val.int_val;
      break;
    case kFLOAT:
      d.floatval = datum.is_null ? NULL_FLOAT : datum.val.real_val;
      break;
    case kDOUBLE:
      d.doubleval = datum.is_null ? NULL_DOUBLE : datum.val.real_val;
      break;
    case kTIME:
    case kTIMESTAMP:
    case kDATE:
      d.bigintval =
          datum.is_null ? inline_fixed_encoding_null_val(ti) : datum.val.int_val;
      break;
    case kPOINT:
    case kMULTIPOINT:
    case kLINESTRING:
    case kMULTILINESTRING:
    case kPOLYGON:
    case kMULTIPOLYGON:
      throw std::runtime_error("Internal error: geometry type in TDatumToDatum.");
    default:
      throw std::runtime_error("Internal error: invalid type in TDatumToDatum.");
  }
  return d;
}

Here is the call graph for this function:

Here is the caller graph for this function:

static const std::string import_export::trim_space ( const char *  field, const size_t  len  )

static

Definition at line 247 of file Importer.cpp.

Referenced by import_export::delimited_parser::get_row(), and StringToArray().

                                                                       {
  size_t i = 0;
  size_t j = len;
  while (i < j && (field[i] == ' ' || field[i] == '\r')) {
    i++;
  }
  while (i < j && (field[j - 1] == ' ' || field[j - 1] == '\r')) {
    j--;
  }
  return std::string(field + i, j - i);
}

Here is the caller graph for this function:

template<class T >

bool import_export::try_cast

(

const std::string &

str

)

Definition at line 3207 of file Importer.cpp.

References heavydb.dtypes::T.

                                    {
  try {
    boost::lexical_cast<T>(str);
  } catch (const boost::bad_lexical_cast& e) {
    return false;
  }
  return true;
}

Variable Documentation

std::map<std::string, ImportStatus> import_export::import_status_map

static

Definition at line 167 of file Importer.cpp.

Referenced by import_export::Importer::get_import_status(), and import_export::Importer::set_import_status().

constexpr size_t import_export::kImportFileBufferSize = (1 << 23)

static

Definition at line 34 of file CopyParams.h.

const size_t import_export::kImportRowLimit = 10000

static

Definition at line 170 of file Importer.cpp.

Referenced by import_export::DataStreamSink::import_compressed(), and import_export::Detector::importDelimited().

constexpr size_t import_export::max_import_buffer_resize_byte_size = 1024 * 1024 * 1024

static

Definition at line 37 of file CopyParams.h.

constexpr bool import_export::PROMOTE_LINESTRING_TO_MULTILINESTRING = false

static

Definition at line 163 of file Importer.cpp.

Referenced by import_export::Detector::detect_sqltype(), and import_export::Importer::gdalToColumnDescriptorsGeo().

constexpr bool import_export::PROMOTE_POINT_TO_MULTIPOINT = false

static

Definition at line 162 of file Importer.cpp.

Referenced by import_export::Detector::detect_sqltype(), and import_export::Importer::gdalToColumnDescriptorsGeo().

constexpr bool import_export::PROMOTE_POLYGON_TO_MULTIPOLYGON = true

static

Definition at line 164 of file Importer.cpp.

Referenced by import_export::Detector::detect_sqltype(), and import_export::Importer::gdalToColumnDescriptorsGeo().

heavyai::shared_mutex import_export::status_mutex

static

Definition at line 166 of file Importer.cpp.

Referenced by import_export::Importer::get_import_status(), and import_export::Importer::set_import_status().