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

Protected Member Functions
virtual bool	regexMatchColumns (const std::string &row_str, const boost::regex &line_regex, size_t logical_column_count, std::vector< std::string > &parsed_columns_str, std::vector< std::string_view > &parsed_columns_sv, const std::string &file_path) const
virtual bool	shouldRemoveNonMatches () const
virtual bool	shouldTruncateStringValues () const

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 152 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 388 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_regex_, line_start_regex_, foreign_storage::anonymous_namespace{RegexFileBufferParser.cpp}::line_starts_with_regex(), import_export::delimited_parser::max_buffer_resize, and shouldRemoveNonMatches().

                                                  {
  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_,
                                     line_regex_,
                                     shouldRemoveNonMatches());
  return end_pos + 1;
}

Here is the call graph for this function:

size_t foreign_storage::RegexFileBufferParser::getMaxBufferResize ( )

staticprivate

Definition at line 479 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, bool  skip_dict_encoding = 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 160 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_, StringDictionary::MAX_STRLEN, import_export::CopyParams::null_str, foreign_storage::ParseBufferRequest::process_row_count, foreign_storage::TextFileBufferParser::processGeoColumn(), regexMatchColumns(), run_benchmark_import::result, shouldRemoveNonMatches(), shouldTruncateStringValues(), 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 {
          parsed_columns_str.clear();
          parsed_columns_sv.clear();
          set_all_nulls = regexMatchColumns(row_str,
                                            line_regex_,
                                            logical_column_count,
                                            parsed_columns_str,
                                            parsed_columns_sv,
                                            request.getFilePath());
          if (set_all_nulls && shouldRemoveNonMatches()) {
            current_row_id = row_count--;
            remaining_row_count++;
            continue;
          }
        } 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());
              } 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 {
                auto& column_sv = parsed_columns_sv[parsed_column_index];
                if (column_type.is_string() && shouldTruncateStringValues() &&
                    column_sv.length() > StringDictionary::MAX_STRLEN) {
                  column_sv = column_sv.substr(0, StringDictionary::MAX_STRLEN);
                }
                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, skip_dict_encoding);
  }
  return result;
}

Here is the call graph for this function:

bool foreign_storage::RegexFileBufferParser::regexMatchColumns ( const std::string &  row_str, const boost::regex &  line_regex, size_t  logical_column_count, std::vector< std::string > &  parsed_columns_str, std::vector< std::string_view > &  parsed_columns_sv, const std::string &  file_path  ) const

protectedvirtual

Reimplemented in foreign_storage::LogFileBufferParser.

Definition at line 330 of file RegexFileBufferParser.cpp.

References CHECK_GT, and foreign_storage::throw_number_of_columns_mismatch_error().

Referenced by parseBuffer(), and foreign_storage::LogFileBufferParser::regexMatchColumns().

                                      {
  boost::smatch match;
  bool set_all_nulls{false};
  if (boost::regex_match(row_str, match, line_regex)) {
    auto matched_column_count = match.size() - 1 + parsed_columns_sv.size();
    if (logical_column_count != matched_column_count) {
      throw_number_of_columns_mismatch_error(
          logical_column_count, matched_column_count, file_path);
    }
    CHECK_GT(match.size(), static_cast<size_t>(1));
    for (size_t i = 1; i < match.size(); i++) {
      parsed_columns_str.emplace_back(match[i].str());
      parsed_columns_sv.emplace_back(parsed_columns_str.back());
    }
  } else {
    parsed_columns_str.clear();
    parsed_columns_sv =
        std::vector<std::string_view>(logical_column_count, std::string_view{});
    set_all_nulls = true;
  }
  return set_all_nulls;
}

Here is the call graph for this function:

Here is the caller graph for this function:

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

static

Definition at line 475 of file RegexFileBufferParser.cpp.

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

                                                                       {
  max_buffer_resize_ = max_buffer_resize;
}

bool foreign_storage::RegexFileBufferParser::shouldRemoveNonMatches ( ) const

protectedvirtual

Reimplemented in foreign_storage::LogFileBufferParser.

Definition at line 483 of file RegexFileBufferParser.cpp.

Referenced by findRowEndPosition(), and parseBuffer().

                                                         {
  return false;
}

Here is the caller graph for this function:

bool foreign_storage::RegexFileBufferParser::shouldTruncateStringValues ( ) const

protectedvirtual

Reimplemented in foreign_storage::LogFileBufferParser.

Definition at line 487 of file RegexFileBufferParser.cpp.

Referenced by parseBuffer().

                                                             {
  return false;
}

Here is the caller graph for this function:

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

References foreign_storage::TextFileBufferParser::BUFFER_SIZE_KEY, foreign_storage::ForeignTable::GEO_VALIDATE_GEOMETRY_KEY, HEADER_KEY, import_export::kHasHeader, import_export::kNoHeader, foreign_storage::OptionsContainer::options, import_export::CopyParams::plain_text, foreign_storage::AbstractFileStorageDataWrapper::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;
    }
  } else {
    // By default, regex parsed files are not assumed to have headers.
    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(AbstractFileStorageDataWrapper::THREADS_KEY);
      it != foreign_table->options.end()) {
    copy_params.threads = std::stoi(it->second);
  }
  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:

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 455 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.empty() &&
          !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 51 of file RegexFileBufferParser.h.

Referenced by validateAndGetCopyParams().

boost::regex foreign_storage::RegexFileBufferParser::line_regex_

private

Definition at line 74 of file RegexFileBufferParser.h.

Referenced by findRowEndPosition(), and parseBuffer().

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

inlinestatic

Definition at line 49 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 75 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 50 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 68 of file RegexFileBufferParser.h.

Referenced by getMaxBufferResize(), and setMaxBufferResize().

bool foreign_storage::RegexFileBufferParser::skip_first_line_ {false}

inlinestaticprivate

Definition at line 72 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