import_export::delimited_parser Namespace Reference

Classes
class	InsufficientBufferSizeException

Functions
size_t	find_beginning (const char *buffer, size_t begin, size_t end, const CopyParams &copy_params)
	Finds the closest possible row beginning in the given buffer. More...
size_t	find_end (const char *buffer, size_t size, const import_export::CopyParams &copy_params, unsigned int &num_rows_this_buffer, size_t buffer_first_row_index, bool &in_quote, size_t offset)
size_t	get_max_buffer_resize ()
	Gets the maximum size to which thread buffers should be automatically resized. More...
void	set_max_buffer_resize (const size_t max_buffer_resize)
	Sets the maximum size to which thread buffers should be automatically resized. This function is only used for testing. More...
size_t	find_row_end_pos (size_t &alloc_size, std::unique_ptr< char[]> &buffer, size_t &buffer_size, const CopyParams &copy_params, const size_t buffer_first_row_index, unsigned int &num_rows_in_buffer, FILE *file, foreign_storage::FileReader *file_reader=nullptr)
	Finds the closest possible row ending to the end of the given buffer. The buffer is resized as needed, with more content read from the file, until an end of row is found or a configured max buffer limit is reached. More...
template<typename T >
const char *	get_row (const char *buf, const char *buf_end, const char *entire_buf_end, const import_export::CopyParams &copy_params, const bool *is_array, std::vector< T > &row, std::vector< std::unique_ptr< char[]>> &tmp_buffers, bool &try_single_thread, bool filter_empty_lines)
	Parses the first row in the given buffer and inserts fields into given vector. More...
template const char *	get_row (const char *buf, const char *buf_end, const char *entire_buf_end, const import_export::CopyParams &copy_params, const bool *is_array, std::vector< std::string > &row, std::vector< std::unique_ptr< char[]>> &tmp_buffers, bool &try_single_thread, bool filter_empty_lines)
template const char *	get_row (const char *buf, const char *buf_end, const char *entire_buf_end, const import_export::CopyParams &copy_params, const bool *is_array, std::vector< std::string_view > &row, std::vector< std::unique_ptr< char[]>> &tmp_buffers, bool &try_single_thread, bool filter_empty_lines)
void	parse_string_array (const std::string &s, const import_export::CopyParams &copy_params, std::vector< std::string > &string_vec, bool truncate_values=false)
	Parses given string array and inserts into given vector of strings. More...
void	extend_buffer (std::unique_ptr< char[]> &buffer, size_t &buffer_size, size_t &alloc_size, FILE *file, foreign_storage::FileReader *file_reader, size_t max_buffer_resize)

Variables
static size_t	max_buffer_resize = max_import_buffer_resize_byte_size

Function Documentation

void import_export::delimited_parser::extend_buffer	(	std::unique_ptr< char[]> &	buffer,
		size_t &	buffer_size,
		size_t &	alloc_size,
		FILE *	file,
		foreign_storage::FileReader *	file_reader,
		size_t	max_buffer_resize
	)

Extends the given buffer to the lesser of max_buffer_resize or twice the given allocation size and reads new content from the file into the newly allocated buffer.

Parameters

buffer	- buffer that will be extended
buffer_size	- current buffer size
alloc_size	- current allocation size
file	- handle for file to be read from (one of file or file_reader must be present)
file_reader	- reader for file to be read from (one of file or file_reader must be present)
max_buffer_resize	- maximum size that the buffer can be extended to

Definition at line 358 of file DelimitedParserUtils.cpp.

References CHECK, logger::INFO, LOG, and foreign_storage::FileReader::read().

Referenced by find_row_end_pos(), and foreign_storage::RegexFileBufferParser::findRowEndPosition().

                                             {
  auto old_buffer = std::move(buffer);
  alloc_size = std::min(max_buffer_resize, alloc_size * 2);
  LOG(INFO) << "Setting import thread buffer allocation size to " << alloc_size
            << " bytes";
  buffer = std::make_unique<char[]>(alloc_size);

  memcpy(buffer.get(), old_buffer.get(), buffer_size);
  size_t fread_size;
  CHECK(file != nullptr || file_reader != nullptr);
  if (file != nullptr) {
    fread_size = fread(buffer.get() + buffer_size, 1, alloc_size - buffer_size, file);
  } else {
    fread_size = file_reader->read(buffer.get() + buffer_size, alloc_size - buffer_size);
  }
  buffer_size += fread_size;
}

Here is the call graph for this function:

Here is the caller graph for this function:

size_t import_export::delimited_parser::find_beginning	(	const char *	buffer,
		size_t	begin,
		size_t	end,
		const CopyParams &	copy_params
	)

Finds the closest possible row beginning in the given buffer.

Parameters

buffer	Given buffer which has the rows in csv format. (NOT OWN)
begin	Start index of buffer to look for the beginning.
end	End index of buffer to look for the beginning.
copy_params	Copy params for the table.

Returns

The position of the closest possible row beginning to the start of the given buffer.

Definition at line 62 of file DelimitedParserUtils.cpp.

References import_export::CopyParams::line_delim.

Referenced by import_export::import_thread_delimited(), and foreign_storage::CsvFileBufferParser::parseBuffer().

                                                                  {
  // @TODO(wei) line_delim is in quotes note supported
  if (begin == 0 || (begin > 0 && buffer[begin - 1] == copy_params.line_delim)) {
    return 0;
  }
  size_t i;
  const char* buf = buffer + begin;
  for (i = 0; i < end - begin; i++) {
    if (buf[i] == copy_params.line_delim) {
      return i + 1;
    }
  }
  return i;
}

Here is the caller graph for this function:

size_t import_export::delimited_parser::find_end	(	const char *	buffer,
		size_t	size,
		const import_export::CopyParams &	copy_params,
		unsigned int &	num_rows_this_buffer,
		size_t	buffer_first_row_index,
		bool &	in_quote,
		size_t	offset
	)

Definition at line 80 of file DelimitedParserUtils.cpp.

References import_export::CopyParams::escape, import_export::CopyParams::line_delim, import_export::CopyParams::quote, import_export::CopyParams::quoted, and to_string().

Referenced by find_row_end_pos().

                               {
  size_t last_line_delim_pos = 0;
  const char* current = buffer + offset;
  if (copy_params.quoted) {
    while (current < buffer + size) {
      while (!in_quote && current < buffer + size) {
        // We are outside of quotes. We have to find the last possible line delimiter.
        if (*current == copy_params.line_delim) {
          last_line_delim_pos = current - buffer;
          ++num_rows_this_buffer;
        } else if (*current == copy_params.quote) {
          in_quote = true;
        }
        ++current;
      }

      while (in_quote && current < buffer + size) {
        // We are in a quoted field. We have to find the ending quote.
        if ((*current == copy_params.escape) && (current < buffer + size - 1) &&
            (*(current + 1) == copy_params.quote)) {
          ++current;
        } else if (*current == copy_params.quote) {
          in_quote = false;
        }
        ++current;
      }
    }
  } else {
    while (current < buffer + size) {
      if (*current == copy_params.line_delim) {
        last_line_delim_pos = current - buffer;
        ++num_rows_this_buffer;
      }
      ++current;
    }
  }

  if (last_line_delim_pos <= 0) {
    size_t excerpt_length = std::min<size_t>(50, size);
    std::string buffer_excerpt{buffer, buffer + excerpt_length};
    std::string error_message =
        "Unable to find an end of line character after reading " + std::to_string(size) +
        " characters. Please ensure that the correct \"line_delimiter\" option is "
        "specified or update the \"buffer_size\" option appropriately. Row number: " +
        std::to_string(buffer_first_row_index + 1) +
        ". First few characters in row: " + buffer_excerpt;
    throw InsufficientBufferSizeException{error_message};
  }

  return last_line_delim_pos + 1;
}

Here is the call graph for this function:

Here is the caller graph for this function:

size_t import_export::delimited_parser::find_row_end_pos	(	size_t &	alloc_size,
		std::unique_ptr< char[]> &	buffer,
		size_t &	buffer_size,
		const CopyParams &	copy_params,
		const size_t	buffer_first_row_index,
		unsigned int &	num_rows_in_buffer,
		FILE *	file,
		foreign_storage::FileReader *	file_reader = nullptr
	)

Finds the closest possible row ending to the end of the given buffer. The buffer is resized as needed, with more content read from the file, until an end of row is found or a configured max buffer limit is reached.

Parameters

alloc_size	Allocation size of subsequent buffer. This is adjusted as needed, if the buffer has to be resized.
buffer	Given buffer which has the rows in csv format.
buffer_size	Size of the buffer.
copy_params	Copy params for the table.
buffer_first_row_index	Index of first row in the buffer.
num_rows_in_buffer	Number of rows until the closest possible row ending.
file	Handle to CSV file being parsed. (optional)
file_reader	Handle to a FileReader class, must be valid if file isnt

Returns

The position of the closest possible row ending to the end of the given buffer.

Definition at line 148 of file DelimitedParserUtils.cpp.

References CHECK, extend_buffer(), find_end(), get_max_buffer_resize(), foreign_storage::FileReader::isScanFinished(), and max_buffer_resize.

Referenced by foreign_storage::CsvFileBufferParser::findRowEndPosition(), and import_export::Importer::importDelimited().

                                                                {
  bool found_end_pos{false};
  bool in_quote{false};
  size_t offset{0};
  size_t end_pos;
  CHECK(file != nullptr || file_reader != nullptr);
  const auto max_buffer_resize = get_max_buffer_resize();
  while (!found_end_pos) {
    try {
      end_pos = delimited_parser::find_end(buffer.get(),
                                           buffer_size,
                                           copy_params,
                                           num_rows_in_buffer,
                                           buffer_first_row_index,
                                           in_quote,
                                           offset);
      found_end_pos = true;
    } catch (InsufficientBufferSizeException& e) {
      if (alloc_size >= max_buffer_resize) {
        throw;
      }
      if (file == nullptr && file_reader->isScanFinished()) {
        throw;
      }
      offset = buffer_size;
      extend_buffer(
          buffer, buffer_size, alloc_size, file, file_reader, max_buffer_resize);
    }
  }
  return end_pos;
}

Here is the call graph for this function:

Here is the caller graph for this function:

size_t import_export::delimited_parser::get_max_buffer_resize

(

)

Gets the maximum size to which thread buffers should be automatically resized.

Definition at line 140 of file DelimitedParserUtils.cpp.

References max_buffer_resize.

Referenced by find_row_end_pos().

                               {
  return max_buffer_resize;
}

Here is the caller graph for this function:

template<typename T >

const char * import_export::delimited_parser::get_row	(	const char *	buf,
		const char *	buf_end,
		const char *	entire_buf_end,
		const import_export::CopyParams &	copy_params,
		const bool *	is_array,
		std::vector< T > &	row,
		std::vector< std::unique_ptr< char[]>> &	tmp_buffers,
		bool &	try_single_thread,
		bool	filter_empty_lines
	)

Parses the first row in the given buffer and inserts fields into given vector.

Parameters

buf	Given buffer which has the rows in csv format. (NOT OWN)
buf_end	End of the sliced buffer for the thread. (NOT OWN)
entire_buf_end	End of the entire buffer. (NOT OWN)
copy_params	Copy params for the table.
is_array	Array of bools which tells if a column is an array type.
row	Given vector to be populated with parsed fields.
try_single_thread	In case of parse errors, this will tell if parsing should continue with single thread.
filter_empty_lines	Whether to skip empty lines (used when parsing single columns returned by s3 select, as nulls may be encoded as empty lines)

Returns

Pointer to the next row after the first row is parsed.

Definition at line 188 of file DelimitedParserUtils.cpp.

References import_export::CopyParams::array_begin, import_export::CopyParams::array_end, import_export::CopyParams::delimiter, logger::ERROR, import_export::CopyParams::escape, field(), anonymous_namespace{DelimitedParserUtils.cpp}::is_eol(), LOG, import_export::CopyParams::quote, import_export::CopyParams::quoted, anonymous_namespace{DelimitedParserUtils.cpp}::trim_quotes(), import_export::trim_space(), and import_export::CopyParams::trim_spaces.

Referenced by import_export::import_thread_delimited(), parse_string_array(), foreign_storage::CsvFileBufferParser::parseBuffer(), import_export::Detector::split_raw_data(), and foreign_storage::CsvFileBufferParser::validateExpectedColumnCount().

                                             {
  const char* field = buf;
  const char* p;
  bool in_quote = false;
  bool in_array = false;
  bool has_escape = false;
  bool strip_quotes = false;
  try_single_thread = false;
  for (p = buf; p < entire_buf_end; ++p) {
    if (*p == copy_params.escape && p < entire_buf_end - 1 &&
        *(p + 1) == copy_params.quote) {
      p++;
      has_escape = true;
    } else if (copy_params.quoted && *p == copy_params.quote) {
      in_quote = !in_quote;
      if (in_quote) {
        strip_quotes = true;
      }
    } else if (!in_quote && is_array != nullptr && *p == copy_params.array_begin &&
               is_array[row.size()]) {
      in_array = true;
      while (p < entire_buf_end - 1) {  // Array type will be parsed separately.
        ++p;
        if (*p == copy_params.array_end) {
          in_array = false;
          break;
        }
      }
    } else if (*p == copy_params.delimiter || is_eol(*p, copy_params)) {
      if (!in_quote) {
        if (!has_escape && !strip_quotes) {
          const char* field_end = p;
          if (copy_params.trim_spaces) {
            trim_space(field, field_end);
          }
          row.emplace_back(field, field_end - field);
        } else {
          tmp_buffers.emplace_back(std::make_unique<char[]>(p - field + 1));
          auto field_buf = tmp_buffers.back().get();
          int j = 0, i = 0;
          for (; i < p - field; i++, j++) {
            if (has_escape && field[i] == copy_params.escape &&
                field[i + 1] == copy_params.quote) {
              field_buf[j] = copy_params.quote;
              i++;
            } else {
              field_buf[j] = field[i];
            }
          }
          const char* field_begin = field_buf;
          const char* field_end = field_buf + j;
          if (copy_params.trim_spaces) {
            trim_space(field_begin, field_end);
          }
          trim_quotes(field_begin, field_end, copy_params);
          row.emplace_back(field_begin, field_end - field_begin);
        }
        field = p + 1;
        has_escape = false;
        strip_quotes = false;

        if (is_eol(*p, copy_params)) {
          // We are at the end of the row. Skip the line endings now.
          if (filter_empty_lines) {
            while (p + 1 < buf_end && is_eol(*(p + 1), copy_params)) {
              p++;
            }
          } else {
            // skip DOS carriage return line feed only
            if (p + 1 < buf_end && *p == '\r' && *(p + 1) == '\n') {
              p++;
            }
          }
          break;
        }
      }
    }
  }
  /*
  @TODO(wei) do error handling
  */
  if (in_quote) {
    LOG(ERROR) << "Unmatched quote.";
    try_single_thread = true;
  }
  if (in_array) {
    LOG(ERROR) << "Unmatched array.";
    try_single_thread = true;
  }
  return p;
}

Here is the call graph for this function:

Here is the caller graph for this function:

template const char* import_export::delimited_parser::get_row	(	const char *	buf,
		const char *	buf_end,
		const char *	entire_buf_end,
		const import_export::CopyParams &	copy_params,
		const bool *	is_array,
		std::vector< std::string > &	row,
		std::vector< std::unique_ptr< char[]>> &	tmp_buffers,
		bool &	try_single_thread,
		bool	filter_empty_lines
	)

template const char* import_export::delimited_parser::get_row	(	const char *	buf,
		const char *	buf_end,
		const char *	entire_buf_end,
		const import_export::CopyParams &	copy_params,
		const bool *	is_array,
		std::vector< std::string_view > &	row,
		std::vector< std::unique_ptr< char[]>> &	tmp_buffers,
		bool &	try_single_thread,
		bool	filter_empty_lines
	)

void import_export::delimited_parser::parse_string_array	(	const std::string &	s,
		const import_export::CopyParams &	copy_params,
		std::vector< std::string > &	string_vec,
		bool	truncate_values = false
	)

Parses given string array and inserts into given vector of strings.

Parameters

s	Given string array
copy_params	Copy params for the table.
string_vec	Given vector to be populated with parsed fields.

Definition at line 308 of file DelimitedParserUtils.cpp.

References import_export::CopyParams::array_begin, import_export::CopyParams::array_delim, import_export::CopyParams::array_end, import_export::CopyParams::delimiter, get_row(), StringDictionary::MAX_STRLEN, import_export::CopyParams::null_str, and to_string().

Referenced by import_export::TypedImportBuffer::add_value(), import_export::TypedImportBuffer::addDefaultValues(), RowToColumnLoader::convert_string_to_column(), and foreign_storage::anonymous_namespace{LogFileBufferParser.cpp}::create_map_from_arrays().

                                              {
  if (s == copy_params.null_str || s == "NULL" || s.size() < 1 || s.empty()) {
    return;
  }
  if (s[0] != copy_params.array_begin || s[s.size() - 1] != copy_params.array_end) {
    throw std::runtime_error("Malformed Array :" + s);
  }

  std::string row(s.c_str() + 1, s.length() - 2);
  if (row.empty()) {  // allow empty arrays
    return;
  }
  row.push_back('\n');

  bool try_single_thread = false;
  import_export::CopyParams array_params = copy_params;
  array_params.delimiter = copy_params.array_delim;
  std::vector<std::unique_ptr<char[]>> tmp_buffers;
  get_row(row.c_str(),
          row.c_str() + row.length(),
          row.c_str() + row.length(),
          array_params,
          nullptr,
          string_vec,
          tmp_buffers,
          try_single_thread,
          true);

  for (size_t i = 0; i < string_vec.size(); ++i) {
    if (string_vec[i].size() > StringDictionary::MAX_STRLEN) {
      if (truncate_values) {
        string_vec[i] = string_vec[i].substr(0, StringDictionary::MAX_STRLEN);
      } else {
        throw std::runtime_error("Array String too long : " + string_vec[i] + " max is " +
                                 std::to_string(StringDictionary::MAX_STRLEN));
      }
    }
  }

  // use empty string to mark nulls
  for (auto& value : string_vec) {
    if (value == copy_params.null_str || value == "NULL" || value.empty()) {
      value.clear();
    }
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

void import_export::delimited_parser::set_max_buffer_resize

(

const size_t

max_buffer_resize_param

)

Sets the maximum size to which thread buffers should be automatically resized. This function is only used for testing.

Definition at line 144 of file DelimitedParserUtils.cpp.

References max_buffer_resize.

                                                                 {
  max_buffer_resize = max_buffer_resize_param;
}

Variable Documentation

size_t import_export::delimited_parser::max_buffer_resize = max_import_buffer_resize_byte_size

static

Definition at line 138 of file DelimitedParserUtils.cpp.

Referenced by find_row_end_pos(), foreign_storage::RegexFileBufferParser::findRowEndPosition(), get_max_buffer_resize(), set_max_buffer_resize(), and foreign_storage::RegexFileBufferParser::setMaxBufferResize().