anonymous_namespace{ForeignDataImporter.cpp} Namespace Reference

Functions
std::string	get_data_wrapper_type (const import_export::CopyParams &copy_params)
ChunkMetadataVector	metadata_scan (foreign_storage::ForeignDataWrapper *data_wrapper, foreign_storage::ForeignTable *foreign_table)
void	validate_copy_params (const import_export::CopyParams &copy_params)
import_export::ImportStatus	import_foreign_data (const ChunkMetadataVector &metadata_vector, Fragmenter_Namespace::InsertDataLoader::InsertConnector *connector, Catalog_Namespace::Catalog &catalog, const TableDescriptor *table, foreign_storage::ForeignDataWrapper *data_wrapper, const Catalog_Namespace::SessionInfo *session_info, const import_export::CopyParams &copy_params, const std::string &copy_from_source)

Function Documentation

std::string anonymous_namespace{ForeignDataImporter.cpp}::get_data_wrapper_type

(

const import_export::CopyParams &

copy_params

)

Definition at line 44 of file ForeignDataImporter.cpp.

References foreign_storage::DataWrapperType::CSV, import_export::kDelimitedFile, import_export::kParquetFile, import_export::kRegexParsedFile, foreign_storage::DataWrapperType::PARQUET, foreign_storage::DataWrapperType::REGEX_PARSER, import_export::CopyParams::source_type, and UNREACHABLE.

Referenced by import_export::ForeignDataImporter::importGeneral().

                                                                            {
  std::string data_wrapper_type;
  if (copy_params.source_type == import_export::SourceType::kDelimitedFile) {
    data_wrapper_type = foreign_storage::DataWrapperType::CSV;
  } else if (copy_params.source_type == import_export::SourceType::kRegexParsedFile) {
    data_wrapper_type = foreign_storage::DataWrapperType::REGEX_PARSER;
#ifdef ENABLE_IMPORT_PARQUET
  } else if (copy_params.source_type == import_export::SourceType::kParquetFile) {
    data_wrapper_type = foreign_storage::DataWrapperType::PARQUET;
#endif
  } else {
    UNREACHABLE();
  }
  return data_wrapper_type;
}

Here is the caller graph for this function:

import_export::ImportStatus anonymous_namespace{ForeignDataImporter.cpp}::import_foreign_data	(	const ChunkMetadataVector &	metadata_vector,
		Fragmenter_Namespace::InsertDataLoader::InsertConnector *	connector,
		Catalog_Namespace::Catalog &	catalog,
		const TableDescriptor *	table,
		foreign_storage::ForeignDataWrapper *	data_wrapper,
		const Catalog_Namespace::SessionInfo *	session_info,
		const import_export::CopyParams &	copy_params,
		const std::string &	copy_from_source
	)

Definition at line 90 of file ForeignDataImporter.cpp.

References CHECK, CHECK_GE, CHECK_LE, CHUNK_KEY_COLUMN_IDX, CHUNK_KEY_FRAGMENT_IDX, CHUNK_KEY_TABLE_IDX, CHUNK_KEY_VARLEN_IDX, foreign_storage::ForeignDataWrapper::createRenderGroupAnalyzers(), g_enable_assign_render_groups, import_export::CopyParams::geo_assign_render_groups, shared::get_from_map(), Chunk_NS::Chunk::getChunk(), Catalog_Namespace::Catalog::getDatabaseId(), Catalog_Namespace::Catalog::getMetadataForColumn(), Fragmenter_Namespace::InsertDataLoader::insertChunks(), import_export::ImportStatus::load_failed, import_export::ImportStatus::load_msg, import_export::CopyParams::max_reject, foreign_storage::ForeignDataWrapper::populateChunkBuffers(), import_export::ImportStatus::rows_completed, import_export::ImportStatus::rows_rejected, import_export::Importer::set_import_status(), and TableDescriptor::tableId.

Referenced by import_export::ForeignDataImporter::importGeneral().

                                       {
  int32_t max_fragment_id = -1;
  for (const auto& [key, _] : metadata_vector) {
    max_fragment_id = std::max(max_fragment_id, key[CHUNK_KEY_FRAGMENT_IDX]);
  }
  CHECK_GE(max_fragment_id, 0);

  if (g_enable_assign_render_groups && copy_params.geo_assign_render_groups) {
    // if render group assignment is enabled, tell the wrapper to create any
    // RenderGroupAnalyzers it may need for any poly columns in the table, if
    // that wrapper type supports it
    data_wrapper->createRenderGroupAnalyzers();
  }

  import_export::ImportStatus import_status;
  Fragmenter_Namespace::InsertDataLoader insert_data_loader(*connector);

  for (int32_t fragment_id = 0; fragment_id <= max_fragment_id; ++fragment_id) {
    // gather applicable keys to load for fragment
    std::set<ChunkKey> fragment_keys;
    for (const auto& [key, _] : metadata_vector) {
      if (key[CHUNK_KEY_FRAGMENT_IDX] == fragment_id) {
        fragment_keys.insert(key);

        const auto col_id = key[CHUNK_KEY_COLUMN_IDX];
        const auto table_id = key[CHUNK_KEY_TABLE_IDX];
        const auto col_desc = catalog.getMetadataForColumn(table_id, col_id);
        if (col_desc->columnType.is_varlen_indeed()) {
          CHECK(key.size() > CHUNK_KEY_VARLEN_IDX);
          if (key[CHUNK_KEY_VARLEN_IDX] == 1) {  // data chunk
            auto index_key = key;
            index_key[CHUNK_KEY_VARLEN_IDX] = 2;
            fragment_keys.insert(index_key);
          }
        }
      }
    }

    // create buffers
    std::map<ChunkKey, std::unique_ptr<foreign_storage::ForeignStorageBuffer>>
        fragment_buffers_owner;
    foreign_storage::ChunkToBufferMap fragment_buffers;
    auto delete_buffer = std::make_unique<foreign_storage::ForeignStorageBuffer>();
    for (const auto& key : fragment_keys) {
      fragment_buffers_owner[key] =
          std::make_unique<foreign_storage::ForeignStorageBuffer>();
      fragment_buffers_owner[key]->resetToEmpty();
      fragment_buffers[key] = shared::get_from_map(fragment_buffers_owner, key).get();
    }

    // get chunks for import
    Fragmenter_Namespace::InsertChunks insert_chunks{
        table->tableId, catalog.getDatabaseId(), {}, {}};

    // get the buffers
    data_wrapper->populateChunkBuffers(fragment_buffers, {}, delete_buffer.get());

    // create chunks from buffers
    for (const auto& [key, buffer] : fragment_buffers) {
      const auto col_id = key[CHUNK_KEY_COLUMN_IDX];
      const auto table_id = key[CHUNK_KEY_TABLE_IDX];
      const auto col_desc = catalog.getMetadataForColumn(table_id, col_id);

      if (col_desc->columnType.is_varlen_indeed()) {
        CHECK(key.size() > CHUNK_KEY_VARLEN_IDX);  // check for varlen key
        if (key[CHUNK_KEY_VARLEN_IDX] == 1) {      // data key
          auto index_key = key;
          index_key[CHUNK_KEY_VARLEN_IDX] = 2;
          insert_chunks.chunks[col_id] = Chunk_NS::Chunk::getChunk(
              col_desc, buffer, shared::get_from_map(fragment_buffers, index_key), false);
        }
      } else {  // regular non-varlen case with no index buffer
        insert_chunks.chunks[col_id] =
            Chunk_NS::Chunk::getChunk(col_desc, buffer, nullptr, false);
      }
    }

    // mark which row indices are valid for import
    auto row_count = fragment_buffers.begin()
                         ->second->getEncoder()
                         ->getNumElems();  // asssume all chunks have same number of rows,
                                           // this is validated at a lower level
    insert_chunks.valid_row_indices.reserve(row_count);
    for (size_t irow = 0; irow < row_count; ++irow) {
      if (delete_buffer->size() > 0) {
        CHECK_LE(irow, delete_buffer->size());
        if (delete_buffer->getMemoryPtr()[irow]) {
          continue;
        }
      }
      insert_chunks.valid_row_indices.emplace_back(irow);
    }

    // import chunks
    insert_data_loader.insertChunks(*session_info, insert_chunks);

    CHECK_LE(insert_chunks.valid_row_indices.size(), row_count);
    import_status.rows_rejected += row_count - insert_chunks.valid_row_indices.size();
    import_status.rows_completed += insert_chunks.valid_row_indices.size();
    if (import_status.rows_rejected > copy_params.max_reject) {
      import_status.load_failed = true;
      import_status.load_msg = "Load was cancelled due to max reject rows being reached";
      import_export::Importer::set_import_status(copy_from_source, import_status);
      break;
    }
    import_export::Importer::set_import_status(copy_from_source, import_status);
  }
  return import_status;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ChunkMetadataVector anonymous_namespace{ForeignDataImporter.cpp}::metadata_scan	(	foreign_storage::ForeignDataWrapper *	data_wrapper,
		foreign_storage::ForeignTable *	foreign_table
	)

Definition at line 60 of file ForeignDataImporter.cpp.

References TableDescriptor::maxFragRows, foreign_storage::MetadataScanInfeasibleFragmentSizeException::min_feasible_fragment_size_, and foreign_storage::ForeignDataWrapper::populateChunkMetadata().

Referenced by import_export::ForeignDataImporter::importGeneral().

                                                                              {
  ChunkMetadataVector metadata_vector;
  try {
    data_wrapper->populateChunkMetadata(
        metadata_vector);  // explicitly invoke a metadata scan on data wrapper
  } catch (const foreign_storage::MetadataScanInfeasibleFragmentSizeException&
               metadata_scan_exception) {
    // if a metadata scan exception is thrown, check to see if we can adjust
    // the fragment size and retry

    auto min_feasible_fragment_size = metadata_scan_exception.min_feasible_fragment_size_;
    if (min_feasible_fragment_size < 0) {
      throw;  // no valid fragment size returned by exception
    }
    foreign_table->maxFragRows = min_feasible_fragment_size;
    data_wrapper->populateChunkMetadata(
        metadata_vector);  // attempt another metadata scan, note, we assume that the
                           // metadata scan can be reentered safely after throwing the
                           // exception
  }
  return metadata_vector;
}

Here is the call graph for this function:

Here is the caller graph for this function:

void anonymous_namespace{ForeignDataImporter.cpp}::validate_copy_params

(

const import_export::CopyParams &

copy_params

)

Definition at line 84 of file ForeignDataImporter.cpp.

References import_export::kRegexParsedFile, import_export::CopyParams::source_type, and foreign_storage::validate_regex_parser_options().

Referenced by import_export::ForeignDataImporter::importGeneral().

                                                                      {
  if (copy_params.source_type == import_export::SourceType::kRegexParsedFile) {
    foreign_storage::validate_regex_parser_options(copy_params);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function: