#include <Importer.h>

Inheritance diagram for import_export::DataStreamSink:

Collaboration diagram for import_export::DataStreamSink:

Public Member Functions
	DataStreamSink ()

	DataStreamSink (const CopyParams &copy_params, const std::string file_path)

virtual	~DataStreamSink ()

virtual ImportStatus	importDelimited (const std::string &file_path, const bool decompressed)=0

const CopyParams &	get_copy_params () const

void	import_compressed (std::vector< std::string > &file_paths)

Protected Member Functions
ImportStatus	archivePlumber ()

Protected Attributes
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 629 of file Importer.h.

Constructor & Destructor Documentation

import_export::DataStreamSink::DataStreamSink ( )

inline

Definition at line 631 of file Importer.h.

631 {}

import_export::DataStreamSink::DataStreamSink	(	const CopyParams &	copy_params,
		const std::string	file_path
	)

inline

Definition at line 632 of file Importer.h.

633 : copy_params(copy_params), file_path(file_path) {}

import_export::DataStreamSink::copy_params

CopyParams copy_params

Definition: Importer.h:647

import_export::DataStreamSink::file_path

const std::string file_path

Definition: Importer.h:648

virtual import_export::DataStreamSink::~DataStreamSink ( )

inlinevirtual

Definition at line 634 of file Importer.h.

634 {}

Member Function Documentation

ImportStatus import_export::DataStreamSink::archivePlumber ( )

protected

Definition at line 3299 of file Importer.cpp.

References copy_params, file_path, import_export::CopyParams::file_type, get_filesize(), omnisci::glob(), import_compressed(), import_status, and total_file_size.

Referenced by import_export::Importer::import(), and import_export::Detector::read_file().

                                             {
   // in generalized importing scheme, reaching here file_path may
   // contain a file path, a url or a wildcard of file paths.
   // see CopyTableStmt::execute.
   auto file_paths = omnisci::glob(file_path);
   if (file_paths.size() == 0) {
     file_paths.push_back(file_path);
   }
 
   // sum up sizes of all local files -- only for local files. if
   // file_path is a s3 url, sizes will be obtained via S3Archive.
   for (const auto& file_path : file_paths) {
     total_file_size += get_filesize(file_path);
   }
 
 #ifdef ENABLE_IMPORT_PARQUET
   // s3 parquet goes different route because the files do not use libarchive
   // but parquet api, and they need to landed like .7z files.
   //
   // note: parquet must be explicitly specified by a WITH parameter "parquet='true'",
   //       because for example spark sql users may specify a output url w/o file
   //       extension like this:
   //                df.write
   //                  .mode("overwrite")
   //                  .parquet("s3://bucket/folder/parquet/mydata")
   //       without the parameter, it means plain or compressed csv files.
   // note: .ORC and AVRO files should follow a similar path to Parquet?
   if (copy_params.file_type == FileType::PARQUET) {
     import_parquet(file_paths);
   } else
 #endif
   {
     import_compressed(file_paths);
   }
   return import_status;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

const CopyParams& import_export::DataStreamSink::get_copy_params ( ) const

inline

Definition at line 641 of file Importer.h.

References copy_params.

Referenced by DBHandler::detect_column_types().

641 { return copy_params; }

import_export::DataStreamSink::copy_params

CopyParams copy_params

Definition: Importer.h:647

Here is the caller graph for this function:

void import_export::DataStreamSink::import_compressed ( std::vector< std::string > & file_paths )

Definition at line 3698 of file Importer.cpp.

References omnisci::close(), copy_params, file_offsets, file_offsets_mutex, import_export::CopyParams::has_header, import_status, importDelimited(), import_export::CopyParams::line_delim, import_export::ImportStatus::load_truncated, LOG, import_export::NO_HEADER, p_file, Archive::parse_url(), import_export::CopyParams::plain_text, import_export::ImportStatus::rows_completed, import_export::CopyParams::s3_access_key, import_export::CopyParams::s3_endpoint, import_export::CopyParams::s3_region, import_export::CopyParams::s3_secret_key, import_export::status_mutex, total_file_size, logger::WARNING, and File_Namespace::write().

Referenced by archivePlumber().

                                                                        {
 #ifdef _MSC_VER
   throw std::runtime_error("CSV Import not yet supported on Windows.");
 #else
   // a new requirement is to have one single input stream into
   // Importer::importDelimited, so need to move pipe related
   // stuff to the outmost block.
   int fd[2];
   if (pipe(fd) < 0) {
     throw std::runtime_error(std::string("failed to create a pipe: ") + strerror(errno));
   }
   signal(SIGPIPE, SIG_IGN);
 
   std::exception_ptr teptr;
   // create a thread to read uncompressed byte stream out of pipe and
   // feed into importDelimited()
   ImportStatus ret;
   auto th_pipe_reader = std::thread([&]() {
     try {
       // importDelimited will read from FILE* p_file
       if (0 == (p_file = fdopen(fd[0], "r"))) {
         throw std::runtime_error(std::string("failed to open a pipe: ") +
                                  strerror(errno));
       }
 
       // in future, depending on data types of this uncompressed stream
       // it can be feed into other function such like importParquet, etc
       ret = importDelimited(file_path, true);
     } catch (...) {
       if (!teptr) {  // no replace
         teptr = std::current_exception();
       }
     }
 
     if (p_file) {
       fclose(p_file);
     }
     p_file = 0;
   });
 
   // create a thread to iterate all files (in all archives) and
   // forward the uncompressed byte stream to fd[1] which is
   // then feed into importDelimited, importParquet, and etc.
   auto th_pipe_writer = std::thread([&]() {
     std::unique_ptr<S3Archive> us3arch;
     bool stop = false;
     for (size_t fi = 0; !stop && fi < file_paths.size(); fi++) {
       try {
         auto file_path = file_paths[fi];
         std::unique_ptr<Archive> uarch;
         std::map<int, std::string> url_parts;
         Archive::parse_url(file_path, url_parts);
         const std::string S3_objkey_url_scheme = "s3ok";
         if ("file" == url_parts[2] || "" == url_parts[2]) {
           uarch.reset(new PosixFileArchive(file_path, copy_params.plain_text));
         } else if ("s3" == url_parts[2]) {
 #ifdef HAVE_AWS_S3
           // new a S3Archive with a shared s3client.
           // should be safe b/c no wildcard with s3 url
           us3arch.reset(new S3Archive(file_path,
                                       copy_params.s3_access_key,
                                       copy_params.s3_secret_key,
                                       copy_params.s3_region,
                                       copy_params.s3_endpoint,
                                       copy_params.plain_text));
           us3arch->init_for_read();
           total_file_size += us3arch->get_total_file_size();
           // not land all files here but one by one in following iterations
           for (const auto& objkey : us3arch->get_objkeys()) {
             file_paths.emplace_back(std::string(S3_objkey_url_scheme) + "://" + objkey);
           }
           continue;
 #else
           throw std::runtime_error("AWS S3 support not available");
 #endif  // HAVE_AWS_S3
         } else if (S3_objkey_url_scheme == url_parts[2]) {
 #ifdef HAVE_AWS_S3
           auto objkey = file_path.substr(3 + S3_objkey_url_scheme.size());
           auto file_path =
               us3arch->land(objkey, teptr, nullptr != dynamic_cast<Detector*>(this));
           if (0 == file_path.size()) {
             throw std::runtime_error(std::string("failed to land s3 object: ") + objkey);
           }
           uarch.reset(new PosixFileArchive(file_path, copy_params.plain_text));
           // file not removed until file closed
           us3arch->vacuum(objkey);
 #else
           throw std::runtime_error("AWS S3 support not available");
 #endif  // HAVE_AWS_S3
         }
 #if 0   // TODO(ppan): implement and enable any other archive class
         else
         if ("hdfs" == url_parts[2])
           uarch.reset(new HdfsArchive(file_path));
 #endif
         else {
           throw std::runtime_error(std::string("unsupported archive url: ") + file_path);
         }
 
         // init the archive for read
         auto& arch = *uarch;
 
         // coming here, the archive of url should be ready to be read, unarchived
         // and uncompressed by libarchive into a byte stream (in csv) for the pipe
         const void* buf;
         size_t size;
         bool just_saw_archive_header;
         bool is_detecting = nullptr != dynamic_cast<Detector*>(this);
         bool first_text_header_skipped = false;
         // start reading uncompressed bytes of this archive from libarchive
         // note! this archive may contain more than one files!
         file_offsets.push_back(0);
         while (!stop && !!(just_saw_archive_header = arch.read_next_header())) {
           bool insert_line_delim_after_this_file = false;
           while (!stop) {
             int64_t offset{-1};
             auto ok = arch.read_data_block(&buf, &size, &offset);
             // can't use (uncompressed) size, so track (max) file offset.
             // also we want to capture offset even on e.o.f.
             if (offset > 0) {
               std::unique_lock<std::mutex> lock(file_offsets_mutex);
               file_offsets.back() = offset;
             }
             if (!ok) {
               break;
             }
             // one subtle point here is now we concatenate all files
             // to a single FILE stream with which we call importDelimited
             // only once. this would make it misunderstand that only one
             // header line is with this 'single' stream, while actually
             // we may have one header line for each of the files.
             // so we need to skip header lines here instead in importDelimited.
             const char* buf2 = (const char*)buf;
             int size2 = size;
             if (copy_params.has_header != import_export::ImportHeaderRow::NO_HEADER &&
                 just_saw_archive_header && (first_text_header_skipped || !is_detecting)) {
               while (size2-- > 0) {
                 if (*buf2++ == copy_params.line_delim) {
                   break;
                 }
               }
               if (size2 <= 0) {
                 LOG(WARNING) << "No line delimiter in block." << std::endl;
               } else {
                 just_saw_archive_header = false;
                 first_text_header_skipped = true;
               }
             }
             // In very rare occasions the write pipe somehow operates in a mode similar to
             // non-blocking while pipe(fds) should behave like pipe2(fds, 0) which means
             // blocking mode. On such a unreliable blocking mode, a possible fix is to
             // loop reading till no bytes left, otherwise the annoying `failed to write
             // pipe: Success`...
             if (size2 > 0) {
               int nremaining = size2;
               while (nremaining > 0) {
                 // try to write the entire remainder of the buffer to the pipe
                 int nwritten = write(fd[1], buf2, nremaining);
                 // how did we do?
                 if (nwritten < 0) {
                   // something bad happened
                   if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {
                     // ignore these, assume nothing written, try again
                     nwritten = 0;
                   } else {
                     // a real error
                     throw std::runtime_error(
                         std::string("failed or interrupted write to pipe: ") +
                         strerror(errno));
                   }
                 } else if (nwritten == nremaining) {
                   // we wrote everything; we're done
                   break;
                 }
                 // only wrote some (or nothing), try again
                 nremaining -= nwritten;
                 buf2 += nwritten;
                 // no exception when too many rejected
                 // @simon.eves how would this get set? from the other thread? mutex
                 // needed?
                 if (import_status.load_truncated) {
                   stop = true;
                   break;
                 }
               }
               // check that this file (buf for size) ended with a line delim
               if (size > 0) {
                 const char* plast = static_cast<const char*>(buf) + (size - 1);
                 insert_line_delim_after_this_file = (*plast != copy_params.line_delim);
               }
             }
           }
           // if that file didn't end with a line delim, we insert one here to terminate
           // that file's stream use a loop for the same reason as above
           if (insert_line_delim_after_this_file) {
             while (true) {
               // write the delim char to the pipe
               int nwritten = write(fd[1], &copy_params.line_delim, 1);
               // how did we do?
               if (nwritten < 0) {
                 // something bad happened
                 if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {
                   // ignore these, assume nothing written, try again
                   nwritten = 0;
                 } else {
                   // a real error
                   throw std::runtime_error(
                       std::string("failed or interrupted write to pipe: ") +
                       strerror(errno));
                 }
               } else if (nwritten == 1) {
                 // we wrote it; we're done
                 break;
               }
             }
           }
         }
       } catch (...) {
         // when import is aborted because too many data errors or because end of a
         // detection, any exception thrown by s3 sdk or libarchive is okay and should be
         // suppressed.
         mapd_shared_lock<mapd_shared_mutex> read_lock(status_mutex);
         if (import_status.load_truncated) {
           break;
         }
         if (import_status.rows_completed > 0) {
           if (nullptr != dynamic_cast<Detector*>(this)) {
             break;
           }
         }
         if (!teptr) {  // no replace
           teptr = std::current_exception();
         }
         break;
       }
     }
     // close writer end
     close(fd[1]);
   });
 
   th_pipe_reader.join();
   th_pipe_writer.join();
 
   // rethrow any exception happened herebefore
   if (teptr) {
     std::rethrow_exception(teptr);
   }
 #endif
 }