The InsertOrderFragmenter is a child class of AbstractFragmenter, and fragments data in insert order. Likely the default fragmenter. More...

#include <InsertOrderFragmenter.h>

Inheritance diagram for Fragmenter_Namespace::InsertOrderFragmenter:

Collaboration diagram for Fragmenter_Namespace::InsertOrderFragmenter:

Public Types
using	ModifyTransactionTracker = UpdelRoll

Public Member Functions
	InsertOrderFragmenter (const std::vector< int > chunkKeyPrefix, std::vector< Chunk_NS::Chunk > &chunkVec, Data_Namespace::DataMgr dataMgr, Catalog_Namespace::Catalog catalog, const int physicalTableId, const int shard, const size_t maxFragmentRows=DEFAULT_FRAGMENT_ROWS, const size_t maxChunkSize=DEFAULT_MAX_CHUNK_SIZE, const size_t pageSize=DEFAULT_PAGE_SIZE, const size_t maxRows=DEFAULT_MAX_ROWS, const Data_Namespace::MemoryLevel defaultInsertLevel=Data_Namespace::DISK_LEVEL)

	~InsertOrderFragmenter () override

TableInfo	getFragmentsForQuery () override
	returns (inside QueryInfo) object all ids and row sizes of fragments More...

void	insertData (InsertData &insertDataStruct) override
	appends data onto the most recently occuring fragment, creating a new one if necessary More...

void	insertDataNoCheckpoint (InsertData &insertDataStruct) override
	Given data wrapped in an InsertData struct, inserts it into the correct partitions No locks and checkpoints taken needs to be managed externally. More...

void	dropFragmentsToSize (const size_t maxRows) override
	Will truncate table to less than maxRows by dropping fragments. More...

void	updateChunkStats (const ColumnDescriptor *cd, std::unordered_map< int, ChunkStats > &stats_map) override
	Update chunk stats. More...

int	getFragmenterId () override
	get fragmenter's id More...

std::vector< int >	getChunkKeyPrefix () const

std::string	getFragmenterType () override
	get fragmenter's type (as string More...

size_t	getNumRows () override

void	setNumRows (const size_t numTuples) override

void	updateColumn (const Catalog_Namespace::Catalog catalog, const TableDescriptor td, const ColumnDescriptor *cd, const int fragment_id, const std::vector< uint64_t > &frag_offsets, const std::vector< ScalarTargetValue > &rhs_values, const SQLTypeInfo &rhs_type, const Data_Namespace::MemoryLevel memory_level, UpdelRoll &updel_roll) override

void	updateColumns (const Catalog_Namespace::Catalog catalog, const TableDescriptor td, const int fragmentId, const std::vector< TargetMetaInfo > sourceMetaInfo, const std::vector< const ColumnDescriptor * > columnDescriptors, const RowDataProvider &sourceDataProvider, const size_t indexOffFragmentOffsetColumn, const Data_Namespace::MemoryLevel memoryLevel, UpdelRoll &updelRoll) override

void	updateColumn (const Catalog_Namespace::Catalog catalog, const TableDescriptor td, const ColumnDescriptor *cd, const int fragment_id, const std::vector< uint64_t > &frag_offsets, const ScalarTargetValue &rhs_value, const SQLTypeInfo &rhs_type, const Data_Namespace::MemoryLevel memory_level, UpdelRoll &updel_roll) override

void	updateColumnMetadata (const ColumnDescriptor *cd, FragmentInfo &fragment, std::shared_ptr< Chunk_NS::Chunk > chunk, const bool null, const double dmax, const double dmin, const int64_t lmax, const int64_t lmin, const SQLTypeInfo &rhs_type, UpdelRoll &updel_roll) override

void	updateMetadata (const Catalog_Namespace::Catalog *catalog, const MetaDataKey &key, UpdelRoll &updel_roll) override

void	compactRows (const Catalog_Namespace::Catalog catalog, const TableDescriptor td, const int fragment_id, const std::vector< uint64_t > &frag_offsets, const Data_Namespace::MemoryLevel memory_level, UpdelRoll &updel_roll) override

const std::vector< uint64_t >	getVacuumOffsets (const std::shared_ptr< Chunk_NS::Chunk > &chunk) override

auto	getChunksForAllColumns (const TableDescriptor *td, const FragmentInfo &fragment, const Data_Namespace::MemoryLevel memory_level)

Public Member Functions inherited from Fragmenter_Namespace::AbstractFragmenter
virtual	~AbstractFragmenter ()

Static Public Member Functions
static void	updateColumn (const Catalog_Namespace::Catalog *catalog, const std::string &tab_name, const std::string &col_name, const int fragment_id, const std::vector< uint64_t > &frag_offsets, const std::vector< ScalarTargetValue > &rhs_values, const SQLTypeInfo &rhs_type, const Data_Namespace::MemoryLevel memory_level, UpdelRoll &updel_roll)

Protected Member Functions
FragmentInfo *	createNewFragment (const Data_Namespace::MemoryLevel memory_level=Data_Namespace::DISK_LEVEL)
	creates new fragment, calling createChunk() method of BufferMgr to make a new chunk for each column of the table. More...

void	deleteFragments (const std::vector< int > &dropFragIds)

void	getChunkMetadata ()

void	lockInsertCheckpointData (const InsertData &insertDataStruct)

void	insertDataImpl (InsertData &insertDataStruct)

void	replicateData (const InsertData &insertDataStruct)

	InsertOrderFragmenter (const InsertOrderFragmenter &)

InsertOrderFragmenter &	operator= (const InsertOrderFragmenter &)

FragmentInfo &	getFragmentInfoFromId (const int fragment_id)

auto	vacuum_fixlen_rows (const FragmentInfo &fragment, const std::shared_ptr< Chunk_NS::Chunk > &chunk, const std::vector< uint64_t > &frag_offsets)

auto	vacuum_varlen_rows (const FragmentInfo &fragment, const std::shared_ptr< Chunk_NS::Chunk > &chunk, const std::vector< uint64_t > &frag_offsets)

Protected Attributes
std::vector< int >	chunkKeyPrefix_

std::map< int, Chunk_NS::Chunk >	columnMap_

std::deque< FragmentInfo >	fragmentInfoVec_

Data_Namespace::DataMgr *	dataMgr_

Catalog_Namespace::Catalog *	catalog_

const int	physicalTableId_

const int	shard_

size_t	maxFragmentRows_

size_t	pageSize_

size_t	numTuples_

int	maxFragmentId_

size_t	maxChunkSize_

size_t	maxRows_

std::string	fragmenterType_

mapd_shared_mutex	fragmentInfoMutex_

mapd_shared_mutex	insertMutex_

Data_Namespace::MemoryLevel	defaultInsertLevel_

bool	hasMaterializedRowId_

int	rowIdColId_

std::unordered_map< int, size_t >	varLenColInfo_

std::shared_ptr< std::mutex >	mutex_access_inmem_states

std::mutex	temp_mutex_

Detailed Description

The InsertOrderFragmenter is a child class of AbstractFragmenter, and fragments data in insert order. Likely the default fragmenter.

InsertOrderFragmenter

Definition at line 54 of file InsertOrderFragmenter.h.

Member Typedef Documentation

using Fragmenter_Namespace::InsertOrderFragmenter::ModifyTransactionTracker = UpdelRoll

Definition at line 56 of file InsertOrderFragmenter.h.

Constructor & Destructor Documentation

Fragmenter_Namespace::InsertOrderFragmenter::InsertOrderFragmenter	(	const std::vector< int >	chunkKeyPrefix,
		std::vector< Chunk_NS::Chunk > &	chunkVec,
		Data_Namespace::DataMgr *	dataMgr,
		Catalog_Namespace::Catalog *	catalog,
		const int	physicalTableId,
		const int	shard,
		const size_t	maxFragmentRows = `DEFAULT_FRAGMENT_ROWS`,
		const size_t	maxChunkSize = `DEFAULT_MAX_CHUNK_SIZE`,
		const size_t	pageSize = `DEFAULT_PAGE_SIZE`,
		const size_t	maxRows = `DEFAULT_MAX_ROWS`,
		const Data_Namespace::MemoryLevel	defaultInsertLevel = `Data_Namespace::DISK_LEVEL`
	)

Fragmenter_Namespace::InsertOrderFragmenter::~InsertOrderFragmenter ( )

override

Definition at line 86 of file InsertOrderFragmenter.cpp.

86 {}

Fragmenter_Namespace::InsertOrderFragmenter::InsertOrderFragmenter ( const InsertOrderFragmenter & )

protected

Member Function Documentation

void Fragmenter_Namespace::InsertOrderFragmenter::compactRows	(	const Catalog_Namespace::Catalog *	catalog,
		const TableDescriptor *	td,
		const int	fragment_id,
		const std::vector< uint64_t > &	frag_offsets,
		const Data_Namespace::MemoryLevel	memory_level,
		UpdelRoll &	updel_roll
	)

overridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 1209 of file UpdelStorage.cpp.

References CHECK(), cpu_threads(), anonymous_namespace{ResultSetReductionInterpreter.cpp}::get_element_size(), getChunksForAllColumns(), getFragmentInfoFromId(), UpdelRoll::numTuples, Fragmenter_Namespace::set_chunk_metadata(), Fragmenter_Namespace::set_chunk_stats(), updateColumnMetadata(), vacuum_fixlen_rows(), vacuum_varlen_rows(), and Fragmenter_Namespace::wait_cleanup_threads().

Referenced by updateColumn().

                                                                {
   auto& fragment = getFragmentInfoFromId(fragment_id);
   auto chunks = getChunksForAllColumns(td, fragment, memory_level);
   const auto ncol = chunks.size();
 
   std::vector<int8_t> has_null_per_thread(ncol, 0);
   std::vector<double> max_double_per_thread(ncol, std::numeric_limits<double>::lowest());
   std::vector<double> min_double_per_thread(ncol, std::numeric_limits<double>::max());
   std::vector<int64_t> max_int64t_per_thread(ncol, std::numeric_limits<uint64_t>::min());
   std::vector<int64_t> min_int64t_per_thread(ncol, std::numeric_limits<uint64_t>::max());
 
   // parallel delete columns
   std::vector<std::future<void>> threads;
   auto nrows_to_vacuum = frag_offsets.size();
   auto nrows_in_fragment = fragment.getPhysicalNumTuples();
   auto nrows_to_keep = nrows_in_fragment - nrows_to_vacuum;
 
   for (size_t ci = 0; ci < chunks.size(); ++ci) {
     auto chunk = chunks[ci];
     const auto cd = chunk->get_column_desc();
     const auto& col_type = cd->columnType;
     auto data_buffer = chunk->get_buffer();
     auto index_buffer = chunk->get_index_buf();
     auto data_addr = data_buffer->getMemoryPtr();
     auto indices_addr = index_buffer ? index_buffer->getMemoryPtr() : nullptr;
     auto index_array = (StringOffsetT*)indices_addr;
     bool is_varlen = col_type.is_varlen_indeed();
 
     auto fixlen_vacuum = [=,
                           &has_null_per_thread,
                           &max_double_per_thread,
                           &min_double_per_thread,
                           &min_int64t_per_thread,
                           &max_int64t_per_thread,
                           &updel_roll,
                           &frag_offsets,
                           &fragment] {
       size_t nbytes_fix_data_to_keep;
       nbytes_fix_data_to_keep = vacuum_fixlen_rows(fragment, chunk, frag_offsets);
 
       data_buffer->encoder->setNumElems(nrows_to_keep);
       data_buffer->setSize(nbytes_fix_data_to_keep);
       data_buffer->setUpdated();
 
       set_chunk_metadata(catalog, fragment, chunk, nrows_to_keep, updel_roll);
 
       auto daddr = data_addr;
       auto element_size =
           col_type.is_fixlen_array() ? col_type.get_size() : get_element_size(col_type);
       for (size_t irow = 0; irow < nrows_to_keep; ++irow, daddr += element_size) {
         if (col_type.is_fixlen_array()) {
           auto encoder =
               dynamic_cast<FixedLengthArrayNoneEncoder*>(data_buffer->encoder.get());
           CHECK(encoder);
           encoder->updateMetadata((int8_t*)daddr);
         } else if (col_type.is_fp()) {
           set_chunk_stats(col_type,
                           data_addr,
                           has_null_per_thread[ci],
                           min_double_per_thread[ci],
                           max_double_per_thread[ci]);
         } else {
           set_chunk_stats(col_type,
                           data_addr,
                           has_null_per_thread[ci],
                           min_int64t_per_thread[ci],
                           max_int64t_per_thread[ci]);
         }
       }
     };
 
     auto varlen_vacuum = [=, &updel_roll, &frag_offsets, &fragment] {
       size_t nbytes_var_data_to_keep;
       nbytes_var_data_to_keep = vacuum_varlen_rows(fragment, chunk, frag_offsets);
 
       data_buffer->encoder->setNumElems(nrows_to_keep);
       data_buffer->setSize(nbytes_var_data_to_keep);
       data_buffer->setUpdated();
 
       index_array[nrows_to_keep] = data_buffer->size();
       index_buffer->setSize(sizeof(*index_array) *
                             (nrows_to_keep ? 1 + nrows_to_keep : 0));
       index_buffer->setUpdated();
 
       set_chunk_metadata(catalog, fragment, chunk, nrows_to_keep, updel_roll);
     };
 
     if (is_varlen) {
       threads.emplace_back(std::async(std::launch::async, varlen_vacuum));
     } else {
       threads.emplace_back(std::async(std::launch::async, fixlen_vacuum));
     }
     if (threads.size() >= (size_t)cpu_threads()) {
       wait_cleanup_threads(threads);
     }
   }
 
   wait_cleanup_threads(threads);
 
   auto key = std::make_pair(td, &fragment);
   updel_roll.numTuples[key] = nrows_to_keep;
   for (size_t ci = 0; ci < chunks.size(); ++ci) {
     auto chunk = chunks[ci];
     auto cd = chunk->get_column_desc();
     if (!cd->columnType.is_fixlen_array()) {
       updateColumnMetadata(cd,
                            fragment,
                            chunk,
                            has_null_per_thread[ci],
                            max_double_per_thread[ci],
                            min_double_per_thread[ci],
                            max_int64t_per_thread[ci],
                            min_int64t_per_thread[ci],
                            cd->columnType,
                            updel_roll);
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

FragmentInfo * Fragmenter_Namespace::InsertOrderFragmenter::createNewFragment ( const Data_Namespace::MemoryLevel memory_level = Data_Namespace::DISK_LEVEL )

protected

creates new fragment, calling createChunk() method of BufferMgr to make a new chunk for each column of the table.

Also unpins the chunks of the previous insert buffer

Definition at line 588 of file InsertOrderFragmenter.cpp.

References Fragmenter_Namespace::FragmentInfo::deviceIds, Fragmenter_Namespace::FragmentInfo::fragmentId, Fragmenter_Namespace::FragmentInfo::physicalTableId, Fragmenter_Namespace::FragmentInfo::setPhysicalNumTuples(), Fragmenter_Namespace::FragmentInfo::shadowNumTuples, and Fragmenter_Namespace::FragmentInfo::shard.

                                                  {
   // also sets the new fragment as the insertBuffer for each column
 
   maxFragmentId_++;
   FragmentInfo newFragmentInfo;
   newFragmentInfo.fragmentId = maxFragmentId_;
   newFragmentInfo.shadowNumTuples = 0;
   newFragmentInfo.setPhysicalNumTuples(0);
   for (const auto levelSize : dataMgr_->levelSizes_) {
     newFragmentInfo.deviceIds.push_back(newFragmentInfo.fragmentId % levelSize);
   }
   newFragmentInfo.physicalTableId = physicalTableId_;
   newFragmentInfo.shard = shard_;
 
   for (map<int, Chunk>::iterator colMapIt = columnMap_.begin();
        colMapIt != columnMap_.end();
        ++colMapIt) {
     // colMapIt->second.unpin_buffer();
     ChunkKey chunkKey = chunkKeyPrefix_;
     chunkKey.push_back(colMapIt->second.get_column_desc()->columnId);
     chunkKey.push_back(maxFragmentId_);
     colMapIt->second.createChunkBuffer(
         dataMgr_,
         chunkKey,
         memoryLevel,
         newFragmentInfo.deviceIds[static_cast<int>(memoryLevel)],
         pageSize_);
     colMapIt->second.init_encoder();
   }
 
   mapd_lock_guard<mapd_shared_mutex> writeLock(fragmentInfoMutex_);
   fragmentInfoVec_.push_back(newFragmentInfo);
   return &(fragmentInfoVec_.back());
 }

Here is the call graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::deleteFragments ( const std::vector< int > & dropFragIds )

protected

Definition at line 200 of file InsertOrderFragmenter.cpp.

References catalog_().

                                                                           {
   // Fix a verified loophole on sharded logical table which is locked using logical
   // tableId while it's its physical tables that can come here when fragments overflow
   // during COPY. Locks on a logical table and its physical tables never intersect, which
   // means potential races. It'll be an overkill to resolve a logical table to physical
   // tables in MapDHandler, ParseNode or other higher layers where the logical table is
   // locked with Table Read/Write locks; it's easier to lock the logical table of its
   // physical tables. A downside of this approach may be loss of parallel execution of
   // deleteFragments across physical tables. Because deleteFragments is a short in-memory
   // operation, the loss seems not a big deal.
   auto chunkKeyPrefix = chunkKeyPrefix_;
   if (shard_ >= 0) {
     chunkKeyPrefix[1] = catalog_->getLogicalTableId(chunkKeyPrefix[1]);
   }
 
   // need to keep lock seq as TableLock >> fragmentInfoMutex_ or
   // SELECT and COPY may enter a deadlock
   using namespace Lock_Namespace;
   WriteLock delete_lock = TableLockMgr::getWriteLockForTable(chunkKeyPrefix);
 
   mapd_unique_lock<mapd_shared_mutex> writeLock(fragmentInfoMutex_);
 
   for (const auto fragId : dropFragIds) {
     for (const auto& col : columnMap_) {
       int colId = col.first;
       vector<int> fragPrefix = chunkKeyPrefix_;
       fragPrefix.push_back(colId);
       fragPrefix.push_back(fragId);
       dataMgr_->deleteChunksWithPrefix(fragPrefix);
     }
   }
 }

Here is the call graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::dropFragmentsToSize ( const size_t maxRows )

overridevirtual

Will truncate table to less than maxRows by dropping fragments.

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 173 of file InsertOrderFragmenter.cpp.

References CHECK_GE, CHECK_GT, DROP_FRAGMENT_FACTOR, logger::INFO, and LOG.

                                                                     {
   // not safe to call from outside insertData
   // b/c depends on insertLock around numTuples_
 
   // don't ever drop the only fragment!
   if (numTuples_ == fragmentInfoVec_.back().getPhysicalNumTuples()) {
     return;
   }
 
   if (numTuples_ > maxRows) {
     size_t preNumTuples = numTuples_;
     vector<int> dropFragIds;
     size_t targetRows = maxRows * DROP_FRAGMENT_FACTOR;
     while (numTuples_ > targetRows) {
       CHECK_GT(fragmentInfoVec_.size(), size_t(0));
       size_t numFragTuples = fragmentInfoVec_[0].getPhysicalNumTuples();
       dropFragIds.push_back(fragmentInfoVec_[0].fragmentId);
       fragmentInfoVec_.pop_front();
       CHECK_GE(numTuples_, numFragTuples);
       numTuples_ -= numFragTuples;
     }
     deleteFragments(dropFragIds);
     LOG(INFO) << "dropFragmentsToSize, numTuples pre: " << preNumTuples
               << " post: " << numTuples_ << " maxRows: " << maxRows;
   }
 }

std::vector<int> Fragmenter_Namespace::InsertOrderFragmenter::getChunkKeyPrefix ( ) const

inline

Definition at line 102 of file InsertOrderFragmenter.h.

References chunkKeyPrefix_.

102 { return chunkKeyPrefix_; }

Fragmenter_Namespace::InsertOrderFragmenter::chunkKeyPrefix_

std::vector< int > chunkKeyPrefix_

Definition: InsertOrderFragmenter.h:180

void Fragmenter_Namespace::InsertOrderFragmenter::getChunkMetadata ( )

protected

Definition at line 88 of file InsertOrderFragmenter.cpp.

References Data_Namespace::DISK_LEVEL, and to_string().

                                              {
   if (defaultInsertLevel_ == Data_Namespace::MemoryLevel::DISK_LEVEL) {
     // memory-resident tables won't have anything on disk
     std::vector<std::pair<ChunkKey, ChunkMetadata>> chunk_metadata;
     dataMgr_->getChunkMetadataVecForKeyPrefix(chunk_metadata, chunkKeyPrefix_);
 
     // data comes like this - database_id, table_id, column_id, fragment_id
     // but lets sort by database_id, table_id, fragment_id, column_id
 
     int fragment_subkey_index = 3;
     std::sort(chunk_metadata.begin(),
               chunk_metadata.end(),
               [&](const std::pair<ChunkKey, ChunkMetadata>& pair1,
                   const std::pair<ChunkKey, ChunkMetadata>& pair2) {
                 return pair1.first[3] < pair2.first[3];
               });
 
     for (auto chunk_itr = chunk_metadata.begin(); chunk_itr != chunk_metadata.end();
          ++chunk_itr) {
       int cur_column_id = chunk_itr->first[2];
       int cur_fragment_id = chunk_itr->first[fragment_subkey_index];
 
       if (fragmentInfoVec_.empty() ||
           cur_fragment_id != fragmentInfoVec_.back().fragmentId) {
         maxFragmentId_ = cur_fragment_id;
         fragmentInfoVec_.emplace_back();
         fragmentInfoVec_.back().fragmentId = cur_fragment_id;
         fragmentInfoVec_.back().setPhysicalNumTuples(chunk_itr->second.numElements);
         numTuples_ += fragmentInfoVec_.back().getPhysicalNumTuples();
         for (const auto level_size : dataMgr_->levelSizes_) {
           fragmentInfoVec_.back().deviceIds.push_back(cur_fragment_id % level_size);
         }
         fragmentInfoVec_.back().shadowNumTuples =
             fragmentInfoVec_.back().getPhysicalNumTuples();
         fragmentInfoVec_.back().physicalTableId = physicalTableId_;
         fragmentInfoVec_.back().shard = shard_;
       } else {
         if (chunk_itr->second.numElements !=
             fragmentInfoVec_.back().getPhysicalNumTuples()) {
           throw std::runtime_error(
               "Inconsistency in num tuples within fragment for table " +
               std::to_string(physicalTableId_) + ", Column " +
               std::to_string(cur_column_id) + ". Fragment Tuples: " +
               std::to_string(fragmentInfoVec_.back().getPhysicalNumTuples()) +
               ", Chunk Tuples: " + std::to_string(chunk_itr->second.numElements));
         }
       }
       fragmentInfoVec_.back().setChunkMetadata(cur_column_id, chunk_itr->second);
     }
   }
 
   ssize_t maxFixedColSize = 0;
 
   for (auto colIt = columnMap_.begin(); colIt != columnMap_.end(); ++colIt) {
     ssize_t size = colIt->second.get_column_desc()->columnType.get_size();
     if (size == -1) {  // variable length
       varLenColInfo_.insert(std::make_pair(colIt->first, 0));
       size = 8;  // b/c we use this for string and array indices - gross to have magic
                  // number here
     }
     maxFixedColSize = std::max(maxFixedColSize, size);
   }
 
   // this is maximum number of rows assuming everything is fixed length
   maxFragmentRows_ = std::min(maxFragmentRows_, maxChunkSize_ / maxFixedColSize);
 
   if (fragmentInfoVec_.size() > 0) {
     // Now need to get the insert buffers for each column - should be last
     // fragment
     int lastFragmentId = fragmentInfoVec_.back().fragmentId;
     int deviceId =
         fragmentInfoVec_.back().deviceIds[static_cast<int>(defaultInsertLevel_)];
     for (auto colIt = columnMap_.begin(); colIt != columnMap_.end(); ++colIt) {
       ChunkKey insertKey = chunkKeyPrefix_;  // database_id and table_id
       insertKey.push_back(colIt->first);     // column id
       insertKey.push_back(lastFragmentId);   // fragment id
       colIt->second.getChunkBuffer(dataMgr_, insertKey, defaultInsertLevel_, deviceId);
       auto varLenColInfoIt = varLenColInfo_.find(colIt->first);
       if (varLenColInfoIt != varLenColInfo_.end()) {
         varLenColInfoIt->second = colIt->second.get_buffer()->size();
       }
     }
   }
 }

Here is the call graph for this function:

auto Fragmenter_Namespace::InsertOrderFragmenter::getChunksForAllColumns	(	const TableDescriptor *	td,
		const FragmentInfo &	fragment,
		const Data_Namespace::MemoryLevel	memory_level
	)

Definition at line 1019 of file UpdelStorage.cpp.

References catalog_, CHECK(), Catalog_Namespace::DBMetadata::dbId, Fragmenter_Namespace::FragmentInfo::fragmentId, Chunk_NS::Chunk::getChunk(), Fragmenter_Namespace::FragmentInfo::getChunkMetadataMapPhysical(), Catalog_Namespace::Catalog::getCurrentDB(), Catalog_Namespace::Catalog::getDataMgr(), Catalog_Namespace::Catalog::getMetadataForColumn(), TableDescriptor::nColumns, and TableDescriptor::tableId.

Referenced by compactRows().

                                                   {
   std::vector<std::shared_ptr<Chunk_NS::Chunk>> chunks;
   // coming from updateColumn (on '$delete$' column) we dont have chunks for all columns
   for (int col_id = 1, ncol = 0; ncol < td->nColumns; ++col_id) {
     if (const auto cd = catalog_->getMetadataForColumn(td->tableId, col_id)) {
       ++ncol;
       if (!cd->isVirtualCol) {
         auto chunk_meta_it = fragment.getChunkMetadataMapPhysical().find(col_id);
         CHECK(chunk_meta_it != fragment.getChunkMetadataMapPhysical().end());
         ChunkKey chunk_key{
             catalog_->getCurrentDB().dbId, td->tableId, col_id, fragment.fragmentId};
         auto chunk = Chunk_NS::Chunk::getChunk(cd,
                                                &catalog_->getDataMgr(),
                                                chunk_key,
                                                memory_level,
                                                0,
                                                chunk_meta_it->second.numBytes,
                                                chunk_meta_it->second.numElements);
         chunks.push_back(chunk);
       }
     }
   }
   return chunks;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

int Fragmenter_Namespace::InsertOrderFragmenter::getFragmenterId ( )

inlineoverridevirtual

get fragmenter's id

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 101 of file InsertOrderFragmenter.h.

References chunkKeyPrefix_.

101 { return chunkKeyPrefix_.back(); }

Fragmenter_Namespace::InsertOrderFragmenter::chunkKeyPrefix_

std::vector< int > chunkKeyPrefix_

Definition: InsertOrderFragmenter.h:180

std::string Fragmenter_Namespace::InsertOrderFragmenter::getFragmenterType ( )

inlineoverridevirtual

get fragmenter's type (as string

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 106 of file InsertOrderFragmenter.h.

References fragmenterType_.

106 { return fragmenterType_; }

Fragmenter_Namespace::InsertOrderFragmenter::fragmenterType_

std::string fragmenterType_

Definition: InsertOrderFragmenter.h:197

FragmentInfo & Fragmenter_Namespace::InsertOrderFragmenter::getFragmentInfoFromId ( const int fragment_id )

protected

Definition at line 48 of file UpdelStorage.cpp.

References CHECK(), and fragmentInfoVec_.

Referenced by compactRows(), updateColumn(), and updateColumns().

                                                                                 {
   auto fragment_it = std::find_if(
       fragmentInfoVec_.begin(), fragmentInfoVec_.end(), [=](const auto& f) -> bool {
         return f.fragmentId == fragment_id;
       });
   CHECK(fragment_it != fragmentInfoVec_.end());
   return *fragment_it;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

TableInfo Fragmenter_Namespace::InsertOrderFragmenter::getFragmentsForQuery ( )

overridevirtual

returns (inside QueryInfo) object all ids and row sizes of fragments

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 624 of file InsertOrderFragmenter.cpp.

                                                       {
   mapd_shared_lock<mapd_shared_mutex> readLock(fragmentInfoMutex_);
   TableInfo queryInfo;
   queryInfo.chunkKeyPrefix = chunkKeyPrefix_;
   // right now we don't test predicate, so just return (copy of) all fragments
   bool fragmentsExist = false;
   if (fragmentInfoVec_.empty()) {
     // If we have no fragments add a dummy empty fragment to make the executor
     // not have separate logic for 0-row tables
     int maxFragmentId = 0;
     FragmentInfo emptyFragmentInfo;
     emptyFragmentInfo.fragmentId = maxFragmentId;
     emptyFragmentInfo.shadowNumTuples = 0;
     emptyFragmentInfo.setPhysicalNumTuples(0);
     emptyFragmentInfo.deviceIds.resize(dataMgr_->levelSizes_.size());
     emptyFragmentInfo.physicalTableId = physicalTableId_;
     emptyFragmentInfo.shard = shard_;
     queryInfo.fragments.push_back(emptyFragmentInfo);
   } else {
     fragmentsExist = true;
     queryInfo.fragments = fragmentInfoVec_;  // makes a copy
   }
   readLock.unlock();
   queryInfo.setPhysicalNumTuples(0);
   auto partIt = queryInfo.fragments.begin();
   if (fragmentsExist) {
     while (partIt != queryInfo.fragments.end()) {
       if (partIt->getPhysicalNumTuples() == 0) {
         // this means that a concurrent insert query inserted tuples into a new fragment
         // but when the query came in we didn't have this fragment. To make sure we don't
         // mess up the executor we delete this fragment from the metadatamap (fixes
         // earlier bug found 2015-05-08)
         partIt = queryInfo.fragments.erase(partIt);
       } else {
         queryInfo.setPhysicalNumTuples(queryInfo.getPhysicalNumTuples() +
                                        partIt->getPhysicalNumTuples());
         ++partIt;
       }
     }
   } else {
     // We added a dummy fragment and know the table is empty
     queryInfo.setPhysicalNumTuples(0);
   }
   return queryInfo;
 }

Here is the call graph for this function:

size_t Fragmenter_Namespace::InsertOrderFragmenter::getNumRows ( )

inlineoverridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 107 of file InsertOrderFragmenter.h.

References numTuples_.

107 { return numTuples_; }

Fragmenter_Namespace::InsertOrderFragmenter::numTuples_

size_t numTuples_

Definition: InsertOrderFragmenter.h:193

const std::vector< uint64_t > Fragmenter_Namespace::InsertOrderFragmenter::getVacuumOffsets ( const std::shared_ptr< Chunk_NS::Chunk > & chunk )

overridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 1048 of file UpdelStorage.cpp.

References CHECK(), cpu_threads(), and Fragmenter_Namespace::wait_cleanup_threads().

Referenced by updateColumn().

                                                {
   const auto data_buffer = chunk->get_buffer();
   const auto data_addr = data_buffer->getMemoryPtr();
   const size_t nrows_in_chunk = data_buffer->size();
   const size_t ncore = cpu_threads();
   const size_t segsz = (nrows_in_chunk + ncore - 1) / ncore;
   std::vector<std::vector<uint64_t>> deleted_offsets;
   deleted_offsets.resize(ncore);
   std::vector<std::future<void>> threads;
   for (size_t rbegin = 0; rbegin < nrows_in_chunk; rbegin += segsz) {
     threads.emplace_back(std::async(std::launch::async, [=, &deleted_offsets] {
       const auto rend = std::min<size_t>(rbegin + segsz, nrows_in_chunk);
       const auto ithread = rbegin / segsz;
       CHECK(ithread < deleted_offsets.size());
       deleted_offsets[ithread].reserve(segsz);
       for (size_t r = rbegin; r < rend; ++r) {
         if (data_addr[r]) {
           deleted_offsets[ithread].push_back(r);
         }
       }
     }));
   }
   wait_cleanup_threads(threads);
   std::vector<uint64_t> all_deleted_offsets;
   for (size_t i = 0; i < ncore; ++i) {
     all_deleted_offsets.insert(
         all_deleted_offsets.end(), deleted_offsets[i].begin(), deleted_offsets[i].end());
   }
   return all_deleted_offsets;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::insertData ( InsertData & insertDataStruct )

overridevirtual

appends data onto the most recently occuring fragment, creating a new one if necessary

Todo:: be able to fill up current fragment in multi-row insert before creating new fragment

Implements Fragmenter_Namespace::AbstractFragmenter.

Reimplemented in Fragmenter_Namespace::SortedOrderFragmenter.

Definition at line 316 of file InsertOrderFragmenter.cpp.

References catalog_(), Fragmenter_Namespace::InsertData::databaseId, Data_Namespace::DISK_LEVEL, and Fragmenter_Namespace::InsertData::tableId.

Referenced by Fragmenter_Namespace::SortedOrderFragmenter::insertData().

                                                                    {
   // TODO: this local lock will need to be centralized when ALTER COLUMN is added, bc
   try {
     mapd_unique_lock<mapd_shared_mutex> insertLock(
         insertMutex_);  // prevent two threads from trying to insert into the same table
                         // simultaneously
 
     insertDataImpl(insertDataStruct);
 
     if (defaultInsertLevel_ ==
         Data_Namespace::DISK_LEVEL) {  // only checkpoint if data is resident on disk
       dataMgr_->checkpoint(
           chunkKeyPrefix_[0],
           chunkKeyPrefix_[1]);  // need to checkpoint here to remove window for corruption
     }
   } catch (...) {
     int32_t tableEpoch =
         catalog_->getTableEpoch(insertDataStruct.databaseId, insertDataStruct.tableId);
 
     // the statement below deletes *this* object!
     // relying on exception propagation at this stage
     // until we can sort this out in a cleaner fashion
     catalog_->setTableEpoch(
         insertDataStruct.databaseId, insertDataStruct.tableId, tableEpoch);
     throw;
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::insertDataImpl ( InsertData & insertDataStruct )

protected

Definition at line 430 of file InsertOrderFragmenter.cpp.

References CHECK(), CHECK_GT, CHECK_LE, Fragmenter_Namespace::InsertData::columnDescriptors, Fragmenter_Namespace::InsertData::columnIds, Fragmenter_Namespace::InsertData::data, Fragmenter_Namespace::FragmentInfo::fragmentId, DataBlockPtr::numbersPtr, Fragmenter_Namespace::InsertData::numRows, Fragmenter_Namespace::InsertData::replicate_count, Fragmenter_Namespace::FragmentInfo::shadowChunkMetadataMap, and Fragmenter_Namespace::FragmentInfo::shadowNumTuples.

                                                                        {
   // populate deleted system column if it should exists, as it will not come from client
   // Do not add this magical column in the replicate ALTER TABLE ADD route as
   // it is not needed and will cause issues
   std::unique_ptr<int8_t[]> data_for_deleted_column;
   for (const auto& cit : columnMap_) {
     if (cit.second.get_column_desc()->isDeletedCol &&
         insertDataStruct.replicate_count == 0) {
       data_for_deleted_column.reset(new int8_t[insertDataStruct.numRows]);
       memset(data_for_deleted_column.get(), 0, insertDataStruct.numRows);
       insertDataStruct.data.emplace_back(DataBlockPtr{data_for_deleted_column.get()});
       insertDataStruct.columnIds.push_back(cit.second.get_column_desc()->columnId);
       insertDataStruct.columnDescriptors[cit.first] = cit.second.get_column_desc();
       break;
     }
   }
   // MAT we need to add a removal of the empty column we pushed onto here
   // for upstream safety.  Should not be a problem but need to fix.
 
   // insert column to columnMap_ if not yet (ALTER ADD COLUMN)
   for (const auto columnId : insertDataStruct.columnIds) {
     if (columnMap_.end() == columnMap_.find(columnId)) {
       columnMap_.emplace(
           columnId, Chunk_NS::Chunk(insertDataStruct.columnDescriptors.at(columnId)));
     }
   }
 
   // when replicate (add) column(s), this path seems wont work; go separate route...
   if (insertDataStruct.replicate_count > 0) {
     replicateData(insertDataStruct);
     return;
   }
 
   std::unordered_map<int, int> inverseInsertDataColIdMap;
   for (size_t insertId = 0; insertId < insertDataStruct.columnIds.size(); ++insertId) {
     inverseInsertDataColIdMap.insert(
         std::make_pair(insertDataStruct.columnIds[insertId], insertId));
   }
 
   size_t numRowsLeft = insertDataStruct.numRows;
   size_t numRowsInserted = 0;
   vector<DataBlockPtr> dataCopy =
       insertDataStruct.data;  // bc append data will move ptr forward and this violates
                               // constness of InsertData
   if (numRowsLeft <= 0) {
     return;
   }
 
   FragmentInfo* currentFragment = 0;
 
   if (fragmentInfoVec_.empty()) {  // if no fragments exist for table
     currentFragment = createNewFragment(defaultInsertLevel_);
   } else {
     currentFragment = &(fragmentInfoVec_.back());
   }
   size_t startFragment = fragmentInfoVec_.size() - 1;
 
   while (numRowsLeft > 0) {  // may have to create multiple fragments for bulk insert
     // loop until done inserting all rows
     CHECK_LE(currentFragment->shadowNumTuples, maxFragmentRows_);
     size_t rowsLeftInCurrentFragment =
         maxFragmentRows_ - currentFragment->shadowNumTuples;
     size_t numRowsToInsert = min(rowsLeftInCurrentFragment, numRowsLeft);
     if (rowsLeftInCurrentFragment != 0) {
       for (auto& varLenColInfoIt : varLenColInfo_) {
         CHECK_LE(varLenColInfoIt.second, maxChunkSize_);
         size_t bytesLeft = maxChunkSize_ - varLenColInfoIt.second;
         auto insertIdIt = inverseInsertDataColIdMap.find(varLenColInfoIt.first);
         if (insertIdIt != inverseInsertDataColIdMap.end()) {
           auto colMapIt = columnMap_.find(varLenColInfoIt.first);
           numRowsToInsert = std::min(
               numRowsToInsert,
               colMapIt->second.getNumElemsForBytesInsertData(dataCopy[insertIdIt->second],
                                                              numRowsToInsert,
                                                              numRowsInserted,
                                                              bytesLeft));
         }
       }
     }
 
     if (rowsLeftInCurrentFragment == 0 || numRowsToInsert == 0) {
       currentFragment = createNewFragment(defaultInsertLevel_);
       if (numRowsInserted == 0) {
         startFragment++;
       }
       rowsLeftInCurrentFragment = maxFragmentRows_;
       for (auto& varLenColInfoIt : varLenColInfo_) {
         varLenColInfoIt.second = 0;  // reset byte counter
       }
       numRowsToInsert = min(rowsLeftInCurrentFragment, numRowsLeft);
       for (auto& varLenColInfoIt : varLenColInfo_) {
         CHECK_LE(varLenColInfoIt.second, maxChunkSize_);
         size_t bytesLeft = maxChunkSize_ - varLenColInfoIt.second;
         auto insertIdIt = inverseInsertDataColIdMap.find(varLenColInfoIt.first);
         if (insertIdIt != inverseInsertDataColIdMap.end()) {
           auto colMapIt = columnMap_.find(varLenColInfoIt.first);
           numRowsToInsert = std::min(
               numRowsToInsert,
               colMapIt->second.getNumElemsForBytesInsertData(dataCopy[insertIdIt->second],
                                                              numRowsToInsert,
                                                              numRowsInserted,
                                                              bytesLeft));
         }
       }
     }
 
     CHECK_GT(numRowsToInsert, size_t(0));  // would put us into an endless loop as we'd
                                            // never be able to insert anything
 
     // for each column, append the data in the appropriate insert buffer
     for (size_t i = 0; i < insertDataStruct.columnIds.size(); ++i) {
       int columnId = insertDataStruct.columnIds[i];
       auto colMapIt = columnMap_.find(columnId);
       CHECK(colMapIt != columnMap_.end());
       currentFragment->shadowChunkMetadataMap[columnId] =
           colMapIt->second.appendData(dataCopy[i], numRowsToInsert, numRowsInserted);
       auto varLenColInfoIt = varLenColInfo_.find(columnId);
       if (varLenColInfoIt != varLenColInfo_.end()) {
         varLenColInfoIt->second = colMapIt->second.get_buffer()->size();
       }
     }
     if (hasMaterializedRowId_) {
       size_t startId = maxFragmentRows_ * currentFragment->fragmentId +
                        currentFragment->shadowNumTuples;
       int64_t* rowIdData = new int64_t[numRowsToInsert];
       for (size_t i = 0; i < numRowsToInsert; ++i) {
         rowIdData[i] = i + startId;
       }
       DataBlockPtr rowIdBlock;
       rowIdBlock.numbersPtr = reinterpret_cast<int8_t*>(rowIdData);
       auto colMapIt = columnMap_.find(rowIdColId_);
       currentFragment->shadowChunkMetadataMap[rowIdColId_] =
           colMapIt->second.appendData(rowIdBlock, numRowsToInsert, numRowsInserted);
       delete[] rowIdData;
     }
 
     currentFragment->shadowNumTuples =
         fragmentInfoVec_.back().getPhysicalNumTuples() + numRowsToInsert;
     numRowsLeft -= numRowsToInsert;
     numRowsInserted += numRowsToInsert;
   }
   {
     // Only take the fragment info lock when updating fragment info map. Otherwise,
     // potential deadlock can occur after SELECT has locked TableReadLock and COPY_FROM
     // has locked fragmentInfoMutex_ while SELECT waits for fragmentInfoMutex_ and
     // COPY_FROM waits for TableWriteLock
     mapd_unique_lock<mapd_shared_mutex> writeLock(fragmentInfoMutex_);
     for (auto partIt = fragmentInfoVec_.begin() + startFragment;
          partIt != fragmentInfoVec_.end();
          ++partIt) {
       partIt->setPhysicalNumTuples(partIt->shadowNumTuples);
       partIt->setChunkMetadataMap(partIt->shadowChunkMetadataMap);
     }
   }
   numTuples_ += insertDataStruct.numRows;
   dropFragmentsToSize(maxRows_);
 }

Here is the call graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::insertDataNoCheckpoint ( InsertData & insertDataStruct )

overridevirtual

Given data wrapped in an InsertData struct, inserts it into the correct partitions No locks and checkpoints taken needs to be managed externally.

Implements Fragmenter_Namespace::AbstractFragmenter.

Reimplemented in Fragmenter_Namespace::SortedOrderFragmenter.

Definition at line 344 of file InsertOrderFragmenter.cpp.

Referenced by Fragmenter_Namespace::SortedOrderFragmenter::insertDataNoCheckpoint(), and updateColumns().

                                                                                {
   // TODO: this local lock will need to be centralized when ALTER COLUMN is added, bc
   mapd_unique_lock<mapd_shared_mutex> insertLock(
       insertMutex_);  // prevent two threads from trying to insert into the same table
                       // simultaneously
   insertDataImpl(insertDataStruct);
 }

Here is the caller graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::lockInsertCheckpointData ( const InsertData & insertDataStruct )

protected

InsertOrderFragmenter& Fragmenter_Namespace::InsertOrderFragmenter::operator= ( const InsertOrderFragmenter & )

protected

void Fragmenter_Namespace::InsertOrderFragmenter::replicateData ( const InsertData & insertDataStruct )

protected

Definition at line 352 of file InsertOrderFragmenter.cpp.

References Fragmenter_Namespace::InsertData::bypass, CHECK(), Fragmenter_Namespace::InsertData::columnDescriptors, Fragmenter_Namespace::InsertData::columnIds, Fragmenter_Namespace::InsertData::data, and Fragmenter_Namespace::InsertData::numRows.

                                                                             {
   size_t numRowsLeft = insertDataStruct.numRows;
   for (auto& fragmentInfo : fragmentInfoVec_) {
     fragmentInfo.shadowChunkMetadataMap = fragmentInfo.getChunkMetadataMapPhysical();
     auto numRowsToInsert = fragmentInfo.getPhysicalNumTuples();  // not getNumTuples()
     size_t numRowsCanBeInserted;
     for (size_t i = 0; i < insertDataStruct.columnIds.size(); i++) {
       if (insertDataStruct.bypass[i]) {
         continue;
       }
       auto columnId = insertDataStruct.columnIds[i];
       auto colDesc = insertDataStruct.columnDescriptors.at(columnId);
       CHECK(columnMap_.find(columnId) != columnMap_.end());
 
       ChunkKey chunkKey = chunkKeyPrefix_;
       chunkKey.push_back(columnId);
       chunkKey.push_back(fragmentInfo.fragmentId);
 
       auto colMapIt = columnMap_.find(columnId);
       auto& chunk = colMapIt->second;
       if (chunk.isChunkOnDevice(
               dataMgr_,
               chunkKey,
               defaultInsertLevel_,
               fragmentInfo.deviceIds[static_cast<int>(defaultInsertLevel_)])) {
         dataMgr_->deleteChunksWithPrefix(chunkKey);
       }
       chunk.createChunkBuffer(
           dataMgr_,
           chunkKey,
           defaultInsertLevel_,
           fragmentInfo.deviceIds[static_cast<int>(defaultInsertLevel_)]);
       chunk.init_encoder();
 
       try {
         DataBlockPtr dataCopy = insertDataStruct.data[i];
         auto size = colDesc->columnType.get_size();
         if (0 > size) {
           std::unique_lock<std::mutex> lck(*mutex_access_inmem_states);
           varLenColInfo_[columnId] = 0;
           numRowsCanBeInserted = chunk.getNumElemsForBytesInsertData(
               dataCopy, numRowsToInsert, 0, maxChunkSize_, true);
         } else {
           numRowsCanBeInserted = maxChunkSize_ / size;
         }
 
         // FIXME: abort a case in which new column is wider than existing columns
         if (numRowsCanBeInserted < numRowsToInsert) {
           throw std::runtime_error("new column '" + colDesc->columnName +
                                    "' wider than existing columns is not supported");
         }
 
         auto chunkMetadata = chunk.appendData(dataCopy, numRowsToInsert, 0, true);
         {
           std::unique_lock<std::mutex> lck(*fragmentInfo.mutex_access_inmem_states);
           fragmentInfo.shadowChunkMetadataMap[columnId] = chunkMetadata;
         }
 
         // update total size of var-len column in (actually the last) fragment
         if (0 > size) {
           std::unique_lock<std::mutex> lck(*mutex_access_inmem_states);
           varLenColInfo_[columnId] = chunk.get_buffer()->size();
         }
       } catch (...) {
         dataMgr_->deleteChunksWithPrefix(chunkKey);
         throw;
       }
     }
     numRowsLeft -= numRowsToInsert;
   }
   CHECK(0 == numRowsLeft);
 
   mapd_unique_lock<mapd_shared_mutex> writeLock(fragmentInfoMutex_);
   for (auto& fragmentInfo : fragmentInfoVec_) {
     fragmentInfo.setChunkMetadataMap(fragmentInfo.shadowChunkMetadataMap);
   }
 }

Here is the call graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::setNumRows ( const size_t numTuples )

inlineoverridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 108 of file InsertOrderFragmenter.h.

References numTuples_.

108 { numTuples_ = numTuples; }

Fragmenter_Namespace::InsertOrderFragmenter::numTuples_

size_t numTuples_

Definition: InsertOrderFragmenter.h:193

void Fragmenter_Namespace::InsertOrderFragmenter::updateChunkStats	(	const ColumnDescriptor *	cd,
		std::unordered_map< int, ChunkStats > &	stats_map
	)

overridevirtual

Update chunk stats.

WARNING: This method is entirely unlocked. Higher level locks are expected to prevent any table read or write during a chunk metadata update, since we need to modify various buffers and metadata maps.

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 233 of file InsertOrderFragmenter.cpp.

References catalog_(), CHECK(), ChunkMetadata::chunkStats, ColumnDescriptor::columnId, ColumnDescriptor::columnType, DatumToString(), Data_Namespace::DISK_LEVEL, get_logical_type_info(), Chunk_NS::Chunk::getChunk(), LOG, showChunk(), VLOG, and logger::WARNING.

                                                     {
   if (shard_ >= 0) {
     LOG(WARNING) << "Skipping chunk stats update for logical table " << physicalTableId_;
   }
 
   CHECK(cd);
   const auto column_id = cd->columnId;
   const auto col_itr = columnMap_.find(column_id);
   CHECK(col_itr != columnMap_.end());
 
   for (auto& fragment : fragmentInfoVec_) {
     auto stats_itr = stats_map.find(fragment.fragmentId);
     if (stats_itr != stats_map.end()) {
       auto chunk_meta_it = fragment.getChunkMetadataMapPhysical().find(column_id);
       CHECK(chunk_meta_it != fragment.getChunkMetadataMapPhysical().end());
       ChunkKey chunk_key{catalog_->getCurrentDB().dbId,
                          physicalTableId_,
                          column_id,
                          fragment.fragmentId};
       auto chunk = Chunk_NS::Chunk::getChunk(cd,
                                              &catalog_->getDataMgr(),
                                              chunk_key,
                                              Data_Namespace::MemoryLevel::DISK_LEVEL,
                                              0,
                                              chunk_meta_it->second.numBytes,
                                              chunk_meta_it->second.numElements);
       auto buf = chunk->get_buffer();
       CHECK(buf);
       auto encoder = buf->encoder.get();
       if (!encoder) {
         throw std::runtime_error("No encoder for chunk " + showChunk(chunk_key));
       }
 
       auto chunk_stats = stats_itr->second;
 
       ChunkMetadata old_chunk_metadata;
       encoder->getMetadata(old_chunk_metadata);
       auto& old_chunk_stats = old_chunk_metadata.chunkStats;
 
       const bool didResetStats = encoder->resetChunkStats(chunk_stats);
       // Use the logical type to display data, since the encoding should be ignored
       const auto logical_ti = get_logical_type_info(cd->columnType);
       if (!didResetStats) {
         VLOG(3) << "Skipping chunk stats reset for " << showChunk(chunk_key);
         VLOG(3) << "Max: " << DatumToString(old_chunk_stats.max, logical_ti) << " -> "
                 << DatumToString(chunk_stats.max, logical_ti);
         VLOG(3) << "Min: " << DatumToString(old_chunk_stats.min, logical_ti) << " -> "
                 << DatumToString(chunk_stats.min, logical_ti);
         VLOG(3) << "Nulls: " << (chunk_stats.has_nulls ? "True" : "False");
         continue;  // move to next fragment
       }
 
       VLOG(2) << "Resetting chunk stats for " << showChunk(chunk_key);
       VLOG(2) << "Max: " << DatumToString(old_chunk_stats.max, logical_ti) << " -> "
               << DatumToString(chunk_stats.max, logical_ti);
       VLOG(2) << "Min: " << DatumToString(old_chunk_stats.min, logical_ti) << " -> "
               << DatumToString(chunk_stats.min, logical_ti);
       VLOG(2) << "Nulls: " << (chunk_stats.has_nulls ? "True" : "False");
 
       // Reset fragment metadata map and set buffer to dirty
       ChunkMetadata new_metadata;
       // Run through fillChunkStats to ensure any transformations to the raw metadata
       // values get applied (e.g. for date in days)
       encoder->getMetadata(new_metadata);
       fragment.setChunkMetadata(column_id, new_metadata);
       fragment.shadowChunkMetadataMap =
           fragment.getChunkMetadataMap();  // TODO(adb): needed?
       buf->setDirty();
     } else {
       LOG(WARNING) << "No chunk stats update found for fragment " << fragment.fragmentId
                    << ", table " << physicalTableId_ << ", "
                    << ", column " << column_id;
     }
   }
 }

Here is the call graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::updateColumn	(	const Catalog_Namespace::Catalog *	catalog,
		const std::string &	tab_name,
		const std::string &	col_name,
		const int	fragment_id,
		const std::vector< uint64_t > &	frag_offsets,
		const std::vector< ScalarTargetValue > &	rhs_values,
		const SQLTypeInfo &	rhs_type,
		const Data_Namespace::MemoryLevel	memory_level,
		UpdelRoll &	updel_roll
	)

static

Definition at line 63 of file UpdelStorage.cpp.

References CHECK(), Catalog_Namespace::Catalog::getMetadataForColumn(), and Catalog_Namespace::Catalog::getMetadataForTable().

Referenced by updateColumn().

                                                                 {
   const auto td = catalog->getMetadataForTable(tab_name);
   CHECK(td);
   const auto cd = catalog->getMetadataForColumn(td->tableId, col_name);
   CHECK(cd);
   td->fragmenter->updateColumn(catalog,
                                td,
                                cd,
                                fragment_id,
                                frag_offsets,
                                rhs_values,
                                rhs_type,
                                memory_level,
                                updel_roll);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::updateColumn	(	const Catalog_Namespace::Catalog *	catalog,
		const TableDescriptor *	td,
		const ColumnDescriptor *	cd,
		const int	fragment_id,
		const std::vector< uint64_t > &	frag_offsets,
		const std::vector< ScalarTargetValue > &	rhs_values,
		const SQLTypeInfo &	rhs_type,
		const Data_Namespace::MemoryLevel	memory_level,
		UpdelRoll &	updel_roll
	)

overridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 610 of file UpdelStorage.cpp.

                                                                 {
   updel_roll.catalog = catalog;
   updel_roll.logicalTableId = catalog->getLogicalTableId(td->tableId);
   updel_roll.memoryLevel = memory_level;
 
   const size_t ncore = cpu_threads();
   const auto nrow = frag_offsets.size();
   const auto n_rhs_values = rhs_values.size();
   if (0 == nrow) {
     return;
   }
   CHECK(nrow == n_rhs_values || 1 == n_rhs_values);
 
   auto& fragment = getFragmentInfoFromId(fragment_id);
   auto chunk_meta_it = fragment.getChunkMetadataMapPhysical().find(cd->columnId);
   CHECK(chunk_meta_it != fragment.getChunkMetadataMapPhysical().end());
   ChunkKey chunk_key{
       catalog->getCurrentDB().dbId, td->tableId, cd->columnId, fragment.fragmentId};
   auto chunk = Chunk_NS::Chunk::getChunk(cd,
                                          &catalog->getDataMgr(),
                                          chunk_key,
                                          Data_Namespace::CPU_LEVEL,
                                          0,
                                          chunk_meta_it->second.numBytes,
                                          chunk_meta_it->second.numElements);
 
   std::vector<int8_t> has_null_per_thread(ncore, 0);
   std::vector<double> max_double_per_thread(ncore, std::numeric_limits<double>::lowest());
   std::vector<double> min_double_per_thread(ncore, std::numeric_limits<double>::max());
   std::vector<int64_t> max_int64t_per_thread(ncore, std::numeric_limits<int64_t>::min());
   std::vector<int64_t> min_int64t_per_thread(ncore, std::numeric_limits<int64_t>::max());
 
   // parallel update elements
   std::vector<std::future<void>> threads;
 
   const auto segsz = (nrow + ncore - 1) / ncore;
   auto dbuf = chunk->get_buffer();
   auto dbuf_addr = dbuf->getMemoryPtr();
   dbuf->setUpdated();
   {
     std::lock_guard<std::mutex> lck(updel_roll.mutex);
     if (updel_roll.dirtyChunks.count(chunk.get()) == 0) {
       updel_roll.dirtyChunks.emplace(chunk.get(), chunk);
     }
 
     ChunkKey chunkey{updel_roll.catalog->getCurrentDB().dbId,
                      cd->tableId,
                      cd->columnId,
                      fragment.fragmentId};
     updel_roll.dirtyChunkeys.insert(chunkey);
   }
   for (size_t rbegin = 0, c = 0; rbegin < nrow; ++c, rbegin += segsz) {
     threads.emplace_back(std::async(
         std::launch::async,
         [=,
          &has_null_per_thread,
          &min_int64t_per_thread,
          &max_int64t_per_thread,
          &min_double_per_thread,
          &max_double_per_thread,
          &frag_offsets,
          &rhs_values] {
           SQLTypeInfo lhs_type = cd->columnType;
 
           // !! not sure if this is a undocumented convention or a bug, but for a sharded
           // table the dictionary id of a encoded string column is not specified by
           // comp_param in physical table but somehow in logical table :) comp_param in
           // physical table is always 0, so need to adapt accordingly...
           auto cdl = (shard_ < 0)
                          ? cd
                          : catalog->getMetadataForColumn(
                                catalog->getLogicalTableId(td->tableId), cd->columnId);
           CHECK(cdl);
           DecimalOverflowValidator decimalOverflowValidator(lhs_type);
           NullAwareValidator<DecimalOverflowValidator> nullAwareDecimalOverflowValidator(
               lhs_type, &decimalOverflowValidator);
           DateDaysOverflowValidator dateDaysOverflowValidator(lhs_type);
           NullAwareValidator<DateDaysOverflowValidator> nullAwareDateOverflowValidator(
               lhs_type, &dateDaysOverflowValidator);
 
           StringDictionary* stringDict{nullptr};
           if (lhs_type.is_string()) {
             CHECK(kENCODING_DICT == lhs_type.get_compression());
             auto dictDesc = const_cast<DictDescriptor*>(
                 catalog->getMetadataForDict(cdl->columnType.get_comp_param()));
             CHECK(dictDesc);
             stringDict = dictDesc->stringDict.get();
             CHECK(stringDict);
           }
 
           for (size_t r = rbegin; r < std::min(rbegin + segsz, nrow); r++) {
             const auto roffs = frag_offsets[r];
             auto data_ptr = dbuf_addr + roffs * get_element_size(lhs_type);
             auto sv = &rhs_values[1 == n_rhs_values ? 0 : r];
             ScalarTargetValue sv2;
 
             // Subtle here is on the two cases of string-to-string assignments, when
             // upstream passes RHS string as a string index instead of a preferred "real
             // string".
             //   case #1. For "SET str_col = str_literal", it is hard to resolve temp str
             //   index
             //            in this layer, so if upstream passes a str idx here, an
             //            exception is thrown.
             //   case #2. For "SET str_col1 = str_col2", RHS str idx is converted to LHS
             //   str idx.
             if (rhs_type.is_string()) {
               if (const auto vp = boost::get<int64_t>(sv)) {
                 auto dictDesc = const_cast<DictDescriptor*>(
                     catalog->getMetadataForDict(rhs_type.get_comp_param()));
                 if (nullptr == dictDesc) {
                   throw std::runtime_error(
                       "UPDATE does not support cast from string literal to string "
                       "column.");
                 }
                 auto stringDict = dictDesc->stringDict.get();
                 CHECK(stringDict);
                 sv2 = NullableString(stringDict->getString(*vp));
                 sv = &sv2;
               }
             }
 
             if (const auto vp = boost::get<int64_t>(sv)) {
               auto v = *vp;
               if (lhs_type.is_string()) {
                 throw std::runtime_error("UPDATE does not support cast to string.");
               }
               put_scalar<int64_t>(data_ptr, lhs_type, v, cd->columnName, &rhs_type);
               if (lhs_type.is_decimal()) {
                 nullAwareDecimalOverflowValidator.validate<int64_t>(v);
                 int64_t decimal_val;
                 get_scalar<int64_t>(data_ptr, lhs_type, decimal_val);
                 tabulate_metadata(lhs_type,
                                   min_int64t_per_thread[c],
                                   max_int64t_per_thread[c],
                                   has_null_per_thread[c],
                                   (v == inline_int_null_value<int64_t>() &&
                                    lhs_type.get_notnull() == false)
                                       ? v
                                       : decimal_val);
                 auto const positive_v_and_negative_d = (v >= 0) && (decimal_val < 0);
                 auto const negative_v_and_positive_d = (v < 0) && (decimal_val >= 0);
                 if (positive_v_and_negative_d || negative_v_and_positive_d) {
                   throw std::runtime_error(
                       "Data conversion overflow on " + std::to_string(v) +
                       " from DECIMAL(" + std::to_string(rhs_type.get_dimension()) + ", " +
                       std::to_string(rhs_type.get_scale()) + ") to (" +
                       std::to_string(lhs_type.get_dimension()) + ", " +
                       std::to_string(lhs_type.get_scale()) + ")");
                 }
               } else if (is_integral(lhs_type)) {
                 if (lhs_type.is_date_in_days()) {
                   // Store meta values in seconds
                   if (lhs_type.get_size() == 2) {
                     nullAwareDateOverflowValidator.validate<int16_t>(v);
                   } else {
                     nullAwareDateOverflowValidator.validate<int32_t>(v);
                   }
                   int64_t days;
                   get_scalar<int64_t>(data_ptr, lhs_type, days);
                   const auto seconds = DateConverters::get_epoch_seconds_from_days(days);
                   tabulate_metadata(lhs_type,
                                     min_int64t_per_thread[c],
                                     max_int64t_per_thread[c],
                                     has_null_per_thread[c],
                                     (v == inline_int_null_value<int64_t>() &&
                                      lhs_type.get_notnull() == false)
                                         ? NullSentinelSupplier()(lhs_type, v)
                                         : seconds);
                 } else {
                   int64_t target_value;
                   if (rhs_type.is_decimal()) {
                     target_value = round(decimal_to_double(rhs_type, v));
                   } else {
                     target_value = v;
                   }
                   tabulate_metadata(lhs_type,
                                     min_int64t_per_thread[c],
                                     max_int64t_per_thread[c],
                                     has_null_per_thread[c],
                                     target_value);
                 }
               } else {
                 tabulate_metadata(
                     lhs_type,
                     min_double_per_thread[c],
                     max_double_per_thread[c],
                     has_null_per_thread[c],
                     rhs_type.is_decimal() ? decimal_to_double(rhs_type, v) : v);
               }
             } else if (const auto vp = boost::get<double>(sv)) {
               auto v = *vp;
               if (lhs_type.is_string()) {
                 throw std::runtime_error("UPDATE does not support cast to string.");
               }
               put_scalar<double>(data_ptr, lhs_type, v, cd->columnName);
               if (lhs_type.is_integer()) {
                 tabulate_metadata(lhs_type,
                                   min_int64t_per_thread[c],
                                   max_int64t_per_thread[c],
                                   has_null_per_thread[c],
                                   int64_t(v));
               } else if (lhs_type.is_fp()) {
                 tabulate_metadata(lhs_type,
                                   min_double_per_thread[c],
                                   max_double_per_thread[c],
                                   has_null_per_thread[c],
                                   double(v));
               } else {
                 UNREACHABLE() << "Unexpected combination of a non-floating or integer "
                                  "LHS with a floating RHS.";
               }
             } else if (const auto vp = boost::get<float>(sv)) {
               auto v = *vp;
               if (lhs_type.is_string()) {
                 throw std::runtime_error("UPDATE does not support cast to string.");
               }
               put_scalar<float>(data_ptr, lhs_type, v, cd->columnName);
               if (lhs_type.is_integer()) {
                 tabulate_metadata(lhs_type,
                                   min_int64t_per_thread[c],
                                   max_int64t_per_thread[c],
                                   has_null_per_thread[c],
                                   int64_t(v));
               } else {
                 tabulate_metadata(lhs_type,
                                   min_double_per_thread[c],
                                   max_double_per_thread[c],
                                   has_null_per_thread[c],
                                   double(v));
               }
             } else if (const auto vp = boost::get<NullableString>(sv)) {
               const auto s = boost::get<std::string>(vp);
               const auto sval = s ? *s : std::string("");
               if (lhs_type.is_string()) {
                 decltype(stringDict->getOrAdd(sval)) sidx;
                 {
                   std::unique_lock<std::mutex> lock(temp_mutex_);
                   sidx = stringDict->getOrAdd(sval);
                 }
                 put_scalar<int64_t>(data_ptr, lhs_type, sidx, cd->columnName);
                 tabulate_metadata(lhs_type,
                                   min_int64t_per_thread[c],
                                   max_int64t_per_thread[c],
                                   has_null_per_thread[c],
                                   int64_t(sidx));
               } else if (sval.size() > 0) {
                 auto dval = std::atof(sval.data());
                 if (lhs_type.is_boolean()) {
                   dval = sval == "t" || sval == "true" || sval == "T" || sval == "True";
                 } else if (lhs_type.is_time()) {
                   throw std::runtime_error(
                       "Date/Time/Timestamp update not supported through translated "
                       "string path.");
                 }
                 if (lhs_type.is_fp() || lhs_type.is_decimal()) {
                   put_scalar<double>(data_ptr, lhs_type, dval, cd->columnName);
                   tabulate_metadata(lhs_type,
                                     min_double_per_thread[c],
                                     max_double_per_thread[c],
                                     has_null_per_thread[c],
                                     double(dval));
                 } else {
                   put_scalar<int64_t>(data_ptr, lhs_type, dval, cd->columnName);
                   tabulate_metadata(lhs_type,
                                     min_int64t_per_thread[c],
                                     max_int64t_per_thread[c],
                                     has_null_per_thread[c],
                                     int64_t(dval));
                 }
               } else {
                 put_null(data_ptr, lhs_type, cd->columnName);
                 has_null_per_thread[c] = true;
               }
             } else {
               CHECK(false);
             }
           }
         }));
     if (threads.size() >= (size_t)cpu_threads()) {
       wait_cleanup_threads(threads);
     }
   }
   wait_cleanup_threads(threads);
 
   // for unit test
   if (Fragmenter_Namespace::FragmentInfo::unconditionalVacuum_) {
     if (cd->isDeletedCol) {
       const auto deleted_offsets = getVacuumOffsets(chunk);
       if (deleted_offsets.size() > 0) {
         compactRows(catalog, td, fragment_id, deleted_offsets, memory_level, updel_roll);
         return;
       }
     }
   }
   bool has_null_per_chunk{false};
   double max_double_per_chunk{std::numeric_limits<double>::lowest()};
   double min_double_per_chunk{std::numeric_limits<double>::max()};
   int64_t max_int64t_per_chunk{std::numeric_limits<int64_t>::min()};
   int64_t min_int64t_per_chunk{std::numeric_limits<int64_t>::max()};
   for (size_t c = 0; c < ncore; ++c) {
     has_null_per_chunk = has_null_per_chunk || has_null_per_thread[c];
     max_double_per_chunk =
         std::max<double>(max_double_per_chunk, max_double_per_thread[c]);
     min_double_per_chunk =
         std::min<double>(min_double_per_chunk, min_double_per_thread[c]);
     max_int64t_per_chunk =
         std::max<int64_t>(max_int64t_per_chunk, max_int64t_per_thread[c]);
     min_int64t_per_chunk =
         std::min<int64_t>(min_int64t_per_chunk, min_int64t_per_thread[c]);
   }
   updateColumnMetadata(cd,
                        fragment,
                        chunk,
                        has_null_per_chunk,
                        max_double_per_chunk,
                        min_double_per_chunk,
                        max_int64t_per_chunk,
                        min_int64t_per_chunk,
                        cd->columnType,
                        updel_roll);
 }

Here is the call graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::updateColumn	(	const Catalog_Namespace::Catalog *	catalog,
		const TableDescriptor *	td,
		const ColumnDescriptor *	cd,
		const int	fragment_id,
		const std::vector< uint64_t > &	frag_offsets,
		const ScalarTargetValue &	rhs_value,
		const SQLTypeInfo &	rhs_type,
		const Data_Namespace::MemoryLevel	memory_level,
		UpdelRoll &	updel_roll
	)

overridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 87 of file UpdelStorage.cpp.

References updateColumn().

                                                                 {
   updateColumn(catalog,
                td,
                cd,
                fragment_id,
                frag_offsets,
                std::vector<ScalarTargetValue>(1, rhs_value),
                rhs_type,
                memory_level,
                updel_roll);
 }

Here is the call graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::updateColumnMetadata	(	const ColumnDescriptor *	cd,
		FragmentInfo &	fragment,
		std::shared_ptr< Chunk_NS::Chunk >	chunk,
		const bool	null,
		const double	dmax,
		const double	dmin,
		const int64_t	lmax,
		const int64_t	lmin,
		const SQLTypeInfo &	rhs_type,
		UpdelRoll &	updel_roll
	)

overridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 940 of file UpdelStorage.cpp.

References UpdelRoll::catalog, UpdelRoll::chunkMetadata, ColumnDescriptor::columnId, ColumnDescriptor::columnType, Fragmenter_Namespace::FragmentInfo::getChunkMetadataMapPhysical(), Catalog_Namespace::Catalog::getMetadataForTable(), SQLTypeInfoCore< TYPE_FACET_PACK >::is_decimal(), Fragmenter_Namespace::is_integral(), kENCODING_DICT, UpdelRoll::mutex, UpdelRoll::numTuples, Fragmenter_Namespace::FragmentInfo::shadowNumTuples, and ColumnDescriptor::tableId.

Referenced by compactRows(), and updateColumn().

                                                                         {
   auto td = updel_roll.catalog->getMetadataForTable(cd->tableId);
   auto key = std::make_pair(td, &fragment);
   std::lock_guard<std::mutex> lck(updel_roll.mutex);
   if (0 == updel_roll.chunkMetadata.count(key)) {
     updel_roll.chunkMetadata[key] = fragment.getChunkMetadataMapPhysical();
   }
   if (0 == updel_roll.numTuples.count(key)) {
     updel_roll.numTuples[key] = fragment.shadowNumTuples;
   }
   auto& chunkMetadata = updel_roll.chunkMetadata[key];
 
   auto buffer = chunk->get_buffer();
   const auto& lhs_type = cd->columnType;
 
   auto update_stats = [& encoder = buffer->encoder](auto min, auto max, auto has_null) {
     static_assert(std::is_same<decltype(min), decltype(max)>::value,
                   "Type mismatch on min/max");
     if (has_null) {
       encoder->updateStats(decltype(min)(), true);
     }
     if (max < min) {
       return;
     }
     encoder->updateStats(min, false);
     encoder->updateStats(max, false);
   };
 
   if (is_integral(lhs_type) || (lhs_type.is_decimal() && rhs_type.is_decimal())) {
     update_stats(min_int64t_per_chunk, max_int64t_per_chunk, has_null_per_chunk);
   } else if (lhs_type.is_fp()) {
     update_stats(min_double_per_chunk, max_double_per_chunk, has_null_per_chunk);
   } else if (lhs_type.is_decimal()) {
     update_stats((int64_t)(min_double_per_chunk * pow(10, lhs_type.get_scale())),
                  (int64_t)(max_double_per_chunk * pow(10, lhs_type.get_scale())),
                  has_null_per_chunk);
   } else if (!lhs_type.is_array() && !lhs_type.is_geometry() &&
              !(lhs_type.is_string() && kENCODING_DICT != lhs_type.get_compression())) {
     update_stats(min_int64t_per_chunk, max_int64t_per_chunk, has_null_per_chunk);
   }
   buffer->encoder->getMetadata(chunkMetadata[cd->columnId]);
 
   // removed as @alex suggests. keep it commented in case of any chance to revisit
   // it once after vacuum code is introduced. fragment.invalidateChunkMetadataMap();
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::updateColumns	(	const Catalog_Namespace::Catalog *	catalog,
		const TableDescriptor *	td,
		const int	fragmentId,
		const std::vector< TargetMetaInfo >	sourceMetaInfo,
		const std::vector< const ColumnDescriptor * >	columnDescriptors,
		const RowDataProvider &	sourceDataProvider,
		const size_t	indexOffFragmentOffsetColumn,
		const Data_Namespace::MemoryLevel	memoryLevel,
		UpdelRoll &	updelRoll
	)

overridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 300 of file UpdelStorage.cpp.

                           {
   if (!is_feature_enabled<VarlenUpdates>()) {
     throw std::runtime_error("varlen UPDATE path not enabled.");
   }
 
   updelRoll.is_varlen_update = true;
   updelRoll.catalog = catalog;
   updelRoll.logicalTableId = catalog->getLogicalTableId(td->tableId);
   updelRoll.memoryLevel = memoryLevel;
 
   size_t num_entries = sourceDataProvider.getEntryCount();
   size_t num_rows = sourceDataProvider.getRowCount();
 
   if (0 == num_rows) {
     // bail out early
     return;
   }
 
   TargetValueConverterFactory factory;
 
   auto& fragment = getFragmentInfoFromId(fragmentId);
   std::vector<std::shared_ptr<Chunk_NS::Chunk>> chunks;
   get_chunks(catalog, td, fragment, memoryLevel, chunks);
   std::vector<std::unique_ptr<TargetValueConverter>> sourceDataConverters(
       columnDescriptors.size());
   std::vector<std::unique_ptr<ChunkToInsertDataConverter>> chunkConverters;
   std::shared_ptr<Executor> executor;
 
   if (g_enable_experimental_string_functions) {
     executor = Executor::getExecutor(catalog->getCurrentDB().dbId);
   }
 
   std::shared_ptr<Chunk_NS::Chunk> deletedChunk;
   for (size_t indexOfChunk = 0; indexOfChunk < chunks.size(); indexOfChunk++) {
     auto chunk = chunks[indexOfChunk];
     const auto chunk_cd = chunk->get_column_desc();
 
     if (chunk_cd->isDeletedCol) {
       deletedChunk = chunk;
       continue;
     }
 
     auto targetColumnIt = std::find_if(columnDescriptors.begin(),
                                        columnDescriptors.end(),
                                        [=](const ColumnDescriptor* cd) -> bool {
                                          return cd->columnId == chunk_cd->columnId;
                                        });
 
     if (targetColumnIt != columnDescriptors.end()) {
       auto indexOfTargetColumn = std::distance(columnDescriptors.begin(), targetColumnIt);
 
       auto sourceDataMetaInfo = sourceMetaInfo[indexOfTargetColumn];
       auto targetDescriptor = columnDescriptors[indexOfTargetColumn];
 
       ConverterCreateParameter param{
           num_rows,
           *catalog,
           sourceDataMetaInfo,
           targetDescriptor,
           targetDescriptor->columnType,
           !targetDescriptor->columnType.get_notnull(),
           sourceDataProvider.getLiteralDictionary(),
           g_enable_experimental_string_functions
               ? executor->getStringDictionaryProxy(
                     sourceDataMetaInfo.get_type_info().get_comp_param(),
                     executor->getRowSetMemoryOwner(),
                     true)
               : nullptr};
       auto converter = factory.create(param);
       sourceDataConverters[indexOfTargetColumn] = std::move(converter);
 
       if (targetDescriptor->columnType.is_geometry()) {
         // geometry columns are composites
         // need to skip chunks, depending on geo type
         switch (targetDescriptor->columnType.get_type()) {
           case kMULTIPOLYGON:
             indexOfChunk += 5;
             break;
           case kPOLYGON:
             indexOfChunk += 4;
             break;
           case kLINESTRING:
             indexOfChunk += 2;
             break;
           case kPOINT:
             indexOfChunk += 1;
             break;
           default:
             CHECK(false);  // not supported
         }
       }
     } else {
       if (chunk_cd->columnType.is_varlen() || chunk_cd->columnType.is_fixlen_array()) {
         std::unique_ptr<ChunkToInsertDataConverter> converter;
 
         if (chunk_cd->columnType.is_fixlen_array()) {
           converter =
               std::make_unique<FixedLenArrayChunkConverter>(num_rows, chunk.get());
         } else if (chunk_cd->columnType.is_string()) {
           converter = std::make_unique<StringChunkConverter>(num_rows, chunk.get());
         } else if (chunk_cd->columnType.is_geometry()) {
           // the logical geo column is a string column
           converter = std::make_unique<StringChunkConverter>(num_rows, chunk.get());
         } else {
           converter = std::make_unique<ArrayChunkConverter>(num_rows, chunk.get());
         }
 
         chunkConverters.push_back(std::move(converter));
 
       } else if (chunk_cd->columnType.is_date_in_days()) {
         /* Q: Why do we need this?
            A: In variable length updates path we move the chunk content of column
            without decoding. Since it again passes through DateDaysEncoder
            the expected value should be in seconds, but here it will be in days.
            Therefore, using DateChunkConverter chunk values are being scaled to
            seconds which then ultimately encoded in days in DateDaysEncoder.
         */
         std::unique_ptr<ChunkToInsertDataConverter> converter;
         const size_t physical_size = chunk_cd->columnType.get_size();
         if (physical_size == 2) {
           converter =
               std::make_unique<DateChunkConverter<int16_t>>(num_rows, chunk.get());
         } else if (physical_size == 4) {
           converter =
               std::make_unique<DateChunkConverter<int32_t>>(num_rows, chunk.get());
         } else {
           CHECK(false);
         }
         chunkConverters.push_back(std::move(converter));
       } else {
         std::unique_ptr<ChunkToInsertDataConverter> converter;
         SQLTypeInfo logical_type = get_logical_type_info(chunk_cd->columnType);
         int logical_size = logical_type.get_size();
         int physical_size = chunk_cd->columnType.get_size();
 
         if (logical_type.is_string()) {
           // for dicts -> logical = physical
           logical_size = physical_size;
         }
 
         if (8 == physical_size) {
           converter = std::make_unique<ScalarChunkConverter<int64_t, int64_t>>(
               num_rows, chunk.get());
         } else if (4 == physical_size) {
           if (8 == logical_size) {
             converter = std::make_unique<ScalarChunkConverter<int32_t, int64_t>>(
                 num_rows, chunk.get());
           } else {
             converter = std::make_unique<ScalarChunkConverter<int32_t, int32_t>>(
                 num_rows, chunk.get());
           }
         } else if (2 == chunk_cd->columnType.get_size()) {
           if (8 == logical_size) {
             converter = std::make_unique<ScalarChunkConverter<int16_t, int64_t>>(
                 num_rows, chunk.get());
           } else if (4 == logical_size) {
             converter = std::make_unique<ScalarChunkConverter<int16_t, int32_t>>(
                 num_rows, chunk.get());
           } else {
             converter = std::make_unique<ScalarChunkConverter<int16_t, int16_t>>(
                 num_rows, chunk.get());
           }
         } else if (1 == chunk_cd->columnType.get_size()) {
           if (8 == logical_size) {
             converter = std::make_unique<ScalarChunkConverter<int8_t, int64_t>>(
                 num_rows, chunk.get());
           } else if (4 == logical_size) {
             converter = std::make_unique<ScalarChunkConverter<int8_t, int32_t>>(
                 num_rows, chunk.get());
           } else if (2 == logical_size) {
             converter = std::make_unique<ScalarChunkConverter<int8_t, int16_t>>(
                 num_rows, chunk.get());
           } else {
             converter = std::make_unique<ScalarChunkConverter<int8_t, int8_t>>(
                 num_rows, chunk.get());
           }
         } else {
           CHECK(false);  // unknown
         }
 
         chunkConverters.push_back(std::move(converter));
       }
     }
   }
 
   static boost_variant_accessor<ScalarTargetValue> SCALAR_TARGET_VALUE_ACCESSOR;
   static boost_variant_accessor<int64_t> OFFSET_VALUE__ACCESSOR;
 
   updelRoll.dirtyChunks[deletedChunk.get()] = deletedChunk;
   ChunkKey chunkey{updelRoll.catalog->getCurrentDB().dbId,
                    deletedChunk->get_column_desc()->tableId,
                    deletedChunk->get_column_desc()->columnId,
                    fragment.fragmentId};
   updelRoll.dirtyChunkeys.insert(chunkey);
   bool* deletedChunkBuffer =
       reinterpret_cast<bool*>(deletedChunk->get_buffer()->getMemoryPtr());
 
   std::atomic<size_t> row_idx{0};
 
   auto row_converter = [&sourceDataProvider,
                         &sourceDataConverters,
                         &indexOffFragmentOffsetColumn,
                         &chunkConverters,
                         &deletedChunkBuffer,
                         &row_idx](size_t indexOfEntry) -> void {
     // convert the source data
     const auto row = sourceDataProvider.getEntryAt(indexOfEntry);
     if (row.empty()) {
       return;
     }
 
     size_t indexOfRow = row_idx.fetch_add(1);
 
     for (size_t col = 0; col < sourceDataConverters.size(); col++) {
       if (sourceDataConverters[col]) {
         const auto& mapd_variant = row[col];
         sourceDataConverters[col]->convertToColumnarFormat(indexOfRow, &mapd_variant);
       }
     }
 
     auto scalar = checked_get(
         indexOfRow, &row[indexOffFragmentOffsetColumn], SCALAR_TARGET_VALUE_ACCESSOR);
     auto indexInChunkBuffer = *checked_get(indexOfRow, scalar, OFFSET_VALUE__ACCESSOR);
 
     // convert the remaining chunks
     for (size_t idx = 0; idx < chunkConverters.size(); idx++) {
       chunkConverters[idx]->convertToColumnarFormat(indexOfRow, indexInChunkBuffer);
     }
 
     // now mark the row as deleted
     deletedChunkBuffer[indexInChunkBuffer] = true;
   };
 
   bool can_go_parallel = num_rows > 20000;
 
   if (can_go_parallel) {
     const size_t num_worker_threads = cpu_threads();
     std::vector<std::future<void>> worker_threads;
     for (size_t i = 0,
                 start_entry = 0,
                 stride = (num_entries + num_worker_threads - 1) / num_worker_threads;
          i < num_worker_threads && start_entry < num_entries;
          ++i, start_entry += stride) {
       const auto end_entry = std::min(start_entry + stride, num_rows);
       worker_threads.push_back(std::async(
           std::launch::async,
           [&row_converter](const size_t start, const size_t end) {
             for (size_t indexOfRow = start; indexOfRow < end; ++indexOfRow) {
               row_converter(indexOfRow);
             }
           },
           start_entry,
           end_entry));
     }
 
     for (auto& child : worker_threads) {
       child.wait();
     }
 
   } else {
     for (size_t entryIdx = 0; entryIdx < num_entries; entryIdx++) {
       row_converter(entryIdx);
     }
   }
 
   Fragmenter_Namespace::InsertData insert_data;
   insert_data.databaseId = catalog->getCurrentDB().dbId;
   insert_data.tableId = td->tableId;
 
   for (size_t i = 0; i < chunkConverters.size(); i++) {
     chunkConverters[i]->addDataBlocksToInsertData(insert_data);
     continue;
   }
 
   for (size_t i = 0; i < sourceDataConverters.size(); i++) {
     if (sourceDataConverters[i]) {
       sourceDataConverters[i]->addDataBlocksToInsertData(insert_data);
     }
     continue;
   }
 
   insert_data.numRows = num_rows;
   insertDataNoCheckpoint(insert_data);
 
   // update metdata
   if (!deletedChunk->get_buffer()->has_encoder) {
     deletedChunk->init_encoder();
   }
   deletedChunk->get_buffer()->encoder->updateStats(static_cast<int64_t>(true), false);
 
   auto& shadowDeletedChunkMeta =
       fragment.shadowChunkMetadataMap[deletedChunk->get_column_desc()->columnId];
   if (shadowDeletedChunkMeta.numElements >
       deletedChunk->get_buffer()->encoder->getNumElems()) {
     // the append will have populated shadow meta data, otherwise use existing num
     // elements
     deletedChunk->get_buffer()->encoder->setNumElems(shadowDeletedChunkMeta.numElements);
   }
   deletedChunk->get_buffer()->setUpdated();
 }

Here is the call graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::updateMetadata	(	const Catalog_Namespace::Catalog *	catalog,
		const MetaDataKey &	key,
		UpdelRoll &	updel_roll
	)

overridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 995 of file UpdelStorage.cpp.

References UpdelRoll::chunkMetadata, fragmentInfoMutex_, and UpdelRoll::numTuples.

                                                                   {
   mapd_unique_lock<mapd_shared_mutex> writeLock(fragmentInfoMutex_);
   if (updel_roll.chunkMetadata.count(key)) {
     auto& fragmentInfo = *key.second;
     const auto& chunkMetadata = updel_roll.chunkMetadata[key];
     fragmentInfo.shadowChunkMetadataMap = chunkMetadata;
     fragmentInfo.setChunkMetadataMap(chunkMetadata);
     fragmentInfo.shadowNumTuples = updel_roll.numTuples[key];
     fragmentInfo.setPhysicalNumTuples(fragmentInfo.shadowNumTuples);
     // TODO(ppan): When fragment-level compaction is enable, the following code
     // should suffice. When not (ie. existing code), we'll revert to update
     // InsertOrderFragmenter::varLenColInfo_
     /*
     for (const auto cit : chunkMetadata) {
       const auto& cd = *catalog->getMetadataForColumn(td->tableId, cit.first);
       if (cd.columnType.get_size() < 0)
         fragmentInfo.varLenColInfox[cd.columnId] = cit.second.numBytes;
     }
     */
   }
 }

auto Fragmenter_Namespace::InsertOrderFragmenter::vacuum_fixlen_rows	(	const FragmentInfo &	fragment,
		const std::shared_ptr< Chunk_NS::Chunk > &	chunk,
		const std::vector< uint64_t > &	frag_offsets
	)

protected

Definition at line 1117 of file UpdelStorage.cpp.

References anonymous_namespace{ResultSetReductionInterpreter.cpp}::get_element_size(), and Fragmenter_Namespace::FragmentInfo::getPhysicalNumTuples().

Referenced by compactRows().

                                              {
   const auto cd = chunk->get_column_desc();
   const auto& col_type = cd->columnType;
   auto data_buffer = chunk->get_buffer();
   auto data_addr = data_buffer->getMemoryPtr();
   auto element_size =
       col_type.is_fixlen_array() ? col_type.get_size() : get_element_size(col_type);
   int64_t irow_of_blk_to_keep = 0;  // head of next row block to keep
   int64_t irow_of_blk_to_fill = 0;  // row offset to fit the kept block
   size_t nbytes_fix_data_to_keep = 0;
   auto nrows_to_vacuum = frag_offsets.size();
   auto nrows_in_fragment = fragment.getPhysicalNumTuples();
   for (size_t irow = 0; irow <= nrows_to_vacuum; irow++) {
     auto is_last_one = irow == nrows_to_vacuum;
     auto irow_to_vacuum = is_last_one ? nrows_in_fragment : frag_offsets[irow];
     auto maddr_to_vacuum = data_addr;
     int64_t nrows_to_keep = irow_to_vacuum - irow_of_blk_to_keep;
     if (nrows_to_keep > 0) {
       auto nbytes_to_keep = nrows_to_keep * element_size;
       if (irow_of_blk_to_fill != irow_of_blk_to_keep) {
         // move curr fixlen row block toward front
         memmove(maddr_to_vacuum + irow_of_blk_to_fill * element_size,
                 maddr_to_vacuum + irow_of_blk_to_keep * element_size,
                 nbytes_to_keep);
       }
       irow_of_blk_to_fill += nrows_to_keep;
       nbytes_fix_data_to_keep += nbytes_to_keep;
     }
     irow_of_blk_to_keep = irow_to_vacuum + 1;
   }
   return nbytes_fix_data_to_keep;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

auto Fragmenter_Namespace::InsertOrderFragmenter::vacuum_varlen_rows	(	const FragmentInfo &	fragment,
		const std::shared_ptr< Chunk_NS::Chunk > &	chunk,
		const std::vector< uint64_t > &	frag_offsets
	)

protected

Definition at line 1153 of file UpdelStorage.cpp.

References Fragmenter_Namespace::FragmentInfo::getPhysicalNumTuples().

Referenced by compactRows().

                                              {
   auto data_buffer = chunk->get_buffer();
   auto index_buffer = chunk->get_index_buf();
   auto data_addr = data_buffer->getMemoryPtr();
   auto indices_addr = index_buffer ? index_buffer->getMemoryPtr() : nullptr;
   auto index_array = (StringOffsetT*)indices_addr;
   int64_t irow_of_blk_to_keep = 0;  // head of next row block to keep
   int64_t irow_of_blk_to_fill = 0;  // row offset to fit the kept block
   size_t nbytes_fix_data_to_keep = 0;
   size_t nbytes_var_data_to_keep = 0;
   auto nrows_to_vacuum = frag_offsets.size();
   auto nrows_in_fragment = fragment.getPhysicalNumTuples();
   for (size_t irow = 0; irow <= nrows_to_vacuum; irow++) {
     auto is_last_one = irow == nrows_to_vacuum;
     auto irow_to_vacuum = is_last_one ? nrows_in_fragment : frag_offsets[irow];
     auto maddr_to_vacuum = data_addr;
     int64_t nrows_to_keep = irow_to_vacuum - irow_of_blk_to_keep;
     if (nrows_to_keep > 0) {
       auto ibyte_var_data_to_keep = nbytes_var_data_to_keep;
       auto nbytes_to_keep =
           (is_last_one ? data_buffer->size() : index_array[irow_to_vacuum]) -
           index_array[irow_of_blk_to_keep];
       if (irow_of_blk_to_fill != irow_of_blk_to_keep) {
         // move curr varlen row block toward front
         memmove(data_addr + ibyte_var_data_to_keep,
                 data_addr + index_array[irow_of_blk_to_keep],
                 nbytes_to_keep);
 
         const auto index_base = index_array[irow_of_blk_to_keep];
         for (int64_t i = 0; i < nrows_to_keep; ++i) {
           auto& index = index_array[irow_of_blk_to_keep + i];
           index = ibyte_var_data_to_keep + (index - index_base);
         }
       }
       nbytes_var_data_to_keep += nbytes_to_keep;
       maddr_to_vacuum = indices_addr;
 
       constexpr static auto index_element_size = sizeof(StringOffsetT);
       nbytes_to_keep = nrows_to_keep * index_element_size;
       if (irow_of_blk_to_fill != irow_of_blk_to_keep) {
         // move curr fixlen row block toward front
         memmove(maddr_to_vacuum + irow_of_blk_to_fill * index_element_size,
                 maddr_to_vacuum + irow_of_blk_to_keep * index_element_size,
                 nbytes_to_keep);
       }
       irow_of_blk_to_fill += nrows_to_keep;
       nbytes_fix_data_to_keep += nbytes_to_keep;
     }
     irow_of_blk_to_keep = irow_to_vacuum + 1;
   }
   return nbytes_var_data_to_keep;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

Catalog_Namespace::Catalog* Fragmenter_Namespace::InsertOrderFragmenter::catalog_

protected

Definition at line 187 of file InsertOrderFragmenter.h.

Referenced by getChunksForAllColumns(), and Fragmenter_Namespace::SortedOrderFragmenter::sortData().

std::vector<int> Fragmenter_Namespace::InsertOrderFragmenter::chunkKeyPrefix_

protected

Definition at line 180 of file InsertOrderFragmenter.h.

Referenced by getChunkKeyPrefix(), and getFragmenterId().

std::map<int, Chunk_NS::Chunk> Fragmenter_Namespace::InsertOrderFragmenter::columnMap_

protected

stores a map of column id to metadata about that column

Definition at line 182 of file InsertOrderFragmenter.h.

Data_Namespace::DataMgr* Fragmenter_Namespace::InsertOrderFragmenter::dataMgr_

protected

Definition at line 186 of file InsertOrderFragmenter.h.

Data_Namespace::MemoryLevel Fragmenter_Namespace::InsertOrderFragmenter::defaultInsertLevel_

protected

Definition at line 203 of file InsertOrderFragmenter.h.

std::string Fragmenter_Namespace::InsertOrderFragmenter::fragmenterType_

protected

Definition at line 197 of file InsertOrderFragmenter.h.

Referenced by getFragmenterType().

mapd_shared_mutex Fragmenter_Namespace::InsertOrderFragmenter::fragmentInfoMutex_

protected

Definition at line 199 of file InsertOrderFragmenter.h.

Referenced by updateMetadata().

std::deque<FragmentInfo> Fragmenter_Namespace::InsertOrderFragmenter::fragmentInfoVec_

protected

data about each fragment stored - id and number of rows

Definition at line 184 of file InsertOrderFragmenter.h.

Referenced by getFragmentInfoFromId().

bool Fragmenter_Namespace::InsertOrderFragmenter::hasMaterializedRowId_

protected

Definition at line 204 of file InsertOrderFragmenter.h.

mapd_shared_mutex Fragmenter_Namespace::InsertOrderFragmenter::insertMutex_

protected

Definition at line 201 of file InsertOrderFragmenter.h.

size_t Fragmenter_Namespace::InsertOrderFragmenter::maxChunkSize_

protected

Definition at line 195 of file InsertOrderFragmenter.h.

int Fragmenter_Namespace::InsertOrderFragmenter::maxFragmentId_

protected

Definition at line 194 of file InsertOrderFragmenter.h.

size_t Fragmenter_Namespace::InsertOrderFragmenter::maxFragmentRows_

protected

Definition at line 190 of file InsertOrderFragmenter.h.

size_t Fragmenter_Namespace::InsertOrderFragmenter::maxRows_

protected

Definition at line 196 of file InsertOrderFragmenter.h.

std::shared_ptr<std::mutex> Fragmenter_Namespace::InsertOrderFragmenter::mutex_access_inmem_states

protected

Definition at line 207 of file InsertOrderFragmenter.h.

size_t Fragmenter_Namespace::InsertOrderFragmenter::numTuples_

protected

Definition at line 193 of file InsertOrderFragmenter.h.

Referenced by getNumRows(), and setNumRows().

size_t Fragmenter_Namespace::InsertOrderFragmenter::pageSize_

protected

Definition at line 191 of file InsertOrderFragmenter.h.

const int Fragmenter_Namespace::InsertOrderFragmenter::physicalTableId_

protected

Definition at line 188 of file InsertOrderFragmenter.h.

Referenced by Fragmenter_Namespace::SortedOrderFragmenter::sortData().

int Fragmenter_Namespace::InsertOrderFragmenter::rowIdColId_

protected

Definition at line 205 of file InsertOrderFragmenter.h.

const int Fragmenter_Namespace::InsertOrderFragmenter::shard_

protected

Definition at line 189 of file InsertOrderFragmenter.h.

Referenced by updateColumn().

std::mutex Fragmenter_Namespace::InsertOrderFragmenter::temp_mutex_

mutableprotected

Definition at line 230 of file InsertOrderFragmenter.h.

Referenced by updateColumn().

std::unordered_map<int, size_t> Fragmenter_Namespace::InsertOrderFragmenter::varLenColInfo_

protected

Definition at line 206 of file InsertOrderFragmenter.h.

The documentation for this class was generated from the following files:

/home/jenkins-slave/workspace/core-os-doxygen/Fragmenter/InsertOrderFragmenter.h
/home/jenkins-slave/workspace/core-os-doxygen/Fragmenter/InsertOrderFragmenter.cpp
/home/jenkins-slave/workspace/core-os-doxygen/Fragmenter/UpdelStorage.cpp

Public Types

Public Member Functions

Static Public Member Functions

Protected Member Functions

Protected Attributes

Detailed Description

Member Typedef Documentation

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation