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	setSkipFirstLineForTesting (bool skip)
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)
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	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 34 of file RegexFileBufferParser.h.

Constructor & Destructor Documentation

foreign_storage::RegexFileBufferParser::RegexFileBufferParser

(

const ForeignTable *

foreign_table

)

Definition at line 155 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 287 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 372 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 163 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::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(), i, 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(), and run_benchmark_import::result.

                                         {
  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 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);

  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;
      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) {
        bool set_all_nulls = regex_match_columns(row_str,
                                                 line_regex_,
                                                 logical_column_count,
                                                 parsed_columns_str,
                                                 parsed_columns_sv,
                                                 request.getFilePath());

        size_t parsed_column_index = 0;
        size_t import_buffer_index = 0;
        auto columns = request.getColumns();
        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 =
                (set_all_nulls || isNullDatum(parsed_columns_sv[parsed_column_index],
                                              cd,
                                              request.copy_params.null_str));
            if (column_type.is_geometry()) {
              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());
              // Skip remaining physical columns
              for (int i = 0; i < cd->columnType.get_physical_cols(); ++i) {
                ++cd_it;
              }
            } else {
              request.import_buffers[import_buffer_index]->add_value(
                  cd,
                  parsed_columns_sv[parsed_column_index],
                  is_null,
                  request.copy_params);
              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()));

  ParseBufferResult result{};
  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 368 of file RegexFileBufferParser.cpp.

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

                                                                       {
  max_buffer_resize_ = max_buffer_resize;
}

void foreign_storage::RegexFileBufferParser::setSkipFirstLineForTesting ( bool  skip )

static

Definition at line 376 of file RegexFileBufferParser.cpp.

References skip_first_line_.

                                                                {
  skip_first_line_ = skip;
}

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

References BUFFER_SIZE_KEY, import_export::HAS_HEADER, import_export::NO_HEADER, foreign_storage::OptionsContainer::options, import_export::CopyParams::plain_text, and skip_first_line_.

                                             {
  import_export::CopyParams copy_params{};
  copy_params.plain_text = true;
  if (skip_first_line_) {
    // This branch should only be executed in tests
    copy_params.has_header = import_export::ImportHeaderRow::HAS_HEADER;
  } else {
    copy_params.has_header = import_export::ImportHeaderRow::NO_HEADER;
  }
  if (auto it = foreign_table->options.find(BUFFER_SIZE_KEY);
      it != foreign_table->options.end()) {
    copy_params.buffer_size = std::stoi(it->second);
  }
  return copy_params;
}

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 349 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::BUFFER_SIZE_KEY = "BUFFER_SIZE"

inlinestatic

Definition at line 62 of file RegexFileBufferParser.h.

Referenced by validateAndGetCopyParams().

boost::regex foreign_storage::RegexFileBufferParser::line_regex_

private

Definition at line 73 of file RegexFileBufferParser.h.

Referenced by parseBuffer().

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

inlinestatic

Definition at line 60 of file RegexFileBufferParser.h.

Referenced by 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 74 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 61 of file RegexFileBufferParser.h.

Referenced by 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 67 of file RegexFileBufferParser.h.

Referenced by getMaxBufferResize(), and setMaxBufferResize().

bool foreign_storage::RegexFileBufferParser::skip_first_line_ {false}

inlinestaticprivate

Definition at line 71 of file RegexFileBufferParser.h.

Referenced by setSkipFirstLineForTesting(), and validateAndGetCopyParams().

---

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