import_export::DataStreamSink Class Reference

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

Constructor & Destructor Documentation

DataStreamSink() [1/2]

import_export::DataStreamSink::DataStreamSink ( )

inline

Definition at line 630 of file Importer.h.

{}

DataStreamSink() [2/2]

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

inline

Definition at line 631 of file Importer.h.

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

~DataStreamSink()

virtual import_export::DataStreamSink::~DataStreamSink ( )

inlinevirtual

Definition at line 633 of file Importer.h.

{}

Member Function Documentation

archivePlumber()

ImportStatus import_export::DataStreamSink::archivePlumber ( )

protected

Definition at line 3300 of file Importer.cpp.

References arrow_throw_if(), import_export::Importer::buffer, copy_params, cpu_threads(), import_export::TypedImportBuffer::del_values(), import_export::CopyParams::delimiter, logger::ERROR, import_export::CopyParams::escape, measure< TimeT >::execution(), import_export::BadRowsTracker::file_name, file_offsets, file_offsets_mutex, file_path, import_export::CopyParams::file_type, ThreadController_NS::SimpleThreadController< FutureReturnType >::finish(), import_export::Importer::get_column_descs(), get_filesize(), import_export::Importer::getLoader(), import_export::HAS_HEADER, import_export::CopyParams::has_header, import_export::Importer::import_buffers_vec, import_compressed(), import_status, import_export::BadRowsTracker::importer, logger::INFO, kARRAY, kBIGINT, kBOOLEAN, kCHAR, kDATE, kDECIMAL, kDOUBLE, kFLOAT, kINT, kLINESTRING, kMULTIPOLYGON, kNUMERIC, kPOINT, kPOLYGON, kSMALLINT, kTEXT, kTIME, kTIMESTAMP, kTINYINT, kVARCHAR, import_export::CopyParams::line_delim, import_export::Importer::load(), load_failed, import_export::ImportStatus::load_truncated, import_export::Importer::loader, LOG, mapd_glob(), import_export::CopyParams::max_reject, import_export::Importer::max_threads, num_rows, foreign_storage::anonymous_namespace{LazyParquetImporter.cpp}::open_parquet_table(), Archive::parse_url(), import_export::CopyParams::plain_text, import_export::CopyParams::quote, import_export::CopyParams::quoted, import_export::BadRowsTracker::row_group, import_export::BadRowsTracker::rows, import_export::ImportStatus::rows_completed, import_export::ImportStatus::rows_estimated, import_export::ImportStatus::rows_rejected, import_export::CopyParams::s3_access_key, import_export::CopyParams::s3_endpoint, import_export::CopyParams::s3_region, import_export::CopyParams::s3_secret_key, ThreadController_NS::SimpleThreadController< FutureReturnType >::startThread(), import_export::CopyParams::threads, to_string(), total_file_size, run_benchmark_import::type, anonymous_namespace{ArrowImporter.h}::value_getter(), and VLOG.

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 = mapd_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:

get_copy_params()

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

inline

Definition at line 640 of file Importer.h.

Referenced by DBHandler::detect_column_types().

{ return copy_params; }

Here is the caller graph for this function:

import_compressed()

void import_export::DataStreamSink::import_compressed

(

std::vector< std::string > &

file_paths

)

Definition at line 3698 of file Importer.cpp.

References File_Namespace::close(), copy_params, file_offsets, file_offsets_mutex, import_export::CopyParams::has_header, import_status, import_export::Importer::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, total_file_size, logger::WARNING, and File_Namespace::write().

Referenced by archivePlumber().

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

Here is the call graph for this function:

Here is the caller graph for this function:

importDelimited()

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

pure virtual

Implemented in import_export::Importer, and import_export::Detector.

Member Data Documentation

copy_params

CopyParams import_export::DataStreamSink::copy_params

protected

Definition at line 646 of file Importer.h.

Referenced by archivePlumber(), import_export::Detector::detect_row_delimiter(), import_export::Detector::find_best_sqltypes(), import_export::Detector::find_best_sqltypes_and_headers(), import_compressed(), import_export::import_thread_delimited(), import_export::import_thread_shapefile(), import_export::Detector::importDelimited(), import_export::Importer::importDelimited(), import_export::Importer::importGDAL(), and import_export::Detector::split_raw_data().

file_offsets

std::vector<size_t> import_export::DataStreamSink::file_offsets

protected

Definition at line 652 of file Importer.h.

Referenced by archivePlumber(), import_compressed(), and import_export::Importer::importDelimited().

file_offsets_mutex

std::mutex import_export::DataStreamSink::file_offsets_mutex

protected

Definition at line 653 of file Importer.h.

Referenced by archivePlumber(), import_compressed(), and import_export::Importer::importDelimited().

file_path

const std::string import_export::DataStreamSink::file_path

protected

Definition at line 647 of file Importer.h.

Referenced by archivePlumber(), import_export::Detector::detect_row_delimiter(), import_export::Detector::find_best_encodings(), import_export::Detector::find_best_sqltypes(), import_export::Importer::Importer(), and foreign_storage::LazyParquetImporter::partialImport().

import_status

ImportStatus import_export::DataStreamSink::import_status

protected

Definition at line 649 of file Importer.h.

Referenced by archivePlumber(), import_compressed(), import_export::import_thread_delimited(), import_export::import_thread_shapefile(), import_export::Detector::importDelimited(), import_export::Importer::importDelimited(), and import_export::Importer::importGDAL().

load_failed

bool import_export::DataStreamSink::load_failed = false

protected

Definition at line 650 of file Importer.h.

Referenced by archivePlumber(), import_export::Importer::checkpoint(), import_export::Detector::importDelimited(), import_export::Importer::importDelimited(), import_export::Importer::importGDAL(), import_export::Importer::load(), and foreign_storage::LazyParquetImporter::partialImport().

p_file

FILE* import_export::DataStreamSink::p_file = nullptr

protected

Definition at line 648 of file Importer.h.

Referenced by import_compressed(), import_export::Detector::importDelimited(), import_export::Importer::importDelimited(), import_export::Importer::Importer(), and import_export::Importer::~Importer().

total_file_size

size_t import_export::DataStreamSink::total_file_size {0}

protected

Definition at line 651 of file Importer.h.

Referenced by archivePlumber(), import_compressed(), and import_export::Importer::importDelimited().

---

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