import_export::Detector Class Reference

#include <Importer.h>

Inheritance diagram for import_export::Detector:

Collaboration diagram for import_export::Detector:

Public Member Functions
	Detector (const boost::filesystem::path &fp, CopyParams &cp)
std::vector< std::string >	get_headers ()
std::vector< std::vector< std::string > >	get_sample_rows (size_t n)
Public Member Functions inherited from import_export::DataStreamSink
	DataStreamSink ()
	DataStreamSink (const CopyParams &copy_params, const std::string file_path)
virtual	~DataStreamSink ()
const CopyParams &	get_copy_params () const
void	import_compressed (std::vector< std::string > &file_paths)

Static Public Member Functions
static SQLTypes	detect_sqltype (const std::string &str)

Public Attributes
std::vector< std::vector< std::string > >	raw_rows
std::vector< SQLTypes >	best_sqltypes
std::vector< EncodingType >	best_encodings
bool	has_headers = false

Private Member Functions
void	init ()
void	read_file ()
void	detect_row_delimiter ()
void	split_raw_data ()
std::vector< SQLTypes >	detect_column_types (const std::vector< std::string > &row)
void	find_best_sqltypes ()
std::vector< SQLTypes >	find_best_sqltypes (const std::vector< std::vector< std::string >> &raw_rows, const CopyParams &copy_params)
std::vector< SQLTypes >	find_best_sqltypes (const std::vector< std::vector< std::string >>::const_iterator &row_begin, const std::vector< std::vector< std::string >>::const_iterator &row_end, const CopyParams &copy_params)
std::vector< EncodingType >	find_best_encodings (const std::vector< std::vector< std::string >>::const_iterator &row_begin, const std::vector< std::vector< std::string >>::const_iterator &row_end, const std::vector< SQLTypes > &best_types)
bool	detect_headers (const std::vector< SQLTypes > &first_types, const std::vector< SQLTypes > &rest_types)
void	find_best_sqltypes_and_headers ()
ImportStatus	importDelimited (const std::string &file_path, const bool decompressed) override

Static Private Member Functions
static bool	more_restrictive_sqltype (const SQLTypes a, const SQLTypes b)

Private Attributes
std::string	raw_data
boost::filesystem::path	file_path
std::chrono::duration< double >	timeout {1}
std::string	line1

Additional Inherited Members
Protected Member Functions inherited from import_export::DataStreamSink
ImportStatus	archivePlumber ()
Protected Attributes inherited from import_export::DataStreamSink
CopyParams	copy_params
const std::string	file_path
FILE *	p_file = nullptr
ImportStatus	import_status
bool	load_failed = false
size_t	total_file_size {0}
std::vector< size_t >	file_offsets
std::mutex	file_offsets_mutex

Detailed Description

Definition at line 656 of file Importer.h.

Constructor & Destructor Documentation

Detector()

import_export::Detector::Detector ( const boost::filesystem::path &  fp, CopyParams &  cp  )

inline

Definition at line 658 of file Importer.h.

References logger::init().

      : DataStreamSink(cp, fp.string()), file_path(fp) {
    read_file();
    init();
  };

Here is the call graph for this function:

Member Function Documentation

detect_column_types()

std::vector< SQLTypes > import_export::Detector::detect_column_types ( const std::vector< std::string > &  row )

private

Definition at line 3085 of file Importer.cpp.

                                                                                 {
  std::vector<SQLTypes> types(row.size());
  for (size_t i = 0; i < row.size(); i++) {
    types[i] = detect_sqltype(row[i]);
  }
  return types;
}

detect_headers()

bool import_export::Detector::detect_headers ( const std::vector< SQLTypes > &  first_types, const std::vector< SQLTypes > &  rest_types  )

private

Definition at line 3227 of file Importer.cpp.

References kTEXT.

                                                                     {
  if (head_types.size() != tail_types.size()) {
    return false;
  }
  bool has_headers = false;
  for (size_t col_idx = 0; col_idx < tail_types.size(); col_idx++) {
    if (head_types[col_idx] != kTEXT) {
      return false;
    }
    has_headers = has_headers || tail_types[col_idx] != kTEXT;
  }
  return has_headers;
}

detect_row_delimiter()

void import_export::Detector::detect_row_delimiter ( )

private

Definition at line 2933 of file Importer.cpp.

References import_export::DataStreamSink::copy_params, import_export::CopyParams::delimiter, and import_export::DataStreamSink::file_path.

                                    {
  if (copy_params.delimiter == '\0') {
    copy_params.delimiter = ',';
    if (boost::filesystem::extension(file_path) == ".tsv") {
      copy_params.delimiter = '\t';
    }
  }
}

detect_sqltype()

SQLTypes import_export::Detector::detect_sqltype ( const std::string &  str )

static

Definition at line 2991 of file Importer.cpp.

References kBIGINT, kDATE, kDOUBLE, kFLOAT, kINT, kLINESTRING, kMULTIPOLYGON, kPOINT, kPOLYGON, kSMALLINT, kTEXT, kTIME, kTIMESTAMP, import_export::try_strptimes(), and run_benchmark_import::type.

                                                      {
  SQLTypes type = kTEXT;
  if (try_cast<double>(str)) {
    type = kDOUBLE;
    /*if (try_cast<bool>(str)) {
      type = kBOOLEAN;
    }*/
    if (try_cast<int16_t>(str)) {
      type = kSMALLINT;
    } else if (try_cast<int32_t>(str)) {
      type = kINT;
    } else if (try_cast<int64_t>(str)) {
      type = kBIGINT;
    } else if (try_cast<float>(str)) {
      type = kFLOAT;
    }
  }

  // check for geo types
  if (type == kTEXT) {
    // convert to upper case
    std::string str_upper_case = str;
    std::transform(
        str_upper_case.begin(), str_upper_case.end(), str_upper_case.begin(), ::toupper);

    // then test for leading words
    if (str_upper_case.find("POINT") == 0) {
      type = kPOINT;
    } else if (str_upper_case.find("LINESTRING") == 0) {
      type = kLINESTRING;
    } else if (str_upper_case.find("POLYGON") == 0) {
      if (PROMOTE_POLYGON_TO_MULTIPOLYGON) {
        type = kMULTIPOLYGON;
      } else {
        type = kPOLYGON;
      }
    } else if (str_upper_case.find("MULTIPOLYGON") == 0) {
      type = kMULTIPOLYGON;
    } else if (str_upper_case.find_first_not_of("0123456789ABCDEF") ==
                   std::string::npos &&
               (str_upper_case.size() % 2) == 0) {
      // simple hex blob (two characters per byte, not uu-encode or base64)
      if (str_upper_case.size() >= 10) {
        // match WKB blobs for supported geometry types
        // the first byte specifies if the data is big-endian or little-endian
        // the next four bytes are the geometry type (1 = POINT etc.)
        // @TODO support eWKB, which has extra bits set in the geometry type
        auto first_five_bytes = str_upper_case.substr(0, 10);
        if (first_five_bytes == "0000000001" || first_five_bytes == "0101000000") {
          type = kPOINT;
        } else if (first_five_bytes == "0000000002" || first_five_bytes == "0102000000") {
          type = kLINESTRING;
        } else if (first_five_bytes == "0000000003" || first_five_bytes == "0103000000") {
          type = kPOLYGON;
        } else if (first_five_bytes == "0000000006" || first_five_bytes == "0106000000") {
          type = kMULTIPOLYGON;
        } else {
          // unsupported WKB type
          return type;
        }
      } else {
        // too short to be WKB
        return type;
      }
    }
  }

  // check for time types
  if (type == kTEXT) {
    // @TODO
    // make these tests more robust so they don't match stuff they should not
    char* buf;
    buf = try_strptimes(str.c_str(),
                        {"%Y-%m-%d", "%m/%d/%Y", "%Y/%m/%d", "%d-%b-%y", "%d/%b/%Y"});
    if (buf) {
      type = kDATE;
      if (*buf == 'T' || *buf == ' ' || *buf == ':') {
        buf++;
      }
    }
    buf = try_strptimes(buf == nullptr ? str.c_str() : buf,
                        {"%T %z", "%T", "%H%M%S", "%R"});
    if (buf) {
      if (type == kDATE) {
        type = kTIMESTAMP;
      } else {
        type = kTIME;
      }
    }
  }

  return type;
}

Here is the call graph for this function:

find_best_encodings()

std::vector< EncodingType > import_export::Detector::find_best_encodings ( const std::vector< std::vector< std::string >>::const_iterator &  row_begin, const std::vector< std::vector< std::string >>::const_iterator &  row_end, const std::vector< SQLTypes > &  best_types  )

private

Definition at line 3191 of file Importer.cpp.

References import_export::DataStreamSink::file_path, IS_STRING, kENCODING_DICT, and kENCODING_NONE.

                                           {
  if (raw_rows.size() < 1) {
    throw std::runtime_error("No rows found in: " +
                             boost::filesystem::basename(file_path));
  }
  size_t num_cols = best_types.size();
  std::vector<EncodingType> best_encodes(num_cols, kENCODING_NONE);
  std::vector<size_t> num_rows_per_col(num_cols, 1);
  std::vector<std::unordered_set<std::string>> count_set(num_cols);
  for (auto row = row_begin; row != row_end; row++) {
    for (size_t col_idx = 0; col_idx < row->size() && col_idx < num_cols; col_idx++) {
      if (IS_STRING(best_types[col_idx])) {
        count_set[col_idx].insert(row->at(col_idx));
        num_rows_per_col[col_idx]++;
      }
    }
  }
  for (size_t col_idx = 0; col_idx < num_cols; col_idx++) {
    if (IS_STRING(best_types[col_idx])) {
      float uniqueRatio =
          static_cast<float>(count_set[col_idx].size()) / num_rows_per_col[col_idx];
      if (uniqueRatio < 0.75) {
        best_encodes[col_idx] = kENCODING_DICT;
      }
    }
  }
  return best_encodes;
}

find_best_sqltypes() [1/3]

void import_export::Detector::find_best_sqltypes ( )

private

Definition at line 3138 of file Importer.cpp.

References import_export::DataStreamSink::copy_params.

                                  {
  best_sqltypes = find_best_sqltypes(raw_rows.begin(), raw_rows.end(), copy_params);
}

find_best_sqltypes() [2/3]

std::vector< SQLTypes > import_export::Detector::find_best_sqltypes ( const std::vector< std::vector< std::string >> &  raw_rows, const CopyParams &  copy_params  )

private

Definition at line 3142 of file Importer.cpp.

References import_export::DataStreamSink::copy_params.

                                   {
  return find_best_sqltypes(raw_rows.begin(), raw_rows.end(), copy_params);
}

find_best_sqltypes() [3/3]

std::vector< SQLTypes > import_export::Detector::find_best_sqltypes ( const std::vector< std::vector< std::string >>::const_iterator &  row_begin, const std::vector< std::vector< std::string >>::const_iterator &  row_end, const CopyParams &  copy_params  )

private

Definition at line 3148 of file Importer.cpp.

References run_benchmark_import::end_time, import_export::DataStreamSink::file_path, kCHAR, kTEXT, and import_export::CopyParams::null_str.

                                   {
  if (raw_rows.size() < 1) {
    throw std::runtime_error("No rows found in: " +
                             boost::filesystem::basename(file_path));
  }
  auto end_time = std::chrono::steady_clock::now() + timeout;
  size_t num_cols = raw_rows.front().size();
  std::vector<SQLTypes> best_types(num_cols, kCHAR);
  std::vector<size_t> non_null_col_counts(num_cols, 0);
  for (auto row = row_begin; row != row_end; row++) {
    while (best_types.size() < row->size() || non_null_col_counts.size() < row->size()) {
      best_types.push_back(kCHAR);
      non_null_col_counts.push_back(0);
    }
    for (size_t col_idx = 0; col_idx < row->size(); col_idx++) {
      // do not count nulls
      if (row->at(col_idx) == "" || !row->at(col_idx).compare(copy_params.null_str)) {
        continue;
      }
      SQLTypes t = detect_sqltype(row->at(col_idx));
      non_null_col_counts[col_idx]++;
      if (!more_restrictive_sqltype(best_types[col_idx], t)) {
        best_types[col_idx] = t;
      }
    }
    if (std::chrono::steady_clock::now() > end_time) {
      break;
    }
  }
  for (size_t col_idx = 0; col_idx < num_cols; col_idx++) {
    // if we don't have any non-null values for this column make it text to be
    // safe b/c that is least restrictive type
    if (non_null_col_counts[col_idx] == 0) {
      best_types[col_idx] = kTEXT;
    }
  }

  return best_types;
}

find_best_sqltypes_and_headers()

void import_export::Detector::find_best_sqltypes_and_headers ( )

private

Definition at line 3115 of file Importer.cpp.

References import_export::AUTODETECT, import_export::DataStreamSink::copy_params, import_export::HAS_HEADER, import_export::CopyParams::has_header, and import_export::NO_HEADER.

                                              {
  best_sqltypes = find_best_sqltypes(raw_rows.begin() + 1, raw_rows.end(), copy_params);
  best_encodings =
      find_best_encodings(raw_rows.begin() + 1, raw_rows.end(), best_sqltypes);
  std::vector<SQLTypes> head_types = detect_column_types(raw_rows.at(0));
  switch (copy_params.has_header) {
    case import_export::ImportHeaderRow::AUTODETECT:
      has_headers = detect_headers(head_types, best_sqltypes);
      if (has_headers) {
        copy_params.has_header = import_export::ImportHeaderRow::HAS_HEADER;
      } else {
        copy_params.has_header = import_export::ImportHeaderRow::NO_HEADER;
      }
      break;
    case import_export::ImportHeaderRow::NO_HEADER:
      has_headers = false;
      break;
    case import_export::ImportHeaderRow::HAS_HEADER:
      has_headers = true;
      break;
  }
}

get_headers()

std::vector< std::string > import_export::Detector::get_headers

(

)

Definition at line 3250 of file Importer.cpp.

References to_string().

Referenced by DBHandler::detect_column_types().

                                           {
  std::vector<std::string> headers(best_sqltypes.size());
  for (size_t i = 0; i < best_sqltypes.size(); i++) {
    if (has_headers && i < raw_rows[0].size()) {
      headers[i] = raw_rows[0][i];
    } else {
      headers[i] = "column_" + std::to_string(i + 1);
    }
  }
  return headers;
}

Here is the call graph for this function:

Here is the caller graph for this function:

get_sample_rows()

std::vector< std::vector< std::string > > import_export::Detector::get_sample_rows

(

size_t

n

)

Definition at line 3242 of file Importer.cpp.

Referenced by DBHandler::detect_column_types().

                                                                  {
  n = std::min(n, raw_rows.size());
  size_t offset = (has_headers && raw_rows.size() > 1) ? 1 : 0;
  std::vector<std::vector<std::string>> sample_rows(raw_rows.begin() + offset,
                                                    raw_rows.begin() + n);
  return sample_rows;
}

Here is the caller graph for this function:

importDelimited()

ImportStatus import_export::Detector::importDelimited ( const std::string &  file_path, const bool  decompressed  )

overrideprivatevirtual

Implements import_export::DataStreamSink.

Definition at line 2866 of file Importer.cpp.

References import_export::DataStreamSink::copy_params, run_benchmark_import::end_time, import_export::DataStreamSink::import_status, import_export::CopyParams::line_delim, import_export::DataStreamSink::load_failed, import_export::ImportStatus::load_truncated, import_export::DataStreamSink::p_file, and import_export::ImportStatus::rows_completed.

                                                                {
  if (!p_file) {
    p_file = fopen(file_path.c_str(), "rb");
  }
  if (!p_file) {
    throw std::runtime_error("failed to open file '" + file_path +
                             "': " + strerror(errno));
  }

  // somehow clang does not support ext/stdio_filebuf.h, so
  // need to diy readline with customized copy_params.line_delim...
  std::string line;
  line.reserve(1 * 1024 * 1024);
  auto end_time = std::chrono::steady_clock::now() +
                  timeout * (boost::istarts_with(file_path, "s3://") ? 3 : 1);
  try {
    while (!feof(p_file)) {
      int c;
      size_t n = 0;
      while (EOF != (c = fgetc(p_file)) && copy_params.line_delim != c) {
        if (n++ >= line.capacity()) {
          break;
        }
        line += c;
      }
      if (0 == n) {
        break;
      }
      // remember the first line, which is possibly a header line, to
      // ignore identical header line(s) in 2nd+ files of a archive;
      // otherwise, 2nd+ header may be mistaken as an all-string row
      // and so be final column types.
      if (line1.empty()) {
        line1 = line;
      } else if (line == line1) {
        line.clear();
        continue;
      }

      raw_data += line;
      raw_data += copy_params.line_delim;
      line.clear();
      ++import_status.rows_completed;
      if (std::chrono::steady_clock::now() > end_time) {
        if (import_status.rows_completed > 10000) {
          break;
        }
      }
    }
  } catch (std::exception& e) {
  }

  // as if load truncated
  import_status.load_truncated = true;
  load_failed = true;

  fclose(p_file);
  p_file = nullptr;
  return import_status;
}

init()

void import_export::Detector::init ( )

private

Definition at line 2860 of file Importer.cpp.

                    {
  detect_row_delimiter();
  split_raw_data();
  find_best_sqltypes_and_headers();
}

more_restrictive_sqltype()

bool import_export::Detector::more_restrictive_sqltype ( const SQLTypes  a, const SQLTypes  b  )

staticprivate

Definition at line 3093 of file Importer.cpp.

References kBIGINT, kBOOLEAN, kCHAR, kDATE, kDOUBLE, kFLOAT, kINT, kLINESTRING, kMULTIPOLYGON, kPOINT, kPOLYGON, kSMALLINT, kTEXT, kTIME, and kTIMESTAMP.

                                                                          {
  static std::array<int, kSQLTYPE_LAST> typeorder;
  typeorder[kCHAR] = 0;
  typeorder[kBOOLEAN] = 2;
  typeorder[kSMALLINT] = 3;
  typeorder[kINT] = 4;
  typeorder[kBIGINT] = 5;
  typeorder[kFLOAT] = 6;
  typeorder[kDOUBLE] = 7;
  typeorder[kTIMESTAMP] = 8;
  typeorder[kTIME] = 9;
  typeorder[kDATE] = 10;
  typeorder[kPOINT] = 11;
  typeorder[kLINESTRING] = 11;
  typeorder[kPOLYGON] = 11;
  typeorder[kMULTIPOLYGON] = 11;
  typeorder[kTEXT] = 12;

  // note: b < a instead of a < b because the map is ordered most to least restrictive
  return typeorder[b] < typeorder[a];
}

read_file()

void import_export::Detector::read_file ( )

private

Definition at line 2928 of file Importer.cpp.

References import_export::DataStreamSink::archivePlumber().

                         {
  // this becomes analogous to Importer::import()
  (void)DataStreamSink::archivePlumber();
}

Here is the call graph for this function:

split_raw_data()

void import_export::Detector::split_raw_data ( )

private

Definition at line 2942 of file Importer.cpp.

References import_export::DataStreamSink::copy_params, import_export::delimited_parser::get_row(), and import_export::CopyParams::threads.

                              {
  const char* buf = raw_data.c_str();
  const char* buf_end = buf + raw_data.size();
  bool try_single_thread = false;
  for (const char* p = buf; p < buf_end; p++) {
    std::vector<std::string> row;
    std::vector<std::unique_ptr<char[]>> tmp_buffers;
    p = import_export::delimited_parser::get_row(
        p, buf_end, buf_end, copy_params, nullptr, row, tmp_buffers, try_single_thread);
    raw_rows.push_back(row);
    if (try_single_thread) {
      break;
    }
  }
  if (try_single_thread) {
    copy_params.threads = 1;
    raw_rows.clear();
    for (const char* p = buf; p < buf_end; p++) {
      std::vector<std::string> row;
      std::vector<std::unique_ptr<char[]>> tmp_buffers;
      p = import_export::delimited_parser::get_row(
          p, buf_end, buf_end, copy_params, nullptr, row, tmp_buffers, try_single_thread);
      raw_rows.push_back(row);
    }
  }
}

Here is the call graph for this function:

Member Data Documentation

best_encodings

std::vector<EncodingType> import_export::Detector::best_encodings

Definition at line 671 of file Importer.h.

Referenced by DBHandler::detect_column_types().

best_sqltypes

std::vector<SQLTypes> import_export::Detector::best_sqltypes

Definition at line 670 of file Importer.h.

Referenced by DBHandler::detect_column_types().

file_path

boost::filesystem::path import_export::Detector::file_path

private

Definition at line 701 of file Importer.h.

has_headers

bool import_export::Detector::has_headers = false

Definition at line 672 of file Importer.h.

line1

std::string import_export::Detector::line1

private

Definition at line 703 of file Importer.h.

raw_data

std::string import_export::Detector::raw_data

private

Definition at line 700 of file Importer.h.

raw_rows

std::vector<std::vector<std::string> > import_export::Detector::raw_rows

Definition at line 668 of file Importer.h.

timeout

std::chrono::duration<double> import_export::Detector::timeout {1}

private

Definition at line 702 of file Importer.h.

---

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

• /home/jenkins-slave/workspace/core-os-doxygen/ImportExport/Importer.h
• /home/jenkins-slave/workspace/core-os-doxygen/ImportExport/Importer.cpp