#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 657 of file Importer.h.

Constructor & Destructor Documentation

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

inline

Definition at line 659 of file Importer.h.

References init(), and read_file().

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

Here is the call graph for this function:

Member Function Documentation

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

private

Definition at line 3083 of file Importer.cpp.

References detect_sqltype().

Referenced by find_best_sqltypes_and_headers().

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

Here is the call graph for this function:

Here is the caller graph for this function:

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

private

Definition at line 3225 of file Importer.cpp.

References has_headers, and kTEXT.

Referenced by find_best_sqltypes_and_headers().

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

Here is the caller graph for this function:

void import_export::Detector::detect_row_delimiter ( )

private

Definition at line 2931 of file Importer.cpp.

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

Referenced by init().

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

Here is the caller graph for this function:

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

static

Definition at line 2989 of file Importer.cpp.

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

Referenced by detect_column_types(), and find_best_sqltypes().

                                                       {
   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:

Here is the caller graph for this function:

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 3189 of file Importer.cpp.

References file_path, IS_STRING, kENCODING_DICT, kENCODING_NONE, and raw_rows.

Referenced by find_best_sqltypes_and_headers().

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

Here is the caller graph for this function:

void import_export::Detector::find_best_sqltypes ( )

private

Definition at line 3136 of file Importer.cpp.

References best_sqltypes, import_export::DataStreamSink::copy_params, and raw_rows.

Referenced by find_best_sqltypes(), and find_best_sqltypes_and_headers().

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

Here is the caller graph for this function:

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 3140 of file Importer.cpp.

References import_export::DataStreamSink::copy_params, find_best_sqltypes(), and raw_rows.

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

Here is the call graph for this function:

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 3146 of file Importer.cpp.

References detect_sqltype(), run_benchmark_import::end_time, file_path, kCHAR, kTEXT, more_restrictive_sqltype(), import_export::CopyParams::null_str, raw_rows, and timeout.

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

Here is the call graph for this function:

void import_export::Detector::find_best_sqltypes_and_headers ( )

private

Definition at line 3113 of file Importer.cpp.

References import_export::AUTODETECT, best_encodings, best_sqltypes, import_export::DataStreamSink::copy_params, detect_column_types(), detect_headers(), find_best_encodings(), find_best_sqltypes(), import_export::HAS_HEADER, import_export::CopyParams::has_header, has_headers, import_export::NO_HEADER, and raw_rows.

Referenced by init().

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

Here is the call graph for this function:

Here is the caller graph for this function:

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

Definition at line 3248 of file Importer.cpp.

References best_sqltypes, has_headers, raw_rows, and 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:

std::vector< std::vector< std::string > > import_export::Detector::get_sample_rows ( size_t n )

Definition at line 3240 of file Importer.cpp.

References has_headers, and raw_rows.

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:

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

overrideprivatevirtual

Implements import_export::DataStreamSink.

Definition at line 2864 of file Importer.cpp.

References import_export::DataStreamSink::copy_params, run_benchmark_import::end_time, omnisci::fopen(), import_export::DataStreamSink::import_status, parse_ast::line, line1, import_export::CopyParams::line_delim, import_export::DataStreamSink::load_failed, import_export::ImportStatus::load_truncated, import_export::DataStreamSink::p_file, raw_data, import_export::ImportStatus::rows_completed, and timeout.

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

Here is the call graph for this function:

void import_export::Detector::init ( )

private

Definition at line 2858 of file Importer.cpp.

References detect_row_delimiter(), find_best_sqltypes_and_headers(), and split_raw_data().

Referenced by Detector().

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

Here is the call graph for this function:

Here is the caller graph for this function:

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

staticprivate

Definition at line 3091 of file Importer.cpp.

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

Referenced by find_best_sqltypes().

                                                                           {
   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];
 }

Here is the caller graph for this function:

void import_export::Detector::read_file ( )

private

Definition at line 2926 of file Importer.cpp.

References import_export::DataStreamSink::archivePlumber().

Referenced by Detector().

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

Here is the call graph for this function:

Here is the caller graph for this function:

void import_export::Detector::split_raw_data ( )

private

Definition at line 2940 of file Importer.cpp.

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

Referenced by init().

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