Classes
struct	ParseBufferRequest

struct	ParseBufferResult

Functions
void	set_array_flags_and_geo_columns_count (std::unique_ptr< bool[]> &array_flags, int &phys_cols, int &point_cols, const std::list< const ColumnDescriptor * > &columns)

void	validate_expected_column_count (std::vector< std::string_view > &row, size_t num_cols, int point_cols, const std::string &file_name)

bool	is_coordinate_scalar (const std::string_view datum)

bool	set_coordinates_from_separate_lon_lat_columns (const std::string_view lon_str, const std::string_view lat_str, std::vector< double > &coords, const bool is_lon_lat_order)

bool	is_null_datum (const std::string_view datum, const ColumnDescriptor *column, const std::string &null_indicator)

void	process_geo_column (std::vector< std::unique_ptr< import_export::TypedImportBuffer >> &import_buffers, size_t &col_idx, const import_export::CopyParams &copy_params, std::list< const ColumnDescriptor * >::iterator &cd_it, std::vector< std::string_view > &row, size_t &import_idx, bool is_null, size_t first_row_index, size_t row_index_plus_one, std::shared_ptr< Catalog_Namespace::Catalog > catalog)

std::map< int, DataBlockPtr >	convert_import_buffers_to_data_blocks (const std::vector< std::unique_ptr< import_export::TypedImportBuffer >> &import_buffers)

bool	skip_column_import (ParseBufferRequest &request, int column_idx)

ParseBufferResult	parse_buffer (ParseBufferRequest &request, bool convert_data_blocks)

Variables
static constexpr bool	PROMOTE_POLYGON_TO_MULTIPOLYGON = true

Function Documentation

std::map<int, DataBlockPtr> foreign_storage::csv_file_buffer_parser::convert_import_buffers_to_data_blocks ( const std::vector< std::unique_ptr< import_export::TypedImportBuffer >> & import_buffers )

Definition at line 199 of file CsvFileBufferParser.h.

References DataBlockPtr::arraysPtr, CHECK, CHECK_EQ, IS_STRING, kARRAY, kBOOLEAN, kENCODING_DICT, kENCODING_NONE, DataBlockPtr::numbersPtr, run_benchmark_import::result, and DataBlockPtr::stringsPtr.

Referenced by parse_buffer().

                         {
   std::map<int, DataBlockPtr> result;
   std::vector<std::pair<const size_t, std::future<int8_t*>>>
       encoded_data_block_ptrs_futures;
   // make all async calls to string dictionary here and then continue execution
   for (const auto& import_buffer : import_buffers) {
     if (import_buffer == nullptr)
       continue;
     DataBlockPtr p;
     if (import_buffer->getTypeInfo().is_number() ||
         import_buffer->getTypeInfo().is_time() ||
         import_buffer->getTypeInfo().get_type() == kBOOLEAN) {
       p.numbersPtr = import_buffer->getAsBytes();
     } else if (import_buffer->getTypeInfo().is_string()) {
       auto string_payload_ptr = import_buffer->getStringBuffer();
       if (import_buffer->getTypeInfo().get_compression() == kENCODING_NONE) {
         p.stringsPtr = string_payload_ptr;
       } else {
         CHECK_EQ(kENCODING_DICT, import_buffer->getTypeInfo().get_compression());
         p.numbersPtr = nullptr;
 
         auto column_id = import_buffer->getColumnDesc()->columnId;
         encoded_data_block_ptrs_futures.emplace_back(std::make_pair(
             column_id,
             std::async(std::launch::async, [&import_buffer, string_payload_ptr] {
               import_buffer->addDictEncodedString(*string_payload_ptr);
               return import_buffer->getStringDictBuffer();
             })));
       }
     } else if (import_buffer->getTypeInfo().is_geometry()) {
       auto geo_payload_ptr = import_buffer->getGeoStringBuffer();
       p.stringsPtr = geo_payload_ptr;
     } else {
       CHECK(import_buffer->getTypeInfo().get_type() == kARRAY);
       if (IS_STRING(import_buffer->getTypeInfo().get_subtype())) {
         CHECK(import_buffer->getTypeInfo().get_compression() == kENCODING_DICT);
         import_buffer->addDictEncodedStringArray(*import_buffer->getStringArrayBuffer());
         p.arraysPtr = import_buffer->getStringArrayDictBuffer();
       } else {
         p.arraysPtr = import_buffer->getArrayBuffer();
       }
     }
     result[import_buffer->getColumnDesc()->columnId] = p;
   }
 
   // wait for the async requests we made for string dictionary
   for (auto& encoded_ptr_future : encoded_data_block_ptrs_futures) {
     result[encoded_ptr_future.first].numbersPtr = encoded_ptr_future.second.get();
   }
   return result;
 }

Here is the caller graph for this function:

bool foreign_storage::csv_file_buffer_parser::is_coordinate_scalar ( const std::string_view datum )

Definition at line 62 of file CsvFileBufferParser.h.

Referenced by parse_buffer(), process_geo_column(), and set_coordinates_from_separate_lon_lat_columns().

                                                       {
   // field looks like a scalar numeric value (and not a hex blob)
   return datum.size() > 0 && (datum[0] == '.' || isdigit(datum[0]) || datum[0] == '-') &&
          datum.find_first_of("ABCDEFabcdef") == std::string_view::npos;
 }

Here is the caller graph for this function:

bool foreign_storage::csv_file_buffer_parser::is_null_datum	(	const std::string_view	datum,
		const ColumnDescriptor *	column,
		const std::string &	null_indicator
	)

Definition at line 98 of file CsvFileBufferParser.h.

References ColumnDescriptor::columnName, ColumnDescriptor::columnType, SQLTypeInfo::get_notnull(), is_null(), and SQLTypeInfo::is_string().

Referenced by parse_buffer().

                                                     {
   bool is_null = (datum == null_indicator);
 
   // Treating empty as NULL
   if (!column->columnType.is_string() && datum.empty()) {
     is_null = true;
   }
 
   if (is_null && column->columnType.get_notnull()) {
     throw std::runtime_error("NULL value provided for column (" + column->columnName +
                              ") with NOT NULL constraint.");
   }
   return is_null;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

ParseBufferResult foreign_storage::csv_file_buffer_parser::parse_buffer	(	ParseBufferRequest &	request,
		bool	convert_data_blocks
	)

Parses a given CSV file buffer and returns data blocks for each column in the file along with metadata related to rows and row offsets within the buffer.

Definition at line 343 of file CsvFileBufferParser.h.

Referenced by foreign_storage::parse_file_regions(), and foreign_storage::scan_metadata().

                                                                                       {
   CHECK(request.buffer);
   size_t begin = import_export::delimited_parser::find_beginning(
       request.buffer.get(), request.begin_pos, request.end_pos, request.copy_params);
   const char* thread_buf = request.buffer.get() + request.begin_pos + begin;
   const char* thread_buf_end = request.buffer.get() + request.end_pos;
   const char* buf_end = request.buffer.get() + request.buffer_size;
 
   std::vector<std::string_view> row;
   size_t row_index_plus_one = 0;
   const char* p = thread_buf;
   bool try_single_thread = false;
   int phys_cols = 0;
   int point_cols = 0;
   std::unique_ptr<bool[]> array_flags;
 
   set_array_flags_and_geo_columns_count(
       array_flags, phys_cols, point_cols, request.getColumns());
   auto num_cols = request.getColumns().size() - phys_cols;
 
   size_t row_count = 0;
   size_t remaining_row_count = request.process_row_count;
   std::vector<size_t> row_offsets{};
   row_offsets.emplace_back(request.file_offset + (p - request.buffer.get()));
 
   std::string file_path = request.getFilePath();
   for (; p < thread_buf_end && remaining_row_count > 0; p++, remaining_row_count--) {
     row.clear();
     row_count++;
     std::vector<std::unique_ptr<char[]>>
         tmp_buffers;  // holds string w/ removed escape chars, etc
 
     p = import_export::delimited_parser::get_row(p,
                                                  thread_buf_end,
                                                  buf_end,
                                                  request.copy_params,
                                                  array_flags.get(),
                                                  row,
                                                  tmp_buffers,
                                                  try_single_thread);
 
     row_index_plus_one++;
     validate_expected_column_count(row, num_cols, point_cols, file_path);
 
     size_t import_idx = 0;
     size_t col_idx = 0;
     try {
       auto columns = request.getColumns();
       for (auto cd_it = columns.begin(); cd_it != columns.end(); cd_it++) {
         auto cd = *cd_it;
         const auto& col_ti = cd->columnType;
         bool is_null = is_null_datum(row[import_idx], cd, request.copy_params.null_str);
 
         if (col_ti.is_geometry()) {
           if (!skip_column_import(request, col_idx)) {
             process_geo_column(request.import_buffers,
                                col_idx,
                                request.copy_params,
                                cd_it,
                                row,
                                import_idx,
                                is_null,
                                request.first_row_index,
                                row_index_plus_one,
                                request.getCatalog());
           } else {
             // update import/col idx according to types
             if (!is_null && cd->columnType == kPOINT &&
                 is_coordinate_scalar(row[import_idx])) {
               ++import_idx;
             }
             ++import_idx;
             ++col_idx;
             col_idx += col_ti.get_physical_cols();
           }
           // skip remaining physical columns
           for (int i = 0; i < cd->columnType.get_physical_cols(); ++i) {
             ++cd_it;
           }
         } else {
           if (!skip_column_import(request, col_idx)) {
             request.import_buffers[col_idx]->add_value(
                 cd, row[import_idx], is_null, request.copy_params);
           }
           ++import_idx;
           ++col_idx;
         }
       }
     } catch (const std::exception& e) {
       // TODO: Appropriate error handling for FSI
       for (size_t col_idx_to_pop = 0; col_idx_to_pop < col_idx; ++col_idx_to_pop) {
         request.import_buffers[col_idx_to_pop]->pop_value();
       }
       LOG(ERROR) << "Input exception thrown: " << e.what()
                  << ". Row discarded. Data: " << shared::printContainer(row);
     }
   }
   row_offsets.emplace_back(request.file_offset + (p - request.buffer.get()));
 
   ParseBufferResult result{};
   result.row_offsets = row_offsets;
   result.row_count = row_count;
   if (convert_data_blocks) {
     result.column_id_to_data_blocks_map =
         convert_import_buffers_to_data_blocks(request.import_buffers);
   }
   return result;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void foreign_storage::csv_file_buffer_parser::process_geo_column	(	std::vector< std::unique_ptr< import_export::TypedImportBuffer >> &	import_buffers,
		size_t &	col_idx,
		const import_export::CopyParams &	copy_params,
		std::list< const ColumnDescriptor * >::iterator &	cd_it,
		std::vector< std::string_view > &	row,
		size_t &	import_idx,
		bool	is_null,
		size_t	first_row_index,
		size_t	row_index_plus_one,
		std::shared_ptr< Catalog_Namespace::Catalog >	catalog
	)

Definition at line 115 of file CsvFileBufferParser.h.

References CHECK, Geospatial::GeoTypesFactory::getGeoColumns(), Geospatial::GeoTypesFactory::getNullGeoColumns(), is_coordinate_scalar(), IS_GEO, kMULTIPOLYGON, kPOINT, kPOLYGON, import_export::CopyParams::lonlat, import_export::CopyParams::null_str, PROMOTE_POLYGON_TO_MULTIPOLYGON, set_coordinates_from_separate_lon_lat_columns(), import_export::Importer::set_geo_physical_import_buffer(), and to_string().

Referenced by parse_buffer().

                                                      {
   auto cd = *cd_it;
   auto col_ti = cd->columnType;
   SQLTypes col_type = col_ti.get_type();
   CHECK(IS_GEO(col_type));
 
   // store null string in the base column
   import_buffers[col_idx]->add_value(cd, copy_params.null_str, true, copy_params);
 
   auto const& geo_string = row[import_idx];
   ++import_idx;
   ++col_idx;
 
   std::vector<double> coords;
   std::vector<double> bounds;
   std::vector<int> ring_sizes;
   std::vector<int> poly_rings;
   int render_group = 0;
 
   if (!is_null && col_type == kPOINT && is_coordinate_scalar(geo_string)) {
     if (!set_coordinates_from_separate_lon_lat_columns(
             geo_string, row[import_idx], coords, copy_params.lonlat)) {
       throw std::runtime_error("Cannot read lon/lat to insert into POINT column " +
                                cd->columnName);
     }
     ++import_idx;
   } else {
     SQLTypeInfo import_ti{col_ti};
     if (is_null) {
       Geospatial::GeoTypesFactory::getNullGeoColumns(import_ti,
                                                      coords,
                                                      bounds,
                                                      ring_sizes,
                                                      poly_rings,
                                                      PROMOTE_POLYGON_TO_MULTIPOLYGON);
     } else {
       // extract geometry directly from WKT
       if (!Geospatial::GeoTypesFactory::getGeoColumns(std::string(geo_string),
                                                       import_ti,
                                                       coords,
                                                       bounds,
                                                       ring_sizes,
                                                       poly_rings,
                                                       PROMOTE_POLYGON_TO_MULTIPOLYGON)) {
         std::string msg = "Failed to extract valid geometry from row " +
                           std::to_string(first_row_index + row_index_plus_one) +
                           " for column " + cd->columnName;
         throw std::runtime_error(msg);
       }
 
       // validate types
       if (col_type != import_ti.get_type()) {
         if (!PROMOTE_POLYGON_TO_MULTIPOLYGON ||
             !(import_ti.get_type() == SQLTypes::kPOLYGON &&
               col_type == SQLTypes::kMULTIPOLYGON)) {
           throw std::runtime_error("Imported geometry doesn't match the type of column " +
                                    cd->columnName);
         }
       }
     }
   }
 
   // import extracted geo
   import_export::Importer::set_geo_physical_import_buffer(*catalog,
                                                           cd,
                                                           import_buffers,
                                                           col_idx,
                                                           coords,
                                                           bounds,
                                                           ring_sizes,
                                                           poly_rings,
                                                           render_group);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void foreign_storage::csv_file_buffer_parser::set_array_flags_and_geo_columns_count	(	std::unique_ptr< bool[]> &	array_flags,
		int &	phys_cols,
		int &	point_cols,
		const std::list< const ColumnDescriptor * > &	columns
	)

Definition at line 29 of file CsvFileBufferParser.h.

References kARRAY, and kPOINT.

Referenced by parse_buffer().

                                                      {
   array_flags = std::unique_ptr<bool[]>(new bool[columns.size()]);
   size_t i = 0;
   for (const auto cd : columns) {
     const auto& col_ti = cd->columnType;
     phys_cols += col_ti.get_physical_cols();
     if (cd->columnType.get_type() == kPOINT) {
       point_cols++;
     }
 
     if (cd->columnType.get_type() == kARRAY) {
       array_flags.get()[i] = true;
     } else {
       array_flags.get()[i] = false;
     }
     i++;
   }
 }

Here is the caller graph for this function:

bool foreign_storage::csv_file_buffer_parser::set_coordinates_from_separate_lon_lat_columns	(	const std::string_view	lon_str,
		const std::string_view	lat_str,
		std::vector< double > &	coords,
		const bool	is_lon_lat_order
	)

Definition at line 68 of file CsvFileBufferParser.h.

References is_coordinate_scalar(), and gpu_enabled::swap().

Referenced by process_geo_column().

                                                                                 {
   double lon = std::atof(std::string(lon_str).c_str());
   double lat = NAN;
 
   if (is_coordinate_scalar(lat_str)) {
     lat = std::atof(std::string(lat_str).c_str());
   }
 
   // Swap coordinates if this table uses a reverse order: lat/lon
   if (!is_lon_lat_order) {
     std::swap(lat, lon);
   }
 
   // TODO: should check if POINT 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");
   // }
 
   if (std::isinf(lat) || std::isnan(lat) || std::isinf(lon) || std::isnan(lon)) {
     return false;
   }
   coords.push_back(lon);
   coords.push_back(lat);
   return true;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool foreign_storage::csv_file_buffer_parser::skip_column_import	(	ParseBufferRequest &	request,
		int	column_idx
	)

Definition at line 335 of file CsvFileBufferParser.h.

References foreign_storage::csv_file_buffer_parser::ParseBufferRequest::import_buffers.

Referenced by parse_buffer().

                                                                      {
   return request.import_buffers[column_idx] == nullptr;
 }

Here is the caller graph for this function:

void foreign_storage::csv_file_buffer_parser::validate_expected_column_count	(	std::vector< std::string_view > &	row,
		size_t	num_cols,
		int	point_cols,
		const std::string &	file_name
	)

Definition at line 52 of file CsvFileBufferParser.h.

References foreign_storage::throw_number_of_columns_mismatch_error().

Referenced by parse_buffer().

                                                                 {
   // Each POINT could consume two separate coords instead of a single WKT
   if (row.size() < num_cols || (num_cols + point_cols) < row.size()) {
     throw_number_of_columns_mismatch_error(num_cols, row.size(), file_name);
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

Variable Documentation

constexpr bool foreign_storage::csv_file_buffer_parser::PROMOTE_POLYGON_TO_MULTIPOLYGON = true

static

Definition at line 27 of file CsvFileBufferParser.h.

Referenced by process_geo_column().

Classes

Functions

Variables

Function Documentation

Variable Documentation