#include <TextFileBufferParser.h>

Inheritance diagram for foreign_storage::TextFileBufferParser:

Public Member Functions
virtual ParseBufferResult	parseBuffer (ParseBufferRequest &request, bool convert_data_blocks, bool columns_are_pre_filtered=false, bool skip_dict_encoding=false) const =0

virtual import_export::CopyParams	validateAndGetCopyParams (const ForeignTable *foreign_table) const =0

virtual size_t	findRowEndPosition (size_t &alloc_size, std::unique_ptr< char[]> &buffer, size_t &buffer_size, const import_export::CopyParams &copy_params, const size_t buffer_first_row_index, unsigned int &num_rows_in_buffer, FileReader *file_reader) const =0

virtual void	validateFiles (const FileReader file_reader, const ForeignTable foreign_table) const =0

Static Public Member Functions
static std::map< int, DataBlockPtr >	convertImportBuffersToDataBlocks (const std::vector< std::unique_ptr< import_export::TypedImportBuffer >> &import_buffers, const bool skip_dict_encoding=false)

static bool	isCoordinateScalar (const std::string_view datum)

static void	processGeoColumn (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)

static void	fillRejectedRowWithInvalidData (const std::list< const ColumnDescriptor * > &columns, std::list< const ColumnDescriptor * >::iterator &cd_it, const size_t col_idx, ParseBufferRequest &request)

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

Static Public Attributes
static const std::string	BUFFER_SIZE_KEY = "BUFFER_SIZE"

Static Private Member Functions
static void	processInvalidGeoColumn (std::vector< std::unique_ptr< import_export::TypedImportBuffer >> &import_buffers, size_t &col_idx, const import_export::CopyParams &copy_params, const ColumnDescriptor *cd, std::shared_ptr< Catalog_Namespace::Catalog > catalog)

Detailed Description

Definition at line 94 of file TextFileBufferParser.h.

Member Function Documentation

std::map< int, DataBlockPtr > foreign_storage::TextFileBufferParser::convertImportBuffersToDataBlocks	(	const std::vector< std::unique_ptr< import_export::TypedImportBuffer >> &	import_buffers,
		const bool	skip_dict_encoding = `false`
	)

static

Definition at line 59 of file TextFileBufferParser.cpp.

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

Referenced by foreign_storage::CsvFileBufferParser::parseBuffer(), foreign_storage::RegexFileBufferParser::parseBuffer(), and foreign_storage::InternalSystemDataWrapper::populateChunkBuffers().

                                    {
   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;
 
         if (!skip_dict_encoding) {
           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;
   }
 
   if (!skip_dict_encoding) {
     // wait for the async requests we made for string dictionary
     for (auto& encoded_ptr_future : encoded_data_block_ptrs_futures) {
       encoded_ptr_future.second.wait();
     }
     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 call graph for this function:

Here is the caller graph for this function:

void foreign_storage::TextFileBufferParser::fillRejectedRowWithInvalidData	(	const std::list< const ColumnDescriptor * > &	columns,
		std::list< const ColumnDescriptor * >::iterator &	cd_it,
		const size_t	col_idx,
		ParseBufferRequest &	request
	)

static

Fill the current row of the request with invalid (null) data as row will be marked as rejected

Definition at line 171 of file TextFileBufferParser.cpp.

References foreign_storage::ParseBufferRequest::copy_params, foreign_storage::ParseBufferRequest::getCatalog(), foreign_storage::ParseBufferRequest::import_buffers, and processInvalidGeoColumn().

Referenced by foreign_storage::CsvFileBufferParser::parseBuffer(), and foreign_storage::RegexFileBufferParser::parseBuffer().

                                  {
   size_t col_idx = starting_col_idx;
 
   for (; cd_it != columns.end(); cd_it++) {
     auto cd = *cd_it;
     const auto& col_ti = cd->columnType;
     if (col_ti.is_geometry()) {
       if (request.import_buffers[col_idx] != nullptr) {
         processInvalidGeoColumn(request.import_buffers,
                                 col_idx,
                                 request.copy_params,
                                 cd,
                                 request.getCatalog());
       } else {
         ++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 (request.import_buffers[col_idx] != nullptr) {
         request.import_buffers[col_idx]->add_value(cd,
                                                    {},
                                                    true,
                                                    request.copy_params,
                                                    /*check_not_null=*/false);
       }
       ++col_idx;
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

virtual size_t foreign_storage::TextFileBufferParser::findRowEndPosition	(	size_t &	alloc_size,
		std::unique_ptr< char[]> &	buffer,
		size_t &	buffer_size,
		const import_export::CopyParams &	copy_params,
		const size_t	buffer_first_row_index,
		unsigned int &	num_rows_in_buffer,
		FileReader *	file_reader
	)		const

pure virtual

Finds and returns the offset of the end of the last row in the given buffer. If the buffer does not contain at least one row, the buffer is extended with more content from the file until a row is read. An exception is thrown if the buffer is extended to a maximum threshold and at least one row has still not been read.

Implemented in foreign_storage::RegexFileBufferParser, and foreign_storage::CsvFileBufferParser.

Referenced by foreign_storage::dispatch_scan_requests().

Here is the caller graph for this function:

bool foreign_storage::TextFileBufferParser::isCoordinateScalar ( const std::string_view datum )

static

Definition at line 121 of file TextFileBufferParser.cpp.

Referenced by foreign_storage::CsvFileBufferParser::parseBuffer(), processGeoColumn(), and foreign_storage::anonymous_namespace{TextFileBufferParser.cpp}::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::TextFileBufferParser::isNullDatum	(	const std::string_view	datum,
		const ColumnDescriptor *	column,
		const std::string &	null_indicator
	)

static

Definition at line 332 of file TextFileBufferParser.cpp.

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

Referenced by foreign_storage::CsvFileBufferParser::parseBuffer(), and foreign_storage::RegexFileBufferParser::parseBuffer().

                                                                         {
   bool is_null = ImportHelpers::is_null_datum(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:

virtual ParseBufferResult foreign_storage::TextFileBufferParser::parseBuffer	(	ParseBufferRequest &	request,
		bool	convert_data_blocks,
		bool	columns_are_pre_filtered = `false`,
		bool	skip_dict_encoding = `false`
	)		const

pure virtual

Parses a given 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	- file buffer passed into parse_buffer only has the necessary columns that are being requested, not all columns.
skip_dict_encoding	- skip dictionary encoding for encoded strings; the encoding will be required to happen later in processing

Implemented in foreign_storage::RegexFileBufferParser, and foreign_storage::CsvFileBufferParser.

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

Here is the caller graph for this function:

void foreign_storage::TextFileBufferParser::processGeoColumn	(	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
	)

static

Definition at line 244 of file TextFileBufferParser.cpp.

References CHECK, geo_promoted_type_match(), import_export::CopyParams::geo_validate_geometry, Geospatial::GeoTypesFactory::getGeoColumns(), Geospatial::GeoTypesFactory::getNullGeoColumns(), IS_GEO, isCoordinateScalar(), kPOINT, import_export::CopyParams::lonlat, import_export::CopyParams::null_str, foreign_storage::anonymous_namespace{TextFileBufferParser.cpp}::set_coordinates_from_separate_lon_lat_columns(), import_export::Importer::set_geo_physical_import_buffer(), import_export::CopyParams::source_srid, and to_string().

Referenced by foreign_storage::CsvFileBufferParser::parseBuffer(), and foreign_storage::RegexFileBufferParser::parseBuffer().

                                                      {
   auto cd = *cd_it;
   auto col_ti = cd->columnType;
   SQLTypes col_type = col_ti.get_type();
   CHECK(IS_GEO(col_type));
 
   auto starting_col_idx = col_idx;
 
   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;
 
   // prepare to transform from another SRID
   SQLTypeInfo import_ti{col_ti};
   if (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 && col_type == kPOINT && isCoordinateScalar(geo_string)) {
     if (!set_coordinates_from_separate_lon_lat_columns(
             geo_string, row[import_idx], import_ti, coords, copy_params.lonlat)) {
       throw std::runtime_error("Cannot read lon/lat to insert into POINT column " +
                                cd->columnName);
     }
     ++import_idx;
   } else {
     if (is_null || geo_string.empty() || geo_string == "NULL") {
       Geospatial::GeoTypesFactory::getNullGeoColumns(
           import_ti, coords, bounds, ring_sizes, poly_rings);
       is_null = true;
     } 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 + 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);
       }
     }
   }
 
   // allowed to be null?
   if (is_null && col_ti.get_notnull()) {
     throw std::runtime_error("NULL value provided for column (" + cd->columnName +
                              ") with NOT NULL constraint.");
   }
 
   // import extracted geo
   import_export::Importer::set_geo_physical_import_buffer(
       *catalog, cd, import_buffers, col_idx, coords, bounds, ring_sizes, poly_rings);
 
   // store null string in the base column
   import_buffers[starting_col_idx]->add_value(
       cd, copy_params.null_str, true, copy_params);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void foreign_storage::TextFileBufferParser::processInvalidGeoColumn	(	std::vector< std::unique_ptr< import_export::TypedImportBuffer >> &	import_buffers,
		size_t &	col_idx,
		const import_export::CopyParams &	copy_params,
		const ColumnDescriptor *	cd,
		std::shared_ptr< Catalog_Namespace::Catalog >	catalog
	)

staticprivate

Definition at line 209 of file TextFileBufferParser.cpp.

References CHECK, ColumnDescriptor::columnType, SQLTypeInfo::get_type(), Geospatial::GeoTypesFactory::getNullGeoColumns(), IS_GEO, import_export::CopyParams::null_str, and import_export::Importer::set_geo_physical_import_buffer().

Referenced by fillRejectedRowWithInvalidData().

                                                      {
   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);
   ++col_idx;
 
   std::vector<double> coords;
   std::vector<double> bounds;
   std::vector<int> ring_sizes;
   std::vector<int> poly_rings;
 
   SQLTypeInfo import_ti{col_ti};
   Geospatial::GeoTypesFactory::getNullGeoColumns(
       import_ti, coords, bounds, ring_sizes, poly_rings);
 
   // import extracted geo
   import_export::Importer::set_geo_physical_import_buffer(*catalog,
                                                           cd,
                                                           import_buffers,
                                                           col_idx,
                                                           coords,
                                                           bounds,
                                                           ring_sizes,
                                                           poly_rings,
                                                           /*force_null=*/true);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

virtual import_export::CopyParams foreign_storage::TextFileBufferParser::validateAndGetCopyParams ( const ForeignTable * foreign_table ) const

pure virtual

Validates foreign table parse options and returns a CopyParams object upon successful validation. An exception is thrown if validation fails.

Implemented in foreign_storage::RegexFileBufferParser, and foreign_storage::CsvFileBufferParser.

Referenced by foreign_storage::AbstractTextFileDataWrapper::iterativeFileScan(), foreign_storage::AbstractTextFileDataWrapper::populateChunkMetadata(), foreign_storage::AbstractTextFileDataWrapper::populateChunks(), and foreign_storage::AbstractTextFileDataWrapper::restoreDataWrapperInternals().

Here is the caller graph for this function:

virtual void foreign_storage::TextFileBufferParser::validateFiles	(	const FileReader *	file_reader,
		const ForeignTable *	foreign_table
	)		const

pure virtual

Performs basic validation of files to be parsed.

Implemented in foreign_storage::CsvFileBufferParser, and foreign_storage::RegexFileBufferParser.

Referenced by foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::initialize_non_append_mode_scan().

Here is the caller graph for this function:

Member Data Documentation

const std::string foreign_storage::TextFileBufferParser::BUFFER_SIZE_KEY = "BUFFER_SIZE"

inlinestatic

Definition at line 170 of file TextFileBufferParser.h.

Referenced by foreign_storage::ForeignDataWrapperFactory::createForeignTableProxy(), foreign_storage::CsvFileBufferParser::validateAndGetCopyParams(), and foreign_storage::RegexFileBufferParser::validateAndGetCopyParams().

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

/home/jenkins-slave/workspace/core-os-doxygen/DataMgr/ForeignStorage/TextFileBufferParser.h
/home/jenkins-slave/workspace/core-os-doxygen/DataMgr/ForeignStorage/TextFileBufferParser.cpp

Public Member Functions

Static Public Member Functions

Static Public Attributes

Static Private Member Functions

Detailed Description

Member Function Documentation

Member Data Documentation