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)

ParseBufferResult	parse_buffer (ParseBufferRequest &request, bool convert_data_blocks, bool columns_are_pre_filtered)

void	parse_and_validate_expected_column_count (const std::string &row, const import_export::CopyParams &copy_params, size_t num_cols, int point_cols, const std::string &file_name)

bool	skip_column_import (ParseBufferRequest &request, int column_idx)

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 195 of file CsvFileBufferParser.cpp.

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 58 of file CsvFileBufferParser.cpp.

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 94 of file CsvFileBufferParser.cpp.

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:

void foreign_storage::csv_file_buffer_parser::parse_and_validate_expected_column_count	(	const std::string &	row,
		const import_export::CopyParams &	copy_params,
		size_t	num_cols,
		int	point_cols,
		const std::string &	file_name
	)

Takes a single row and verifies number of columns is valid for num_cols and point_cols (number of point columns)

Definition at line 416 of file CsvFileBufferParser.cpp.

References import_export::delimited_parser::get_row(), and validate_expected_column_count().

                                 {
   bool is_array = false;
   bool try_single_thread = false;
   std::vector<std::unique_ptr<char[]>> tmp_buffers;
   std::vector<std::string_view> fields;
   // parse columns in row into fields (other return values are intentionally ignored)
   import_export::delimited_parser::get_row(row.c_str(),
                                            row.c_str() + row.size(),
                                            row.c_str() + row.size(),
                                            copy_params,
                                            &is_array,
                                            fields,
                                            tmp_buffers,
                                            try_single_thread,
                                            false  // Don't filter empty lines
   );
   // Check we have right number of columns
   validate_expected_column_count(fields, num_cols, point_cols, file_name);
 }

Here is the call graph for this function:

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

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.

Parameters

convert_data_blocks	- convert import buffers to data blocks
columns_are_pre_filtered	- CSV buffer passed into parse_buffer only has the necessary columns that are being requested, not all columns.

Definition at line 287 of file CsvFileBufferParser.cpp.

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;
   if (columns_are_pre_filtered) {
     for (size_t col_idx = 0; col_idx < request.getColumns().size(); ++col_idx) {
       if (skip_column_import(request, col_idx)) {
         --num_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
     const char* line_start = p;
     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,
                                                  !columns_are_pre_filtered);
 
     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 column_is_present =
             !(skip_column_import(request, col_idx) && columns_are_pre_filtered);
         CHECK(row.size() > import_idx || !column_is_present);
         bool is_null =
             column_is_present
                 ? is_null_datum(row[import_idx], cd, request.copy_params.null_str)
                 : true;
 
         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])) {
               if (!columns_are_pre_filtered)
                 ++import_idx;
             }
             if (!columns_are_pre_filtered)
               ++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);
           }
           if (column_is_present) {
             ++import_idx;
           }
           ++col_idx;
         }
       }
     } catch (const std::exception& e) {
       throw ForeignStorageException("Parsing failure \"" + std::string(e.what()) +
                                     "\" in row \"" + std::string(line_start, p) +
                                     "\" in file \"" + file_path + "\"");
     }
   }
   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 111 of file CsvFileBufferParser.cpp.

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 25 of file CsvFileBufferParser.cpp.

References i, 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 64 of file CsvFileBufferParser.cpp.

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
	)

inline

Definition at line 92 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 48 of file CsvFileBufferParser.cpp.

References foreign_storage::throw_number_of_columns_mismatch_error().

Referenced by parse_and_validate_expected_column_count(), and 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