#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)

std::vector< SQLTypeInfo >	getBestColumnTypes () const

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, const Catalog_Namespace::SessionInfo *session_info)

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

Public Attributes
std::vector< std::vector < std::string > >	raw_rows

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, const Catalog_Namespace::SessionInfo *session_info) 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

std::vector< SQLTypes >	best_sqltypes

std::vector< EncodingType >	best_encodings

Additional Inherited Members
Protected Member Functions inherited from import_export::DataStreamSink
ImportStatus	archivePlumber (const Catalog_Namespace::SessionInfo *session_info)

Protected Attributes inherited from import_export::DataStreamSink
CopyParams	copy_params

const std::string	file_path

FILE *	p_file = nullptr

ImportStatus	import_status_

heavyai::shared_mutex	import_mutex_

size_t	total_file_size {0}

std::vector< size_t >	file_offsets

std::mutex	file_offsets_mutex

Detailed Description

Definition at line 729 of file Importer.h.

Constructor & Destructor Documentation

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

Definition at line 3670 of file Importer.cpp.

References g_enable_fsi, init(), import_export::kParquetFile, read_file(), and import_export::CopyParams::source_type.

     : DataStreamSink(cp, fp.string()), file_path(fp) {
 #ifdef ENABLE_IMPORT_PARQUET
   if (cp.source_type == import_export::SourceType::kParquetFile && g_enable_fsi &&
       !g_enable_legacy_parquet_import) {
     data_preview_ = get_parquet_data_preview(fp.string(), cp);
   } else
 #endif
   {
     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 3305 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 3449 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 3156 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::path(file_path).extension() == ".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 3216 of file Importer.cpp.

References dateTimeParseOptional< kDATE >(), dateTimeParseOptional< kTIME >(), dateTimeParseOptional< kTIMESTAMP >(), kBIGINT, kDATE, kDOUBLE, kFLOAT, kINT, kLINESTRING, kMULTILINESTRING, kMULTIPOINT, kMULTIPOLYGON, kPOINT, kPOLYGON, kSMALLINT, kTEXT, kTIME, kTIMESTAMP, import_export::PROMOTE_LINESTRING_TO_MULTILINESTRING, import_export::PROMOTE_POINT_TO_MULTIPOINT, import_export::PROMOTE_POLYGON_TO_MULTIPOLYGON, shared::transform(), 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) {
       // promote column type?
       type = PROMOTE_POINT_TO_MULTIPOINT ? kMULTIPOINT : kPOINT;
     } else if (str_upper_case.find("MULTIPOINT") == 0) {
       type = kMULTIPOINT;
     } else if (str_upper_case.find("LINESTRING") == 0) {
       // promote column type?
       type = PROMOTE_LINESTRING_TO_MULTILINESTRING ? kMULTILINESTRING : kLINESTRING;
     } else if (str_upper_case.find("MULTILINESTRING") == 0) {
       type = kMULTILINESTRING;
     } else if (str_upper_case.find("POLYGON") == 0) {
       // promote column type?
       type = PROMOTE_POLYGON_TO_MULTIPOLYGON ? kMULTIPOLYGON : 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 == "0000000004" || first_five_bytes == "0104000000") {
           type = kMULTIPOINT;
         } else if (first_five_bytes == "0000000002" || first_five_bytes == "0102000000") {
           type = kLINESTRING;
         } else if (first_five_bytes == "0000000005" || first_five_bytes == "0105000000") {
           type = kMULTILINESTRING;
         } 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) {
     // This won't match unix timestamp, since floats and ints were checked above.
     if (dateTimeParseOptional<kTIME>(str, 0)) {
       type = kTIME;
     } else if (dateTimeParseOptional<kTIMESTAMP>(str, 0)) {
       type = kTIMESTAMP;
     } else if (dateTimeParseOptional<kDATE>(str, 0)) {
       type = kDATE;
     }
   }
 
   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 3413 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::path(file_path).stem().string());
   }
   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 3360 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 3364 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 3370 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::path(file_path).stem().string());
   }
   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 3337 of file Importer.cpp.

References best_encodings, best_sqltypes, import_export::DataStreamSink::copy_params, detect_column_types(), detect_headers(), find_best_encodings(), find_best_sqltypes(), import_export::CopyParams::has_header, has_headers, import_export::kAutoDetect, import_export::kHasHeader, import_export::kNoHeader, 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::kAutoDetect:
       has_headers = detect_headers(head_types, best_sqltypes);
       if (has_headers) {
         copy_params.has_header = import_export::ImportHeaderRow::kHasHeader;
       } else {
         copy_params.has_header = import_export::ImportHeaderRow::kNoHeader;
       }
       break;
     case import_export::ImportHeaderRow::kNoHeader:
       has_headers = false;
       break;
     case import_export::ImportHeaderRow::kHasHeader:
       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 3479 of file Importer.cpp.

References best_sqltypes, has_headers, raw_rows, and to_string().

Referenced by DBHandler::detect_column_types().

                                            {
 #if defined(ENABLE_IMPORT_PARQUET)
   if (data_preview_.has_value()) {
     return data_preview_.value().column_names;
   } else
 #endif
   {
     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 3464 of file Importer.cpp.

References has_headers, anonymous_namespace{Utm.h}::n, and raw_rows.

Referenced by DBHandler::detect_column_types().

                                                                   {
 #if defined(ENABLE_IMPORT_PARQUET)
   if (data_preview_.has_value()) {
     return data_preview_.value().sample_rows;
   } else
 #endif
   {
     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:

std::vector< SQLTypeInfo > import_export::Detector::getBestColumnTypes ( ) const

Definition at line 3498 of file Importer.cpp.

References best_encodings, best_sqltypes, and CHECK_EQ.

Referenced by DBHandler::detect_column_types().

                                                           {
 #if defined(ENABLE_IMPORT_PARQUET)
   if (data_preview_.has_value()) {
     return data_preview_.value().column_types;
   } else
 #endif
   {
     std::vector<SQLTypeInfo> types;
     CHECK_EQ(best_sqltypes.size(), best_encodings.size());
     for (size_t i = 0; i < best_sqltypes.size(); i++) {
       types.emplace_back(best_sqltypes[i], false, best_encodings[i]);
     }
     return types;
   }
 }

Here is the caller graph for this function:

ImportStatus import_export::Detector::importDelimited	(	const std::string &	file_path,
		const bool	decompressed,
		const Catalog_Namespace::SessionInfo *	session_info
	)

overrideprivatevirtual

Implements import_export::DataStreamSink.

Definition at line 3086 of file Importer.cpp.

References import_export::DataStreamSink::copy_params, run_benchmark_import::end_time, heavyai::fopen(), import_export::DataStreamSink::import_mutex_, import_export::DataStreamSink::import_status_, import_export::kImportRowLimit, parse_ast::line, line1, import_export::CopyParams::line_delim, import_export::ImportStatus::load_failed, anonymous_namespace{Utm.h}::n, import_export::DataStreamSink::p_file, raw_data, import_export::ImportStatus::rows_completed, and timeout.

                                                       {
   // we do not check interrupt status for this detection
   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 >= kImportRowLimit) {
           // stop import when row limit reached
           break;
         }
       }
     }
   } catch (std::exception& e) {
   }
 
   heavyai::lock_guard<heavyai::shared_mutex> write_lock(import_mutex_);
   import_status_.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 3080 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 3313 of file Importer.cpp.

References anonymous_namespace{Utm.h}::a, kBIGINT, kBOOLEAN, kCHAR, kDATE, kDOUBLE, kFLOAT, kINT, kLINESTRING, kMULTILINESTRING, kMULTIPOINT, 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[kMULTIPOINT] = 11;
   typeorder[kLINESTRING] = 11;
   typeorder[kMULTILINESTRING] = 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 3151 of file Importer.cpp.

References import_export::DataStreamSink::archivePlumber().

Referenced by Detector().

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

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 3165 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,
                                                  true);
     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,
                                                    true);
       raw_rows.push_back(row);
     }
   }
 }