foreign_storage::CsvFileBufferParser Class Reference

#include <CsvFileBufferParser.h>

Inheritance diagram for foreign_storage::CsvFileBufferParser:

Collaboration diagram for foreign_storage::CsvFileBufferParser:

Public Member Functions
ParseBufferResult	parseBuffer (ParseBufferRequest &request, bool convert_data_blocks, bool columns_are_pre_filtered=false, bool skip_dict_encoding=false) const override
import_export::CopyParams	validateAndGetCopyParams (const ForeignTable *foreign_table) const override
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, foreign_storage::FileReader *file_reader) const override
void	validateExpectedColumnCount (const std::string &row, const import_export::CopyParams &copy_params, size_t num_cols, int point_cols, const std::string &file_name) const
void	validateFiles (const FileReader *file_reader, const ForeignTable *foreign_table) const override

Static Public Attributes
static const std::string	DELIMITER_KEY = "DELIMITER"
static const std::string	NULLS_KEY = "NULLS"
static const std::string	HEADER_KEY = "HEADER"
static const std::string	QUOTED_KEY = "QUOTED"
static const std::string	QUOTE_KEY = "QUOTE"
static const std::string	ESCAPE_KEY = "ESCAPE"
static const std::string	LINE_DELIMITER_KEY = "LINE_DELIMITER"
static const std::string	ARRAY_DELIMITER_KEY = "ARRAY_DELIMITER"
static const std::string	ARRAY_MARKER_KEY = "ARRAY_MARKER"
static const std::string	LONLAT_KEY = "LONLAT"
static const std::string	GEO_EXPLODE_COLLECTIONS_KEY = "GEO_EXPLODE_COLLECTIONS"
static const std::string	SOURCE_SRID_KEY = "SOURCE_SRID"
static const std::string	TRIM_SPACES_KEY = "TRIM_SPACES"
Static Public Attributes inherited from foreign_storage::TextFileBufferParser
static const std::string	BUFFER_SIZE_KEY = "BUFFER_SIZE"

Additional Inherited Members
Static Public Member Functions inherited from foreign_storage::TextFileBufferParser
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)

Detailed Description

Definition at line 22 of file CsvFileBufferParser.h.

Member Function Documentation

size_t foreign_storage::CsvFileBufferParser::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, foreign_storage::FileReader *  file_reader  ) const

overridevirtual

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.

Implements foreign_storage::TextFileBufferParser.

Definition at line 408 of file CsvFileBufferParser.cpp.

References import_export::delimited_parser::find_row_end_pos().

                                                  {
  return import_export::delimited_parser::find_row_end_pos(alloc_size,
                                                           buffer,
                                                           buffer_size,
                                                           copy_params,
                                                           buffer_first_row_index,
                                                           num_rows_in_buffer,
                                                           nullptr,
                                                           file_reader);
}

Here is the call graph for this function:

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

overridevirtual

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.

Implements foreign_storage::TextFileBufferParser.

Definition at line 122 of file CsvFileBufferParser.cpp.

References foreign_storage::ParseBufferRequest::begin_pos, foreign_storage::ParseBufferRequest::buffer, foreign_storage::ParseBufferRequest::buffer_size, CHECK, foreign_storage::TextFileBufferParser::convertImportBuffersToDataBlocks(), foreign_storage::ParseBufferRequest::copy_params, foreign_storage::ParseBufferRequest::end_pos, foreign_storage::ParseBufferRequest::file_offset, foreign_storage::TextFileBufferParser::fillRejectedRowWithInvalidData(), import_export::delimited_parser::find_beginning(), foreign_storage::ParseBufferRequest::first_row_index, foreign_storage::ParseBufferRequest::foreign_table_schema, import_export::delimited_parser::get_row(), foreign_storage::ParseBufferRequest::getCatalog(), foreign_storage::ParseBufferRequest::getColumns(), foreign_storage::ParseBufferRequest::getFilePath(), foreign_storage::ParseBufferRequest::import_buffers, is_null(), foreign_storage::TextFileBufferParser::isCoordinateScalar(), foreign_storage::TextFileBufferParser::isNullDatum(), kPOINT, import_export::CopyParams::null_str, foreign_storage::ParseBufferRequest::process_row_count, foreign_storage::TextFileBufferParser::processGeoColumn(), run_benchmark_import::result, foreign_storage::anonymous_namespace{CsvFileBufferParser.cpp}::set_array_flags_and_geo_columns_count(), foreign_storage::anonymous_namespace{CsvFileBufferParser.cpp}::skip_column_import(), sv_strip(), foreign_storage::ParseBufferRequest::track_rejected_rows, import_export::CopyParams::trim_spaces, and foreign_storage::anonymous_namespace{CsvFileBufferParser.cpp}::validate_expected_column_count().

                                                                                  {
  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.foreign_table_schema->getLogicalColumns());
  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 current_row_id = 0;
  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()));

  ParseBufferResult result{};

  std::string file_path = request.getFilePath();
  for (; p < thread_buf_end && remaining_row_count > 0; p++, remaining_row_count--) {
    row.clear();
    current_row_id = row_count;
    row_count++;
    std::vector<std::unique_ptr<char[]>>
        tmp_buffers;  // holds string w/ removed escape chars, etc
    const char* line_start = p;
    row_index_plus_one++;
    bool incorrect_column_count = false;
    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);
    try {
      validate_expected_column_count(row, num_cols, point_cols, file_path);
    } catch (const ForeignStorageException& e) {
      if (request.track_rejected_rows) {
        result.rejected_rows.insert(current_row_id);
        incorrect_column_count = true;
      } else {
        throw;
      }
    }

    size_t import_idx = 0;
    size_t col_idx = 0;

    try {
      auto columns = request.getColumns();
      if (incorrect_column_count) {
        auto cd_it = columns.begin();
        fillRejectedRowWithInvalidData(columns, cd_it, 0, request);
        continue;
      }
      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 = false;
        try {
          is_null = column_is_present
                        ? isNullDatum(row[import_idx], cd, request.copy_params.null_str)
                        : true;
        } catch (const std::exception& e) {
          if (request.track_rejected_rows) {
            result.rejected_rows.insert(current_row_id);
            fillRejectedRowWithInvalidData(columns, cd_it, col_idx, request);
            break;  // skip rest of row
          } else {
            throw;
          }
        }
        if (!col_ti.is_string() && !request.copy_params.trim_spaces) {
          // everything but strings should be always trimmed
          row[import_idx] = sv_strip(row[import_idx]);
        }
        if (col_ti.is_geometry()) {
          if (!skip_column_import(request, col_idx)) {
            auto starting_col_idx = col_idx;
            try {
              processGeoColumn(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());
            } catch (const std::exception& e) {
              if (request.track_rejected_rows) {
                result.rejected_rows.insert(current_row_id);
                fillRejectedRowWithInvalidData(columns, cd_it, starting_col_idx, request);
                break;  // skip rest of row
              } else {
                throw;
              }
            }
          } else {
            // update import/col idx according to types
            if (!is_null && cd->columnType == kPOINT &&
                isCoordinateScalar(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)) {
            try {
              request.import_buffers[col_idx]->add_value(
                  cd, row[import_idx], is_null, request.copy_params);
            } catch (const std::exception& e) {
              if (request.track_rejected_rows) {
                result.rejected_rows.insert(current_row_id);
                fillRejectedRowWithInvalidData(columns, cd_it, col_idx, request);
                break;  // skip rest of row
              } else {
                throw;
              }
            }
          }
          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()));

  result.row_offsets = row_offsets;
  result.row_count = row_count;
  if (convert_data_blocks) {
    result.column_id_to_data_blocks_map =
        convertImportBuffersToDataBlocks(request.import_buffers, skip_dict_encoding);
  }
  return result;
}

Here is the call graph for this function:

import_export::CopyParams foreign_storage::CsvFileBufferParser::validateAndGetCopyParams ( const ForeignTable *  foreign_table ) const

overridevirtual

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

Implements foreign_storage::TextFileBufferParser.

Definition at line 334 of file CsvFileBufferParser.cpp.

References ARRAY_DELIMITER_KEY, ARRAY_MARKER_KEY, foreign_storage::TextFileBufferParser::BUFFER_SIZE_KEY, DELIMITER_KEY, ESCAPE_KEY, GEO_EXPLODE_COLLECTIONS_KEY, foreign_storage::ForeignTable::GEO_VALIDATE_GEOMETRY_KEY, HEADER_KEY, import_export::kHasHeader, import_export::kNoHeader, LINE_DELIMITER_KEY, LONLAT_KEY, NULLS_KEY, foreign_storage::OptionsContainer::options, import_export::CopyParams::plain_text, QUOTE_KEY, QUOTED_KEY, SOURCE_SRID_KEY, foreign_storage::AbstractFileStorageDataWrapper::THREADS_KEY, TRIM_SPACES_KEY, foreign_storage::anonymous_namespace{CsvFileBufferParser.cpp}::validate_and_get_bool_value(), foreign_storage::anonymous_namespace{CsvFileBufferParser.cpp}::validate_and_get_delimiter(), and foreign_storage::anonymous_namespace{CsvFileBufferParser.cpp}::validate_and_get_string_with_length().

Referenced by foreign_storage::CsvDataWrapper::validateTableOptions().

                                             {
  import_export::CopyParams copy_params{};
  copy_params.plain_text = true;
  if (const auto& value = validate_and_get_delimiter(foreign_table, DELIMITER_KEY);
      !value.empty()) {
    copy_params.delimiter = value[0];
  }
  if (auto it = foreign_table->options.find(NULLS_KEY);
      it != foreign_table->options.end()) {
    copy_params.null_str = it->second;
  }
  auto has_header = validate_and_get_bool_value(foreign_table, HEADER_KEY);
  if (has_header.has_value()) {
    if (has_header.value()) {
      copy_params.has_header = import_export::ImportHeaderRow::kHasHeader;
    } else {
      copy_params.has_header = import_export::ImportHeaderRow::kNoHeader;
    }
  }
  copy_params.quoted =
      validate_and_get_bool_value(foreign_table, QUOTED_KEY).value_or(copy_params.quoted);
  if (const auto& value =
          validate_and_get_string_with_length(foreign_table, QUOTE_KEY, 1);
      !value.empty()) {
    copy_params.quote = value[0];
  }
  if (const auto& value =
          validate_and_get_string_with_length(foreign_table, ESCAPE_KEY, 1);
      !value.empty()) {
    copy_params.escape = value[0];
  }
  if (const auto& value = validate_and_get_delimiter(foreign_table, LINE_DELIMITER_KEY);
      !value.empty()) {
    copy_params.line_delim = value[0];
  }
  if (const auto& value =
          validate_and_get_string_with_length(foreign_table, ARRAY_DELIMITER_KEY, 1);
      !value.empty()) {
    copy_params.array_delim = value[0];
  }
  if (const auto& value =
          validate_and_get_string_with_length(foreign_table, ARRAY_MARKER_KEY, 2);
      !value.empty()) {
    copy_params.array_begin = value[0];
    copy_params.array_end = value[1];
  }
  copy_params.lonlat =
      validate_and_get_bool_value(foreign_table, LONLAT_KEY).value_or(copy_params.lonlat);
  copy_params.geo_explode_collections =
      validate_and_get_bool_value(foreign_table, GEO_EXPLODE_COLLECTIONS_KEY)
          .value_or(copy_params.geo_explode_collections);
  if (auto it = foreign_table->options.find(SOURCE_SRID_KEY);
      it != foreign_table->options.end()) {
    copy_params.source_srid = std::stoi(it->second);
  }

  if (auto it = foreign_table->options.find(BUFFER_SIZE_KEY);
      it != foreign_table->options.end()) {
    copy_params.buffer_size = std::stoi(it->second);
  }
  if (auto it = foreign_table->options.find(AbstractFileStorageDataWrapper::THREADS_KEY);
      it != foreign_table->options.end()) {
    copy_params.threads = std::stoi(it->second);
  }
  copy_params.trim_spaces = validate_and_get_bool_value(foreign_table, TRIM_SPACES_KEY)
                                .value_or(copy_params.trim_spaces);
  copy_params.geo_validate_geometry =
      validate_and_get_bool_value(foreign_table, ForeignTable::GEO_VALIDATE_GEOMETRY_KEY)
          .value_or(copy_params.geo_validate_geometry);

  return copy_params;
}

Here is the call graph for this function:

Here is the caller graph for this function:

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

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

Definition at line 309 of file CsvFileBufferParser.cpp.

References import_export::delimited_parser::get_row(), and foreign_storage::anonymous_namespace{CsvFileBufferParser.cpp}::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:

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

inlineoverridevirtual

Performs basic validation of files to be parsed.

Implements foreign_storage::TextFileBufferParser.

Definition at line 46 of file CsvFileBufferParser.h.

                                                                      {};

Member Data Documentation

const std::string foreign_storage::CsvFileBufferParser::ARRAY_DELIMITER_KEY = "ARRAY_DELIMITER"

inlinestatic

Definition at line 57 of file CsvFileBufferParser.h.

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

const std::string foreign_storage::CsvFileBufferParser::ARRAY_MARKER_KEY = "ARRAY_MARKER"

inlinestatic

Definition at line 58 of file CsvFileBufferParser.h.

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

const std::string foreign_storage::CsvFileBufferParser::DELIMITER_KEY = "DELIMITER"

inlinestatic

Definition at line 50 of file CsvFileBufferParser.h.

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

const std::string foreign_storage::CsvFileBufferParser::ESCAPE_KEY = "ESCAPE"

inlinestatic

Definition at line 55 of file CsvFileBufferParser.h.

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

const std::string foreign_storage::CsvFileBufferParser::GEO_EXPLODE_COLLECTIONS_KEY = "GEO_EXPLODE_COLLECTIONS"

inlinestatic

Definition at line 60 of file CsvFileBufferParser.h.

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

const std::string foreign_storage::CsvFileBufferParser::HEADER_KEY = "HEADER"

inlinestatic

Definition at line 52 of file CsvFileBufferParser.h.

Referenced by foreign_storage::anonymous_namespace{ForeignDataWrapperFactory.cpp}::set_header_option(), and validateAndGetCopyParams().

const std::string foreign_storage::CsvFileBufferParser::LINE_DELIMITER_KEY = "LINE_DELIMITER"

inlinestatic

Definition at line 56 of file CsvFileBufferParser.h.

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

const std::string foreign_storage::CsvFileBufferParser::LONLAT_KEY = "LONLAT"

inlinestatic

Definition at line 59 of file CsvFileBufferParser.h.

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

const std::string foreign_storage::CsvFileBufferParser::NULLS_KEY = "NULLS"

inlinestatic

Definition at line 51 of file CsvFileBufferParser.h.

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

const std::string foreign_storage::CsvFileBufferParser::QUOTE_KEY = "QUOTE"

inlinestatic

Definition at line 54 of file CsvFileBufferParser.h.

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

const std::string foreign_storage::CsvFileBufferParser::QUOTED_KEY = "QUOTED"

inlinestatic

Definition at line 53 of file CsvFileBufferParser.h.

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

const std::string foreign_storage::CsvFileBufferParser::SOURCE_SRID_KEY = "SOURCE_SRID"

inlinestatic

Definition at line 61 of file CsvFileBufferParser.h.

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

const std::string foreign_storage::CsvFileBufferParser::TRIM_SPACES_KEY = "TRIM_SPACES"

inlinestatic

Definition at line 62 of file CsvFileBufferParser.h.

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

---

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

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