foreign_storage::RegexFileBufferParser Class Reference

#include <RegexFileBufferParser.h>

Inheritance diagram for foreign_storage::RegexFileBufferParser:

Collaboration diagram for foreign_storage::RegexFileBufferParser:

Public Member Functions
	RegexFileBufferParser (const ForeignTable *foreign_table)
ParseBufferResult	parseBuffer (ParseBufferRequest &request, bool convert_data_blocks, bool columns_are_pre_filtered=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, FileReader *file_reader) const override
void	validateFiles (const FileReader *file_reader, const ForeignTable *foreign_table) const override

Static Public Member Functions
static void	setMaxBufferResize (size_t max_buffer_resize)
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)
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, const RenderGroupAnalyzerMap *render_group_analyzer_map)
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	LINE_REGEX_KEY = "LINE_REGEX"
static const std::string	LINE_START_REGEX_KEY = "LINE_START_REGEX"
static const std::string	HEADER_KEY = "HEADER"
Static Public Attributes inherited from foreign_storage::TextFileBufferParser
static const std::string	THREADS_KEY = "THREADS"
static const std::string	BUFFER_SIZE_KEY = "BUFFER_SIZE"

Static Private Member Functions
static size_t	getMaxBufferResize ()

Private Attributes
boost::regex	line_regex_
std::optional< boost::regex >	line_start_regex_

Static Private Attributes
static size_t	max_buffer_resize_
static bool	skip_first_line_ {false}

Detailed Description

Definition at line 23 of file RegexFileBufferParser.h.

Constructor & Destructor Documentation

foreign_storage::RegexFileBufferParser::RegexFileBufferParser

(

const ForeignTable *

foreign_table

)

Definition at line 172 of file RegexFileBufferParser.cpp.

    : line_regex_(get_line_regex(foreign_table))
    , line_start_regex_(get_line_start_regex(foreign_table)) {}

Member Function Documentation

size_t foreign_storage::RegexFileBufferParser::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

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 362 of file RegexFileBufferParser.cpp.

References CHECK, CHECK_EQ, CHECK_GT, import_export::delimited_parser::extend_buffer(), foreign_storage::anonymous_namespace{RegexFileBufferParser.cpp}::find_last_end_of_line(), foreign_storage::anonymous_namespace{RegexFileBufferParser.cpp}::get_row_count(), getMaxBufferResize(), foreign_storage::FileReader::isEndOfLastFile(), foreign_storage::FileReader::isScanFinished(), import_export::CopyParams::line_delim, line_start_regex_, foreign_storage::anonymous_namespace{RegexFileBufferParser.cpp}::line_starts_with_regex(), and import_export::delimited_parser::max_buffer_resize.

                                                  {
  CHECK_GT(buffer_size, static_cast<size_t>(0));
  size_t start_pos{0};
  size_t end_pos = buffer_size - 1;
  bool found_end_pos{false};
  while (!found_end_pos) {
    try {
      end_pos = find_last_end_of_line(
          buffer.get(), buffer_size, start_pos, end_pos, copy_params.line_delim);
      if (file_reader->isEndOfLastFile()) {
        CHECK_EQ(end_pos, buffer_size - 1);
        found_end_pos = true;
      } else if (line_start_regex_.has_value()) {
        // When a LINE_START_REGEX option is present and the file reader is not at the end
        // of file, return the position of the end of line before the last line that
        // matches the line start regex, since the last line that matches the line start
        // regex in this buffer may still have to include/concatenate lines beyond this
        // buffer.
        CHECK_GT(end_pos, static_cast<size_t>(0));
        auto old_end_pos = end_pos;
        end_pos = find_last_end_of_line(buffer.get(),
                                        buffer_size,
                                        start_pos,
                                        old_end_pos - 1,
                                        copy_params.line_delim);
        while (!line_starts_with_regex(
            buffer.get(), end_pos + 1, old_end_pos, line_start_regex_.value())) {
          old_end_pos = end_pos;
          end_pos = find_last_end_of_line(buffer.get(),
                                          buffer_size,
                                          start_pos,
                                          old_end_pos - 1,
                                          copy_params.line_delim);
        }
        found_end_pos = true;
      } else {
        found_end_pos = true;
      }
    } catch (InsufficientBufferSizeException& e) {
      auto max_buffer_resize = getMaxBufferResize();
      if (alloc_size >= max_buffer_resize || file_reader->isScanFinished()) {
        throw;
      }
      start_pos = buffer_size;
      import_export::delimited_parser::extend_buffer(
          buffer, buffer_size, alloc_size, nullptr, file_reader, max_buffer_resize);
      end_pos = buffer_size - 1;
    }
  }
  CHECK(found_end_pos);
  num_rows_in_buffer =
      get_row_count(buffer.get(), 0, end_pos, copy_params.line_delim, line_start_regex_);
  return end_pos + 1;
}

Here is the call graph for this function:

size_t foreign_storage::RegexFileBufferParser::getMaxBufferResize ( )

staticprivate

Definition at line 447 of file RegexFileBufferParser.cpp.

References max_buffer_resize_.

Referenced by findRowEndPosition().

                                                 {
  return max_buffer_resize_;
}

Here is the caller graph for this function:

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

overridevirtual

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.

Implements foreign_storage::TextFileBufferParser.

Definition at line 180 of file RegexFileBufferParser.cpp.

References foreign_storage::ParseBufferRequest::begin_pos, foreign_storage::ParseBufferRequest::buffer, CHECK, foreign_storage::TextFileBufferParser::convertImportBuffersToDataBlocks(), foreign_storage::ParseBufferRequest::copy_params, foreign_storage::ParseBufferRequest::end_pos, foreign_storage::ParseBufferRequest::file_offset, foreign_storage::TextFileBufferParser::fillRejectedRowWithInvalidData(), foreign_storage::ParseBufferRequest::first_row_index, foreign_storage::ParseBufferRequest::foreign_table_schema, foreign_storage::anonymous_namespace{RegexFileBufferParser.cpp}::get_next_row(), foreign_storage::ParseBufferRequest::getCatalog(), foreign_storage::ParseBufferRequest::getColumns(), foreign_storage::ParseBufferRequest::getFilePath(), foreign_storage::ParseBufferRequest::import_buffers, is_null(), foreign_storage::TextFileBufferParser::isNullDatum(), import_export::CopyParams::line_delim, line_regex_, line_start_regex_, import_export::CopyParams::null_str, foreign_storage::ParseBufferRequest::process_row_count, foreign_storage::TextFileBufferParser::processGeoColumn(), foreign_storage::anonymous_namespace{RegexFileBufferParser.cpp}::regex_match_columns(), foreign_storage::ParseBufferRequest::render_group_analyzer_map, run_benchmark_import::result, and foreign_storage::ParseBufferRequest::track_rejected_rows.

                                         {
  CHECK(request.buffer);
  char* buffer_start = request.buffer.get() + request.begin_pos;
  const char* buffer_end = request.buffer.get() + request.end_pos;

  std::vector<size_t> row_offsets;
  row_offsets.emplace_back(request.file_offset + request.begin_pos);

  size_t current_row_id = 0;
  size_t row_count = 0;
  auto logical_column_count = request.foreign_table_schema->getLogicalColumns().size();
  std::vector<std::string> parsed_columns_str;
  parsed_columns_str.reserve(logical_column_count);
  std::vector<std::string_view> parsed_columns_sv;
  parsed_columns_sv.reserve(logical_column_count);

  ParseBufferResult result{};

  std::string row_str;
  size_t remaining_row_count = request.process_row_count;
  auto curr = buffer_start;
  while (curr < buffer_end && remaining_row_count > 0) {
    try {
      row_str = get_next_row(
          curr, buffer_end - 1, request.copy_params.line_delim, line_start_regex_);
      curr += row_str.length() + 1;
      current_row_id = row_count++;
      remaining_row_count--;

      bool skip_all_columns =
          std::all_of(request.import_buffers.begin(),
                      request.import_buffers.end(),
                      [](const auto& import_buffer) { return !import_buffer; });
      if (!skip_all_columns) {
        auto columns = request.getColumns();

        bool set_all_nulls = false;
        try {
          set_all_nulls = regex_match_columns(row_str,
                                              line_regex_,
                                              logical_column_count,
                                              parsed_columns_str,
                                              parsed_columns_sv,
                                              request.getFilePath());
        } catch (const ForeignStorageException& e) {
          if (request.track_rejected_rows) {
            result.rejected_rows.insert(current_row_id);
            auto cd_it = columns.begin();
            fillRejectedRowWithInvalidData(columns, cd_it, 0, request);
            continue;
          } else {
            throw;
          }
        }

        size_t parsed_column_index = 0;
        size_t import_buffer_index = 0;

        for (auto cd_it = columns.begin(); cd_it != columns.end(); ++cd_it) {
          auto cd = *cd_it;
          const auto& column_type = cd->columnType;
          if (request.import_buffers[import_buffer_index]) {
            bool is_null = false;
            try {
              is_null =
                  (set_all_nulls || isNullDatum(parsed_columns_sv[parsed_column_index],
                                                cd,
                                                request.copy_params.null_str));
            } catch (const std::exception& e) {
              if (request.track_rejected_rows) {
                result.rejected_rows.insert(current_row_id);
                fillRejectedRowWithInvalidData(
                    columns, cd_it, import_buffer_index, request);
                break;  // skip rest of row
              } else {
                throw;
              }
            }
            if (column_type.is_geometry()) {
              auto starting_import_buffer_index = import_buffer_index;
              try {
                processGeoColumn(request.import_buffers,
                                 import_buffer_index,
                                 request.copy_params,
                                 cd_it,
                                 parsed_columns_sv,
                                 parsed_column_index,
                                 is_null,
                                 request.first_row_index,
                                 row_count,
                                 request.getCatalog(),
                                 request.render_group_analyzer_map);
              } catch (const std::exception& e) {
                if (request.track_rejected_rows) {
                  result.rejected_rows.insert(current_row_id);
                  fillRejectedRowWithInvalidData(
                      columns, cd_it, starting_import_buffer_index, request);
                  break;  // skip rest of row
                } else {
                  throw;
                }
              }
              // Skip remaining physical columns
              for (int i = 0; i < cd->columnType.get_physical_cols(); ++i) {
                ++cd_it;
              }
            } else {
              try {
                request.import_buffers[import_buffer_index]->add_value(
                    cd,
                    parsed_columns_sv[parsed_column_index],
                    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, import_buffer_index, request);
                  break;  // skip rest of row
                } else {
                  throw;
                }
              }
              parsed_column_index++;
              import_buffer_index++;
            }
          } else {
            // Skip column
            for (int i = 0; i < column_type.get_physical_cols(); i++) {
              import_buffer_index++;
              cd_it++;
            }
            parsed_column_index++;
            import_buffer_index++;
          }
        }
      }
    } catch (const ForeignStorageException& e) {
      throw;
    } catch (const std::exception& e) {
      throw ForeignStorageException("Parsing failure \"" + std::string(e.what()) +
                                    "\" in row \"" + row_str + "\" in file \"" +
                                    request.getFilePath() + "\"");
    }
  }
  row_offsets.emplace_back(request.file_offset + (curr - 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);
  }
  return result;
}

Here is the call graph for this function:

void foreign_storage::RegexFileBufferParser::setMaxBufferResize ( size_t  max_buffer_resize )

static

Definition at line 443 of file RegexFileBufferParser.cpp.

References import_export::delimited_parser::max_buffer_resize, and max_buffer_resize_.

                                                                       {
  max_buffer_resize_ = max_buffer_resize;
}

import_export::CopyParams foreign_storage::RegexFileBufferParser::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 339 of file RegexFileBufferParser.cpp.

References foreign_storage::TextFileBufferParser::BUFFER_SIZE_KEY, HEADER_KEY, import_export::kHasHeader, import_export::kNoHeader, foreign_storage::OptionsContainer::options, import_export::CopyParams::plain_text, foreign_storage::TextFileBufferParser::THREADS_KEY, and foreign_storage::anonymous_namespace{CsvFileBufferParser.cpp}::validate_and_get_bool_value().

                                             {
  import_export::CopyParams copy_params{};
  copy_params.plain_text = true;
  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;
    }
  }
  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(THREADS_KEY);
      it != foreign_table->options.end()) {
    copy_params.threads = std::stoi(it->second);
  }
  return copy_params;
}

Here is the call graph for this function:

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

overridevirtual

Performs basic validation of files to be parsed.

Implements foreign_storage::TextFileBufferParser.

Definition at line 424 of file RegexFileBufferParser.cpp.

References CHECK, foreign_storage::anonymous_namespace{RegexFileBufferParser.cpp}::get_line_start_regex(), foreign_storage::FileReader::getFirstLineForEachFile(), parse_ast::line, line_start_regex_, and foreign_storage::anonymous_namespace{RegexFileBufferParser.cpp}::line_starts_with_regex().

                                                                                   {
  if (line_start_regex_.has_value()) {
    // When a LINE_START_REGEX option is specified, at least the first line in each file
    // has to start with the specified regex.
    auto first_line_by_file_path = file_reader->getFirstLineForEachFile();
    for (const auto& [file_path, line] : first_line_by_file_path) {
      if (!line_starts_with_regex(
              line.c_str(), 0, line.length() - 1, line_start_regex_.value())) {
        auto line_start_regex = get_line_start_regex(foreign_table);
        CHECK(line_start_regex.has_value());
        throw ForeignStorageException{"First line in file \"" + file_path +
                                      "\" does not match line start regex \"" +
                                      line_start_regex.value() + "\""};
      }
    }
  }
}

Here is the call graph for this function:

Member Data Documentation

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

inlinestatic

Definition at line 50 of file RegexFileBufferParser.h.

Referenced by validateAndGetCopyParams().

boost::regex foreign_storage::RegexFileBufferParser::line_regex_

private

Definition at line 61 of file RegexFileBufferParser.h.

Referenced by parseBuffer().

const std::string foreign_storage::RegexFileBufferParser::LINE_REGEX_KEY = "LINE_REGEX"

inlinestatic

Definition at line 48 of file RegexFileBufferParser.h.

Referenced by foreign_storage::ForeignDataWrapperFactory::createForeignTableProxy(), foreign_storage::anonymous_namespace{RegexFileBufferParser.cpp}::get_line_regex(), and foreign_storage::RegexParserDataWrapper::validateTableOptions().

std::optional<boost::regex> foreign_storage::RegexFileBufferParser::line_start_regex_

private

Definition at line 62 of file RegexFileBufferParser.h.

Referenced by findRowEndPosition(), parseBuffer(), and validateFiles().

const std::string foreign_storage::RegexFileBufferParser::LINE_START_REGEX_KEY = "LINE_START_REGEX"

inlinestatic

Definition at line 49 of file RegexFileBufferParser.h.

Referenced by foreign_storage::ForeignDataWrapperFactory::createForeignTableProxy(), foreign_storage::anonymous_namespace{RegexFileBufferParser.cpp}::get_line_start_regex(), and foreign_storage::RegexParserDataWrapper::validateTableOptions().

size_t foreign_storage::RegexFileBufferParser::max_buffer_resize_

inlinestaticprivate

Initial value:

{
      import_export::max_import_buffer_resize_byte_size}

Definition at line 55 of file RegexFileBufferParser.h.

Referenced by getMaxBufferResize(), and setMaxBufferResize().

bool foreign_storage::RegexFileBufferParser::skip_first_line_ {false}

inlinestaticprivate

Definition at line 59 of file RegexFileBufferParser.h.

---

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

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