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, const bool uses_foreign_storage=false)

	~InsertOrderFragmenter () override

size_t	getNumFragments () override
	returns the number of fragments in a table More...

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

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

void	insertChunks (const InsertChunks &insert_chunk) override
	Insert chunks into minimal number of fragments. More...

void	insertDataNoCheckpoint (InsertData &insert_data_struct) 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	insertChunksNoCheckpoint (const InsertChunks &insert_chunk) override
	Insert chunks into minimal number of fragments; no locks or checkpoints taken. More...

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

void	updateColumnChunkMetadata (const ColumnDescriptor *cd, const int fragment_id, const std::shared_ptr< ChunkMetadata > metadata) override
	Updates the metadata for a column chunk. More...

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

FragmentInfo *	getFragmentInfo (const int fragment_id) const override
	Retrieve the fragment info object for an individual fragment for editing. 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

std::optional< ChunkUpdateStats >	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, Executor *executor) 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 UpdateValuesStats &update_values_stats, 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)

void	dropColumns (const std::vector< int > &columnIds) override

bool	hasDeletedRows (const int delete_column_id) override
	Iterates through chunk metadata to return whether any rows have been deleted. More...

void	resetSizesFromFragments () override

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

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	conditionallyInstantiateFileMgrWithParams ()

void	getChunkMetadata ()

void	lockInsertCheckpointData (const InsertData &insertDataStruct)

void	insertDataImpl (InsertData &insert_data)

void	insertChunksImpl (const InsertChunks &insert_chunk)

void	addColumns (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::vector< std::unique_ptr < Chunk_NS::Chunk > >	tracked_in_memory_chunks_

std::deque< std::unique_ptr < 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_

heavyai::shared_mutex	fragmentInfoMutex_

heavyai::shared_mutex	insertMutex_

Data_Namespace::MemoryLevel	defaultInsertLevel_

const bool	uses_foreign_storage_

bool	hasMaterializedRowId_

int	rowIdColId_

std::unordered_map< int, size_t >	varLenColInfo_

std::shared_ptr< std::mutex >	mutex_access_inmem_states

std::mutex	temp_mutex_

Private Member Functions
bool	isAddingNewColumns (const InsertData &insert_data) const

void	dropFragmentsToSizeNoInsertLock (const size_t max_rows)

void	setLastFragmentVarLenColumnSizes ()

void	insertChunksIntoFragment (const InsertChunks &insert_chunks, const std::optional< int > delete_column_id, FragmentInfo *current_fragment, const size_t num_rows_to_insert, size_t &num_rows_inserted, size_t &num_rows_left, std::vector< size_t > &valid_row_indices, const size_t start_fragment)

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`,
		const bool	uses_foreign_storage = `false`
	)

Fragmenter_Namespace::InsertOrderFragmenter::~InsertOrderFragmenter ( )

override

Definition at line 95 of file InsertOrderFragmenter.cpp.

95 {}

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

protected

Member Function Documentation

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

protected

Definition at line 499 of file InsertOrderFragmenter.cpp.

References CHECK, Fragmenter_Namespace::InsertData::columnIds, Fragmenter_Namespace::InsertData::data, and Fragmenter_Namespace::InsertData::numRows.

                                                                          {
   // synchronize concurrent accesses to fragmentInfoVec_
   heavyai::unique_lock<heavyai::shared_mutex> writeLock(fragmentInfoMutex_);
   size_t numRowsLeft = insertDataStruct.numRows;
   for (const auto columnId : insertDataStruct.columnIds) {
     CHECK(columnMap_.end() == columnMap_.find(columnId));
     const auto columnDesc = catalog_->getMetadataForColumn(physicalTableId_, columnId);
     CHECK(columnDesc);
     columnMap_.emplace(columnId, Chunk_NS::Chunk(columnDesc));
   }
   try {
     for (auto const& fragmentInfo : fragmentInfoVec_) {
       fragmentInfo->shadowChunkMetadataMap =
           fragmentInfo->getChunkMetadataMapPhysicalCopy();
       auto numRowsToInsert = fragmentInfo->getPhysicalNumTuples();  // not getNumTuples()
       size_t numRowsCanBeInserted;
       for (size_t i = 0; i < insertDataStruct.columnIds.size(); i++) {
         auto columnId = insertDataStruct.columnIds[i];
         auto colDesc = catalog_->getMetadataForColumn(physicalTableId_, columnId);
         CHECK(colDesc);
         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.initEncoder();
 
         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);
           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.getBuffer()->size();
           }
         } catch (...) {
           dataMgr_->deleteChunksWithPrefix(chunkKey);
           throw;
         }
       }
       numRowsLeft -= numRowsToInsert;
     }
     CHECK(0 == numRowsLeft);
   } catch (const std::exception& e) {
     for (const auto columnId : insertDataStruct.columnIds) {
       columnMap_.erase(columnId);
     }
     throw e;
   }
 
   for (auto const& fragmentInfo : fragmentInfoVec_) {
     fragmentInfo->setChunkMetadataMap(fragmentInfo->shadowChunkMetadataMap);
   }
 }

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 1278 of file UpdelStorage.cpp.

References threading_serial::async(), CHECK, cpu_threads(), anonymous_namespace{ResultSetReductionInterpreter.cpp}::get_element_size(), DateConverters::get_epoch_seconds_from_days(), getChunksForAllColumns(), getFragmentInfo(), Fragmenter_Namespace::set_chunk_metadata(), Fragmenter_Namespace::set_chunk_stats(), UpdelRoll::setNumTuple(), updateColumnMetadata(), vacuum_fixlen_rows(), vacuum_varlen_rows(), and Fragmenter_Namespace::wait_cleanup_threads().

Referenced by updateColumn().

                                                                {
   auto fragment_ptr = getFragmentInfo(fragment_id);
   auto& fragment = *fragment_ptr;
   auto chunks = getChunksForAllColumns(td, fragment, memory_level);
   const auto ncol = chunks.size();
 
   std::vector<ChunkUpdateStats> update_stats_per_thread(ncol);
 
   // 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->getColumnDesc();
     const auto& col_type = cd->columnType;
     auto data_buffer = chunk->getBuffer();
     auto index_buffer = chunk->getIndexBuf();
     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 =
         [=, &update_stats_per_thread, &updel_roll, &frag_offsets, &fragment] {
           size_t nbytes_fix_data_to_keep;
           if (nrows_to_keep == 0) {
             nbytes_fix_data_to_keep = 0;
           } else {
             nbytes_fix_data_to_keep = vacuum_fixlen_rows(fragment, chunk, frag_offsets);
           }
 
           data_buffer->getEncoder()->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);
           data_buffer->getEncoder()->resetChunkStats();
           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->getEncoder());
               CHECK(encoder);
               encoder->updateMetadata((int8_t*)daddr);
             } else if (col_type.is_fp()) {
               set_chunk_stats(col_type,
                               daddr,
                               update_stats_per_thread[ci].new_values_stats.has_null,
                               update_stats_per_thread[ci].new_values_stats.min_double,
                               update_stats_per_thread[ci].new_values_stats.max_double);
             } else {
               set_chunk_stats(col_type,
                               daddr,
                               update_stats_per_thread[ci].new_values_stats.has_null,
                               update_stats_per_thread[ci].new_values_stats.min_int64t,
                               update_stats_per_thread[ci].new_values_stats.max_int64t);
             }
           }
         };
 
     auto varlen_vacuum = [=, &updel_roll, &frag_offsets, &fragment] {
       size_t nbytes_var_data_to_keep;
       if (nrows_to_keep == 0) {
         nbytes_var_data_to_keep = 0;
       } else {
         nbytes_var_data_to_keep = vacuum_varlen_rows(fragment, chunk, frag_offsets);
       }
 
       data_buffer->getEncoder()->setNumElems(nrows_to_keep);
       data_buffer->setSize(nbytes_var_data_to_keep);
       data_buffer->setUpdated();
 
       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);
 
   updel_roll.setNumTuple({td, &fragment}, nrows_to_keep);
   for (size_t ci = 0; ci < chunks.size(); ++ci) {
     auto chunk = chunks[ci];
     auto cd = chunk->getColumnDesc();
     if (!cd->columnType.is_fixlen_array()) {
       // For DATE_IN_DAYS encoded columns, data is stored in days but the metadata is
       // stored in seconds. Do the metadata conversion here before updating the chunk
       // stats.
       if (cd->columnType.is_date_in_days()) {
         auto& stats = update_stats_per_thread[ci].new_values_stats;
         stats.min_int64t = DateConverters::get_epoch_seconds_from_days(stats.min_int64t);
         stats.max_int64t = DateConverters::get_epoch_seconds_from_days(stats.max_int64t);
       }
       updateColumnMetadata(cd,
                            fragment,
                            chunk,
                            update_stats_per_thread[ci].new_values_stats,
                            cd->columnType,
                            updel_roll);
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::conditionallyInstantiateFileMgrWithParams ( )

protected

Definition at line 148 of file InsertOrderFragmenter.cpp.

References Data_Namespace::DISK_LEVEL, File_Namespace::FileMgrParams::max_rollback_epochs, and TableDescriptor::maxRollbackEpochs.

                                                                       {
   // Somewhat awkward to do this in Fragmenter, but FileMgrs are not instantiated until
   // first use by Fragmenter, and until maxRollbackEpochs param, no options were set in
   // storage per table
   if (!uses_foreign_storage_ &&
       defaultInsertLevel_ == Data_Namespace::MemoryLevel::DISK_LEVEL) {
     const TableDescriptor* td =
         catalog_->getMetadataForTable(physicalTableId_, false /*populateFragmenter*/);
     File_Namespace::FileMgrParams fileMgrParams;
     fileMgrParams.max_rollback_epochs = td->maxRollbackEpochs;
     dataMgr_->getGlobalFileMgr()->setFileMgrParams(
         chunkKeyPrefix_[0], chunkKeyPrefix_[1], fileMgrParams);
   }
 }

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 936 of file InsertOrderFragmenter.cpp.

References Fragmenter_Namespace::anonymous_namespace{InsertOrderFragmenter.cpp}::compute_device_for_fragment(), Data_Namespace::CPU_LEVEL, and Fragmenter_Namespace::FragmentInfo::fragmentId.

                                                  {
   // also sets the new fragment as the insertBuffer for each column
 
   maxFragmentId_++;
   auto newFragmentInfo = std::make_unique<FragmentInfo>();
   newFragmentInfo->fragmentId = maxFragmentId_;
   newFragmentInfo->shadowNumTuples = 0;
   newFragmentInfo->setPhysicalNumTuples(0);
   for (const auto levelSize : dataMgr_->levelSizes_) {
     newFragmentInfo->deviceIds.push_back(compute_device_for_fragment(
         physicalTableId_, newFragmentInfo->fragmentId, levelSize));
   }
   newFragmentInfo->physicalTableId = physicalTableId_;
   newFragmentInfo->shard = shard_;
 
   for (map<int, Chunk>::iterator colMapIt = columnMap_.begin();
        colMapIt != columnMap_.end();
        ++colMapIt) {
     auto& chunk = colMapIt->second;
     if (memoryLevel == Data_Namespace::MemoryLevel::CPU_LEVEL) {
       /* At the end of this function chunks from the previous fragment become 'rolled
        * off', temporaray tables will lose reference to any 'rolled off' chunks and are
        * not able to unpin these chunks. Keep reference to 'rolled off' chunks and unpin
        * at ~InsertOrderFragmenter, chunks wrapped by unique_ptr to avoid extraneous
        * ~Chunk calls with temporary chunks.*/
       tracked_in_memory_chunks_.emplace_back(std::make_unique<Chunk_NS::Chunk>(chunk));
     }
     ChunkKey chunkKey = chunkKeyPrefix_;
     chunkKey.push_back(chunk.getColumnDesc()->columnId);
     chunkKey.push_back(maxFragmentId_);
     chunk.createChunkBuffer(dataMgr_,
                             chunkKey,
                             memoryLevel,
                             newFragmentInfo->deviceIds[static_cast<int>(memoryLevel)],
                             pageSize_);
     chunk.initEncoder();
   }
 
   heavyai::lock_guard<heavyai::shared_mutex> writeLock(fragmentInfoMutex_);
   fragmentInfoVec_.push_back(std::move(newFragmentInfo));
   return fragmentInfoVec_.back().get();
 }

Here is the call graph for this function:

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

protected

Definition at line 269 of file InsertOrderFragmenter.cpp.

References lockmgr::TableLockMgrImpl< TableDataLockMgr >::getWriteLockForTable().

                                                                           {
   // 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 DBHandler, 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
   const auto delete_lock =
       lockmgr::TableDataLockMgr::getWriteLockForTable(chunkKeyPrefix);
 
   heavyai::unique_lock<heavyai::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::dropColumns ( const std::vector< int > & columnIds )

overridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 587 of file InsertOrderFragmenter.cpp.

                                                                        {
   // prevent concurrent insert rows and drop column
   heavyai::unique_lock<heavyai::shared_mutex> insertLock(insertMutex_);
   // synchronize concurrent accesses to fragmentInfoVec_
   heavyai::unique_lock<heavyai::shared_mutex> writeLock(fragmentInfoMutex_);
   for (auto const& fragmentInfo : fragmentInfoVec_) {
     fragmentInfo->shadowChunkMetadataMap =
         fragmentInfo->getChunkMetadataMapPhysicalCopy();
   }
 
   for (const auto columnId : columnIds) {
     auto cit = columnMap_.find(columnId);
     if (columnMap_.end() != cit) {
       columnMap_.erase(cit);
     }
 
     vector<int> fragPrefix = chunkKeyPrefix_;
     fragPrefix.push_back(columnId);
     dataMgr_->deleteChunksWithPrefix(fragPrefix);
 
     for (const auto& fragmentInfo : fragmentInfoVec_) {
       auto cmdit = fragmentInfo->shadowChunkMetadataMap.find(columnId);
       if (fragmentInfo->shadowChunkMetadataMap.end() != cmdit) {
         fragmentInfo->shadowChunkMetadataMap.erase(cmdit);
       }
     }
   }
   for (const auto& fragmentInfo : fragmentInfoVec_) {
     fragmentInfo->setChunkMetadataMap(fragmentInfo->shadowChunkMetadataMap);
   }
 }

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 236 of file InsertOrderFragmenter.cpp.

                                                                      {
   heavyai::unique_lock<heavyai::shared_mutex> insert_lock(insertMutex_);
   dropFragmentsToSizeNoInsertLock(max_rows);
 }

void Fragmenter_Namespace::InsertOrderFragmenter::dropFragmentsToSizeNoInsertLock ( const size_t max_rows )

private

Definition at line 241 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 (fragmentInfoVec_.empty() ||
       numTuples_ == fragmentInfoVec_.back()->getPhysicalNumTuples()) {
     return;
   }
 
   if (numTuples_ > max_rows) {
     size_t preNumTuples = numTuples_;
     vector<int> dropFragIds;
     size_t targetRows = max_rows * 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: " << max_rows;
   }
 }

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

inline

Definition at line 120 of file InsertOrderFragmenter.h.

References chunkKeyPrefix_.

120 { return chunkKeyPrefix_; }

Fragmenter_Namespace::InsertOrderFragmenter::chunkKeyPrefix_

std::vector< int > chunkKeyPrefix_

Definition: InsertOrderFragmenter.h:192

void Fragmenter_Namespace::InsertOrderFragmenter::getChunkMetadata ( )

protected

Definition at line 163 of file InsertOrderFragmenter.cpp.

References CHECK, CHECK_GE, Fragmenter_Namespace::anonymous_namespace{InsertOrderFragmenter.cpp}::compute_device_for_fragment(), Data_Namespace::DISK_LEVEL, logger::FATAL, LOG, gpu_enabled::sort(), and to_string().

                                              {
   if (uses_foreign_storage_ ||
       defaultInsertLevel_ == Data_Namespace::MemoryLevel::DISK_LEVEL) {
     // memory-resident tables won't have anything on disk
     ChunkMetadataVector 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 auto& pair1, const auto& 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) {
         auto new_fragment_info = std::make_unique<Fragmenter_Namespace::FragmentInfo>();
         CHECK(new_fragment_info);
         maxFragmentId_ = cur_fragment_id;
         new_fragment_info->fragmentId = cur_fragment_id;
         new_fragment_info->setPhysicalNumTuples(chunk_itr->second->numElements);
         numTuples_ += new_fragment_info->getPhysicalNumTuples();
         for (const auto level_size : dataMgr_->levelSizes_) {
           new_fragment_info->deviceIds.push_back(
               compute_device_for_fragment(physicalTableId_, cur_fragment_id, level_size));
         }
         new_fragment_info->shadowNumTuples = new_fragment_info->getPhysicalNumTuples();
         new_fragment_info->physicalTableId = physicalTableId_;
         new_fragment_info->shard = shard_;
         fragmentInfoVec_.emplace_back(std::move(new_fragment_info));
       } else {
         if (chunk_itr->second->numElements !=
             fragmentInfoVec_.back()->getPhysicalNumTuples()) {
           LOG(FATAL) << "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);
         }
       }
       CHECK(fragmentInfoVec_.back().get());
       fragmentInfoVec_.back().get()->setChunkMetadata(cur_column_id, chunk_itr->second);
     }
   }
 
   size_t maxFixedColSize = 0;
 
   for (auto colIt = columnMap_.begin(); colIt != columnMap_.end(); ++colIt) {
     auto size = colIt->second.getColumnDesc()->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
     }
     CHECK_GE(size, 0);
     maxFixedColSize = std::max(maxFixedColSize, static_cast<size_t>(size));
   }
 
   // this is maximum number of rows assuming everything is fixed length
   maxFragmentRows_ = std::min(maxFragmentRows_, maxChunkSize_ / maxFixedColSize);
   setLastFragmentVarLenColumnSizes();
 }

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 989 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 119 of file InsertOrderFragmenter.h.

References chunkKeyPrefix_.

119 { return chunkKeyPrefix_.back(); }

Fragmenter_Namespace::InsertOrderFragmenter::chunkKeyPrefix_

std::vector< int > chunkKeyPrefix_

Definition: InsertOrderFragmenter.h:192

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

inlineoverridevirtual

get fragmenter's type (as string

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 124 of file InsertOrderFragmenter.h.

References fragmenterType_.

124 { return fragmenterType_; }

Fragmenter_Namespace::InsertOrderFragmenter::fragmenterType_

std::string fragmenterType_

Definition: InsertOrderFragmenter.h:210

FragmentInfo * Fragmenter_Namespace::InsertOrderFragmenter::getFragmentInfo ( const int fragment_id ) const

overridevirtual

Retrieve the fragment info object for an individual fragment for editing.

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 406 of file InsertOrderFragmenter.cpp.

References CHECK.

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

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

Here is the caller graph for this function:

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

protected

TableInfo Fragmenter_Namespace::InsertOrderFragmenter::getFragmentsForQuery ( )

overridevirtual

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

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 985 of file InsertOrderFragmenter.cpp.

                                                       {
   heavyai::shared_lock<heavyai::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;
     std::for_each(
         fragmentInfoVec_.begin(),
         fragmentInfoVec_.end(),
         [&queryInfo](const auto& fragment_owned_ptr) {
           queryInfo.fragments.emplace_back(*fragment_owned_ptr);  // 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::getNumFragments ( )

overridevirtual

returns the number of fragments in a table

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 980 of file InsertOrderFragmenter.cpp.

                                               {
   heavyai::shared_lock<heavyai::shared_mutex> readLock(fragmentInfoMutex_);
   return fragmentInfoVec_.size();
 }

size_t Fragmenter_Namespace::InsertOrderFragmenter::getNumRows ( )

inlineoverridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 125 of file InsertOrderFragmenter.h.

References numTuples_.

125 { return numTuples_; }

Fragmenter_Namespace::InsertOrderFragmenter::numTuples_

size_t numTuples_

Definition: InsertOrderFragmenter.h:206

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 1018 of file UpdelStorage.cpp.

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

Referenced by updateColumn().

                                                {
   const auto data_buffer = chunk->getBuffer();
   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:

bool Fragmenter_Namespace::InsertOrderFragmenter::hasDeletedRows ( const int delete_column_id )

overridevirtual

Iterates through chunk metadata to return whether any rows have been deleted.

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 619 of file InsertOrderFragmenter.cpp.

References CHECK.

                                                                      {
   heavyai::shared_lock<heavyai::shared_mutex> read_lock(fragmentInfoMutex_);
 
   for (auto const& fragment : fragmentInfoVec_) {
     auto chunk_meta_it = fragment->getChunkMetadataMapPhysical().find(delete_column_id);
     CHECK(chunk_meta_it != fragment->getChunkMetadataMapPhysical().end());
     const auto& chunk_stats = chunk_meta_it->second->chunkStats;
     if (chunk_stats.max.tinyintval == 1) {
       return true;
     }
   }
   return false;
 }

void Fragmenter_Namespace::InsertOrderFragmenter::insertChunks ( const InsertChunks & insert_chunk )

overridevirtual

Insert chunks into minimal number of fragments.

Parameters

insert_chunk - the chunks to insert

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 431 of file InsertOrderFragmenter.cpp.

References Fragmenter_Namespace::InsertChunks::db_id, Data_Namespace::DISK_LEVEL, and Fragmenter_Namespace::InsertChunks::table_id.

                                                                          {
   try {
     // prevent two threads from trying to insert into the same table simultaneously
     heavyai::unique_lock<heavyai::shared_mutex> insertLock(insertMutex_);
     insertChunksImpl(insert_chunk);
     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 (...) {
     auto db_id = insert_chunk.db_id;
     auto table_epochs = catalog_->getTableEpochs(db_id, insert_chunk.table_id);
     // the statement below deletes *this* object!
     // relying on exception propagation at this stage
     // until we can sort this out in a cleaner fashion
     catalog_->setTableEpochs(db_id, table_epochs);
     throw;
   }
 }

void Fragmenter_Namespace::InsertOrderFragmenter::insertChunksImpl ( const InsertChunks & insert_chunk )

protected

Definition at line 703 of file InsertOrderFragmenter.cpp.

References CHECK, CHECK_EQ, CHECK_GT, CHECK_LE, Fragmenter_Namespace::InsertChunks::chunks, Fragmenter_Namespace::anonymous_namespace{InsertOrderFragmenter.cpp}::get_num_rows_to_insert(), and Fragmenter_Namespace::InsertChunks::valid_row_indices.

                                                                               {
   std::optional<int> delete_column_id{std::nullopt};
   for (const auto& cit : columnMap_) {
     if (cit.second.getColumnDesc()->isDeletedCol) {
       delete_column_id = cit.second.getColumnDesc()->columnId;
     }
   }
 
   // verify that all chunks to be inserted have same number of rows, otherwise the input
   // data is malformed
   std::optional<size_t> num_rows{std::nullopt};
   for (const auto& [column_id, chunk] : insert_chunks.chunks) {
     auto buffer = chunk->getBuffer();
     CHECK(buffer);
     CHECK(buffer->hasEncoder());
     if (!num_rows.has_value()) {
       num_rows = buffer->getEncoder()->getNumElems();
     } else {
       CHECK_EQ(num_rows.value(), buffer->getEncoder()->getNumElems());
     }
   }
 
   auto valid_row_indices = insert_chunks.valid_row_indices;
   size_t num_rows_left = valid_row_indices.size();
   size_t num_rows_inserted = 0;
 
   if (num_rows_left == 0) {
     return;
   }
 
   FragmentInfo* current_fragment{nullptr};
 
   // Access to fragmentInfoVec_ is protected as we are under the insertMutex_ lock but it
   // feels fragile
   if (fragmentInfoVec_.empty()) {  // if no fragments exist for table
     current_fragment = createNewFragment(defaultInsertLevel_);
   } else {
     current_fragment = fragmentInfoVec_.back().get();
   }
   CHECK(current_fragment);
 
   size_t start_fragment = fragmentInfoVec_.size() - 1;
 
   while (num_rows_left > 0) {  // may have to create multiple fragments for bulk insert
     // loop until done inserting all rows
     CHECK_LE(current_fragment->shadowNumTuples, maxFragmentRows_);
     size_t rows_left_in_current_fragment =
         maxFragmentRows_ - current_fragment->shadowNumTuples;
     size_t num_rows_to_insert = get_num_rows_to_insert(rows_left_in_current_fragment,
                                                        num_rows_left,
                                                        num_rows_inserted,
                                                        varLenColInfo_,
                                                        maxChunkSize_,
                                                        insert_chunks,
                                                        columnMap_,
                                                        valid_row_indices);
 
     if (rows_left_in_current_fragment == 0 || num_rows_to_insert == 0) {
       current_fragment = createNewFragment(defaultInsertLevel_);
       if (num_rows_inserted == 0) {
         start_fragment++;
       }
       rows_left_in_current_fragment = maxFragmentRows_;
       for (auto& varLenColInfoIt : varLenColInfo_) {
         varLenColInfoIt.second = 0;  // reset byte counter
       }
       num_rows_to_insert = get_num_rows_to_insert(rows_left_in_current_fragment,
                                                   num_rows_left,
                                                   num_rows_inserted,
                                                   varLenColInfo_,
                                                   maxChunkSize_,
                                                   insert_chunks,
                                                   columnMap_,
                                                   valid_row_indices);
     }
 
     CHECK_GT(num_rows_to_insert, size_t(0));  // would put us into an endless loop as we'd
                                               // never be able to insert anything
 
     insertChunksIntoFragment(insert_chunks,
                              delete_column_id,
                              current_fragment,
                              num_rows_to_insert,
                              num_rows_inserted,
                              num_rows_left,
                              valid_row_indices,
                              start_fragment);
   }
   numTuples_ += *num_rows;
   dropFragmentsToSizeNoInsertLock(maxRows_);
 }

Here is the call graph for this function:

void Fragmenter_Namespace::InsertOrderFragmenter::insertChunksIntoFragment	(	const InsertChunks &	insert_chunks,
		const std::optional< int >	delete_column_id,
		FragmentInfo *	current_fragment,
		const size_t	num_rows_to_insert,
		size_t &	num_rows_inserted,
		size_t &	num_rows_left,
		std::vector< size_t > &	valid_row_indices,
		const size_t	start_fragment
	)

private

Definition at line 633 of file InsertOrderFragmenter.cpp.

References CHECK, CHECK_EQ, CHECK_GE, CHECK_LE, Fragmenter_Namespace::InsertChunks::chunks, Fragmenter_Namespace::FragmentInfo::fragmentId, DataBlockPtr::numbersPtr, Fragmenter_Namespace::FragmentInfo::shadowChunkMetadataMap, and Fragmenter_Namespace::FragmentInfo::shadowNumTuples.

                                  {
   heavyai::unique_lock<heavyai::shared_mutex> write_lock(fragmentInfoMutex_);
   // for each column, append the data in the appropriate insert buffer
   auto insert_row_indices = valid_row_indices;
   CHECK_GE(insert_row_indices.size(), num_rows_to_insert);
   insert_row_indices.erase(insert_row_indices.begin() + num_rows_to_insert,
                            insert_row_indices.end());
   CHECK_EQ(insert_row_indices.size(), num_rows_to_insert);
   for (auto& [column_id, chunk] : insert_chunks.chunks) {
     auto col_map_it = columnMap_.find(column_id);
     CHECK(col_map_it != columnMap_.end());
     current_fragment->shadowChunkMetadataMap[column_id] =
         col_map_it->second.appendEncodedDataAtIndices(*chunk, insert_row_indices);
     auto var_len_col_info_it = varLenColInfo_.find(column_id);
     if (var_len_col_info_it != varLenColInfo_.end()) {
       var_len_col_info_it->second = col_map_it->second.getBuffer()->size();
       CHECK_LE(var_len_col_info_it->second, maxChunkSize_);
     }
   }
   if (hasMaterializedRowId_) {
     size_t start_id = maxFragmentRows_ * current_fragment->fragmentId +
                       current_fragment->shadowNumTuples;
     std::vector<int64_t> row_id_data(num_rows_to_insert);
     for (size_t i = 0; i < num_rows_to_insert; ++i) {
       row_id_data[i] = i + start_id;
     }
     DataBlockPtr row_id_block;
     row_id_block.numbersPtr = reinterpret_cast<int8_t*>(row_id_data.data());
     auto col_map_it = columnMap_.find(rowIdColId_);
     CHECK(col_map_it != columnMap_.end());
     current_fragment->shadowChunkMetadataMap[rowIdColId_] = col_map_it->second.appendData(
         row_id_block, num_rows_to_insert, num_rows_inserted);
   }
 
   if (delete_column_id) {  // has delete column
     std::vector<int8_t> delete_data(num_rows_to_insert, false);
     DataBlockPtr delete_block;
     delete_block.numbersPtr = reinterpret_cast<int8_t*>(delete_data.data());
     auto col_map_it = columnMap_.find(*delete_column_id);
     CHECK(col_map_it != columnMap_.end());
     current_fragment->shadowChunkMetadataMap[*delete_column_id] =
         col_map_it->second.appendData(
             delete_block, num_rows_to_insert, num_rows_inserted);
   }
 
   current_fragment->shadowNumTuples =
       fragmentInfoVec_.back()->getPhysicalNumTuples() + num_rows_to_insert;
   num_rows_left -= num_rows_to_insert;
   num_rows_inserted += num_rows_to_insert;
   for (auto part_it = fragmentInfoVec_.begin() + start_fragment;
        part_it != fragmentInfoVec_.end();
        ++part_it) {
     auto fragment_ptr = part_it->get();
     fragment_ptr->setPhysicalNumTuples(fragment_ptr->shadowNumTuples);
     fragment_ptr->setChunkMetadataMap(fragment_ptr->shadowChunkMetadataMap);
   }
 
   // truncate the first `num_rows_to_insert` rows in `valid_row_indices`
   valid_row_indices.erase(valid_row_indices.begin(),
                           valid_row_indices.begin() + num_rows_to_insert);
 }

void Fragmenter_Namespace::InsertOrderFragmenter::insertChunksNoCheckpoint ( const InsertChunks & insert_chunk )

overridevirtual

Insert chunks into minimal number of fragments; no locks or checkpoints taken.

Parameters

chunk - the chunks to insert

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 479 of file InsertOrderFragmenter.cpp.

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

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

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 453 of file InsertOrderFragmenter.cpp.

References Fragmenter_Namespace::InsertData::databaseId, Data_Namespace::DISK_LEVEL, and Fragmenter_Namespace::InsertData::tableId.

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

                                                                      {
   try {
     // prevent two threads from trying to insert into the same table simultaneously
     heavyai::unique_lock<heavyai::shared_mutex> insertLock(insertMutex_);
     if (!isAddingNewColumns(insert_data_struct)) {
       insertDataImpl(insert_data_struct);
     } else {
       addColumns(insert_data_struct);
     }
     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 (...) {
     auto table_epochs = catalog_->getTableEpochs(insert_data_struct.databaseId,
                                                  insert_data_struct.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_->setTableEpochs(insert_data_struct.databaseId, table_epochs);
     throw;
   }
 }

Here is the caller graph for this function:

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

protected

Definition at line 795 of file InsertOrderFragmenter.cpp.

References CHECK, CHECK_GT, CHECK_LE, Fragmenter_Namespace::InsertData::columnIds, Fragmenter_Namespace::InsertData::data, Fragmenter_Namespace::InsertData::is_default, DataBlockPtr::numbersPtr, and Fragmenter_Namespace::InsertData::numRows.

                                                                   {
   // populate deleted system column if it should exist, as it will not come from client
   std::unique_ptr<int8_t[]> data_for_deleted_column;
   for (const auto& cit : columnMap_) {
     if (cit.second.getColumnDesc()->isDeletedCol) {
       data_for_deleted_column.reset(new int8_t[insert_data.numRows]);
       memset(data_for_deleted_column.get(), 0, insert_data.numRows);
       insert_data.data.emplace_back(DataBlockPtr{data_for_deleted_column.get()});
       insert_data.columnIds.push_back(cit.second.getColumnDesc()->columnId);
       insert_data.is_default.push_back(false);
       break;
     }
   }
   CHECK(insert_data.is_default.size() == insert_data.columnIds.size());
   std::unordered_map<int, int> inverseInsertDataColIdMap;
   for (size_t insertId = 0; insertId < insert_data.columnIds.size(); ++insertId) {
     inverseInsertDataColIdMap.insert(
         std::make_pair(insert_data.columnIds[insertId], insertId));
   }
 
   size_t numRowsLeft = insert_data.numRows;
   size_t numRowsInserted = 0;
   vector<DataBlockPtr> dataCopy =
       insert_data.data;  // bc append data will move ptr forward and this violates
                          // constness of InsertData
   if (numRowsLeft <= 0) {
     return;
   }
 
   FragmentInfo* currentFragment{nullptr};
 
   // Access to fragmentInfoVec_ is protected as we are under the insertMutex_ lock but it
   // feels fragile
   if (fragmentInfoVec_.empty()) {  // if no fragments exist for table
     currentFragment = createNewFragment(defaultInsertLevel_);
   } else {
     currentFragment = fragmentInfoVec_.back().get();
   }
   CHECK(currentFragment);
 
   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,
                                          insert_data.is_default[insertIdIt->second]));
         }
       }
     }
 
     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,
                                          insert_data.is_default[insertIdIt->second]));
         }
       }
     }
 
     CHECK_GT(numRowsToInsert, size_t(0));  // would put us into an endless loop as we'd
                                            // never be able to insert anything
 
     {
       heavyai::unique_lock<heavyai::shared_mutex> writeLock(fragmentInfoMutex_);
       // for each column, append the data in the appropriate insert buffer
       for (size_t i = 0; i < insert_data.columnIds.size(); ++i) {
         int columnId = insert_data.columnIds[i];
         auto colMapIt = columnMap_.find(columnId);
         CHECK(colMapIt != columnMap_.end());
         currentFragment->shadowChunkMetadataMap[columnId] = colMapIt->second.appendData(
             dataCopy[i], numRowsToInsert, numRowsInserted, insert_data.is_default[i]);
         auto varLenColInfoIt = varLenColInfo_.find(columnId);
         if (varLenColInfoIt != varLenColInfo_.end()) {
           varLenColInfoIt->second = colMapIt->second.getBuffer()->size();
         }
       }
       if (hasMaterializedRowId_) {
         size_t startId = maxFragmentRows_ * currentFragment->fragmentId +
                          currentFragment->shadowNumTuples;
         auto row_id_data = std::make_unique<int64_t[]>(numRowsToInsert);
         for (size_t i = 0; i < numRowsToInsert; ++i) {
           row_id_data[i] = i + startId;
         }
         DataBlockPtr rowIdBlock;
         rowIdBlock.numbersPtr = reinterpret_cast<int8_t*>(row_id_data.get());
         auto colMapIt = columnMap_.find(rowIdColId_);
         currentFragment->shadowChunkMetadataMap[rowIdColId_] =
             colMapIt->second.appendData(rowIdBlock, numRowsToInsert, numRowsInserted);
       }
 
       currentFragment->shadowNumTuples =
           fragmentInfoVec_.back()->getPhysicalNumTuples() + numRowsToInsert;
       numRowsLeft -= numRowsToInsert;
       numRowsInserted += numRowsToInsert;
       for (auto partIt = fragmentInfoVec_.begin() + startFragment;
            partIt != fragmentInfoVec_.end();
            ++partIt) {
         auto fragment_ptr = partIt->get();
         fragment_ptr->setPhysicalNumTuples(fragment_ptr->shadowNumTuples);
         fragment_ptr->setChunkMetadataMap(fragment_ptr->shadowChunkMetadataMap);
       }
     }
   }
   numTuples_ += insert_data.numRows;
   dropFragmentsToSizeNoInsertLock(maxRows_);
 }

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

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 487 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
   heavyai::unique_lock<heavyai::shared_mutex> insertLock(
       insertMutex_);  // prevent two threads from trying to insert into the same table
                       // simultaneously
   if (!isAddingNewColumns(insert_data_struct)) {
     insertDataImpl(insert_data_struct);
   } else {
     addColumns(insert_data_struct);
   }
 }

Here is the caller graph for this function:

bool Fragmenter_Namespace::InsertOrderFragmenter::isAddingNewColumns ( const InsertData & insert_data ) const

private

Definition at line 416 of file InsertOrderFragmenter.cpp.

References CHECK, and Fragmenter_Namespace::InsertData::columnIds.

                                                                                   {
   bool all_columns_already_exist = true, all_columns_are_new = true;
   for (const auto column_id : insert_data.columnIds) {
     if (columnMap_.find(column_id) == columnMap_.end()) {
       all_columns_already_exist = false;
     } else {
       all_columns_are_new = false;
     }
   }
   // only one should be TRUE
   bool either_all_exist_or_all_new = all_columns_already_exist ^ all_columns_are_new;
   CHECK(either_all_exist_or_all_new);
   return all_columns_are_new;
 }

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

protected

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

protected

void Fragmenter_Namespace::InsertOrderFragmenter::resetSizesFromFragments ( )

overridevirtual

Resets the fragmenter's size related metadata using the internal fragment info vector. This is typically done after operations, such as vacuuming, which can change fragment sizes.

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 1036 of file InsertOrderFragmenter.cpp.

                                                     {
   heavyai::shared_lock<heavyai::shared_mutex> read_lock(fragmentInfoMutex_);
   numTuples_ = 0;
   for (const auto& fragment_info : fragmentInfoVec_) {
     numTuples_ += fragment_info->getPhysicalNumTuples();
   }
   setLastFragmentVarLenColumnSizes();
 }

void Fragmenter_Namespace::InsertOrderFragmenter::setLastFragmentVarLenColumnSizes ( )

private

Definition at line 1045 of file InsertOrderFragmenter.cpp.

                                                              {
   if (!uses_foreign_storage_ && fragmentInfoVec_.size() > 0) {
     // Now need to get the insert buffers for each column - should be last
     // fragment
     int lastFragmentId = fragmentInfoVec_.back()->fragmentId;
     // TODO: add accessor here for safe indexing
     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.getBuffer()->size();
       }
     }
   }
 }

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

inlineoverridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 126 of file InsertOrderFragmenter.h.

References numTuples_.

126 { numTuples_ = numTuples; }

Fragmenter_Namespace::InsertOrderFragmenter::numTuples_

size_t numTuples_

Definition: InsertOrderFragmenter.h:206

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

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 315 of file InsertOrderFragmenter.cpp.

References CHECK, ColumnDescriptor::columnId, ColumnDescriptor::columnType, DatumToString(), Data_Namespace::DISK_LEVEL, get_logical_type_info(), Chunk_NS::Chunk::getChunk(), SQLTypeInfo::is_dict_encoded_string(), kBIGINT, LOG, show_chunk(), VLOG, and logger::WARNING.

                                                          {
   // synchronize concurrent accesses to fragmentInfoVec_
   heavyai::unique_lock<heavyai::shared_mutex> writeLock(fragmentInfoMutex_);
   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 const& 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,
                                              memory_level.value_or(defaultInsertLevel_),
                                              0,
                                              chunk_meta_it->second->numBytes,
                                              chunk_meta_it->second->numElements);
       auto buf = chunk->getBuffer();
       CHECK(buf);
       if (!buf->hasEncoder()) {
         throw std::runtime_error("No encoder for chunk " + show_chunk(chunk_key));
       }
       auto encoder = buf->getEncoder();
 
       auto chunk_stats = stats_itr->second;
 
       auto old_chunk_metadata = std::make_shared<ChunkMetadata>();
       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 = cd->columnType.is_dict_encoded_string()
                                   ? SQLTypeInfo(kBIGINT)
                                   : get_logical_type_info(cd->columnType);
       if (!didResetStats) {
         VLOG(3) << "Skipping chunk stats reset for " << show_chunk(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 " << show_chunk(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
       auto new_metadata = std::make_shared<ChunkMetadata>();
       // 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->getChunkMetadataMapPhysicalCopy();  // TODO(adb): needed?
       if (defaultInsertLevel_ == Data_Namespace::DISK_LEVEL) {
         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:

std::optional< ChunkUpdateStats > 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 640 of file UpdelStorage.cpp.

Referenced by updateColumn().

                            {
   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_ptr = getFragmentInfo(fragment_id);
   auto& fragment = *fragment_ptr;
   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<ChunkUpdateStats> update_stats_per_thread(ncore);
 
   // parallel update elements
   std::vector<std::future<void>> threads;
 
   const auto segsz = (nrow + ncore - 1) / ncore;
   auto dbuf = chunk->getBuffer();
   auto dbuf_addr = dbuf->getMemoryPtr();
   dbuf->setUpdated();
   updel_roll.addDirtyChunk(chunk, fragment.fragmentId);
   for (size_t rbegin = 0, c = 0; rbegin < nrow; ++c, rbegin += segsz) {
     threads.emplace_back(std::async(
         std::launch::async, [=, &update_stats_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.");
               }
               int64_t old_val;
               get_scalar<int64_t>(data_ptr, lhs_type, old_val);
               // Handle special case where date column with date in days encoding stores
               // metadata in epoch seconds.
               if (lhs_type.is_date_in_days()) {
                 old_val = DateConverters::get_epoch_seconds_from_days(old_val);
               }
               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);
                 int64_t target_value = (v == inline_int_null_value<int64_t>() &&
                                         lhs_type.get_notnull() == false)
                                            ? v
                                            : decimal_val;
                 update_metadata(
                     lhs_type, update_stats_per_thread[c], target_value, old_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);
                   int64_t target_value = (v == inline_int_null_value<int64_t>() &&
                                           lhs_type.get_notnull() == false)
                                              ? NullSentinelSupplier()(lhs_type, v)
                                              : seconds;
                   update_metadata(
                       lhs_type, update_stats_per_thread[c], target_value, old_val);
                 } else {
                   int64_t target_value;
                   if (rhs_type.is_decimal()) {
                     target_value = round(decimal_to_double(rhs_type, v));
                   } else {
                     target_value = v;
                   }
                   update_metadata(
                       lhs_type, update_stats_per_thread[c], target_value, old_val);
                 }
               } else {
                 if (rhs_type.is_decimal()) {
                   update_metadata(lhs_type,
                                   update_stats_per_thread[c],
                                   decimal_to_double(rhs_type, v),
                                   double(old_val));
                 } else {
                   update_metadata(lhs_type, update_stats_per_thread[c], v, old_val);
                 }
               }
             } 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.");
               }
               double old_val;
               get_scalar<double>(data_ptr, lhs_type, old_val);
               put_scalar<double>(data_ptr, lhs_type, v, cd->columnName);
               if (lhs_type.is_integer()) {
                 update_metadata(
                     lhs_type, update_stats_per_thread[c], int64_t(v), int64_t(old_val));
               } else if (lhs_type.is_fp()) {
                 update_metadata(
                     lhs_type, update_stats_per_thread[c], double(v), double(old_val));
               } 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.");
               }
               float old_val;
               get_scalar<float>(data_ptr, lhs_type, old_val);
               put_scalar<float>(data_ptr, lhs_type, v, cd->columnName);
               if (lhs_type.is_integer()) {
                 update_metadata(
                     lhs_type, update_stats_per_thread[c], int64_t(v), int64_t(old_val));
               } else {
                 update_metadata(lhs_type, update_stats_per_thread[c], double(v), old_val);
               }
             } 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);
                 }
                 int64_t old_val;
                 get_scalar<int64_t>(data_ptr, lhs_type, old_val);
                 put_scalar<int64_t>(data_ptr, lhs_type, sidx, cd->columnName);
                 update_metadata(
                     lhs_type, update_stats_per_thread[c], int64_t(sidx), old_val);
               } 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()) {
                   double old_val;
                   get_scalar<double>(data_ptr, lhs_type, old_val);
                   put_scalar<double>(data_ptr, lhs_type, dval, cd->columnName);
                   update_metadata(
                       lhs_type, update_stats_per_thread[c], double(dval), old_val);
                 } else {
                   int64_t old_val;
                   get_scalar<int64_t>(data_ptr, lhs_type, old_val);
                   put_scalar<int64_t>(data_ptr, lhs_type, dval, cd->columnName);
                   update_metadata(
                       lhs_type, update_stats_per_thread[c], int64_t(dval), old_val);
                 }
               } else {
                 put_null(data_ptr, lhs_type, cd->columnName);
                 update_stats_per_thread[c].new_values_stats.has_null = 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 {};
       }
     }
   }
   ChunkUpdateStats update_stats;
   for (size_t c = 0; c < ncore; ++c) {
     update_metadata(update_stats.new_values_stats,
                     update_stats_per_thread[c].new_values_stats);
     update_metadata(update_stats.old_values_stats,
                     update_stats_per_thread[c].old_values_stats);
   }
 
   CHECK_GT(fragment.shadowNumTuples, size_t(0));
   updateColumnMetadata(
       cd, fragment, chunk, update_stats.new_values_stats, cd->columnType, updel_roll);
   update_stats.updated_rows_count = nrow;
   update_stats.fragment_rows_count = fragment.shadowNumTuples;
   update_stats.chunk = chunk;
   return update_stats;
 }

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 ScalarTargetValue &	rhs_value,
		const SQLTypeInfo &	rhs_type,
		const Data_Namespace::MemoryLevel	memory_level,
		UpdelRoll &	updel_roll
	)

overridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 55 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::updateColumnChunkMetadata	(	const ColumnDescriptor *	cd,
		const int	fragment_id,
		const std::shared_ptr< ChunkMetadata >	metadata
	)

overridevirtual

Updates the metadata for a column chunk.

Parameters

cd	- ColumnDescriptor for the column
fragment_id	- Fragment id of the chunk within the column
metadata	- shared_ptr of the metadata to update column chunk with

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 302 of file InsertOrderFragmenter.cpp.

References CHECK, and ColumnDescriptor::columnId.

                                                  {
   // synchronize concurrent accesses to fragmentInfoVec_
   heavyai::unique_lock<heavyai::shared_mutex> writeLock(fragmentInfoMutex_);
 
   CHECK(metadata.get());
   auto fragment_info = getFragmentInfo(fragment_id);
   CHECK(fragment_info);
   fragment_info->setChunkMetadata(cd->columnId, metadata);
 }

void Fragmenter_Namespace::InsertOrderFragmenter::updateColumnMetadata	(	const ColumnDescriptor *	cd,
		FragmentInfo &	fragment,
		std::shared_ptr< Chunk_NS::Chunk >	chunk,
		const UpdateValuesStats &	update_values_stats,
		const SQLTypeInfo &	rhs_type,
		UpdelRoll &	updel_roll
	)

overridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 928 of file UpdelStorage.cpp.

References UpdelRoll::catalog, ColumnDescriptor::columnId, ColumnDescriptor::columnType, fragmentInfoMutex_, UpdelRoll::getChunkMetadata(), Catalog_Namespace::Catalog::getMetadataForTable(), Fragmenter_Namespace::UpdateValuesStats::has_null, SQLTypeInfo::is_decimal(), Fragmenter_Namespace::is_integral(), kENCODING_DICT, Fragmenter_Namespace::UpdateValuesStats::max_double, Fragmenter_Namespace::UpdateValuesStats::max_int64t, Fragmenter_Namespace::UpdateValuesStats::min_double, Fragmenter_Namespace::UpdateValuesStats::min_int64t, ColumnDescriptor::tableId, and foreign_storage::update_stats().

Referenced by compactRows(), and updateColumn().

                            {
   heavyai::unique_lock<heavyai::shared_mutex> write_lock(fragmentInfoMutex_);
   auto buffer = chunk->getBuffer();
   const auto& lhs_type = cd->columnType;
 
   auto encoder = buffer->getEncoder();
   auto update_stats = [&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(new_values_stats.min_int64t,
                  new_values_stats.max_int64t,
                  new_values_stats.has_null);
   } else if (lhs_type.is_fp()) {
     update_stats(new_values_stats.min_double,
                  new_values_stats.max_double,
                  new_values_stats.has_null);
   } else if (lhs_type.is_decimal()) {
     update_stats((int64_t)(new_values_stats.min_double * pow(10, lhs_type.get_scale())),
                  (int64_t)(new_values_stats.max_double * pow(10, lhs_type.get_scale())),
                  new_values_stats.has_null);
   } else if (!lhs_type.is_array() && !lhs_type.is_geometry() &&
              !(lhs_type.is_string() && kENCODING_DICT != lhs_type.get_compression())) {
     update_stats(new_values_stats.min_int64t,
                  new_values_stats.max_int64t,
                  new_values_stats.has_null);
   }
   auto td = updel_roll.catalog->getMetadataForTable(cd->tableId);
   auto chunk_metadata =
       updel_roll.getChunkMetadata({td, &fragment}, cd->columnId, fragment);
   buffer->getEncoder()->getMetadata(chunk_metadata);
 }

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,
		Executor *	executor
	)

overridevirtual

Implements Fragmenter_Namespace::AbstractFragmenter.

Definition at line 268 of file UpdelStorage.cpp.

References UpdelRoll::addDirtyChunk(), threading_serial::async(), UpdelRoll::catalog, CHECK, checked_get(), Fragmenter_Namespace::InsertData::columnIds, cpu_threads(), TargetValueConverterFactory::create(), Fragmenter_Namespace::InsertData::databaseId, Catalog_Namespace::DBMetadata::dbId, g_enable_string_functions, Fragmenter_Namespace::get_chunks(), get_logical_type_info(), SQLTypeInfo::get_size(), UpdelRoll::getChunkMetadata(), Catalog_Namespace::Catalog::getCurrentDB(), Fragmenter_Namespace::RowDataProvider::getEntryAt(), Fragmenter_Namespace::RowDataProvider::getEntryCount(), getFragmentInfo(), Fragmenter_Namespace::RowDataProvider::getLiteralDictionary(), Catalog_Namespace::Catalog::getLogicalTableId(), Fragmenter_Namespace::RowDataProvider::getRowCount(), insertDataNoCheckpoint(), Fragmenter_Namespace::InsertData::is_default, SQLTypeInfo::is_string(), UpdelRoll::is_varlen_update, kLINESTRING, kMULTILINESTRING, kMULTIPOINT, kMULTIPOLYGON, kPOINT, kPOLYGON, UpdelRoll::logicalTableId, UpdelRoll::memoryLevel, Fragmenter_Namespace::InsertData::numRows, TableDescriptor::tableId, and Fragmenter_Namespace::InsertData::tableId.

                         {
   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_ptr = getFragmentInfo(fragmentId);
   auto& fragment = *fragment_ptr;
   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;
   size_t indexOfDeletedColumn{0};
   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->getColumnDesc();
 
     if (chunk_cd->isDeletedCol) {
       indexOfDeletedColumn = chunk_cd->columnId;
       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,
           sourceDataMetaInfo,
           targetDescriptor,
           *catalog,
           targetDescriptor->columnType,
           !targetDescriptor->columnType.get_notnull(),
           sourceDataProvider.getLiteralDictionary(),
           g_enable_string_functions &&
                   sourceDataMetaInfo.get_type_info().is_dict_encoded_string()
               ? executor->getStringDictionaryProxy(
                     sourceDataMetaInfo.get_type_info().getStringDictKey(),
                     executor->getRowSetMemoryOwner(),
                     true)
               : nullptr,
           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 kMULTILINESTRING:
             indexOfChunk += 3;
             break;
           case kLINESTRING:
           case kMULTIPOINT:
             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.addDirtyChunk(deletedChunk, fragment.fragmentId);
   bool* deletedChunkBuffer =
       reinterpret_cast<bool*>(deletedChunk->getBuffer()->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;
   insert_data.is_default.resize(insert_data.columnIds.size(), false);
   insertDataNoCheckpoint(insert_data);
 
   // update metdata for deleted chunk as we are doing special handling
   auto chunkMetadata =
       updelRoll.getChunkMetadata({td, &fragment}, indexOfDeletedColumn, fragment);
   chunkMetadata->chunkStats.max.boolval = 1;
 
   // Im not completely sure that we need to do this in fragmented and on the buffer
   // but leaving this alone for now
   if (!deletedChunk->getBuffer()->hasEncoder()) {
     deletedChunk->initEncoder();
   }
   deletedChunk->getBuffer()->getEncoder()->updateStats(static_cast<int64_t>(true), false);
 
   if (fragment.shadowNumTuples > deletedChunk->getBuffer()->getEncoder()->getNumElems()) {
     // An append to the same fragment will increase shadowNumTuples.
     // Update NumElems in this case. Otherwise, use existing NumElems.
     deletedChunk->getBuffer()->getEncoder()->setNumElems(fragment.shadowNumTuples);
   }
   deletedChunk->getBuffer()->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 977 of file UpdelStorage.cpp.

References fragmentInfoMutex_, UpdelRoll::getChunkMetadataMap(), and UpdelRoll::getNumTuple().

                                                                   {
   heavyai::unique_lock<heavyai::shared_mutex> writeLock(fragmentInfoMutex_);
   const auto chunk_metadata_map = updel_roll.getChunkMetadataMap(key);
   auto& fragmentInfo = *key.second;
   fragmentInfo.setChunkMetadataMap(chunk_metadata_map);
   fragmentInfo.shadowChunkMetadataMap = fragmentInfo.getChunkMetadataMapPhysicalCopy();
   fragmentInfo.shadowNumTuples = updel_roll.getNumTuple(key);
   fragmentInfo.setPhysicalNumTuples(fragmentInfo.shadowNumTuples);
 }

Here is the call graph for this function:

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 1081 of file UpdelStorage.cpp.

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

Referenced by compactRows().

                                              {
   const auto cd = chunk->getColumnDesc();
   const auto& col_type = cd->columnType;
   auto data_buffer = chunk->getBuffer();
   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 1195 of file UpdelStorage.cpp.

References CHECK, Fragmenter_Namespace::get_buffer_offset(), Fragmenter_Namespace::get_null_padding(), Fragmenter_Namespace::get_var_len_null_array_indexes(), and Fragmenter_Namespace::FragmentInfo::getPhysicalNumTuples().

Referenced by compactRows().

                                              {
   auto is_varlen_array = chunk->getColumnDesc()->columnType.is_varlen_array();
   auto data_buffer = chunk->getBuffer();
   CHECK(data_buffer);
   auto index_buffer = chunk->getIndexBuf();
   CHECK(index_buffer);
   auto data_addr = data_buffer->getMemoryPtr();
   auto indices_addr = index_buffer->getMemoryPtr();
   CHECK(indices_addr);
   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;
   auto nrows_in_fragment = fragment.getPhysicalNumTuples();
   size_t null_padding =
       get_null_padding(is_varlen_array, frag_offsets, index_array, nrows_in_fragment);
   size_t nbytes_var_data_to_keep = null_padding;
   auto null_array_indexes = get_var_len_null_array_indexes(
       chunk->getColumnDesc()->columnType, frag_offsets, index_array, nrows_in_fragment);
   auto nrows_to_vacuum = frag_offsets.size();
   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 deleted_row_start_offset =
           get_buffer_offset(is_varlen_array, index_array, irow_to_vacuum);
       auto kept_row_start_offset =
           get_buffer_offset(is_varlen_array, index_array, irow_of_blk_to_keep);
       auto nbytes_to_keep =
           (is_last_one ? data_buffer->size() : deleted_row_start_offset) -
           kept_row_start_offset;
       if (irow_of_blk_to_fill != irow_of_blk_to_keep) {
         if (nbytes_to_keep > 0) {
           CHECK(data_addr);
           // move curr varlen row block toward front
           memmove(data_addr + ibyte_var_data_to_keep,
                   data_addr + kept_row_start_offset,
                   nbytes_to_keep);
         }
 
         const auto base_offset = kept_row_start_offset;
         for (int64_t i = 0; i < nrows_to_keep; ++i) {
           auto update_index = irow_of_blk_to_keep + i;
           auto offset = get_buffer_offset(is_varlen_array, index_array, update_index);
           index_array[update_index] = ibyte_var_data_to_keep + (offset - base_offset);
         }
       }
       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;
   }
 
   // Set expected null padding, last offset, and negative values for null array offsets.
   index_array[0] = null_padding;
   auto post_vacuum_row_count = nrows_in_fragment - nrows_to_vacuum;
   index_array[post_vacuum_row_count] = nbytes_var_data_to_keep;
   if (!is_varlen_array) {
     CHECK(null_array_indexes.empty());
   }
   for (auto index : null_array_indexes) {
     index_array[index + 1] = -1 * std::abs(index_array[index + 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 200 of file InsertOrderFragmenter.h.

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

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

protected

Definition at line 192 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 194 of file InsertOrderFragmenter.h.

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

protected

Definition at line 199 of file InsertOrderFragmenter.h.

Data_Namespace::MemoryLevel Fragmenter_Namespace::InsertOrderFragmenter::defaultInsertLevel_

protected

Definition at line 216 of file InsertOrderFragmenter.h.

std::string Fragmenter_Namespace::InsertOrderFragmenter::fragmenterType_

protected

Definition at line 210 of file InsertOrderFragmenter.h.

Referenced by getFragmenterType().

heavyai::shared_mutex Fragmenter_Namespace::InsertOrderFragmenter::fragmentInfoMutex_

protected

Definition at line 212 of file InsertOrderFragmenter.h.

Referenced by updateColumnMetadata(), and updateMetadata().

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

protected

data about each fragment stored - id and number of rows

Definition at line 197 of file InsertOrderFragmenter.h.

bool Fragmenter_Namespace::InsertOrderFragmenter::hasMaterializedRowId_

protected

Definition at line 218 of file InsertOrderFragmenter.h.

heavyai::shared_mutex Fragmenter_Namespace::InsertOrderFragmenter::insertMutex_

protected

Definition at line 214 of file InsertOrderFragmenter.h.

size_t Fragmenter_Namespace::InsertOrderFragmenter::maxChunkSize_

protected

Definition at line 208 of file InsertOrderFragmenter.h.

int Fragmenter_Namespace::InsertOrderFragmenter::maxFragmentId_

protected

Definition at line 207 of file InsertOrderFragmenter.h.

size_t Fragmenter_Namespace::InsertOrderFragmenter::maxFragmentRows_

protected

Definition at line 203 of file InsertOrderFragmenter.h.

size_t Fragmenter_Namespace::InsertOrderFragmenter::maxRows_

protected

Definition at line 209 of file InsertOrderFragmenter.h.

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

protected

Definition at line 221 of file InsertOrderFragmenter.h.

size_t Fragmenter_Namespace::InsertOrderFragmenter::numTuples_

protected

Definition at line 206 of file InsertOrderFragmenter.h.

Referenced by getNumRows(), and setNumRows().

size_t Fragmenter_Namespace::InsertOrderFragmenter::pageSize_

protected

Definition at line 204 of file InsertOrderFragmenter.h.

const int Fragmenter_Namespace::InsertOrderFragmenter::physicalTableId_

protected

Definition at line 201 of file InsertOrderFragmenter.h.

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

int Fragmenter_Namespace::InsertOrderFragmenter::rowIdColId_

protected

Definition at line 219 of file InsertOrderFragmenter.h.

const int Fragmenter_Namespace::InsertOrderFragmenter::shard_

protected

Definition at line 202 of file InsertOrderFragmenter.h.

Referenced by updateColumn().

std::mutex Fragmenter_Namespace::InsertOrderFragmenter::temp_mutex_

mutableprotected

Definition at line 246 of file InsertOrderFragmenter.h.

Referenced by updateColumn().

std::vector<std::unique_ptr<Chunk_NS::Chunk> > Fragmenter_Namespace::InsertOrderFragmenter::tracked_in_memory_chunks_

protected

Definition at line 195 of file InsertOrderFragmenter.h.

const bool Fragmenter_Namespace::InsertOrderFragmenter::uses_foreign_storage_

protected

Definition at line 217 of file InsertOrderFragmenter.h.

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

protected

Definition at line 220 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

Protected Member Functions

Protected Attributes

Private Member Functions

Detailed Description

Member Typedef Documentation

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation